restict connPool task size

src/main/java/org/tron/p2p/base/Parameter.java

@@ -24,6 +24,8 @@ public class Parameter {
 
   public static final int UDP_NETTY_WORK_THREAD_NUM = 1;
 
+  public static final int CONN_MAX_QUEUE_SIZE = 10;
+
   public static final int NODE_CONNECTION_TIMEOUT = 2000;
 
   public static final int KEEP_ALIVE_TIMEOUT = 20_000;

src/main/java/org/tron/p2p/connection/business/detect/NodeDetectService.java

@@ -36,7 +36,7 @@ public class NodeDetectService implements MessageProcess {
       .newBuilder().maximumSize(5000).expireAfterWrite(1, TimeUnit.HOURS).build();
 
   private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
-      new BasicThreadFactory.Builder().namingPattern("nodeDetectService").build());
+      BasicThreadFactory.builder().namingPattern("nodeDetectService").build());
 
   private final long NODE_DETECT_THRESHOLD = 5 * 60 * 1000;
 

src/main/java/org/tron/p2p/connection/business/keepalive/KeepAliveService.java

@@ -21,7 +21,7 @@
 public class KeepAliveService implements MessageProcess {
 
   private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
-      new BasicThreadFactory.Builder().namingPattern("keepAlive").build());
+      BasicThreadFactory.builder().namingPattern("keepAlive").build());
 
   public void init() {
     executor.scheduleWithFixedDelay(() -> {

src/main/java/org/tron/p2p/connection/business/pool/ConnPoolService.java

@@ -14,6 +14,7 @@
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
@@ -42,29 +43,31 @@
 public class ConnPoolService extends P2pEventHandler {
 
   private final List<Channel> activePeers = Collections.synchronizedList(new ArrayList<>());
-  private Cache<InetAddress, Long> peerClientCache = CacheBuilder.newBuilder()
+  private final Cache<InetAddress, Long> peerClientCache = CacheBuilder.newBuilder()
       .maximumSize(1000).expireAfterWrite(120, TimeUnit.SECONDS).recordStats().build();
   @Getter
   private final AtomicInteger passivePeersCount = new AtomicInteger(0);
   @Getter
   private final AtomicInteger activePeersCount = new AtomicInteger(0);
   @Getter
   private final AtomicInteger connectingPeersCount = new AtomicInteger(0);
-  private final ScheduledExecutorService poolLoopExecutor = Executors.newSingleThreadScheduledExecutor(
-      new BasicThreadFactory.Builder().namingPattern("connPool").build());
-  private final ScheduledExecutorService disconnectExecutor = Executors.newSingleThreadScheduledExecutor(
-      new BasicThreadFactory.Builder().namingPattern("randomDisconnect").build());
+  private final ScheduledThreadPoolExecutor poolLoopExecutor = new ScheduledThreadPoolExecutor(1,
+      BasicThreadFactory.builder().namingPattern("connPool").build());
+  private final ScheduledExecutorService disconnectExecutor =
+      Executors.newSingleThreadScheduledExecutor(
+          BasicThreadFactory.builder().namingPattern("randomDisconnect").build());
 
   public P2pConfig p2pConfig = Parameter.p2pConfig;
   private PeerClient peerClient;
-  private List<InetSocketAddress> configActiveNodes = new ArrayList<>();
+  private final List<InetSocketAddress> configActiveNodes = new ArrayList<>();
 
   public ConnPoolService() {
     this.messageTypes = new HashSet<>(); //no message type registers
     try {
       Parameter.addP2pEventHandle(this);
       configActiveNodes.addAll(p2pConfig.getActiveNodes());
     } catch (P2pException e) {
+      //no exception will throw
     }
   }
 
@@ -268,6 +271,10 @@ public void triggerConnect(InetSocketAddress address) {
       return;
     }
     connectingPeersCount.decrementAndGet();
+    if (poolLoopExecutor.getQueue().size() >= Parameter.CONN_MAX_QUEUE_SIZE) {
+      log.warn("ConnPool task' size is greater than or equal to {}", Parameter.CONN_MAX_QUEUE_SIZE);
+      return;
+    }
     try {
       if (!ChannelManager.isShutdown) {
         poolLoopExecutor.submit(() -> {
@@ -311,6 +318,7 @@ public synchronized void onDisconnect(Channel peer) {
 
   @Override
   public void onMessage(Channel channel, byte[] data) {
+    //do nothing
   }
 
   public void close() {
@@ -323,6 +331,7 @@ public void close() {
         }
       });
       poolLoopExecutor.shutdownNow();
+      disconnectExecutor.shutdownNow();
     } catch (Exception e) {
       log.warn("Problems shutting down executor", e);
     }

src/main/java/org/tron/p2p/connection/socket/PeerClient.java

@@ -23,7 +23,7 @@ public class PeerClient {
 
   public void init() {
     workerGroup = new NioEventLoopGroup(0,
-        new BasicThreadFactory.Builder().namingPattern("peerClient-%d").build());
+        BasicThreadFactory.builder().namingPattern("peerClient-%d").build());
   }
 
   public void close() {

src/main/java/org/tron/p2p/connection/socket/PeerServer.java

@@ -39,10 +39,10 @@ public void close() {
 
   public void start(int port) {
     EventLoopGroup bossGroup = new NioEventLoopGroup(1,
-        new BasicThreadFactory.Builder().namingPattern("peerBoss").build());
+        BasicThreadFactory.builder().namingPattern("peerBoss").build());
     //if threads = 0, it is number of core * 2
     EventLoopGroup workerGroup = new NioEventLoopGroup(Parameter.TCP_NETTY_WORK_THREAD_NUM,
-        new BasicThreadFactory.Builder().namingPattern("peerWorker-%d").build());
+        BasicThreadFactory.builder().namingPattern("peerWorker-%d").build());
     P2pChannelInitializer p2pChannelInitializer = new P2pChannelInitializer("", false, true);
     try {
       ServerBootstrap b = new ServerBootstrap();

src/main/java/org/tron/p2p/discover/Node.java

@@ -16,6 +16,8 @@ public class Node implements Serializable, Cloneable {
 
   private static final long serialVersionUID = -4267600517925770636L;
 
+  @Setter
+  @Getter
   private byte[] id;
 
   @Getter
@@ -24,6 +26,8 @@ public class Node implements Serializable, Cloneable {
   @Getter
   protected String hostV6;
 
+  @Setter
+  @Getter
   protected int port;
 
   @Setter
@@ -32,6 +36,7 @@ public class Node implements Serializable, Cloneable {
   @Setter
   private int p2pVersion;
 
+  @Getter
   private long updateTime;
 
   public Node(InetSocketAddress address) {
@@ -111,37 +116,17 @@ public String getHexIdShort() {
     return getIdShort(getHexId());
   }
 
-  public byte[] getId() {
-    return id;
-  }
-
-  public void setId(byte[] id) {
-    this.id = id;
-  }
-
   public String getHostKey() {
     return getPreferInetSocketAddress().getAddress().getHostAddress();
   }
 
-  public int getPort() {
-    return port;
-  }
-
-  public void setPort(int port) {
-    this.port = port;
-  }
-
   public String getIdString() {
     if (id == null) {
       return null;
     }
     return new String(id);
   }
 
-  public long getUpdateTime() {
-    return updateTime;
-  }
-
   public void touch() {
     updateTime = System.currentTimeMillis();
   }
@@ -179,8 +164,8 @@ public boolean equals(Object o) {
     return false;
   }
 
-  private String getIdShort(String Id) {
-    return Id == null ? "<null>" : Id.substring(0, 8);
+  private String getIdShort(String hexId) {
+    return hexId == null ? "<null>" : hexId.substring(0, 8);
   }
 
   public InetSocketAddress getInetSocketAddressV4() {

src/main/java/org/tron/p2p/discover/protocol/kad/DiscoverTask.java

@@ -15,7 +15,7 @@
 public class DiscoverTask {
 
   private ScheduledExecutorService discoverer = Executors.newSingleThreadScheduledExecutor(
-      new BasicThreadFactory.Builder().namingPattern("discoverTask").build());
+      BasicThreadFactory.builder().namingPattern("discoverTask").build());
 
   private KadService kadService;
 

src/main/java/org/tron/p2p/discover/protocol/kad/KadService.java

@@ -57,7 +57,7 @@ public void init() {
       bootNodes.add(new Node(address));
     }
     this.pongTimer = Executors.newSingleThreadScheduledExecutor(
-        new BasicThreadFactory.Builder().namingPattern("pongTimer").build());
+        BasicThreadFactory.builder().namingPattern("pongTimer").build());
     this.homeNode = new Node(Parameter.p2pConfig.getNodeID(), Parameter.p2pConfig.getIp(),
         Parameter.p2pConfig.getIpv6(), Parameter.p2pConfig.getPort());
     this.table = new NodeTable(homeNode);

src/main/java/org/tron/p2p/dns/sync/Client.java

@@ -10,7 +10,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -42,7 +41,7 @@ public class Client {
   private final Map<String, ClientTree> clientTrees = new HashMap<>();
 
   private final ScheduledExecutorService syncer = Executors.newSingleThreadScheduledExecutor(
-      new BasicThreadFactory.Builder().namingPattern("dnsSyncer").build());
+      BasicThreadFactory.builder().namingPattern("dnsSyncer").build());
 
   public Client() {
     this.cache = CacheBuilder.newBuilder()

src/main/java/org/tron/p2p/dns/update/PublishService.java

@@ -25,7 +25,7 @@ public class PublishService {
   private static final long publishDelay = 1 * 60 * 60;
 
   private ScheduledExecutorService publisher = Executors.newSingleThreadScheduledExecutor(
-      new BasicThreadFactory.Builder().namingPattern("publishService").build());
+      BasicThreadFactory.builder().namingPattern("publishService").build());
   private Publish publish;
 
   public void init() {

src/main/java/org/tron/p2p/utils/NetUtil.java

@@ -214,7 +214,7 @@ public static InetSocketAddress parseInetSocketAddress(String para) {
 
   private static String getIp(List<String> multiSrcUrls) {
     ExecutorService executor = Executors.newCachedThreadPool(
-        new BasicThreadFactory.Builder().namingPattern("getIp").build());
+        BasicThreadFactory.builder().namingPattern("getIp").build());
     CompletionService<String> completionService = new ExecutorCompletionService<>(executor);
 
     List<Callable<String>> tasks = new ArrayList<>();