Merge pull request #333 from tronprotocol/terminateInactivePeer

Terminate inactive peer

Author: sasaxie

src/main/java/org/tron/common/overlay/message/ReasonCode.java

@@ -80,6 +80,8 @@ public enum ReasonCode {
 
     USER_REASON(0x10),
 
+    RESET(0x11),
+
     /**
      * [0xFF] Reason not specified
      */

src/main/java/org/tron/common/utils/Time.java

@@ -1,9 +1,14 @@
 package org.tron.common.utils;
 
+import java.sql.Timestamp;
+
 public class Time {
 
   public static long getCurrentMillis() {
     return System.currentTimeMillis();
+  }
 
+  public static String getTimeString(long time) {
+    return new Timestamp(time).toString();
   }
 }

src/main/java/org/tron/core/config/Parameter.java

@@ -21,5 +21,17 @@ interface NodeConstant {
 
   }
 
+  interface BlockConstant {
+    long BLOCK_INTERVAL = 5000L;
+  }
+
+  interface NetConstants {
+    long ADV_TIME_OUT = 1000L;
+    long SYNC_TIME_OUT = 5000L;
+    long HEAD_NUM_MAX_DELTA = 1000L;
+    long HEAD_NUM_CHECK_TIME = 60000L;
+  }
+
+
 
 }

src/main/java/org/tron/core/net/node/NodeImpl.java

@@ -13,21 +13,26 @@
 import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import javafx.util.Pair;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.tron.common.overlay.discover.NodeHandler;
 import org.tron.common.overlay.message.Message;
+import org.tron.common.overlay.message.ReasonCode;
 import org.tron.common.overlay.server.Channel.TronState;
-import org.tron.common.overlay.server.ChannelManager;
 import org.tron.common.overlay.server.SyncPool;
 import org.tron.common.utils.ExecutorLoop;
 import org.tron.common.utils.Sha256Hash;
 import org.tron.common.utils.Time;
 import org.tron.core.capsule.BlockCapsule;
 import org.tron.core.capsule.BlockCapsule.BlockId;
+import org.tron.core.config.Parameter.BlockConstant;
+import org.tron.core.config.Parameter.NetConstants;
 import org.tron.core.config.Parameter.NodeConstant;
 import org.tron.core.exception.BadBlockException;
 import org.tron.core.exception.BadTransactionException;
