Merge pull request #332 from tronprotocol/p2p_retry

mdf message queue support retry

Author: olenheim

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

@@ -51,7 +51,7 @@ public class ChannelManager {
 
   private Args args = Args.getInstance();
 
-  private int maxActivePeers = args.getNodeMaxActiveNodes() > 0 ? args.getNodeMaxActiveNodes() : 3000;;
+  private int maxActivePeers = args.getNodeMaxActiveNodes() > 0 ? args.getNodeMaxActiveNodes() : 30;
 
   private PeerServer peerServer;
 

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

@@ -19,23 +19,15 @@
 
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 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;
-import org.tron.common.overlay.message.Message;
-import org.tron.common.overlay.message.PingMessage;
-import org.tron.common.overlay.message.ReasonCode;
-import org.tron.common.overlay.message.StaticMessages;
+import org.tron.common.overlay.message.*;
+
+import java.util.Queue;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * This class contains the logic for sending messages in a queue
@@ -121,6 +113,8 @@ private void disconnect(DisconnectMessage msg) {
 
   public void receivedMessage(Message msg) throws InterruptedException {
 
+    logger.info("rcv from peer[{}], size:{} data:{}", ctx.channel().remoteAddress(), msg.getSendData().readableBytes(), msg.toString());
+
     if (requestQueue.peek() != null) {
       MessageRoundtrip messageRoundtrip = requestQueue.peek();
       Message waitingMessage = messageRoundtrip.getMsg();
@@ -135,7 +129,7 @@ public void receivedMessage(Message msg) throws InterruptedException {
   }
 
   private void removeAnsweredMessage(MessageRoundtrip messageRoundtrip) {
-    if (messageRoundtrip != null && messageRoundtrip.isAnswered())
+    if (messageRoundtrip != null && messageRoundtrip.isAnswered()) 
       requestQueue.remove();
   }
 
@@ -149,22 +143,32 @@ private void nudgeQueue() {
 
   private void sendToWire(MessageRoundtrip messageRoundtrip) {
 
-    if (messageRoundtrip != null && messageRoundtrip.getRetryTimes() == 0) {
-      // TODO: retry logic || messageRoundtrip.hasToRetry()){
+    if (messageRoundtrip == null){
+      return;
+    }
+
+    if (messageRoundtrip.getRetryTimes() > 0 && !messageRoundtrip.hasToRetry()){
+      return;
+    }
 
-      Message msg = messageRoundtrip.getMsg();
+    if (messageRoundtrip.getRetryTimes() > 0){
+      logger.info("send msg timeout. close channel {}.", ctx.channel().remoteAddress());
+      ctx.close();
+      return;
+    }
 
-      //TODO#p2p#peerDel : let node know
-      logger.info(msg.toString());
+    Message msg = messageRoundtrip.getMsg();
 
-      ctx.writeAndFlush(msg.getSendData())
-          .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+    ctx.writeAndFlush(msg.getSendData())
+            .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
 
-      if (msg.getAnswerMessage() != null) {
-        messageRoundtrip.incRetryTimes();
-        messageRoundtrip.saveTime();
-      }
+    if (msg.getAnswerMessage() != null) {
+      messageRoundtrip.incRetryTimes();
+      messageRoundtrip.saveTime();
     }
+
+    logger.info("send to peer[{}] retry[{}], length:{} data:{}", ctx.channel().remoteAddress(),
+            messageRoundtrip.getRetryTimes(), msg.getSendData().readableBytes(), msg.toString());
   }
 
   public void close() {

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

@@ -59,6 +59,10 @@ public void saveTime() {
         lastTimestamp = System.currentTimeMillis();
     }
 
+    public long getTime() {
+        return lastTimestamp;
+    }
+
     public boolean hasToRetry() {
         return 20000 < System.currentTimeMillis() - lastTimestamp;
     }

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

@@ -1,9 +1,10 @@
 package org.tron.core.net.message;
 
-import java.util.List;
 import org.tron.core.capsule.BlockCapsule.BlockId;
 import org.tron.protos.Protocol.BlockInventory.Type;
 
+import java.util.List;
+
 public class SyncBlockChainMessage extends BlockInventoryMessage {
 
   public SyncBlockChainMessage(byte[] packed) {
@@ -20,4 +21,9 @@ public SyncBlockChainMessage(List<BlockId> blockIds) {
   public MessageTypes getType() {
     return MessageTypes.fromByte(this.type);
   }
+
+  @Override
+  public Class<?> getAnswerMessage() {
+    return ChainInventoryMessage.class;
+  }
 }

src/main/resources/config.conf

@@ -28,7 +28,7 @@ node {
   maxActiveNodes = 30
 
   p2p {
-    version = 1 # 1: testnet; 101: debug
+    version = 31 # 1: testnet; 101: debug
   }
 
 }