Merge pull request #700 from tronprotocol/develop

merge develop

Author: zhaohong

src/main/java/org/tron/common/overlay/client/PeerClient.java

@@ -2,25 +2,27 @@
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultMessageSizeEstimator;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
+import org.tron.common.overlay.discover.Node;
+import org.tron.common.overlay.discover.NodeHandler;
+import org.tron.common.overlay.message.ReasonCode;
 import org.tron.common.overlay.server.TronChannelInitializer;
 import org.tron.core.config.args.Args;
 import org.tron.core.net.node.NodeImpl;
 
-import java.io.IOException;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.atomic.AtomicInteger;
-
 @Component
 public class PeerClient {
 
@@ -54,6 +56,20 @@ public void connect(String host, int port, String remoteId) {
         }
     }
 
+    public ChannelFuture connectAsync(NodeHandler nodeHandler, boolean discoveryMode) {
+        Node node = nodeHandler.getNode();
+        return connectAsync(node.getHost(), node.getPort(), node.getHexId(), discoveryMode)
+            .addListener((ChannelFutureListener) future -> {
+                if (!future.isSuccess()) {
+                    logger.error("connect to {}:{} fail,cause:{}", node.getHost(), node.getPort(),
+                        future.cause().getMessage());
+                    nodeHandler.getNodeStatistics().nodeDisconnectedLocal(ReasonCode.CONNECT_FAIL);
+                    nodeHandler.getNodeStatistics().notifyDisconnect();
+                    future.channel().close();
+                }
+            });
+    }
+
     public ChannelFuture connectAsync(String host, int port, String remoteId, boolean discoveryMode) {
 
         logger.info("connect peer {} {} {}", host, port, remoteId);

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

@@ -21,13 +21,13 @@
 import static java.lang.Math.min;
 
 import java.util.concurrent.atomic.AtomicLong;
-
 import org.tron.common.overlay.message.ReasonCode;
 
 public class NodeStatistics {
 
   public final static int REPUTATION_PREDEFINED = 100000;
   public final static long TOO_MANY_PEERS_PENALIZE_TIMEOUT = 60 * 1000;
+  private static final long CLEAR_CYCLE_TIME = 60 * 60 * 1000;
 
   public class StatHandler {
 
@@ -92,7 +92,7 @@ private int getSessionFairReputation() {
     int discoverReput = 0;
 
     discoverReput +=
-            min(discoverInPong.get(), 1) * (discoverOutPing.get() == discoverInPong.get() ? 50 : 1);
+        min(discoverInPong.get(), 1) * (discoverOutPing.get() == discoverInPong.get() ? 51 : 1);
     discoverReput += min(discoverInNeighbours.get(), 10) * 10;
     discoverReput += min(discoverInFind.get(), 50);
 
@@ -117,7 +117,10 @@ private int getSessionFairReputation() {
         }
       }
     }
-    return discoverReput + 10 * reput;
+    int score =
+        discoverReput + 10 * reput - (int) Math.pow(2, disconnectTimes) * (disconnectTimes > 0 ? 10
+            : 0);
+    return score > 0 ? score : 0;
   }
 
   public int getReputation() {
@@ -146,7 +149,15 @@ public boolean isReputationPenalized() {
       return true;
     }
 
-    return tronLastLocalDisconnectReason == ReasonCode.NULL_IDENTITY ||
+    if (lastDisconnectedTime > 0
+        && (System.currentTimeMillis() - lastDisconnectedTime) > CLEAR_CYCLE_TIME) {
+      tronLastLocalDisconnectReason = null;
+      tronLastRemoteDisconnectReason = null;
+      disconnectTimes = 0;
+      persistedReputation = 0;
+    }
+
+    if (tronLastLocalDisconnectReason == ReasonCode.NULL_IDENTITY ||
         tronLastRemoteDisconnectReason == ReasonCode.NULL_IDENTITY ||
         tronLastLocalDisconnectReason == ReasonCode.INCOMPATIBLE_PROTOCOL ||
         tronLastRemoteDisconnectReason == ReasonCode.INCOMPATIBLE_PROTOCOL ||
@@ -163,26 +174,38 @@ public boolean isReputationPenalized() {
         tronLastLocalDisconnectReason == ReasonCode.INCOMPATIBLE_VERSION ||
         tronLastRemoteDisconnectReason == ReasonCode.INCOMPATIBLE_VERSION ||
         tronLastLocalDisconnectReason == ReasonCode.INCOMPATIBLE_CHAIN ||
-        tronLastRemoteDisconnectReason == ReasonCode.INCOMPATIBLE_CHAIN;
+        tronLastRemoteDisconnectReason == ReasonCode.INCOMPATIBLE_CHAIN ||
+        tronLastRemoteDisconnectReason == ReasonCode.SYNC_FAIL ||
+        tronLastLocalDisconnectReason == ReasonCode.SYNC_FAIL) {
+      persistedReputation = 0;
+      return true;
+    }
+    return false;
   }
 
   public boolean isPenalized() {
     return tronLastLocalDisconnectReason == ReasonCode.NULL_IDENTITY ||
-            tronLastRemoteDisconnectReason == ReasonCode.NULL_IDENTITY ||
-            tronLastLocalDisconnectReason == ReasonCode.BAD_PROTOCOL ||
-            tronLastRemoteDisconnectReason == ReasonCode.BAD_PROTOCOL;
+        tronLastRemoteDisconnectReason == ReasonCode.NULL_IDENTITY ||
+        tronLastLocalDisconnectReason == ReasonCode.BAD_PROTOCOL ||
+        tronLastRemoteDisconnectReason == ReasonCode.BAD_PROTOCOL ||
+        tronLastLocalDisconnectReason == ReasonCode.SYNC_FAIL ||
+        tronLastRemoteDisconnectReason == ReasonCode.SYNC_FAIL;
   }
 
   public void nodeDisconnectedRemote(ReasonCode reason) {
     lastDisconnectedTime = System.currentTimeMillis();
     tronLastRemoteDisconnectReason = reason;
-    disconnectTimes++;
   }
 
   public void nodeDisconnectedLocal(ReasonCode reason) {
     lastDisconnectedTime = System.currentTimeMillis();
     tronLastLocalDisconnectReason = reason;
+  }
+
+  public void notifyDisconnect() {
+    lastDisconnectedTime = System.currentTimeMillis();
     disconnectTimes++;
+    persistedReputation = persistedReputation / 2;
   }
 
   public boolean wasDisconnected() {

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

@@ -9,10 +9,9 @@
 import org.tron.common.utils.Sha256Hash;
 import org.tron.core.net.message.MessageTypes;
 
-
 public abstract class Message {
 
-  protected static final Logger logger = LoggerFactory.getLogger("Net");
+  protected static final Logger logger = LoggerFactory.getLogger("Message");
 
   protected boolean unpacked;
   protected byte[] data;
@@ -32,7 +31,6 @@ public Message(byte type, byte[] packed) {
     unpacked = false;
   }
 
-
   public ByteBuf getSendData(){
      return Unpooled.wrappedBuffer(ArrayUtils.add(this.getData(), 0 ,type));
   }
@@ -41,13 +39,25 @@ public Sha256Hash getMessageId() {
     return Sha256Hash.of(getData());
   }
 
-  public abstract byte[] getData();
+  public byte[] getData(){
+    return this.data;
+  }
+
+  public MessageTypes getType(){
+    return MessageTypes.fromByte(this.type);
+  }
+
+  public abstract Class<?> getAnswerMessage();
 
+  @Override
   public String toString() {
     return "[Message Type: " + getType() + ", Message Hash: " + getMessageId() + "]";
   }
 
-  public abstract Class<?> getAnswerMessage();
+  @Override
+  public int hashCode() {
+    return Arrays.hashCode(data);
+  }
 
   @Override
   public boolean equals(Object o) {
@@ -60,12 +70,4 @@ public boolean equals(Object o) {
     Message message = (Message) o;
     return Arrays.equals(data, message.data);
   }
-
-  @Override
-  public int hashCode() {
-    return Arrays.hashCode(data);
-  }
-
-  public abstract MessageTypes getType();
-
 }

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

@@ -16,9 +16,6 @@
 
 public abstract class MessageFactory {
 
-  public static String ERR_NO_SUCH_MSG = "No such message";
-  public static String ERR_PARSE_FAILED = "parse message failed";
-
   protected abstract Message create(byte[] data) throws Exception;
 
 }

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

@@ -89,6 +89,8 @@ public enum ReasonCode {
 
     TIME_OUT(0x20),
 
+    CONNECT_FAIL(0x21),
+
     /**
      * [0xFF] Reason not specified
      */

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

@@ -162,7 +162,7 @@ public void processException(Throwable throwable){
         if (throwable instanceof ReadTimeoutException){
             logger.error("Read timeout, {}", address);
         }else if(baseThrowable instanceof P2pException){
-            logger.error("type: {}, info: {}, {}", ((P2pException) throwable).getType(), errMsg, address);
+            logger.error("type: {}, info: {}, {}", ((P2pException) baseThrowable).getType(), errMsg, address);
         }else if (errMsg != null && errMsg.contains("Connection reset by peer")){
             logger.error("{}, {}", errMsg, address);
         }else {

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

@@ -150,6 +150,7 @@ public void notifyDisconnect(Channel channel) {
         || channel.getChannelHandlerContext().channel() == null) {
       return;
     }
+    channel.getNodeStatistics().notifyDisconnect();
     InetSocketAddress socketAddress = (InetSocketAddress) channel.getChannelHandlerContext()
         .channel().remoteAddress();
     recentlyDisconnected.put(socketAddress.getAddress(), new Date());

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

@@ -26,7 +26,7 @@ public class MessageQueue {
 
   private static final Logger logger = LoggerFactory.getLogger("MessageQueue");
 
-  private boolean sendMsgFlag = false;
+  private volatile boolean sendMsgFlag = false;
 
   private Thread sendMsgThread;
 
@@ -70,7 +70,7 @@ public void activate(ChannelHandlerContext ctx) {
          Message msg = msgQueue.take();
          ctx.writeAndFlush(msg.getSendData()).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
        }catch (Exception e) {
-         logger.error("send message failed, {}, error info: {}", ctx.channel().remoteAddress(), e.getMessage());
+         logger.error("Send message failed, {}, error info: {}", ctx.channel().remoteAddress(), e.getMessage());
        }
      }
     });
@@ -102,6 +102,15 @@ public void close() {
     sendMsgFlag = false;
     if(sendTask != null && !sendTask.isCancelled()){
       sendTask.cancel(false);
+      sendTask = null;
+    }
+    if (sendMsgThread != null){
+      try{
+        sendMsgThread.join(20);
+        sendMsgThread = null;
+      }catch (Exception e){
+        logger.warn("Join send thread failed, peer {}", ctx.channel().remoteAddress());
+      }
     }
   }
 

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

@@ -102,8 +102,7 @@ private void fillUp() {
     nodesInUse.add(nodeManager.getPublicHomeNode().getHexId());
 
     List<NodeHandler> newNodes = nodeManager.getNodes(new NodeSelector(nodesInUse), lackSize);
-    newNodes.forEach(n ->  peerClient.connectAsync(n.getNode().getHost(), n.getNode().getPort(),
-                n.getNode().getHexId(), false));
+    newNodes.forEach(n -> peerClient.connectAsync(n, false));
   }
 
   // for test only

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

@@ -155,10 +155,6 @@ public static int byteArrayToInt(byte[] b) {
     return new BigInteger(1, b).intValue();
   }
 
-  public static boolean isNullOrZeroArray(byte[] array) {
-    return (array == null) || (array.length == 0);
-  }
-
   public static boolean isSingleZero(byte[] array) {
     return (array.length == 1 && array[0] == 0);
   }