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 = 1000;
     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) : 1000;
+    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,10 +251,12 @@ private synchronized void handleSyncBlock() {
     }
 
     final boolean[] isProcessed = {true};
+    final long[] solidNum = {0};
 
     while (isProcessed[0]) {
 
       isProcessed[0] = false;
+      solidNum[0] = tronNetDelegate.getSolidBlockId().getNum();
 
       blockWaitToProcess.forEach((msg, peerConnection) -> {
         synchronized (tronNetDelegate.getBlockLock()) {
@@ -262,6 +265,10 @@ private synchronized void handleSyncBlock() {
             invalid(msg.getBlockId(), peerConnection);
             return;
           }
+          if (msg.getBlockId().getNum() <= solidNum[0]) {
+            blockWaitToProcess.remove(msg);
+            return;
+          }
           final boolean[] isFound = {false};
           tronNetDelegate.getActivePeer().stream()
               .filter(peer -> msg.getBlockId().equals(peer.getSyncBlockToFetch().peek()))