@@ -56,9 +61,6 @@ public class NodeImpl extends PeerConnectionDelegate implements Node {
   @Autowired
   private SyncPool pool;
 
-  @Autowired
-  private ChannelManager channelManager;
-
   class InvToSend {
 
     private HashMap<PeerConnection, HashMap<InventoryType, LinkedList<Sha256Hash>>> send
@@ -112,6 +114,8 @@ public void sendFetch() {
 
   private volatile boolean isHandleSyncBlockActive;
 
+  private ScheduledExecutorService disconnectInactiveExecutor = Executors.newSingleThreadScheduledExecutor();
+
   //broadcast
   private ConcurrentHashMap<Sha256Hash, InventoryType> advObjToSpread = new ConcurrentHashMap<>();
 
@@ -225,6 +229,7 @@ public void close() throws InterruptedException {
     advertiseLoopThread.join();
     advObjFetchLoopThread.join();
     handleSyncBlockLoop.join();
+    disconnectInactiveExecutor.shutdown();
   }
 
   @Override
@@ -379,8 +384,46 @@ private void activeTronPump() {
     advertiseLoopThread.start();
     advObjFetchLoopThread.start();
     handleSyncBlockLoop.start();
+
+    //terminate inactive loop
+    disconnectInactiveExecutor.scheduleWithFixedDelay(() -> {
+      disconnectInactive();
+    }, 30000, BlockConstant.BLOCK_INTERVAL / 2, TimeUnit.MILLISECONDS);
+
+  }
+
+  private void disconnectInactive() {
+    getActivePeer().forEach(peer -> {
+
+      final boolean[] isDisconnected = {false};
+
+      peer.getAdvObjWeRequested().values().stream()
+          .filter(time -> time < Time.getCurrentMillis() - NetConstants.ADV_TIME_OUT)
+          .findFirst().ifPresent(time -> isDisconnected[0] = true);
+
+      if (!isDisconnected[0]) {
+        peer.getSyncBlockRequested().values().stream()
+            .filter(time -> time < Time.getCurrentMillis() - NetConstants.SYNC_TIME_OUT)
+            .findFirst().ifPresent(time -> isDisconnected[0] = true);
+      }
+
+      //TODO:optimize here
+      if (!isDisconnected[0]) {
+        if (del.getHeadBlockId().getNum() - peer.getHeadBlockWeBothHave().getNum() > NetConstants.HEAD_NUM_MAX_DELTA
+            && peer.getConnectTime() < Time.getCurrentMillis() - NetConstants.HEAD_NUM_CHECK_TIME) {
+          isDisconnected[0] = true;
+        }
+      }
+
+
+      if (isDisconnected[0]) {
+        disconnectPeer(peer, ReasonCode.RESET);
+      }
+    });
   }
 
+
+
   private void onHandleInventoryMessage(PeerConnection peer, InventoryMessage msg) {
     //logger.info("on handle advertise inventory message");
     peer.cleanInvGarbage();
@@ -613,7 +656,7 @@ private void onHandleFetchDataMessage(PeerConnection peer, FetchInvDataMessage f
   }
 
   private void banTraitorPeer(PeerConnection peer) {
-    onDisconnectPeer(peer);
+    disconnectPeer(peer, ReasonCode.BAD_PROTOCOL); //TODO: ban it
   }
 
   private void onHandleChainInventoryMessage(PeerConnection peer, ChainInventoryMessage msg) {
@@ -848,7 +891,7 @@ private void syncNextBatchChainIds(PeerConnection peer) {
       peer.sendMessage(new SyncBlockChainMessage((LinkedList<BlockId>) chainSummary));
     } catch (Exception e) { //TODO: use tron excpetion here
       logger.debug(e.getMessage(), e);
-      onDisconnectPeer(peer);
+      disconnectPeer(peer, ReasonCode.BAD_PROTOCOL);//TODO: unlink?
     }
 
   }
@@ -858,6 +901,7 @@ public void onConnectPeer(PeerConnection peer) {
     //TODO:when use new p2p framework, remove this
     logger.info("start sync with::" + peer);
     peer.setTronState(TronState.START_TO_SYNC);
+    peer.setConnectTime(Time.getCurrentMillis());
     startSyncWithPeer(peer);
 //    if (mapPeer.containsKey(peer.getAddress())) {
 //      return;
@@ -873,7 +917,11 @@ public void onConnectPeer(PeerConnection peer) {
   @Override
   public void onDisconnectPeer(PeerConnection peer) {
     //TODO:when use new p2p framework, remove this
-    //mapPeer.remove(peer.getAddress());
+    //peer.disconnect(reason);
+  }
+
+  private void disconnectPeer(PeerConnection peer, ReasonCode reason) {
+    peer.disconnect(reason);
   }
 }
 

src/main/java/org/tron/core/net/peer/PeerConnection.java

@@ -14,6 +14,7 @@
 import org.tron.common.overlay.message.Message;
 import org.tron.common.overlay.server.Channel;
 import org.tron.common.utils.Sha256Hash;
+import org.tron.common.utils.Time;
 import org.tron.core.capsule.BlockCapsule.BlockId;
 
 @Slf4j
@@ -26,6 +27,16 @@ public int hashCode() {
     return super.hashCode();
   }
 
+  public long getConnectTime() {
+    return connectTime;
+  }
+
+  public void setConnectTime(long connectTime) {
+    this.connectTime = connectTime;
+  }
+
+  private long connectTime;
+
   //broadcast
   private Queue<Sha256Hash> invToUs = new LinkedBlockingQueue<>();
 
@@ -188,22 +199,26 @@ public String logSyncStats() {
 //    long lifeTime = System.currentTimeMillis() - connectedTime;
     return String.format(
         "Peer %s: [ %18s, ping %6s ms]-----------\n"
+            + "connect time: %s\n"
             + "last know block num: %s\n "
             + "needSyncFromPeer:%b\n "
             + "needSyncFromUs:%b\n"
             + "syncToFetchSize:%d\n"
             + "syncBlockRequestedSize:%d\n"
             + "unFetchSynNum:%d\n"
+            + "syncChainRequested:%s\n"
             + "blockInPorc:%d\n",
         this.getNode().getHost() + ":" + this.getNode().getPort(),
         this.getPeerIdShort(),
         (int)this.getPeerStats().getAvgLatency(),
+        Time.getTimeString(getConnectTime()),
         headBlockWeBothHave.getNum(),
         isNeedSyncFromPeer(),
         isNeedSyncFromUs(),
         syncBlockToFetch.size(),
         syncBlockRequested.size(),
         unfetchSyncNum,
+        syncChainRequested == null ? "NULL" : Time.getTimeString(syncChainRequested.getValue()),
         blockInProc.size());
   }