Merge pull request #769 from tronprotocol/sync_trx

p2p: Dealing with duplication trx

Author: lvs007

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

@@ -46,7 +46,7 @@ public class ChannelManager {
 
   private static final Logger logger = LoggerFactory.getLogger("ChannelManager");
 
-  private static final int inboundConnectionBanTimeout = 60 * 1000;
+  private static final int inboundConnectionBanTimeout = 30 * 1000;
 
   private final Map<ByteArrayWrapper, Channel> activePeers = new ConcurrentHashMap<>();
 

src/main/java/org/tron/core/net/node/NodeImpl.java

@@ -476,25 +476,19 @@ private void consumerAdvObjToFetch() {
     long now = Time.getCurrentMillis();
     advObjToFetch.values().stream().sorted(PriorItem::compareTo).forEach(idToFetch -> {
       Sha256Hash hash = idToFetch.getHash();
-      if ( (idToFetch.getType().equals(InventoryType.TRX) && TrxCache.getIfPresent(hash) != null) ||
-              (idToFetch.getType().equals(InventoryType.BLOCK) && BlockCache.getIfPresent(hash) != null) ){
-        logger.info("{} {}  Already exist.", idToFetch.getType(), hash);
-        advObjToFetch.remove(hash);
-        return;
-      }
       if (idToFetch.getTime() < now - MSG_CACHE_DURATION_IN_BLOCKS * BLOCK_PRODUCED_INTERVAL) {
         logger.info("This obj is too late to fetch: " + idToFetch);
         advObjToFetch.remove(hash);
-      }else {
-        filterActivePeer.stream()
-                .filter(peer -> peer.getAdvObjSpreadToUs().containsKey(hash) && sendPackage.getSize(peer) < MAX_TRX_PER_PEER)
-                .sorted(Comparator.comparingInt(peer -> sendPackage.getSize(peer)))
-                .findFirst().ifPresent(peer -> {
-          sendPackage.add(idToFetch, peer);
-          peer.getAdvObjWeRequested().put(idToFetch.getItem(), now);
-          advObjToFetch.remove(hash);
-        });
+        return;
       }
+      filterActivePeer.stream()
+              .filter(peer -> peer.getAdvObjSpreadToUs().containsKey(hash) && sendPackage.getSize(peer) < MAX_TRX_PER_PEER)
+              .sorted(Comparator.comparingInt(peer -> sendPackage.getSize(peer)))
+              .findFirst().ifPresent(peer -> {
+        sendPackage.add(idToFetch, peer);
+        peer.getAdvObjWeRequested().put(idToFetch.getItem(), now);
+        advObjToFetch.remove(hash);
+      });
     });
 
     sendPackage.sendFetch();
@@ -648,8 +642,7 @@ public synchronized void disconnectInactive() {
 
   private void onHandleInventoryMessage(PeerConnection peer, InventoryMessage msg) {
     for (Sha256Hash id : msg.getHashList()){
-      if ( (msg.getInventoryType().equals(InventoryType.TRX) && TrxCache.getIfPresent(id) != null) ||
-              (msg.getInventoryType().equals(InventoryType.BLOCK) && BlockCache.getIfPresent(id) != null) ){
+      if (msg.getInventoryType().equals(InventoryType.TRX) && TrxCache.getIfPresent(id) != null) {
         logger.info("{} {} from peer {} Already exist.", msg.getInventoryType(), id, peer.getNode().getHost());
         continue;
       }
@@ -854,17 +847,27 @@ private void cleanUpSyncPeer(PeerConnection peer, ReasonCode reasonCode) {
     disconnectPeer(peer, reasonCode);
   }
 
+  synchronized boolean isTrxExist(TransactionMessage trxMsg){
+    if (TrxCache.getIfPresent(trxMsg.getMessageId()) != null){
+      return true;
+    }
+    TrxCache.put(trxMsg.getMessageId(), trxMsg);
+    return false;
+  }
+
   private void onHandleTransactionMessage(PeerConnection peer, TransactionMessage trxMsg) {
-    //logger.info("on handle transaction message");
     try {
       Item item = new Item(trxMsg.getMessageId(), InventoryType.TRX);
       if (!peer.getAdvObjWeRequested().containsKey(item)) {
         throw new TraitorPeerException("We don't send fetch request to" + peer);
-      } else {
-        peer.getAdvObjWeRequested().remove(item);
-        del.handleTransaction(trxMsg.getTransactionCapsule());
-        broadcast(trxMsg);
       }
+      peer.getAdvObjWeRequested().remove(item);
+      if (isTrxExist(trxMsg)){
+        logger.info("Trx {} from Peer {} already processed.", trxMsg.getMessageId(), peer.getNode().getHost());
+        return;
+      }
+      del.handleTransaction(trxMsg.getTransactionCapsule());
+      broadcast(trxMsg);
     } catch (TraitorPeerException e) {
       logger.error(e.getMessage());
       banTraitorPeer(peer, ReasonCode.BAD_PROTOCOL);