Merge pull request tronprotocol#1097 from tronprotocol/add_message

Add message

Author: zhaohong

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

@@ -87,6 +87,7 @@ 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;
 
@@ -204,6 +205,11 @@ 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())) {
@@ -269,6 +275,7 @@ 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,10 +138,6 @@ 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/node/NodeStatistics.java

@@ -85,10 +85,6 @@ public NodeStatistics(Node node) {
     discoverMessageLatency = new SimpleStatter(node.getIdString());
   }
 
-  private int getSessionReputation() {
-    return getSessionFairReputation() + (isPredefined ? REPUTATION_PREDEFINED : 0);
-  }
-
   private int getSessionFairReputation() {
     int discoverReput = 0;
 
@@ -129,7 +125,14 @@ private int getSessionFairReputation() {
   }
 
   public int getReputation() {
-    return isReputationPenalized() ? 0 : persistedReputation / 2 + getSessionReputation();
+    int score = 0;
+    if (!isReputationPenalized()){
+      score += persistedReputation / 2 + getSessionFairReputation();
+    }
+    if (isPredefined){
+      score += REPUTATION_PREDEFINED;
+    }
+    return score;
   }
 
   public ReasonCode getDisconnectReason() {

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.toString();
+    entryId = n.getHost();
     distance = distance(ownerId, n.getId());
     touch();
   }
 
   public NodeEntry(byte[] ownerId, Node n) {
     this.node = n;
     this.ownerId = ownerId;
-    entryId = n.toString();
+    entryId = n.getHost();
     distance = distance(ownerId, n.getId());
     touch();
   }

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

@@ -59,6 +59,14 @@ 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,38 +1,25 @@
-/*
- * 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;
 import static org.tron.protos.Protocol.ReasonCode.TOO_MANY_PEERS;
+import static org.tron.protos.Protocol.ReasonCode.TOO_MANY_PEERS_WITH_SAME_IP;
 import static org.tron.protos.Protocol.ReasonCode.UNKNOWN;
 
 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;
 import java.util.concurrent.TimeUnit;
+import lombok.Getter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 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;
@@ -53,9 +40,14 @@ public class ChannelManager {
   private Cache<InetAddress, ReasonCode> recentlyDisconnected = CacheBuilder.newBuilder().maximumSize(1000)
       .expireAfterWrite(30, TimeUnit.SECONDS).recordStats().build();
 
+  @Getter
+  private Map<InetAddress, Node> trustPeers = new ConcurrentHashMap();
+
   private Args args = Args.getInstance();
 
-  private int maxActivePeers = args.getNodeMaxActiveNodes() > 0 ? args.getNodeMaxActiveNodes() : 30;
+  private int maxActivePeers = args.getNodeMaxActiveNodes();
+
+  private int getMaxActivePeersWithSameIp = args.getNodeMaxActiveNodesWithSameIp();
 
   private PeerServer peerServer;
 
@@ -73,6 +65,11 @@ private ChannelManager(final PeerServer peerServer, final PeerClient peerClient)
       new Thread(() -> peerServer.start(Args.getInstance().getNodeListenPort()),
           "PeerServerThread").start();
     }
+
+    for (Node node : args.getPassiveNodes()){
+      trustPeers.put(new InetSocketAddress(node.getHost(), node.getPort()).getAddress() , node);
+    }
+    logger.info("Trust peer size {}", trustPeers.size());
   }
 
   public void processDisconnect(Channel channel, ReasonCode reason){
@@ -110,19 +107,26 @@ public void notifyDisconnect(Channel channel) {
 
   public synchronized boolean processPeer(Channel peer) {
 
-    if (recentlyDisconnected.getIfPresent(peer) != null){
-      logger.info("Peer {} recently disconnected.", peer.getInetAddress());
-      return false;
-    }
+    if (!trustPeers.containsKey(peer.getInetAddress())){
+      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 (getConnectionNum(peer.getInetAddress()) >= getMaxActivePeersWithSameIp){
+        peer.disconnect(TOO_MANY_PEERS_WITH_SAME_IP);
+        return false;
+      }
     }
 
     if (activePeers.containsKey(peer.getNodeIdWrapper())) {
@@ -140,6 +144,16 @@ public synchronized boolean processPeer(Channel peer) {
     return true;
   }
 
+  public int getConnectionNum(InetAddress inetAddress){
+    int cnt = 0;
+    for (Channel channel: activePeers.values()){
+      if (channel.getInetAddress().equals(inetAddress)){
+        cnt++;
+      }
+    }
+    return cnt;
+  }
+
   public Collection<Channel> getActivePeers() {
     return activePeers.values();
   }

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

@@ -20,6 +20,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
 import java.util.List;
@@ -125,7 +126,8 @@ private void sendHelloMsg(ChannelHandlerContext ctx, long time){
   private void handleHelloMsg(ChannelHandlerContext ctx, HelloMessage msg) {
     if (remoteId.length != 64) {
       channel.initNode(msg.getFrom().getId(), msg.getFrom().getPort());
-      if (!syncPool.isCanConnect()) {
+      InetAddress address = ((InetSocketAddress)ctx.channel().remoteAddress()).getAddress();
+      if (!channelManager.getTrustPeers().keySet().contains(address) && !syncPool.isCanConnect()) {
         channel.disconnect(ReasonCode.TOO_MANY_PEERS);
         return;
       }

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

@@ -19,6 +19,7 @@
 
 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;
@@ -31,8 +32,6 @@
 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;
@@ -42,6 +41,7 @@
 import org.tron.common.overlay.discover.node.Node;
 import org.tron.common.overlay.discover.node.NodeHandler;
 import org.tron.common.overlay.discover.node.NodeManager;
+import org.tron.common.overlay.discover.node.NodeStatistics;
 import org.tron.core.config.args.Args;
 import org.tron.core.net.peer.PeerConnection;
 import org.tron.core.net.peer.PeerConnectionDelegate;
@@ -52,8 +52,10 @@ 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);
 
@@ -72,7 +74,9 @@ public class SyncPool {
 
   private Args args = Args.getInstance();
 
-  private int maxActiveNodes = args.getNodeMaxActiveNodes() > 0 ? args.getNodeMaxActiveNodes() : 30;
+  private int maxActiveNodes = args.getNodeMaxActiveNodes();
+
+  private int getMaxActivePeersWithSameIp = args.getNodeMaxActiveNodesWithSameIp();
 
   private ScheduledExecutorService poolLoopExecutor = Executors.newSingleThreadScheduledExecutor();
 
@@ -87,6 +91,10 @@ 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();
@@ -104,8 +112,11 @@ public void init(PeerConnectionDelegate peerDel) {
   }
 
   private void fillUp() {
-    int lackSize = (int) (maxActiveNodes * factor) - activePeers.size();
-    if(lackSize <= 0) return;
+    int lackSize = Math.max((int) (maxActiveNodes * factor) - activePeers.size(),
+        (int) (maxActiveNodes * activeFactor - activePeersCount.get()));
+    if (lackSize <= 0) {
+      return;
+    }
 
     final Set<String> nodesInUse = new HashSet<>();
     channelManager.getActivePeers().forEach(channel -> nodesInUse.add(channel.getPeerId()));
@@ -189,7 +200,7 @@ public synchronized void onDisconnect(Channel peer) {
   }
 
   public boolean isCanConnect() {
-    if (activePeers.size() >= maxActiveNodes) {
+    if (passivePeersCount.get() >= maxActiveNodes * (1 - activeFactor)) {
       return false;
     }
     return true;
@@ -220,15 +231,22 @@ public boolean test(NodeHandler handler) {
         return false;
       }
 
+      if (nodesInUse != null && nodesInUse.contains(handler.getNode().getHexId())) {
+        return false;
+      }
+
+      if (handler.getNodeStatistics().getReputation() >= NodeStatistics.REPUTATION_PREDEFINED){
+        return true;
+      }
+
       InetAddress inetAddress = handler.getInetSocketAddress().getAddress();
       if (channelManager.getRecentlyDisconnected().getIfPresent(inetAddress) != null) {
         return false;
       }
       if (channelManager.getBadPeers().getIfPresent(inetAddress) != null) {
         return false;
       }
-
-      if (nodesInUse != null && nodesInUse.contains(handler.getNode().getHexId())) {
+      if (channelManager.getConnectionNum(inetAddress) >= getMaxActivePeersWithSameIp){
         return false;
       }
 

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

@@ -45,9 +45,6 @@ public class Sha256Hash implements Serializable, Comparable<Sha256Hash> {
 
   private final byte[] bytes;
 
-  private long blockNum;
-
-
   private byte[] generateBlockId(long blockNum, Sha256Hash blockHash) {
     byte[] numBytes = Longs.toByteArray(blockNum);
     byte[] hash = blockHash.getBytes();
@@ -62,22 +59,16 @@ private byte[] generateBlockId(long blockNum, byte[] blockHash) {
     return hash;
   }
 
-  public long getBlockNum(){
-    return blockNum;
-  }
-
   public Sha256Hash(long num, byte[] hash) {
     byte[] rawHashBytes = this.generateBlockId(num, hash);
     checkArgument(rawHashBytes.length == LENGTH);
     this.bytes = rawHashBytes;
-    this.blockNum = num;
   }
 
   public Sha256Hash(long num, Sha256Hash hash) {
     byte[] rawHashBytes = this.generateBlockId(num, hash);
     checkArgument(rawHashBytes.length == LENGTH);
     this.bytes = rawHashBytes;
-    this.blockNum = num;
   }
 
   /**