Synchronized Broadcasting Block

src/main/java/org/tron/common/overlay/discover/DiscoverServer.java

@@ -18,6 +18,7 @@
 
 package org.tron.common.overlay.discover;
 
+import com.sun.org.apache.xpath.internal.Arg;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
@@ -58,16 +59,15 @@ public class DiscoverServer {
   @Autowired
   public DiscoverServer(final NodeManager nodeManager) {
     this.nodeManager = nodeManager;
-    if (args.isNodeDiscoveryEnable()) {
+    if (args.isNodeDiscoveryEnable() && !args.isFastForward()) {
       if (port == 0) {
         logger.error("Discovery can't be started while listen port == 0");
       } else {
         new Thread(() -> {
           try {
             start();
           } catch (Exception e) {
-            logger.debug(e.getMessage(), e);
-            throw new RuntimeException(e);
+            logger.error("Discovery server start failed.", e);
           }
         }, "DiscoverServer").start();
       }

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

@@ -4,7 +4,6 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.tron.common.utils.Sha256Hash;
-import org.tron.core.config.args.Args;
 import org.tron.core.net.TronNetDelegate;
 import org.tron.core.net.message.InventoryMessage;
 import org.tron.core.net.message.TronMessage;
@@ -28,17 +27,11 @@ public class InventoryMsgHandler implements TronMsgHandler {
 
   private int maxCountIn10s = 10_000;
 
-  private boolean fastForward = Args.getInstance().isFastForward();
-
   @Override
   public void processMessage(PeerConnection peer, TronMessage msg) {
     InventoryMessage inventoryMessage = (InventoryMessage) msg;
     InventoryType type = inventoryMessage.getInventoryType();
 
-    if (fastForward && inventoryMessage.getInventoryType().equals(InventoryType.TRX)) {
-      return;
-    }
-
     if (!check(peer, inventoryMessage)) {
       return;
     }
@@ -54,10 +47,6 @@ private boolean check(PeerConnection peer, InventoryMessage inventoryMessage) {
     InventoryType type = inventoryMessage.getInventoryType();
     int size = inventoryMessage.getHashList().size();
 
-//    if (size > NetConstants.MAX_INV_FETCH_PER_PEER) {
-//      throw new P2pException(TypeEnum.BAD_MESSAGE, "size: " + size);
-//    }
-
     if (peer.isNeedSyncFromPeer() || peer.isNeedSyncFromUs()) {
       logger.warn("Drop inv: {} size: {} from Peer {}, syncFromUs: {}, syncFromPeer: {}.",
           type, size, peer.getInetAddress(), peer.isNeedSyncFromUs(), peer.isNeedSyncFromPeer());

src/main/java/org/tron/core/net/service/AdvService.java

@@ -65,23 +65,25 @@ public class AdvService {
   private boolean fastForward = Args.getInstance().isFastForward();
 
   public void init() {
-    if (!fastForward) {
-      spreadExecutor.scheduleWithFixedDelay(() -> {
-        try {
-          consumerInvToSpread();
-        } catch (Throwable t) {
-          logger.error("Spread thread error.", t);
-        }
-      }, 100, 10, TimeUnit.MILLISECONDS);
+    if (fastForward) {
+      return;
     }
 
+    spreadExecutor.scheduleWithFixedDelay(() -> {
+      try {
+        consumerInvToSpread();
+      } catch (Throwable t) {
+        logger.error("Spread thread error.", t);
+      }
+    }, 100, 30, TimeUnit.MILLISECONDS);
+
     fetchExecutor.scheduleWithFixedDelay(() -> {
       try {
         consumerInvToFetch();
       } catch (Throwable t) {
         logger.error("Fetch thread error.", t);
       }
-    }, 100, 10, TimeUnit.MILLISECONDS);
+    }, 100, 30, TimeUnit.MILLISECONDS);
   }
 
   public void close() {
@@ -90,6 +92,11 @@ public void close() {
   }
 
   synchronized public boolean addInv(Item item) {
+
+    if (fastForward && !InventoryType.BLOCK.equals(item.getType())) {
+      return false;
+    }
+
     if (invToFetchCache.getIfPresent(item) != null) {
       return false;
     }
@@ -106,6 +113,11 @@ synchronized public boolean addInv(Item item) {
 
     invToFetchCache.put(item, System.currentTimeMillis());
     invToFetch.put(item, System.currentTimeMillis());
+
+    if (InventoryType.BLOCK.equals(item.getType())){
+      consumerInvToFetch();
+    }
+
     return true;
   }
 
@@ -118,6 +130,10 @@ public Message getMessage(Item item) {
   }
 
   public void broadcast(Message msg) {
+    if (fastForward && !(msg instanceof BlockMessage)) {
+      return;
+    }
+
     Item item;
     if (msg instanceof BlockMessage) {
       BlockMessage blockMsg = (BlockMessage) msg;
@@ -140,11 +156,12 @@ public void broadcast(Message msg) {
       logger.error("Adv item is neither block nor trx, type: {}", msg.getType());
       return;
     }
+
     synchronized (invToSpread) {
       invToSpread.put(item, System.currentTimeMillis());
     }
 
-    if (fastForward) {
+    if (InventoryType.BLOCK.equals(item.getType())) {
       consumerInvToSpread();
     }
   }
@@ -160,6 +177,10 @@ public void onDisconnect(PeerConnection peer) {
         }
       });
     }
+
+    if (invToFetch.size() > 0) {
+      consumerInvToFetch();
+    }
   }
 
   private void consumerInvToFetch() {

src/test/java/org/tron/core/net/services/AdvServiceTest.java

@@ -1,6 +1,7 @@
 package org.tron.core.net.services;
 
 import org.junit.Assert;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.tron.common.utils.Sha256Hash;
 import org.tron.core.capsule.BlockCapsule;
@@ -9,6 +10,7 @@
 import org.tron.core.net.service.AdvService;
 import org.tron.protos.Protocol.Inventory.InventoryType;
 
+@Ignore
 public class AdvServiceTest {
 
   AdvService service = new AdvService();