Refactor of sync & adv module

src/main/java/org/tron/common/application/Application.java

@@ -18,7 +18,6 @@
 import org.tron.core.config.args.Args;
 import org.tron.core.db.BlockStore;
 import org.tron.core.db.Manager;
-import org.tron.core.net.node.Node;
 
 public interface Application {
 
@@ -36,8 +35,6 @@ public interface Application {
 
   void shutdownServices();
 
-  Node getP2pNode();
-
   BlockStore getBlockStoreS();
 
   void addService(Service service);

src/main/java/org/tron/common/application/ApplicationImpl.java

@@ -7,33 +7,23 @@
 import org.tron.core.config.args.Args;
 import org.tron.core.db.BlockStore;
 import org.tron.core.db.Manager;
-import org.tron.core.net.node.Node;
-import org.tron.core.net.node.NodeDelegate;
-import org.tron.core.net.node.NodeDelegateImpl;
-import org.tron.core.net.node.NodeImpl;
+import org.tron.core.net.TronNetService;
 
 @Slf4j(topic = "app")
 @Component
 public class ApplicationImpl implements Application {
 
-  @Autowired
-  private NodeImpl p2pNode;
-
   private BlockStore blockStoreDb;
   private ServiceContainer services;
-  private NodeDelegate nodeDelegate;
+
+  @Autowired
+  private TronNetService tronNetService;
 
   @Autowired
   private Manager dbManager;
 
   private boolean isProducer;
 
-
-  private void resetP2PNode() {
-    p2pNode.listen();
-    p2pNode.syncFrom(null);
-  }
-
   @Override
   public void setOptions(Args args) {
     // not used
@@ -44,7 +34,6 @@ public void setOptions(Args args) {
   public void init(Args args) {
     blockStoreDb = dbManager.getBlockStore();
     services = new ServiceContainer();
-    nodeDelegate = new NodeDelegateImpl(dbManager);
   }
 
   @Override
@@ -61,19 +50,17 @@ public void initServices(Args args) {
    * start up the app.
    */
   public void startup() {
-    p2pNode.setNodeDelegate(nodeDelegate);
-    resetP2PNode();
+    tronNetService.start();
   }
 
   @Override
   public void shutdown() {
     logger.info("******** begin to shutdown ********");
-    //p2pNode.shutDown();
+    tronNetService.close();
     synchronized (dbManager.getRevokingStore()) {
       closeRevokingStore();
       closeAllStore();
     }
-    closeConnection();
     dbManager.stopRepushThread();
     dbManager.stopRepushTriggerThread();
     EventPluginLoader.getInstance().stopPlugin();
@@ -90,11 +77,6 @@ public void shutdownServices() {
     services.stop();
   }
 
-  @Override
-  public Node getP2pNode() {
-    return p2pNode;
-  }
-
   @Override
   public BlockStore getBlockStoreS() {
     return blockStoreDb;
@@ -113,17 +95,6 @@ public void setIsProducer(boolean producer) {
     isProducer = producer;
   }
 
-  private void closeConnection() {
-    logger.info("******** begin to shutdown connection ********");
-    try {
-      p2pNode.close();
-    } catch (Exception e) {
-      logger.info("failed to close p2pNode. " + e);
-    } finally {
-      logger.info("******** end to shutdown connection ********");
-    }
-  }
-
   private void closeRevokingStore() {
     logger.info("******** begin to closeRevokingStore ********");
     dbManager.getRevokingStore().shutdown();

src/main/java/org/tron/common/application/CliApplication.java

@@ -17,7 +17,6 @@
 import org.tron.core.config.args.Args;
 import org.tron.core.db.BlockStore;
 import org.tron.core.db.Manager;
-import org.tron.core.net.node.Node;
 
 public class CliApplication implements Application {
 
@@ -56,11 +55,6 @@ public void shutdownServices() {
 
   }
 
-  @Override
-  public Node getP2pNode() {
-    return null;
-  }
-
   @Override
   public BlockStore getBlockStoreS() {
     return null;

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

@@ -13,13 +13,11 @@
 import lombok.extern.slf4j.Slf4j;
 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.Node;
 import org.tron.common.overlay.discover.node.NodeHandler;
 import org.tron.common.overlay.server.TronChannelInitializer;
 import org.tron.core.config.args.Args;
-import org.tron.core.net.node.NodeImpl;
 import org.tron.protos.Protocol.ReasonCode;
 
 @Slf4j(topic = "net")
@@ -29,10 +27,6 @@ public class PeerClient {
   @Autowired
   private ApplicationContext ctx;
 
-  @Autowired
-  @Lazy
-  private NodeImpl node;
-
   private EventLoopGroup workerGroup;
 
   public PeerClient() {
@@ -61,7 +55,7 @@ public ChannelFuture connectAsync(NodeHandler nodeHandler, boolean discoveryMode
     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(),
+            logger.warn("connect to {}:{} fail,cause:{}", node.getHost(), node.getPort(),
                 future.cause().getMessage());
             nodeHandler.getNodeStatistics().nodeDisconnectedLocal(ReasonCode.CONNECT_FAIL);
             nodeHandler.getNodeStatistics().notifyDisconnect();
@@ -70,14 +64,13 @@ public ChannelFuture connectAsync(NodeHandler nodeHandler, boolean discoveryMode
         });
   }
 
-  public ChannelFuture connectAsync(String host, int port, String remoteId, boolean discoveryMode) {
+  private ChannelFuture connectAsync(String host, int port, String remoteId, boolean discoveryMode) {
 
     logger.info("connect peer {} {} {}", host, port, remoteId);
 
     TronChannelInitializer tronChannelInitializer = ctx
         .getBean(TronChannelInitializer.class, remoteId);
     tronChannelInitializer.setPeerDiscoveryMode(discoveryMode);
-    tronChannelInitializer.setNodeImpl(node);
 
     Bootstrap b = new Bootstrap();
     b.group(workerGroup);

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

@@ -22,6 +22,7 @@
 import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
 import io.netty.handler.timeout.ReadTimeoutException;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -39,8 +40,7 @@
 import org.tron.common.overlay.message.StaticMessages;
 import org.tron.core.db.ByteArrayWrapper;
 import org.tron.core.exception.P2pException;
-import org.tron.core.net.peer.PeerConnectionDelegate;
-import org.tron.core.net.peer.TronHandler;
+import org.tron.core.net.TronNetHandler;
 import org.tron.protos.Protocol.ReasonCode;
 
 @Slf4j(topic = "net")
@@ -70,7 +70,7 @@ public class Channel {
   private P2pHandler p2pHandler;
 
   @Autowired
-  private TronHandler tronHandler;
+  private TronNetHandler tronNetHandler;
 
   private ChannelManager channelManager;
 
@@ -82,8 +82,6 @@ public class Channel {
 
   private long startTime;
 
-  private PeerConnectionDelegate peerDel;
-
   private TronState tronState = TronState.INIT;
 
   protected NodeStatistics nodeStatistics;
@@ -99,7 +97,7 @@ public class Channel {
   private boolean isTrustPeer;
 
   public void init(ChannelPipeline pipeline, String remoteId, boolean discoveryMode,
-      ChannelManager channelManager, PeerConnectionDelegate peerDel) {
+      ChannelManager channelManager) {
 
     this.channelManager = channelManager;
 
@@ -118,18 +116,14 @@ public void init(ChannelPipeline pipeline, String remoteId, boolean discoveryMod
     //handshake first
     pipeline.addLast("handshakeHandler", handshakeHandler);
 
-    this.peerDel = peerDel;
-
     messageCodec.setChannel(this);
     msgQueue.setChannel(this);
     handshakeHandler.setChannel(this, remoteId);
     p2pHandler.setChannel(this);
-    tronHandler.setChannel(this);
+    tronNetHandler.setChannel(this);
 
     p2pHandler.setMsgQueue(msgQueue);
-    tronHandler.setMsgQueue(msgQueue);
-    tronHandler.setPeerDel(peerDel);
-
+    tronNetHandler.setMsgQueue(msgQueue);
   }
 
   public void publicHandshakeFinished(ChannelHandlerContext ctx, HelloMessage msg) {
@@ -138,7 +132,7 @@ public void publicHandshakeFinished(ChannelHandlerContext ctx, HelloMessage msg)
     msgQueue.activate(ctx);
     ctx.pipeline().addLast("messageCodec", messageCodec);
     ctx.pipeline().addLast("p2p", p2pHandler);
-    ctx.pipeline().addLast("data", tronHandler);
+    ctx.pipeline().addLast("data", tronNetHandler);
     setStartTime(msg.getTimestamp());
     setTronState(TronState.HANDSHAKE_FINISHED);
     getNodeStatistics().p2pHandShake.add();
@@ -158,7 +152,10 @@ public void disconnect(ReasonCode reason) {
     this.isDisconnect = true;
     channelManager.processDisconnect(this, reason);
     DisconnectMessage msg = new DisconnectMessage(reason);
-    logger.info("Send to {}, {}", ctx.channel().remoteAddress(), msg);
+    logger.info("Send to {} online-time {}s, {}",
+        ctx.channel().remoteAddress(),
+        (System.currentTimeMillis() - startTime) / 1000,
+        msg);
     getNodeStatistics().nodeDisconnectedLocal(reason);
     ctx.writeAndFlush(msg.getSendData()).addListener(future -> close());
   }
@@ -168,17 +165,15 @@ public void processException(Throwable throwable) {
     while (baseThrowable.getCause() != null) {
       baseThrowable = baseThrowable.getCause();
     }
-    String errMsg = throwable.getMessage();
     SocketAddress address = ctx.channel().remoteAddress();
-    if (throwable instanceof ReadTimeoutException) {
-      logger.error("Read timeout, {}", address);
+    if (throwable instanceof ReadTimeoutException ||
+        throwable instanceof IOException) {
+      logger.warn("Close peer {}, reason: {}", address, throwable.getMessage());
     } else if (baseThrowable instanceof P2pException) {
-      logger.error("type: {}, info: {}, {}", ((P2pException) baseThrowable).getType(),
-          baseThrowable.getMessage(), address);
-    } else if (errMsg != null && errMsg.contains("Connection reset by peer")) {
-      logger.error("{}, {}", errMsg, address);
+      logger.warn("Close peer {}, type: {}, info: {}",
+          address,((P2pException) baseThrowable).getType(), baseThrowable.getMessage());
     } else {
-      logger.error("exception caught, {}", address, throwable);
+      logger.error("Close peer {}, exception caught", address, throwable);
     }
     close();
   }
@@ -224,10 +219,6 @@ public void setChannelHandlerContext(ChannelHandlerContext ctx) {
     this.inetSocketAddress = ctx == null ? null : (InetSocketAddress) ctx.channel().remoteAddress();
   }
 
-  public ChannelHandlerContext getChannelHandlerContext() {
-    return this.ctx;
-  }
-
   public InetAddress getInetAddress() {
     return ctx == null ? null : ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress();
   }
@@ -246,10 +237,7 @@ public long getStartTime() {
 
   public void setTronState(TronState tronState) {
     this.tronState = tronState;
-  }
-
-  public TronState getTronState() {
-    return tronState;
+    logger.info("Peer {} status change to {}.", inetSocketAddress, tronState);
   }
 
   public boolean isActive() {
@@ -260,10 +248,6 @@ public boolean isDisconnect() {
     return isDisconnect;
   }
 
-  public boolean isProtocolsInitialized() {
-    return tronState.ordinal() > TronState.INIT.ordinal();
-  }
-
   public boolean isTrustPeer() {
     return isTrustPeer;
   }

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

@@ -27,6 +27,17 @@
 @Component
 public class ChannelManager {
 
+  @Autowired
+  private PeerServer peerServer;
+
+  @Autowired
+  private PeerClient peerClient;
+
+  @Autowired
+  private SyncPool syncPool;
+
+  private Args args = Args.getInstance();
+
   private final Map<ByteArrayWrapper, Channel> activePeers = new ConcurrentHashMap<>();
 
   private Cache<InetAddress, ReasonCode> badPeers = CacheBuilder.newBuilder().maximumSize(10000)
@@ -38,25 +49,10 @@ public class ChannelManager {
   @Getter
   private Map<InetAddress, Node> trustPeers = new ConcurrentHashMap();
 
-  private Args args = Args.getInstance();
-
   private int maxActivePeers = args.getNodeMaxActiveNodes();
 
   private int getMaxActivePeersWithSameIp = args.getNodeMaxActiveNodesWithSameIp();
 
-  private PeerServer peerServer;
-
-  private PeerClient peerClient;
-
-  @Autowired
-  private SyncPool syncPool;
-
-  @Autowired
-  private ChannelManager(final PeerServer peerServer, final PeerClient peerClient) {
-    this.peerServer = peerServer;
-    this.peerClient = peerClient;
-  }
-
   public void init() {
     if (this.args.getNodeListenPort() > 0) {
       new Thread(() -> peerServer.start(Args.getInstance().getNodeListenPort()),
@@ -67,6 +63,8 @@ public void init() {
       trustPeers.put(new InetSocketAddress(node.getHost(), node.getPort()).getAddress(), node);
     }
     logger.info("Trust peer size {}", trustPeers.size());
+
+    syncPool.init();
   }
 
   public void processDisconnect(Channel channel, ReasonCode reason) {
@@ -164,5 +162,6 @@ public Cache<InetAddress, ReasonCode> getBadPeers() {
   public void close() {
     peerServer.close();
     peerClient.close();
+    syncPool.close();
   }
 }

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

@@ -71,7 +71,7 @@ public void activate(ChannelHandlerContext ctx) {
           }
           Message msg = msgQueue.take();
           ctx.writeAndFlush(msg.getSendData()).addListener((ChannelFutureListener) future -> {
-            if (!future.isSuccess()) {
+            if (!future.isSuccess() && !channel.isDisconnect()) {
               logger.error("Fail send to {}, {}", ctx.channel().remoteAddress(), msg);
             }
           });