fix conflict

Author: Yrp

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/Wallet.java

@@ -41,8 +41,10 @@
 import org.tron.core.db.BlockStore;
 import org.tron.core.db.Manager;
 import org.tron.core.db.UtxoStore;
+import org.tron.core.exception.BadItemException;
 import org.tron.core.exception.ContractExeException;
 import org.tron.core.exception.ContractValidateException;
+import org.tron.core.exception.ItemNotFoundException;
 import org.tron.core.exception.ValidateSignatureException;
 import org.tron.core.net.message.TransactionMessage;
 import org.tron.core.net.node.Node;
@@ -232,12 +234,35 @@ public Transaction createTransaction(WitnessUpdateContract witnessUpdateContract
 
   public Block getNowBlock() {
     Sha256Hash headBlockId = dbManager.getHeadBlockId();
-    return dbManager.getBlockById(headBlockId).getInstance();
+    try {
+      return dbManager.getBlockById(headBlockId).getInstance();
+    } catch (BadItemException e) {
+      logger.info(e.getMessage());
+      return null;
+    } catch (ItemNotFoundException e) {
+      logger.info(e.getMessage());
+      return null;
+    }
   }
 
   public Block getBlockByNum(long blockNum) {
-    Sha256Hash headBlockId = dbManager.getBlockIdByNum(blockNum);
-    return dbManager.getBlockById(headBlockId).getInstance();
+    Sha256Hash headBlockId = null;
+    try {
+      headBlockId = dbManager.getBlockIdByNum(blockNum);
+    } catch (BadItemException e) {
+      logger.info(e.getMessage());
+    } catch (ItemNotFoundException e) {
+      logger.info(e.getMessage());
+    }
+    try {
+      return dbManager.getBlockById(headBlockId).getInstance();
+    } catch (BadItemException e) {
+      logger.info(e.getMessage());
+      return null;
+    } catch (ItemNotFoundException e) {
+      logger.info(e.getMessage());
+      return null;
+    }
   }
 
   public AccountList getAllAccounts() {

src/main/java/org/tron/core/capsule/BlockCapsule.java

@@ -28,6 +28,7 @@
 import org.tron.common.crypto.ECKey.ECDSASignature;
 import org.tron.common.utils.Sha256Hash;
 import org.tron.core.capsule.utils.MerkleTree;
+import org.tron.core.exception.BadItemException;
 import org.tron.core.exception.ValidateSignatureException;
 import org.tron.protos.Protocol.Block;
 import org.tron.protos.Protocol.BlockHeader;
@@ -108,26 +109,25 @@ public long getNum() {
 
   private BlockId blockId = new BlockId(Sha256Hash.ZERO_HASH, 0);
 
-  private byte[] data;
-
   private Block block;
-
-  private boolean unpacked;
-
   public boolean generatedByMyself = false;
 
-  private synchronized void unPack() {
-    if (unpacked) {
-      return;
-    }
+  public BlockCapsule(long number, Sha256Hash hash, long when, ByteString witnessAddress) {
+    // blockheader raw
+    BlockHeader.raw.Builder blockHeaderRawBuild = BlockHeader.raw.newBuilder();
+    BlockHeader.raw blockHeaderRaw = blockHeaderRawBuild
+        .setNumber(number)
+        .setParentHash(hash.getByteString())
+        .setTimestamp(when)
+        .setWitnessAddress(witnessAddress).build();
 
-    try {
-      this.block = Block.parseFrom(data);
-    } catch (InvalidProtocolBufferException e) {
-      logger.debug(e.getMessage());
-    }
+    // block header
+    BlockHeader.Builder blockHeaderBuild = BlockHeader.newBuilder();
+    BlockHeader blockHeader = blockHeaderBuild.setRawData(blockHeaderRaw).build();
 
-    unpacked = true;
+    // block
+    Block.Builder blockBuild = Block.newBuilder();
+    this.block = blockBuild.setBlockHeader(blockHeader).build();
   }
 
   public BlockCapsule(long number, ByteString hash, long when, ByteString witnessAddress) {
@@ -146,7 +146,6 @@ public BlockCapsule(long number, ByteString hash, long when, ByteString witnessA
     // block
     Block.Builder blockBuild = Block.newBuilder();
     this.block = blockBuild.setBlockHeader(blockHeader).build();
-    unpacked = true;
   }
 
   public BlockCapsule(long timestamp, ByteString parentHash, long number,
@@ -167,7 +166,7 @@ public BlockCapsule(long timestamp, ByteString parentHash, long number,
     Block.Builder blockBuild = Block.newBuilder();
     transactionList.forEach(trx -> blockBuild.addTransactions(trx));
     this.block = blockBuild.setBlockHeader(blockHeader).build();
-    unpacked = true;
+
   }
 
   public void addTransaction(TransactionCapsule pendingTrx) {
@@ -193,7 +192,6 @@ public void sign(byte[] privateKey) {
   }
 
   private Sha256Hash getRawHash() {
-    unPack();
     return Sha256Hash.of(this.block.getBlockHeader().getRawData().toByteArray());
   }
 
@@ -210,15 +208,10 @@ public boolean validateSignature() throws ValidateSignatureException {
   }
 
   public BlockId getBlockId() {
-    unPack();
     if (blockId.equals(Sha256Hash.ZERO_HASH)) {
       blockId = new BlockId(Sha256Hash.of(this.block.getBlockHeader().toByteArray()), getNum());
     }
-
     return blockId;
-//    return blockId.equals(Sha256Hash.ZERO_HASH)
-//        ? blockId = new BlockId(Sha256Hash.of(this.block.getBlockHeader().toByteArray()), getNum())
-//        : blockId;
   }
 
   public Sha256Hash calcMerkleRoot() {
@@ -246,41 +239,29 @@ public void setMerkleRoot() {
   }
 
   public Sha256Hash getMerkleRoot() {
-    unPack();
     return Sha256Hash.wrap(this.block.getBlockHeader().getRawData().getTxTrieRoot());
   }
 
-  public ByteString getWitnessAddress(){
-    unPack();
+  public ByteString getWitnessAddress() {
     return this.block.getBlockHeader().getRawData().getWitnessAddress();
   }
 
 
-  private void pack() {
-    if (data == null) {
-      this.data = this.block.toByteArray();
-    }
-  }
-
-  public boolean validate() {
-    unPack();
-    return true;
-  }
-
   public BlockCapsule(Block block) {
     this.block = block;
-    unpacked = true;
   }
 
-  public BlockCapsule(byte[] data) {
-    this.data = data;
-    unPack();
+  public BlockCapsule(byte[] data) throws BadItemException {
+    try {
+      this.block = Block.parseFrom(data);
+    } catch (InvalidProtocolBufferException e) {
+      throw new BadItemException();
+    }
   }
 
   @Override
   public byte[] getData() {
-    pack();
-    return data;
+    return this.block.toByteArray();
   }
 
   @Override
@@ -289,28 +270,23 @@ public Block getInstance() {
   }
 
   public Sha256Hash getParentHash() {
-    unPack();
     return Sha256Hash.wrap(this.block.getBlockHeader().getRawData().getParentHash());
   }
 
   public ByteString getParentHashStr() {
-    unPack();
     return this.block.getBlockHeader().getRawData().getParentHash();
   }
 
   public long getNum() {
-    unPack();
     return this.block.getBlockHeader().getRawData().getNumber();
   }
 
   public long getTimeStamp() {
-    unPack();
     return this.block.getBlockHeader().getRawData().getTimestamp();
   }
 
   @Override
   public String toString() {
-    unPack();
     return "BlockCapsule{" +
         "blockId=" + blockId +
         ", num=" + getNum() +