Merge pull request #1036 from tronprotocol/revert-1023-w_p2p

Revert "p2p: add trust nodes & active nodes function and solving eclipse attack"

Author: zhaohong

src/main/java/org/tron/common/overlay/discover/node/NodeHandler.java

@@ -87,7 +87,6 @@ public enum State {
   private NodeStatistics nodeStatistics;
   private NodeHandler replaceCandidate;
   private volatile boolean waitForPong = false;
-  private volatile boolean waitForNeighbors = false;
   private volatile int pingTrials = 3;
   private long pingSent;
 
@@ -205,11 +204,6 @@ public void handlePong(PongMessage msg) {
   }
 
   public void handleNeighbours(NeighborsMessage msg) {
-    if (!waitForNeighbors){
-      logger.warn("Receive neighbors from {} without send find nodes.", node.getHost());
-      return;
-    }
-    waitForNeighbors = false;
     getNodeStatistics().discoverInNeighbours.add();
     for (Node n : msg.getNodes()) {
       if (!nodeManager.getPublicHomeNode().getHexId().equals(n.getHexId())) {
@@ -275,7 +269,6 @@ public void sendNeighbours(List<Node> neighbours) {
   }
 
   public void sendFindNode(byte[] target) {
-    waitForNeighbors = true;
     Message findNode = new FindNodeMessage(nodeManager.getPublicHomeNode(), target);
     sendMessage(findNode);
     getNodeStatistics().discoverOutFind.add();

src/main/java/org/tron/common/overlay/discover/node/NodeManager.java

@@ -138,6 +138,10 @@ public void run() {
       for (Node node : bootNodes) {
         getNodeHandler(node);
       }
+
+      for (Node node : args.getNodeActive()) {
+        getNodeHandler(node).getNodeStatistics().setPredefined(true);
+      }
     }
   }
 

src/main/java/org/tron/common/overlay/discover/table/NodeEntry.java

@@ -33,15 +33,15 @@ public class NodeEntry {
   public NodeEntry(Node n) {
     this.node = n;
     this.ownerId = n.getId();
-    entryId = n.getHost();
+    entryId = n.toString();
     distance = distance(ownerId, n.getId());
     touch();
   }
 
   public NodeEntry(byte[] ownerId, Node n) {
     this.node = n;
     this.ownerId = ownerId;
-    entryId = n.getHost();
+    entryId = n.toString();
     distance = distance(ownerId, n.getId());
     touch();
   }

src/main/java/org/tron/common/overlay/discover/table/NodeTable.java

@@ -59,14 +59,6 @@ public final void initialize() {
 
   public synchronized Node addNode(Node n) {
     NodeEntry e = new NodeEntry(node.getId(), n);
-    if (nodes.contains(e)) {
-      nodes.forEach(nodeEntry -> {
-        if (nodeEntry.equals(e)) {
-          nodeEntry.touch();
-        }
-      });
-      return null;
-    }
     NodeEntry lastSeen = buckets[getBucketId(e)].addNode(e);
     if (lastSeen != null) {
       return lastSeen.getNode();

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

@@ -1,3 +1,20 @@
+/*
+ * Copyright (c) [2016] [ <ether.camp> ]
+ * This file is part of the ethereumJ library.
+ *
+ * The ethereumJ library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The ethereumJ library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the ethereumJ library. If not, see <http://www.gnu.org/licenses/>.
+ */
 package org.tron.common.overlay.server;
 
 import static org.tron.protos.Protocol.ReasonCode.DUPLICATE_PEER;
@@ -7,7 +24,6 @@
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import java.net.InetAddress;
-import java.net.InetSocketAddress;
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -17,7 +33,6 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.tron.common.overlay.client.PeerClient;
-import org.tron.common.overlay.discover.node.Node;
 import org.tron.core.config.args.Args;
 import org.tron.core.db.ByteArrayWrapper;
 import org.tron.protos.Protocol.ReasonCode;
@@ -38,8 +53,6 @@ public class ChannelManager {
   private Cache<InetAddress, ReasonCode> recentlyDisconnected = CacheBuilder.newBuilder().maximumSize(1000)
       .expireAfterWrite(30, TimeUnit.SECONDS).recordStats().build();
 
-  private Map<InetAddress, Node> trustPeers = new ConcurrentHashMap();
-
   private Args args = Args.getInstance();
 
   private int maxActivePeers = args.getNodeMaxActiveNodes() > 0 ? args.getNodeMaxActiveNodes() : 30;
@@ -60,11 +73,6 @@ private ChannelManager(final PeerServer peerServer, final PeerClient peerClient)
       new Thread(() -> peerServer.start(Args.getInstance().getNodeListenPort()),
           "PeerServerThread").start();
     }
-
-    for (Node node : args.getTrustNodes()){
-      trustPeers.put(new InetSocketAddress(node.getHost(), node.getPort()).getAddress() , node);
-    }
-    logger.info("Trust peer size {}", trustPeers.size());
   }
 
   public void processDisconnect(Channel channel, ReasonCode reason){
@@ -102,21 +110,19 @@ public void notifyDisconnect(Channel channel) {
 
   public synchronized boolean processPeer(Channel peer) {
 
-    if (!trustPeers.containsKey(peer.getInetAddress())){
-      if (recentlyDisconnected.getIfPresent(peer) != null){
-        logger.info("Peer {} recently disconnected.", peer.getInetAddress());
-        return false;
-      }
+    if (recentlyDisconnected.getIfPresent(peer) != null){
+      logger.info("Peer {} recently disconnected.", peer.getInetAddress());
+      return false;
+    }
 
-      if (badPeers.getIfPresent(peer) != null) {
-        peer.disconnect(peer.getNodeStatistics().getDisconnectReason());
-        return false;
-      }
+    if (badPeers.getIfPresent(peer) != null) {
+      peer.disconnect(peer.getNodeStatistics().getDisconnectReason());
+      return false;
+    }
 
-      if (!peer.isActive() && activePeers.size() >= maxActivePeers) {
-        peer.disconnect(TOO_MANY_PEERS);
-        return false;
-      }
+    if (!peer.isActive() && activePeers.size() >= maxActivePeers) {
+      peer.disconnect(TOO_MANY_PEERS);
+      return false;
     }
 
     if (activePeers.containsKey(peer.getNodeIdWrapper())) {

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

@@ -19,7 +19,6 @@
 
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
-import com.google.common.collect.Lists;
 import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -32,6 +31,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
+
+import com.google.common.collect.Lists;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -51,10 +52,8 @@ public class SyncPool {
   public static final Logger logger = LoggerFactory.getLogger("SyncPool");
 
   private static final double factor = 0.4;
-  private static final double activeFactor = 0.2;
 
-  private final List<PeerConnection> activePeers = Collections
-      .synchronizedList(new ArrayList<PeerConnection>());
+  private final List<PeerConnection> activePeers = Collections.synchronizedList(new ArrayList<PeerConnection>());
   private final AtomicInteger passivePeersCount = new AtomicInteger(0);
   private final AtomicInteger activePeersCount = new AtomicInteger(0);
 
@@ -88,10 +87,6 @@ public void init(PeerConnectionDelegate peerDel) {
 
     peerClient = ctx.getBean(PeerClient.class);
 
-    for (Node node : args.getActiveNodes()) {
-      nodeManager.getNodeHandler(node).getNodeStatistics().setPredefined(true);
-    }
-
     poolLoopExecutor.scheduleWithFixedDelay(() -> {
       try {
         fillUp();
@@ -109,11 +104,8 @@ public void init(PeerConnectionDelegate peerDel) {
   }
 
   private void fillUp() {
-    int lackSize = Math.max((int) (maxActiveNodes * factor) - activePeers.size(),
-        (int) (maxActiveNodes * activeFactor - activePeersCount.get()));
-    if (lackSize <= 0) {
-      return;
-    }
+    int lackSize = (int) (maxActiveNodes * factor) - activePeers.size();
+    if(lackSize <= 0) return;
 
     final Set<String> nodesInUse = new HashSet<>();
     channelManager.getActivePeers().forEach(channel -> nodesInUse.add(channel.getPeerId()));
@@ -197,7 +189,7 @@ public synchronized void onDisconnect(Channel peer) {
   }
 
   public boolean isCanConnect() {
-    if (passivePeersCount.get() >= maxActiveNodes * (1 - activeFactor)) {
+    if (activePeers.size() >= maxActiveNodes) {
       return false;
     }
     return true;

src/main/java/org/tron/core/config/args/Args.java

@@ -121,11 +121,7 @@ public class Args {
 
   @Getter
   @Setter
-  private List<Node> activeNodes;
-
-  @Getter
-  @Setter
-  private List<Node> trustNodes;
+  private List<Node> nodeActive;
 
   @Getter
   @Setter
@@ -272,8 +268,7 @@ public static void clearParam() {
     INSTANCE.nodeDiscoveryEnable = false;
     INSTANCE.nodeDiscoveryPersist = false;
     INSTANCE.nodeConnectionTimeout = 0;
-    INSTANCE.activeNodes = Collections.emptyList();
-    INSTANCE.trustNodes = Collections.emptyList();
+    INSTANCE.nodeActive = Collections.emptyList();
     INSTANCE.nodeChannelReadTimeout = 0;
     INSTANCE.nodeMaxActiveNodes = 0;
     INSTANCE.minParticipationRate = 0;
@@ -407,9 +402,7 @@ public static void setParam(final String[] args, final String confFileName) {
         config.hasPath("node.connection.timeout") ? config.getInt("node.connection.timeout") * 1000
             : 0;
 
-    INSTANCE.activeNodes = getNodes(config, "active.node");
-
-    INSTANCE.trustNodes = getNodes(config, "trust.node");
+    INSTANCE.nodeActive = nodeActive(config);
 
     INSTANCE.nodeChannelReadTimeout =
         config.hasPath("node.channel.read.timeout") ? config.getInt("node.channel.read.timeout")
@@ -551,12 +544,12 @@ public String getOutputDirectory() {
     return this.outputDirectory;
   }
 
-  private static List<Node> getNodes(final com.typesafe.config.Config config, String path) {
-    if (!config.hasPath(path)) {
+  private static List<Node> nodeActive(final com.typesafe.config.Config config) {
+    if (!config.hasPath("node.active")) {
       return Collections.EMPTY_LIST;
     }
     List<Node> ret = new ArrayList<>();
-    List<String> list = config.getStringList(path);
+    List<String> list = config.getStringList("node.active");
     for (String configString : list) {
       Node n = Node.instanceOf(configString);
       ret.add(n);

src/main/resources/config.conf

@@ -31,7 +31,7 @@ storage {
     //      writeBufferSize = 10485760, // 10 MB = 10 * 1024 * 1024 B
     //      cacheSize = 10485760,       // 10 MB = 10 * 1024 * 1024 B
     //      maxOpenFiles = 100
-//    }, 
+//    },
 //    {
 //      name = "account-index",
 //      path = "storage_directory_test",
@@ -123,20 +123,6 @@ node {
 
 }
 
-active.node = [
-  # Active establish connection in any case
-  # Sample entries:
-  # "ip:port",
-  # "ip:port"
-]
-
-trust.node = [
-  # Passive accept connection in any case
-  # Sample entries:
-  # "ip:port",
-  # "ip:port"
-]
-
 seed.node = {
   # List of the seed nodes
   # Seed nodes are stable full nodes

src/test/java/org/tron/core/config/args/ArgsTest.java

@@ -62,7 +62,7 @@ public void get() {
     Assert.assertEquals("46.168.1.1", args.getNodeExternalIp());
     Assert.assertEquals(18888, args.getNodeListenPort());
     Assert.assertEquals(2000, args.getNodeConnectionTimeout());
-    Assert.assertEquals(0, args.getActiveNodes().size());
+    Assert.assertEquals(0, args.getNodeActive().size());
     Assert.assertEquals(30, args.getNodeMaxActiveNodes());
     Assert.assertEquals(43, args.getNodeP2pVersion());
     //Assert.assertEquals(30, args.getSyncNodeCount());