Merge branch 'develop' of https://github.com/tronprotocol/java-tron into develop

# Conflicts:
#	src/main/java/org/tron/core/net/node/NodeDelegateImpl.java

Author: zhaohong

src/main/java/org/tron/common/overlay/discover/NodeManager.java

@@ -281,6 +281,9 @@ public List<NodeHandler> getNodes(Predicate<NodeHandler> predicate,  int limit)
 
     logger.debug("nodeHandlerMap size {} filter peer  size {}", nodeHandlerMap.size(), filtered.size());
 
+    //TODO: here can use head num sort.
+    filtered.sort((o1, o2) -> o2.getNodeStatistics().getReputation() - o1.getNodeStatistics().getReputation());
+
     return CollectionUtils.truncate(filtered, limit);
   }
 

src/main/java/org/tron/common/overlay/discover/UDPListener.java

@@ -61,7 +61,7 @@ public UDPListener(final NodeManager nodeManager) {
       } else {
         new Thread(() -> {
           try {
-            UDPListener.this.start();
+            start();
           } catch (Exception e) {
             logger.debug(e.getMessage(), e);
             throw new RuntimeException(e);

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

@@ -58,12 +58,15 @@ public Thread newThread(Runnable r) {
     }
   });
 
+  private static Thread sendMsgThread;
+
   private Queue<MessageRoundtrip> requestQueue = new ConcurrentLinkedQueue<>();
-  private Queue<MessageRoundtrip> respondQueue = new ConcurrentLinkedQueue<>();
+  //private Queue<MessageRoundtrip> respondQueue = new ConcurrentLinkedQueue<>();
+
+  private BlockingQueue<Message> msgQueue = new LinkedBlockingQueue<>();
+
   private ChannelHandlerContext ctx = null;
 
-//  @Autowired
-//  EthereumListener ethereumListener;
   boolean hasPing = false;
   private ScheduledFuture<?> timerTask;
   private Channel channel;
@@ -73,13 +76,29 @@ public MessageQueue() {
 
   public void activate(ChannelHandlerContext ctx) {
     this.ctx = ctx;
+
     timerTask = timer.scheduleAtFixedRate(() -> {
       try {
         nudgeQueue();
       } catch (Throwable t) {
         logger.error("Unhandled exception", t);
       }
     }, 10, 10, TimeUnit.MILLISECONDS);
+
+    sendMsgThread = new Thread(()->{
+     while (true) {
+       try {
+         Message msg = msgQueue.take();
+         ctx.writeAndFlush(msg.getSendData())
+                 .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+       }catch (InterruptedException e){
+         break;
+       }catch (Exception e) {
+         logger.error("send message failed, {}, error info: {}", ctx.channel().remoteAddress(), e.getMessage());
+       }
+     }
+    });
+    sendMsgThread.start();
   }
 
   public void setChannel(Channel channel) {
@@ -95,7 +114,7 @@ public void sendMessage(Message msg) {
     if (msg.getAnswerMessage() != null)
       requestQueue.add(new MessageRoundtrip(msg));
     else
-      respondQueue.add(new MessageRoundtrip(msg));
+      msgQueue.offer(msg);
   }
 
   public void disconnect() {
@@ -138,7 +157,7 @@ private void nudgeQueue() {
     // remove last answered message on the queue
     removeAnsweredMessage(requestQueue.peek());
     // Now send the next message
-    sendToWire(respondQueue.poll());
+    //sendToWire(respondQueue.poll());
     sendToWire(requestQueue.peek());
   }
 
@@ -173,8 +192,7 @@ private void sendToWire(MessageRoundtrip messageRoundtrip) {
   }
 
   public void close() {
-    if (!timerTask.isCancelled()) {
-      timerTask.cancel(false);
-    }
+    sendMsgThread.interrupt();
+    timerTask.cancel(false);
   }
 }

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

@@ -116,6 +116,7 @@ private synchronized void prepareActive() {
 
     if (active.isEmpty()) return;
 
+    //sort by latency
     active.sort(Comparator.comparingDouble(c -> c.getPeerStats().getAvgLatency()));
 
     for (PeerConnection channel : active) {
@@ -129,10 +130,6 @@ private synchronized void prepareActive() {
   }
 
   synchronized void logActivePeers() {
-    logger.info("-------- active node.");
-    for (NodeHandler handler : nodeManager.dumpActiveNodes()) {
-      logger.info(handler.getNode().toString());
-    }
     logger.info("-------- active channel {}, node in user size {}", channelManager.getActivePeers().size(), channelManager.nodesInUse().size());
     for (Channel channel: channelManager.getActivePeers()){
       logger.info(channel.toString());

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

@@ -7,7 +7,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
-import java.util.function.Function;
 
 public class ExecutorLoop<In> {
 

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

@@ -17,7 +17,11 @@ interface ChainConstant {
 
   interface NodeConstant {
 
-    long SYNC_FETCH_BATCH_NUM = 1000;
+    long SYNC_RETURN_BATCH_NUM = 1000;
+    long SYNC_FETCH_BATCH_NUM = 500;
+    long MAX_BLOCKS_IN_PROCESS = 400;
+    long MAX_BLOCKS_ALREADY_FETCHED = 800;
+    long MAX_BLOCKS_SYNC_FROM_ONE_PEER = 200;
     long SYNC_CHAIN_LIMIT_NUM = 500;
 
   }

src/main/java/org/tron/core/config/args/Args.java

@@ -6,6 +6,7 @@
 import com.beust.jcommander.Parameter;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigObject;
+
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileReader;
@@ -23,6 +24,7 @@
 import java.util.Optional;
 import java.util.Properties;
 import java.util.stream.Collectors;
+
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
@@ -154,11 +156,11 @@ public class Args {
   @Getter
   @Setter
   //If you are running a solidity node for java tron, this flag is set to true
-  private boolean solidityNode;
+  private boolean solidityNode = false;
 
   @Getter
   @Setter
-  //If you are running a solidity node for java tron, solidity node must assign a trust node for sync solidity block
+  @Parameter(names = {"--trust-node"}, description = "Trust node addr")
   private String trustNodeAddr;
 
   public static void clearParam() {
@@ -299,8 +301,9 @@ public static void setParam(final String[] args, final String configFilePath) {
     INSTANCE.nodeP2pVersion =
         config.hasPath("node.p2p.version") ? config.getInt("node.p2p.version") : 0;
 
-    INSTANCE.trustNodeAddr =
-        config.hasPath("node.trust") ? config.getString("node.trust") : null;
+    if (StringUtils.isEmpty(INSTANCE.trustNodeAddr)) {
+      INSTANCE.trustNodeAddr = config.hasPath("node.trustNode") ? config.getString("node.trustNode") : null;
+    }
   }
 
 

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

@@ -243,6 +243,11 @@ private void addIfEmtpy() {
     }
   }
 
+  @Override
+  public int size() {
+    return stack.size();
+  }
+
   public void setMaxSize(int maxSize) {
     this.maxSize.set(maxSize);
   }

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

@@ -817,6 +817,14 @@ public void updateLatestSolidifiedBlock() {
     long latestSolidifiedBlockNum = numbers.get(solidifiedPosition);
 
     getDynamicPropertiesStore().saveLatestSolidifiedBlockNum(latestSolidifiedBlockNum);
+    ((AbstractRevokingStore) revokingStore).setMaxSize((int) (
+        dynamicPropertiesStore.getLatestBlockHeaderNumber()
+            - dynamicPropertiesStore.getLatestSolidifiedBlockNum() + 1)
+    );
+  }
+
+  public long getSyncBeginNumber() {
+    return dynamicPropertiesStore.getLatestBlockHeaderNumber() - revokingStore.size();
   }
 
   /**
@@ -847,6 +855,14 @@ public void updateSignedWitness(BlockCapsule block) {
     witnessCapsule.setLatestBlockNum(block.getNum());
     witnessCapsule.setLatestSlotNum(witnessController.getAbSlotAtTime(block.getTimeStamp()));
 
+    //Update memory witness status
+    WitnessCapsule wit = witnessController.getWitnesseByAddress(block.getWitnessAddress());
+    if (wit != null) {
+      wit.setTotalProduced(witnessCapsule.getTotalProduced() + 1);
+      wit.setLatestBlockNum(block.getNum());
+      wit.setLatestSlotNum(witnessController.getAbSlotAtTime(block.getTimeStamp()));
+    }
+
     this.getWitnessStore().put(witnessCapsule.getAddress().toByteArray(), witnessCapsule);
 
     AccountCapsule sun = accountStore.getSun();
@@ -861,7 +877,7 @@ public void updateSignedWitness(BlockCapsule block) {
       logger.debug(e.getMessage(), e);
     }
 
-    logger.info("updateSignedWitness. witness address:{}, blockNum:{}, totalProduced:{}",
+    logger.debug("updateSignedWitness. witness address:{}, blockNum:{}, totalProduced:{}",
         witnessCapsule.createReadableString(), block.getNum(), witnessCapsule.getTotalProduced());
 
   }

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

@@ -29,5 +29,7 @@ public interface RevokingDatabase {
 
   void enable();
 
+  int size();
+
   void disable();
 }