feat(net): optimize InventoryMessage processing logic

Author: xxo1shine

common/src/main/java/org/tron/common/parameter/CommonParameter.java

@@ -165,6 +165,9 @@ public class CommonParameter {
   public int maxConnectionsWithSameIp;
   @Getter
   @Setter
+  public int maxTps;
+  @Getter
+  @Setter
   public int minParticipationRate;
   @Getter
   @Setter

common/src/main/java/org/tron/core/Constant.java

@@ -93,6 +93,7 @@ public class Constant {
 
   public static final String NODE_MAX_ACTIVE_NODES = "node.maxActiveNodes";
   public static final String NODE_MAX_ACTIVE_NODES_WITH_SAME_IP = "node.maxActiveNodesWithSameIp";
+  public static final String NODE_MAX_TPS = "node.maxTps";
   public static final String NODE_CONNECT_FACTOR = "node.connectFactor";
   public static final String NODE_ACTIVE_CONNECT_FACTOR = "node.activeConnectFactor";
 

framework/src/main/java/org/tron/core/config/args/Args.java

@@ -135,6 +135,7 @@ public static void clearParam() {
     PARAMETER.minConnections = 8;
     PARAMETER.minActiveConnections = 3;
     PARAMETER.maxConnectionsWithSameIp = 2;
+    PARAMETER.maxTps = 1000;
     PARAMETER.minParticipationRate = 0;
     PARAMETER.nodeListenPort = 0;
     PARAMETER.nodeDiscoveryBindIp = "";
@@ -621,6 +622,9 @@ public static void setParam(final String[] args, final String confFileName) {
                       .getInt(Constant.NODE_MAX_CONNECTIONS_WITH_SAME_IP) : 2;
     }
 
+    PARAMETER.maxTps = config.hasPath(Constant.NODE_MAX_TPS)
+            ? config.getInt(Constant.NODE_MAX_TPS) : 1000;
+
     PARAMETER.minParticipationRate =
         config.hasPath(Constant.NODE_MIN_PARTICIPATION_RATE)
             ? config.getInt(Constant.NODE_MIN_PARTICIPATION_RATE)

framework/src/main/java/org/tron/core/net/P2pEventHandlerImpl.java

@@ -1,5 +1,8 @@
 package org.tron.core.net;
 
+import static org.tron.core.net.message.MessageTypes.INVENTORY;
+import static org.tron.core.net.message.MessageTypes.TRX;
+
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -11,11 +14,13 @@
 import org.tron.common.prometheus.MetricKeys;
 import org.tron.common.prometheus.Metrics;
 import org.tron.consensus.pbft.message.PbftMessage;
+import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
 import org.tron.core.net.message.MessageTypes;
 import org.tron.core.net.message.PbftMessageFactory;
 import org.tron.core.net.message.TronMessage;
 import org.tron.core.net.message.TronMessageFactory;
+import org.tron.core.net.message.adv.InventoryMessage;
 import org.tron.core.net.message.base.DisconnectMessage;
 import org.tron.core.net.message.handshake.HelloMessage;
 import org.tron.core.net.messagehandler.BlockMsgHandler;
@@ -84,6 +89,8 @@ public class P2pEventHandlerImpl extends P2pEventHandler {
 
   private byte MESSAGE_MAX_TYPE = 127;
 
+  private int maxCountIn10s = Args.getInstance().getMaxTps() * 10;
+
   public P2pEventHandlerImpl() {
     Set<Byte> set = new HashSet<>();
     for (byte i = 0; i < MESSAGE_MAX_TYPE; i++) {
@@ -140,10 +147,27 @@ private void processMessage(PeerConnection peer, byte[] data) {
     try {
       msg = TronMessageFactory.create(data);
       type = msg.getType();
+
+      if (INVENTORY.equals(type)) {
+        InventoryMessage message = (InventoryMessage) msg;
+        Protocol.Inventory.InventoryType inventoryType = message.getInventoryType();
+        int count = peer.getPeerStatistics().messageStatistics.tronInTrxInventoryElement
+                .getCount(10);
+        if (inventoryType.equals(Protocol.Inventory.InventoryType.TRX) && count > maxCountIn10s) {
+          logger.warn("Drop inventory from Peer {}, cur:{}, max:{}",
+                  peer.getInetAddress(), count, maxCountIn10s);
+          if (Args.getInstance().isOpenPrintLog()) {
+            logger.warn("[overload]Drop tx list is: {}", ((InventoryMessage) msg).getHashList());
+          }
+          return;
+        }
+      }
+
       peer.getPeerStatistics().messageStatistics.addTcpInMessage(msg);
       if (PeerConnection.needToLog(msg)) {
         logger.info("Receive message from  peer: {}, {}", peer.getInetSocketAddress(), msg);
       }
+
       switch (type) {
         case P2P_PING:
         case P2P_PONG:

framework/src/main/java/org/tron/core/net/messagehandler/InventoryMsgHandler.java

@@ -26,8 +26,6 @@ public class InventoryMsgHandler implements TronMsgHandler {
   @Autowired
   private TransactionsMsgHandler transactionsMsgHandler;
 
-  private int maxCountIn10s = 10_000;
-
   @Override
   public void processMessage(PeerConnection peer, TronMessage msg) {
     InventoryMessage inventoryMessage = (InventoryMessage) msg;
@@ -54,25 +52,13 @@ private boolean check(PeerConnection peer, InventoryMessage inventoryMessage) {
       return false;
     }
 
-    if (type.equals(InventoryType.TRX)) {
-      int count = peer.getPeerStatistics().messageStatistics.tronInTrxInventoryElement.getCount(10);
-      if (count > maxCountIn10s) {
-        logger.warn("Drop inv: {} size: {} from Peer {}, Inv count: {} is overload",
-            type, size, peer.getInetAddress(), count);
-        if (Args.getInstance().isOpenPrintLog()) {
-          logger.warn("[overload]Drop tx list is: {}", inventoryMessage.getHashList());
-        }
-        return false;
-      }
-
-      if (transactionsMsgHandler.isBusy()) {
-        logger.warn("Drop inv: {} size: {} from Peer {}, transactionsMsgHandler is busy",
-            type, size, peer.getInetAddress());
-        if (Args.getInstance().isOpenPrintLog()) {
-          logger.warn("[isBusy]Drop tx list is: {}", inventoryMessage.getHashList());
-        }
-        return false;
+    if (type.equals(InventoryType.TRX) && transactionsMsgHandler.isBusy()) {
+      logger.warn("Drop inv: {} size: {} from Peer {}, transactionsMsgHandler is busy",
+              type, size, peer.getInetAddress());
+      if (Args.getInstance().isOpenPrintLog()) {
+        logger.warn("[isBusy]Drop tx list is: {}", inventoryMessage.getHashList());
       }
+      return false;
     }
 
     return true;

framework/src/test/java/org/tron/common/config/args/ArgsTest.java

@@ -37,5 +37,6 @@ public void testConfig() {
     Assert.assertEquals(Args.getInstance().getRateLimiterGlobalQps(), 50000);
     Assert.assertEquals(Args.getInstance().getRateLimiterGlobalIpQps(), 10000);
     Assert.assertEquals(Args.getInstance().p2pDisable, true);
+    Assert.assertEquals(Args.getInstance().getMaxTps(), 1000);
   }
 }