Merge pull request #766 from tronprotocol/sync_trx

p2p: mdf duce trx

Author: lvs007

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

@@ -80,7 +80,7 @@ public class NodeImpl extends PeerConnectionDelegate implements Node {
   private SyncPool pool;
 
   private Cache<Sha256Hash, TransactionMessage> TrxCache = CacheBuilder.newBuilder()
-      .maximumSize(100_000).expireAfterWrite(1, TimeUnit.HOURS)
+      .maximumSize(100_000).expireAfterWrite(1, TimeUnit.HOURS).initialCapacity(100_000)
       .recordStats().build();
 
   private Cache<Sha256Hash, BlockMessage> BlockCache = CacheBuilder.newBuilder()
@@ -475,17 +475,24 @@ private void consumerAdvObjToFetch() {
     InvToSend sendPackage = new InvToSend();
     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(idToFetch.getHash());
+        advObjToFetch.remove(hash);
       }else {
         filterActivePeer.stream()
-                .filter(peer -> peer.getAdvObjSpreadToUs().containsKey(idToFetch.getHash()) && sendPackage.getSize(peer) < MAX_TRX_PER_PEER)
+                .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(idToFetch.getHash());
+          advObjToFetch.remove(hash);
         });
       }
     });
@@ -640,11 +647,11 @@ public synchronized void disconnectInactive() {
 
 
   private void onHandleInventoryMessage(PeerConnection peer, InventoryMessage msg) {
-    msg.getHashList().forEach(id -> {
+    for (Sha256Hash id : msg.getHashList()){
       if ( (msg.getInventoryType().equals(InventoryType.TRX) && TrxCache.getIfPresent(id) != null) ||
               (msg.getInventoryType().equals(InventoryType.BLOCK) && BlockCache.getIfPresent(id) != null) ){
         logger.info("{} {} from peer {} Already exist.", msg.getInventoryType(), id, peer.getNode().getHost());
-        return;
+        continue;
       }
       final boolean[] spreaded = {false};
       final boolean[] requested = {false};
@@ -681,7 +688,7 @@ private void onHandleInventoryMessage(PeerConnection peer, InventoryMessage msg)
           }
         }
       }
-    });
+    }
   }
 
   @Override