Merge pull request #941 from tronprotocol/addTxWaterLine

Add tx water line

Author: olenheim

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

@@ -0,0 +1,82 @@
+package org.tron.common.utils;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Created by olivier on 2018/06/01
+ */
+class SlotBaseCounter {
+  private int slotSize;
+  private AtomicInteger[] slotCounter;
+
+  public SlotBaseCounter(int slotSize) {
+    slotSize = slotSize < 1 ? 1 : slotSize;
+    this.slotSize = slotSize;
+    this.slotCounter = new AtomicInteger[slotSize];
+    for (int i = 0; i < this.slotSize; i++) {
+      slotCounter[i] = new AtomicInteger(0);
+    }
+  }
+
+  public void increaseSlot(int slotSize) {
+    slotCounter[slotSize].incrementAndGet();
+  }
+
+  public void wipeSlot(int slotSize) {
+    slotCounter[slotSize].set(0);
+  }
+
+  public int totalCount() {
+    return Arrays.stream(slotCounter).mapToInt(slotCounter -> slotCounter.get()).sum();
+  }
+
+  @Override
+  public String toString() {
+    return Arrays.toString(slotCounter);
+  }
+}
+
+/**
+ * Created by olivier on 2018/06/01
+ */
+public class SlidingWindowCounter {
+  private volatile SlotBaseCounter slotBaseCounter;
+  private volatile int windowSize;
+  private volatile int head;
+
+  public SlidingWindowCounter(int windowSize) {
+    resizeWindow(windowSize);
+  }
+
+  public synchronized void resizeWindow(int windowSize) {
+    this.windowSize = windowSize;
+    this.slotBaseCounter = new SlotBaseCounter(windowSize);
+    this.head = 0;
+  }
+
+  public void increase() {
+    slotBaseCounter.increaseSlot(head);
+  }
+
+  public int totalAndAdvance() {
+    int total = totalCount();
+    advance();
+    return total;
+  }
+
+  public void advance() {
+    int tail = (head + 1) % windowSize;
+    slotBaseCounter.wipeSlot(tail);
+    head = tail;
+  }
+
+  public int totalCount() {
+    return slotBaseCounter.totalCount();
+  }
+
+  @Override
+  public String toString() {
+    return "total = " + totalCount() + " head = " + head + " >> " + slotBaseCounter;
+  }
+}

src/main/java/org/tron/core/config/Parameter.java

