Merge pull request #493 from tronprotocol/sync_cache

sync: add send cache & solve synchro information inconsistency

Author: olenheim

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

@@ -95,8 +95,9 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 
   @Override
   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-    logger.info("exception caught, {} {}", ctx.channel().remoteAddress(), cause.getMessage());
+    logger.error("exception caught, {}", ctx.channel().remoteAddress(), cause);
     ctx.close();
+    closeChannel(ctx);
   }
 
   public void closeChannel(ChannelHandlerContext ctx) {

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

@@ -4,6 +4,8 @@
 import static org.tron.core.config.Parameter.NodeConstant.MAX_BLOCKS_IN_PROCESS;
 import static org.tron.core.config.Parameter.NodeConstant.MAX_BLOCKS_SYNC_FROM_ONE_PEER;
 
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
 import com.google.common.collect.Iterables;
 import io.netty.util.internal.ConcurrentSet;
 import java.util.ArrayList;
@@ -70,6 +72,14 @@ public class NodeImpl extends PeerConnectionDelegate implements Node {
   @Lazy
   private SyncPool pool;
 
+  Cache<Sha256Hash, TransactionMessage> TrxCache = CacheBuilder.newBuilder()
+          .maximumSize(10000).expireAfterWrite(60, TimeUnit.SECONDS)
+          .recordStats().build();
+
+  Cache<Sha256Hash, BlockMessage> BlockCache = CacheBuilder.newBuilder()
+          .maximumSize(10).expireAfterWrite(60, TimeUnit.SECONDS)
+          .recordStats().build();
+
   class InvToSend {
 
     private HashMap<PeerConnection, HashMap<InventoryType, LinkedList<Sha256Hash>>> send
@@ -241,10 +251,10 @@ public void broadcast(Message msg) {
     if (msg instanceof BlockMessage) {
       logger.info("Ready to broadcast a block, Its hash is " + msg.getMessageId());
       freshBlockId.offer(((BlockMessage) msg).getBlockId());
-      blockToAdvertise.add(((BlockMessage) msg).getBlockId());
+      BlockCache.put(msg.getMessageId(), (BlockMessage) msg);
       type = InventoryType.BLOCK;
     } else if (msg instanceof TransactionMessage) {
-      trxToAdvertise.add(msg.getMessageId());
+      TrxCache.put(msg.getMessageId(), (TransactionMessage)msg);
       type = InventoryType.TRX;
     } else {
       return;
@@ -776,24 +786,36 @@ private void onHandleFetchDataMessage(PeerConnection peer, FetchInvDataMessage f
     MessageTypes type = fetchInvDataMsg.getInvMessageType();
 
     //TODO:maybe can use message cache here
-    final BlockCapsule[] blocks = {del.getGenesisBlock()};
+    BlockCapsule block = null;
     //get data and send it one by one
-    fetchInvDataMsg.getHashList()
-        .forEach(hash -> {
-          if (del.contain(hash, type)) {
-            Message msg = del.getData(hash, type);
-            if (type.equals(MessageTypes.BLOCK)) {
-              blocks[0] = ((BlockMessage) msg).getBlockCapsule();
-            }
-            peer.sendMessage(msg);
-          } else {
-            peer.sendMessage(new ItemNotFound());
-          }
-        });
+    for (Sha256Hash hash : fetchInvDataMsg.getHashList()){
+
+      Message msg;
+
+      if (type == MessageTypes.BLOCK){
+        msg = BlockCache.getIfPresent(hash);
+      }else {
+        msg = TrxCache.getIfPresent(hash);
+      }
+
+      if (msg == null){
+        msg = del.getData(hash, type);
+      }
+
+      if (msg != null) {
+        if (type.equals(MessageTypes.BLOCK)) {
+          block = ((BlockMessage) msg).getBlockCapsule();
+        }
+        peer.sendMessage(msg);
+      } else {
+        logger.error("fetch message {} {} failed.", type, hash);
+        peer.sendMessage(new ItemNotFound());
+      }
+    }
 
-    if (blocks[0] != null) {
-      peer.setHeadBlockWeBothHave(blocks[0].getBlockId());
-      peer.setHeadBlockTimeWeBothHave(blocks[0].getTimeStamp());
+    if (block != null) {
+      peer.setHeadBlockWeBothHave(block.getBlockId());
+      peer.setHeadBlockTimeWeBothHave(block.getTimeStamp());
     }
   }
 

src/main/java/org/tron/core/net/peer/TronHandler.java

@@ -61,7 +61,7 @@ public void channelRead0(final ChannelHandlerContext ctx, TronMessage msg) throw
 
   @Override
   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-    logger.info("exception caught, {} {}", ctx.channel().remoteAddress(), cause.getMessage());
+    logger.error("exception caught, {}", ctx.channel().remoteAddress(), cause);
     ctx.close();
   }