Merge pull request #5195 from chengtx01/optimize_sync

feat(net):optimize sync service

Author: jwrct

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

@@ -199,6 +199,9 @@ public class CommonParameter {
   @Getter
   @Setter
   public PublishConfig dnsPublishConfig;
+  @Getter
+  @Setter
+  public long syncFetchBatchNum;
 
   //If you are running a solidity node for java tron, this flag is set to true
   @Getter

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

@@ -90,6 +90,7 @@ public class Constant {
   public static final String NODE_MAX_CONNECTIONS = "node.maxConnections";
   public static final String NODE_MIN_CONNECTIONS = "node.minConnections";
   public static final String NODE_MIN_ACTIVE_CONNECTIONS = "node.minActiveConnections";
+  public static final String NODE_SYNC_FETCH_BATCH_NUM = "node.syncFetchBatchNum";
 
   public static final String NODE_MAX_ACTIVE_NODES = "node.maxActiveNodes";
   public static final String NODE_MAX_ACTIVE_NODES_WITH_SAME_IP = "node.maxActiveNodesWithSameIp";

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

@@ -146,6 +146,7 @@ public static void clearParam() {
     PARAMETER.nodeEnableIpv6 = false;
     PARAMETER.dnsTreeUrls = new ArrayList<>();
     PARAMETER.dnsPublishConfig = null;
+    PARAMETER.syncFetchBatchNum = 2000;
     PARAMETER.rpcPort = 0;
     PARAMETER.rpcOnSolidityPort = 0;
     PARAMETER.rpcOnPBFTPort = 0;
@@ -657,6 +658,15 @@ public static void setParam(final String[] args, final String confFileName) {
 
     PARAMETER.dnsPublishConfig = loadDnsPublishConfig(config);
 
+    PARAMETER.syncFetchBatchNum = config.hasPath(Constant.NODE_SYNC_FETCH_BATCH_NUM) ? config
+        .getInt(Constant.NODE_SYNC_FETCH_BATCH_NUM) : 2000;
+    if (PARAMETER.syncFetchBatchNum > 2000) {
+      PARAMETER.syncFetchBatchNum = 2000;
+    }
+    if (PARAMETER.syncFetchBatchNum < 100) {
+      PARAMETER.syncFetchBatchNum = 100;
+    }
+
     PARAMETER.rpcPort =
         config.hasPath(Constant.NODE_RPC_PORT)
             ? config.getInt(Constant.NODE_RPC_PORT) : 50051;

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

@@ -13,6 +13,7 @@
 import org.tron.core.capsule.BlockCapsule.BlockId;
 import org.tron.core.config.Parameter.ChainConstant;
 import org.tron.core.config.Parameter.NetConstants;
+import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
 import org.tron.core.exception.P2pException.TypeEnum;
 import org.tron.core.net.TronNetDelegate;
@@ -32,6 +33,8 @@ public class ChainInventoryMsgHandler implements TronMsgHandler {
   @Autowired
   private SyncService syncService;
 
+  private final long syncFetchBatchNum = Args.getInstance().getSyncFetchBatchNum();
+
   @Override
   public void processMessage(PeerConnection peer, TronMessage msg) throws P2pException {
 
@@ -88,7 +91,7 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
     peer.setFetchAble(true);
     if ((chainInventoryMessage.getRemainNum() == 0 && !peer.getSyncBlockToFetch().isEmpty())
         || (chainInventoryMessage.getRemainNum() != 0
-        && peer.getSyncBlockToFetch().size() > NetConstants.SYNC_FETCH_BATCH_NUM)) {
+        && peer.getSyncBlockToFetch().size() > syncFetchBatchNum)) {
       syncService.setFetchFlag(true);
     } else {
       syncService.syncNext(peer);

framework/src/main/java/org/tron/core/net/service/sync/SyncService.java

@@ -21,7 +21,6 @@
 import org.tron.common.utils.Pair;
 import org.tron.core.capsule.BlockCapsule;
 import org.tron.core.capsule.BlockCapsule.BlockId;
-import org.tron.core.config.Parameter.NetConstants;
 import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
 import org.tron.core.exception.P2pException.TypeEnum;
@@ -65,6 +64,8 @@ public class SyncService {
   @Setter
   private volatile boolean fetchFlag = false;
 
+  private final long syncFetchBatchNum = Args.getInstance().getSyncFetchBatchNum();
+
   public void init() {
     fetchExecutor.scheduleWithFixedDelay(() -> {
       try {
@@ -132,7 +133,7 @@ public void processBlock(PeerConnection peer, BlockMessage blockMessage) {
     handleFlag = true;
     if (peer.isIdle()) {
       if (peer.getRemainNum() > 0
-          && peer.getSyncBlockToFetch().size() <= NetConstants.SYNC_FETCH_BATCH_NUM) {
+          && peer.getSyncBlockToFetch().size() <= syncFetchBatchNum) {
         syncNext(peer);
       } else {
         fetchFlag = true;
@@ -250,6 +251,7 @@ private synchronized void handleSyncBlock() {
     }
 
     final boolean[] isProcessed = {true};
+    long solidNum = tronNetDelegate.getSolidBlockId().getNum();
 
     while (isProcessed[0]) {
 
@@ -262,6 +264,10 @@ private synchronized void handleSyncBlock() {
             invalid(msg.getBlockId(), peerConnection);
             return;
           }
+          if (msg.getBlockId().getNum() <= solidNum) {
+            blockWaitToProcess.remove(msg);
+            return;
+          }
           final boolean[] isFound = {false};
           tronNetDelegate.getActivePeer().stream()
               .filter(peer -> msg.getBlockId().equals(peer.getSyncBlockToFetch().peek()))

framework/src/test/java/org/tron/core/net/services/SyncServiceTest.java

@@ -183,11 +183,21 @@ public void testHandleSyncBlock() throws Exception {
     Map<BlockMessage, PeerConnection> blockJustReceived =
             (Map<BlockMessage, PeerConnection>)
             ReflectUtils.getFieldObject(service, "blockJustReceived");
+    Protocol.BlockHeader.raw.Builder blockHeaderRawBuild = Protocol.BlockHeader.raw.newBuilder();
+    Protocol.BlockHeader.raw blockHeaderRaw = blockHeaderRawBuild
+        .setNumber(100000)
+        .build();
 
-    BlockCapsule blockCapsule = new BlockCapsule(Protocol.Block.newBuilder().build());
+    // block header
+    Protocol.BlockHeader.Builder blockHeaderBuild = Protocol.BlockHeader.newBuilder();
+    Protocol.BlockHeader blockHeader = blockHeaderBuild.setRawData(blockHeaderRaw).build();
+
+    BlockCapsule blockCapsule = new BlockCapsule(Protocol.Block.newBuilder()
+        .setBlockHeader(blockHeader).build());
 
     BlockCapsule.BlockId blockId = blockCapsule.getBlockId();
 
+
     InetSocketAddress a1 = new InetSocketAddress("127.0.0.1", 10001);
     Channel c1 = mock(Channel.class);
     Mockito.when(c1.getInetSocketAddress()).thenReturn(a1);