@@ -45,7 +45,7 @@ interface NetConstants {
     long HEAD_NUM_MAX_DELTA = 1000L;
     long HEAD_NUM_CHECK_TIME = 60000L;
     int MAX_INVENTORY_SIZE_IN_MINUTES = 2;
-    long NET_MAX_TRX_PER_SECOND = 1000L;
+    long NET_MAX_TRX_PER_SECOND = 700L;
     long MAX_TRX_PER_PEER = 200L;
     int NET_MAX_INV_SIZE_IN_MINUTES = 2;
     int MSG_CACHE_DURATION_IN_BLOCKS = 5;

src/main/java/org/tron/core/db/Manager.java

@@ -934,7 +934,7 @@ public synchronized BlockCapsule generateBlock(
       TransactionCapsule trx = (TransactionCapsule) iterator.next();
       if (DateTime.now().getMillis() - when
           > ChainConstant.BLOCK_PRODUCED_INTERVAL * 0.5 * ChainConstant.BLOCK_PRODUCED_TIME_OUT) {
-        logger.debug("Processing transaction time exceeds the 50% producing time。");
+        logger.warn("Processing transaction time exceeds the 50% producing time。");
         break;
       }
       // check the block size

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

@@ -102,7 +102,7 @@ public synchronized LinkedList<Sha256Hash> handleBlock(BlockCapsule block, boole
 
   @Override
   public void handleTransaction(TransactionCapsule trx) throws BadTransactionException {
-    logger.info("handle transaction");
+    logger.debug("handle transaction");
     if (dbManager.getTransactionIdCache().getIfPresent(trx.getTransactionId()) != null) {
       logger.warn("This transaction has been processed");
       return;

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

@@ -3,6 +3,7 @@
 import static org.tron.core.config.Parameter.ChainConstant.BLOCK_PRODUCED_INTERVAL;
 import static org.tron.core.config.Parameter.NetConstants.MAX_TRX_PER_PEER;
 import static org.tron.core.config.Parameter.NetConstants.MSG_CACHE_DURATION_IN_BLOCKS;
+import static org.tron.core.config.Parameter.NetConstants.NET_MAX_TRX_PER_SECOND;
 import static org.tron.core.config.Parameter.NodeConstant.MAX_BLOCKS_ALREADY_FETCHED;
 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;
@@ -43,6 +44,7 @@
 import org.tron.common.overlay.server.SyncPool;
 import org.tron.common.utils.ExecutorLoop;
 import org.tron.common.utils.Sha256Hash;
+import org.tron.common.utils.SlidingWindowCounter;
 import org.tron.common.utils.Time;
 import org.tron.core.capsule.BlockCapsule;
 import org.tron.core.capsule.BlockCapsule.BlockId;
@@ -88,11 +90,13 @@ public class NodeImpl extends PeerConnectionDelegate implements Node {
       .maximumSize(10).expireAfterWrite(60, TimeUnit.SECONDS)
       .recordStats().build();
 
+  private SlidingWindowCounter fetchWaterLine =
+      new SlidingWindowCounter(BLOCK_PRODUCED_INTERVAL * MSG_CACHE_DURATION_IN_BLOCKS / 100);
+
   private int maxTrxsSize = 1_000_000;
 
   private int maxTrxsCnt = 100;
 
-
   @Getter
   class PriorItem implements java.lang.Comparable<PriorItem> {
 
@@ -281,6 +285,9 @@ public Thread newThread(Runnable r) {
   private ScheduledExecutorService handleSyncBlockExecutor = Executors
       .newSingleThreadScheduledExecutor();
 
+  private ScheduledExecutorService fetchWaterLineExecutor = Executors
+      .newSingleThreadScheduledExecutor();
+
   private volatile boolean isHandleSyncBlockActive = false;
 
   private AtomicLong fetchSequenceCounter = new AtomicLong(0L);
@@ -467,6 +474,15 @@ private void activeTronPump() {
         logger.error("Unhandled exception", t);
       }
     }, 10, 1, TimeUnit.SECONDS);
+
+    //fetchWaterLine:
+    fetchWaterLineExecutor.scheduleWithFixedDelay(() -> {
+      try {
+        fetchWaterLine.advance();
+      } catch (Throwable t) {
+        logger.error("Unhandled exception", t);
+      }
+    }, 1000, 100, TimeUnit.MILLISECONDS);
   }
 
   private void consumerAdvObjToFetch() {
@@ -687,15 +703,18 @@ private void onHandleInventoryMessage(PeerConnection peer, InventoryMessage msg)
 
         //avoid TRX flood attack here.
         if (msg.getInventoryType().equals(InventoryType.TRX)
-            && peer.isAdvInvFull()) {
-          logger.info("A peer is flooding us, stop handle inv, the peer is:" + peer);
+            && (peer.isAdvInvFull()
+            || isFlooded())) {
+          logger.warn("A peer is flooding us, stop handle inv, the peer is: " + peer);
           return;
         }
 
         peer.getAdvObjSpreadToUs().put(id, System.currentTimeMillis());
         if (!requested[0]) {
           if (!badAdvObj.containsKey(id)) {
             if (!advObjToFetch.contains(id)) {
+              fetchWaterLine.increase();
+              logger.info("water line:" + fetchWaterLine.totalCount());
               this.advObjToFetch.put(id, new PriorItem(new Item(id, msg.getInventoryType()),
                   fetchSequenceCounter.incrementAndGet()));
             } else {
@@ -708,6 +727,11 @@ private void onHandleInventoryMessage(PeerConnection peer, InventoryMessage msg)
     }
   }
 
+  private boolean isFlooded() {
+    return fetchWaterLine.totalCount()
+        > BLOCK_PRODUCED_INTERVAL * NET_MAX_TRX_PER_SECOND * MSG_CACHE_DURATION_IN_BLOCKS / 1000;
+  }
+
   @Override
   public void syncFrom(Sha256Hash myHeadBlockHash) {
     try {
@@ -721,7 +745,6 @@ public void syncFrom(Sha256Hash myHeadBlockHash) {
     logger.info("wait end");
   }
 
-
   private void onHandleBlockMessage(PeerConnection peer, BlockMessage blkMsg) {
     Map<Item, Long> advObjWeRequested = peer.getAdvObjWeRequested();
     Map<BlockId, Long> syncBlockRequested = peer.getSyncBlockRequested();

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

@@ -1,6 +1,5 @@
 package org.tron.core.net.peer;
 
-import static org.tron.core.config.Parameter.ChainConstant.BLOCK_PRODUCED_INTERVAL;
 import static org.tron.core.config.Parameter.NetConstants.MAX_INVENTORY_SIZE_IN_MINUTES;
 import static org.tron.core.config.Parameter.NetConstants.NET_MAX_TRX_PER_SECOND;
 
@@ -169,8 +168,7 @@ private void removeIterator(Iterator<Entry<Sha256Hash, Long>> iterator, long old
   }
 
   public boolean isAdvInvFull() {
-    return advObjSpreadToUs.size() > MAX_INVENTORY_SIZE_IN_MINUTES * 60 * NET_MAX_TRX_PER_SECOND
-        + (MAX_INVENTORY_SIZE_IN_MINUTES + 1) * 60 / BLOCK_PRODUCED_INTERVAL;
+   return advObjSpreadToUs.size() > MAX_INVENTORY_SIZE_IN_MINUTES * 60 * NET_MAX_TRX_PER_SECOND;
   }
 
   public boolean isBanned() {