Merge pull request #770 from tronprotocol/p2p_msg

P2p: mdf P2pHandler hasping to volatile

Author: lvs007

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

@@ -26,7 +26,7 @@ public class MessageQueue {
 
   private volatile boolean sendMsgFlag = false;
 
-  private long sendTime;
+  private volatile long sendTime;
 
   private Thread sendMsgThread;
 

src/main/java/org/tron/common/overlay/server/P2pHandler.java

@@ -26,8 +26,6 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.context.annotation.Scope;
 import org.springframework.stereotype.Component;
 import org.tron.common.overlay.message.DisconnectMessage;
@@ -38,8 +36,6 @@
 @Scope("prototype")
 public class P2pHandler extends SimpleChannelInboundHandler<P2pMessage> {
 
-  private final static Logger logger = LoggerFactory.getLogger("P2pHandler");
-
   private static ScheduledExecutorService pingTimer =
       Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "P2pPingTimer"));
 
@@ -49,9 +45,9 @@ public class P2pHandler extends SimpleChannelInboundHandler<P2pMessage> {
 
   private ScheduledFuture<?> pingTask;
 
-  private boolean hasPing = false;
+  private volatile boolean hasPing = false;
 
-  private long sendPingTime;
+  private volatile long sendPingTime;
 
   private ChannelHandlerContext ctx;
 

src/main/java/org/tron/common/overlay/server/SyncPool.java

@@ -27,6 +27,9 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,7 +39,9 @@
 import org.tron.common.overlay.discover.Node;
 import org.tron.common.overlay.discover.NodeHandler;
 import org.tron.common.overlay.discover.NodeManager;
+import org.tron.common.utils.Sha256Hash;
 import org.tron.core.config.args.Args;
+import org.tron.core.net.message.TransactionMessage;
 import org.tron.core.net.peer.PeerConnection;
 import org.tron.core.net.peer.PeerConnectionDelegate;
 
@@ -49,6 +54,9 @@ public class SyncPool {
 
   private final List<PeerConnection> activePeers = Collections.synchronizedList(new ArrayList<PeerConnection>());
 
+  private Cache<NodeHandler, Long> nodeHandlerCache = CacheBuilder.newBuilder()
+          .maximumSize(1000).expireAfterWrite(120, TimeUnit.SECONDS).recordStats().build();
+
   @Autowired
   private NodeManager nodeManager;
 
@@ -102,7 +110,10 @@ private void fillUp() {
     nodesInUse.add(nodeManager.getPublicHomeNode().getHexId());
 
     List<NodeHandler> newNodes = nodeManager.getNodes(new NodeSelector(nodesInUse), lackSize);
-    newNodes.forEach(n -> peerClient.connectAsync(n, false));
+    newNodes.forEach(n -> {
+      peerClient.connectAsync(n, false);
+      nodeHandlerCache.put(n, System.currentTimeMillis());
+    });
   }
 
   // for test only
@@ -112,17 +123,17 @@ public void addActivePeers(PeerConnection p) {
 
 
   synchronized void logActivePeers() {
+
     logger.info("-------- active node {}", nodeManager.dumpActiveNodes().size());
-    nodeManager.dumpActiveNodes().forEach(handler -> {
+      nodeManager.dumpActiveNodes().forEach(handler -> {
       if (handler.getNode().getPort() == 18888) {
         logger.info("address: {}:{}, ID:{} {}",
-            handler.getNode().getHost(), handler.getNode().getPort(),
-            handler.getNode().getHexIdShort(), handler.getNodeStatistics().toString());
+        handler.getNode().getHost(), handler.getNode().getPort(),
+        handler.getNode().getHexIdShort(), handler.getNodeStatistics().toString());
       }
-    });
+     });
 
-    logger.info("-------- active channel {}, node in user size {}", channelManager.getActivePeers().size(),
-            channelManager.nodesInUse().size());
+    logger.info("-------- active channel {}", channelManager.getActivePeers().size());
     for (Channel channel: channelManager.getActivePeers()){
       logger.info(channel.toString());
     }
@@ -203,6 +214,10 @@ public boolean test(NodeHandler handler) {
         return false;
       }
 
+      if (nodeHandlerCache.getIfPresent(handler) != null){
+        return false;
+      }
+
       if (handler.getNodeStatistics().getReputation() < 100) {
         return false;
       }

src/main/java/org/tron/core/net/message/TransactionMessage.java

@@ -20,6 +20,12 @@ public TransactionMessage(Transaction trx) {
     this.data = trx.toByteArray();
   }
 
+  @Override
+  public String toString(){
+    return  new StringBuilder().append(super.toString())
+            .append("messageId: ").append(super.getMessageId()).toString();
+  }
+
   @Override
   public Class<?> getAnswerMessage() {
     return null;