optimize blockLock; MessageQueue does not send message if channel is disconnected

Author: 317787106

framework/src/main/java/org/tron/common/overlay/server/ChannelManager.java

@@ -34,7 +34,7 @@
 @Component
 public class ChannelManager {
 
-  private final Map<ByteArrayWrapper, Channel> activePeers = new ConcurrentHashMap<>();
+  private final Map<ByteArrayWrapper, Channel> activeChannels = new ConcurrentHashMap<>();
   @Autowired
   private PeerServer peerServer;
   @Autowired
@@ -121,7 +121,7 @@ public void processDisconnect(Channel channel, ReasonCode reason) {
 
   public void notifyDisconnect(Channel channel) {
     syncPool.onDisconnect(channel);
-    activePeers.values().remove(channel);
+    activeChannels.values().remove(channel);
     if (channel != null) {
       if (channel.getNodeStatistics() != null) {
         channel.getNodeStatistics().notifyDisconnect();
@@ -146,7 +146,7 @@ public synchronized boolean processPeer(Channel peer) {
         return false;
       }
 
-      if (!peer.isActive() && activePeers.size() >= maxActivePeers) {
+      if (!peer.isActive() && activeChannels.size() >= maxActivePeers) {
         peer.disconnect(TOO_MANY_PEERS);
         return false;
       }
@@ -157,7 +157,7 @@ public synchronized boolean processPeer(Channel peer) {
       }
     }
 
-    Channel channel = activePeers.get(peer.getNodeIdWrapper());
+    Channel channel = activeChannels.get(peer.getNodeIdWrapper());
     if (channel != null) {
       if (channel.getStartTime() > peer.getStartTime()) {
         logger.info("Disconnect connection established later, {}", channel.getNode());
@@ -167,14 +167,14 @@ public synchronized boolean processPeer(Channel peer) {
         return false;
       }
     }
-    activePeers.put(peer.getNodeIdWrapper(), peer);
-    logger.info("Add active peer {}, total active peers: {}", peer, activePeers.size());
+    activeChannels.put(peer.getNodeIdWrapper(), peer);
+    logger.info("Add active peer {}, total active peers: {}", peer, activeChannels.size());
     return true;
   }
 
   public int getConnectionNum(InetAddress inetAddress) {
     int cnt = 0;
-    for (Channel channel : activePeers.values()) {
+    for (Channel channel : activeChannels.values()) {
       if (channel.getInetAddress().equals(inetAddress)) {
         cnt++;
       }
@@ -183,7 +183,7 @@ public int getConnectionNum(InetAddress inetAddress) {
   }
 
   public Collection<Channel> getActivePeers() {
-    return activePeers.values();
+    return activeChannels.values();
   }
 
   public Cache<InetAddress, ReasonCode> getRecentlyDisconnected() {

framework/src/main/java/org/tron/common/overlay/server/MessageQueue.java

@@ -69,6 +69,11 @@ public void activate(ChannelHandlerContext ctx) {
             continue;
           }
           Message msg = msgQueue.take();
+          if (channel.isDisconnect()) {
+            logger.warn("Failed to send to {} as channel has closed, {}",
+                ctx.channel().remoteAddress(), msg);
+            return;
+          }
           ctx.writeAndFlush(msg.getSendData()).addListener((ChannelFutureListener) future -> {
             if (!future.isSuccess() && !channel.isDisconnect()) {
               logger.warn("Failed to send to {}, {}", ctx.channel().remoteAddress(), msg);
@@ -92,6 +97,11 @@ public void setChannel(Channel channel) {
   }
 
   public void fastSend(Message msg) {
+    if (channel.isDisconnect()) {
+      logger.warn("Fast send to {} failed as channel has closed, {} ",
+          ctx.channel().remoteAddress(), msg);
+      return;
+    }
     logger.info("Fast send to {}, {} ", ctx.channel().remoteAddress(), msg);
     ctx.writeAndFlush(msg.getSendData()).addListener((ChannelFutureListener) future -> {
       if (!future.isSuccess() && !channel.isDisconnect()) {

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

@@ -234,8 +234,8 @@ private synchronized void handleSyncBlock() {
 
       isProcessed[0] = false;
 
-      synchronized (tronNetDelegate.getBlockLock()) {
-        blockWaitToProcess.forEach((msg, peerConnection) -> {
+      blockWaitToProcess.forEach((msg, peerConnection) -> {
+        synchronized (tronNetDelegate.getBlockLock()) {
           if (peerConnection.isDisconnect()) {
             blockWaitToProcess.remove(msg);
             invalid(msg.getBlockId());
@@ -254,8 +254,8 @@ private synchronized void handleSyncBlock() {
             isProcessed[0] = true;
             processSyncBlock(msg.getBlockCapsule());
           }
-        });
-      }
+        }
+      });
     }
   }