Merge pull request #308 from Junze888/develop

Develop

src/main/java/io/xdag/config/AbstractConfig.java

@@ -80,7 +80,7 @@ public class AbstractConfig implements Config, AdminSpec, NodeSpec, WalletSpec,
     protected int netMaxOutboundConnections = 128;
     protected int netMaxInboundConnections = 512;
     protected int netMaxInboundConnectionsPerIp = 5;
-    protected int netMaxMessageQueueSize = 4096;
+//    protected int netMaxMessageQueueSize = 4096;
     protected int netMaxFrameBodySize = 128 * 1024;
     protected int netMaxPacketSize = 16 * 1024 * 1024;
     protected int netRelayRedundancy = 8;

src/main/java/io/xdag/config/spec/NodeSpec.java

@@ -44,7 +44,7 @@ public interface NodeSpec {
 
     int getWaitEpoch();
 
-    int getNetMaxMessageQueueSize();
+//    int getNetMaxMessageQueueSize();
 
     int getNetHandshakeExpiry();
 

src/main/java/io/xdag/core/BlockchainImpl.java

@@ -866,6 +866,7 @@ public XAmount unApplyBlock(Block block) {
         for (Address link : links) {
             if (!link.isAddress) {
                 Block ref = getBlockByHash(link.getAddress(), false);
+                //even mainBlock duplicate link the TX_block which other mainBlock is handled, we could check the TX ref if this mainBlock.
                 if (ref.getInfo().getRef() != null
                         && equalBytes(ref.getInfo().getRef(), block.getHashLow().toArray())
                         && ((ref.getInfo().flags & BI_MAIN_REF) != 0)) {

src/main/java/io/xdag/net/XdagP2pHandler.java

@@ -338,10 +338,10 @@ private ReasonCode checkPeer(Peer peer, boolean newHandShake) {
 //        }
 
         // validator can't share IP address
-        if (channelMgr.isActiveIP(channel.getRemoteIp()) // already connected
-                && nodeSpec.getNetwork() == Network.MAINNET) { // on main net
-            return ReasonCode.VALIDATOR_IP_LIMITED;
-        }
+//        if (channelMgr.isActiveIP(channel.getRemoteIp()) // already connected
+//                && nodeSpec.getNetwork() == Network.MAINNET) { // on main net
+//            return ReasonCode.VALIDATOR_IP_LIMITED;
+//        }
 
         return null;
     }

src/main/java/io/xdag/net/message/MessageQueue.java

@@ -30,11 +30,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.xdag.config.Config;
 import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.xdag.net.message.p2p.DisconnectMessage;
@@ -50,8 +46,8 @@ public class MessageQueue {
                     .daemon(true)
                     .build());
     private final Config config;
-
-    private final Queue<Message> queue = new ConcurrentLinkedQueue<>();
+    //'8192' is a value obtained from testing experience, not a standard value.Looking forward to optimization.
+    private final BlockingQueue<Message> queue = new LinkedBlockingQueue<>(8192);
     private final Queue<Message> prioritized = new ConcurrentLinkedQueue<>();
     private ChannelHandlerContext ctx;
     private ScheduledFuture<?> timerTask;
@@ -95,26 +91,27 @@ public void disconnect(ReasonCode code) {
         }
     }
 
-    public boolean sendMessage(Message msg) {
-        if (size() >= config.getNodeSpec().getNetMaxMessageQueueSize()) {
-            disconnect(ReasonCode.MESSAGE_QUEUE_FULL);
-            return false;
-        }
-
+    public void sendMessage(Message msg) {
+    //when full message queue, whitelist don't need to disconnect.
         if (config.getNodeSpec().getNetPrioritizedMessages().contains(msg.getCode())) {
             prioritized.add(msg);
         } else {
-            queue.add(msg);
+            try {
+                //update to BlockingQueue, capacity 8192
+                queue.put(msg);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
         }
-        return true;
     }
 
     public int size() {
         return queue.size() + prioritized.size();
     }
 
     private void nudgeQueue() {
-        int n = Math.min(5, size());
+        //Increase bandwidth consumption of a full used single sync thread to 3 Mbps.
+        int n = Math.min(8, size());
         if (n == 0) {
             return;
         }

src/test/java/io/xdag/BlockBuilder.java

@@ -100,6 +100,18 @@ public static Block generateNewTransactionBlock(Config config, KeyPair key, long
         return b;
     }
 
+    public static Block generateNewTransactionBlock(Config config, KeyPair key, long xdagTime, Address from, Address to,
+                                                    XAmount amount, XAmount VariableFee) {
+        List<Address> refs = Lists.newArrayList();
+        List<KeyPair> keys = Lists.newArrayList();
+        refs.add(new Address(from.getAddress(), XDAG_FIELD_INPUT, amount,true)); // key1
+        refs.add(new Address(to.getAddress(), XDAG_FIELD_OUTPUT, amount,true));
+        keys.add(key);
+        Block b = new Block(config, xdagTime, refs, null, false, keys, null, 0, VariableFee); // orphan
+        b.signOut(key);
+        return b;
+    }
+
     public static Block generateWalletTransactionBlock(Config config, KeyPair key, long xdagTime, Address from, Address to,
                                                     XAmount amount) {
         List<Address> refs = Lists.newArrayList();