Merge pull request #1094 from ethereum/feature/private-auto-sync

Feature/private auto sync

Author: mkalinin

ethereumj-core/src/main/java/org/ethereum/config/SystemProperties.java

@@ -825,6 +825,12 @@ public boolean fastSyncSkipHistory() {
         return config.getBoolean("sync.fast.skipHistory");
     }
 
+    @ValidateMe
+    public int makeDoneByTimeout() {
+        return config.getInt("sync.makeDoneByTimeout");
+    }
+
+
     @ValidateMe
     public boolean isPublicHomeNode() { return config.getBoolean("peer.discovery.public.home.node");}
 

ethereumj-core/src/main/java/org/ethereum/sync/BlockDownloader.java

@@ -20,6 +20,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.apache.commons.collections4.queue.CircularFifoQueue;
 import org.ethereum.core.*;
 import org.ethereum.net.server.Channel;
@@ -196,7 +197,7 @@ public void onFailure(Throwable t) {
                                         logger.debug("{}: Error receiving headers. Dropping the peer.", name, t);
                                         any.getEthHandler().dropConnection();
                                     }
-                                });
+                                }, MoreExecutors.directExecutor());
                                 it.remove();
                                 reqHeadersCounter++;
                             }
@@ -252,6 +253,7 @@ public void onFailure(Throwable t) {
                 if (blocksToAsk >= MAX_IN_REQUEST) {
 //                    SyncQueueIfc.BlocksRequest bReq = syncQueue.requestBlocks(maxBlocks);
 
+                    boolean fewHeadersReqMode = false;
                     if (bReqs.size() == 1 && bReqs.get(0).getBlockHeaders().size() <= 3) {
                         // new blocks are better to request from the header senders first
                         // to get more chances to receive block body promptly
@@ -261,7 +263,9 @@ public void onFailure(Throwable t) {
                                 ListenableFuture<List<Block>> futureBlocks =
                                         channel.getEthHandler().sendGetBlockBodies(singletonList(blockHeaderWrapper));
                                 if (futureBlocks != null) {
-                                    Futures.addCallback(futureBlocks, new BlocksCallback(channel));
+                                    Futures.addCallback(futureBlocks, new BlocksCallback(channel),
+                                            MoreExecutors.directExecutor());
+                                    fewHeadersReqMode = true;
                                 }
                             }
                         }
@@ -285,12 +289,21 @@ public void onFailure(Throwable t) {
                                     any.getEthHandler().sendGetBlockBodies(blocksRequest.getBlockHeaders());
                             blocksRequested += blocksRequest.getBlockHeaders().size();
                             if (futureBlocks != null) {
-                                Futures.addCallback(futureBlocks, new BlocksCallback(any));
+                                Futures.addCallback(futureBlocks, new BlocksCallback(any),
+                                        MoreExecutors.directExecutor());
                                 reqBlocksCounter++;
                                 it.remove();
                             }
                         }
                     }
+
+                    // Case when we have requested few headers and was not able
+                    // to remove request from the list in above cycle because
+                    // there were no idle peers or whatever
+                    if (fewHeadersReqMode && !bReqs.isEmpty()) {
+                        bReqs.clear();
+                    }
+
                     receivedBlocksLatch = new CountDownLatch(max(reqBlocksCounter - 2, 1));
                     receivedBlocksLatch.await(1000, TimeUnit.MILLISECONDS);
                 } else {

ethereumj-core/src/main/java/org/ethereum/sync/SyncManager.java

@@ -34,6 +34,7 @@
 
 import java.text.DecimalFormat;
 import java.text.DecimalFormatSymbols;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -107,6 +108,7 @@ public void accept(BlockWrapper blockWrapper) {
     private long importStart;
     private EthereumListener.SyncState syncDoneType = EthereumListener.SyncState.COMPLETE;
     private ScheduledExecutorService logExecutor = Executors.newSingleThreadScheduledExecutor();
+    private LocalDateTime initRegularTime;
 
     private AtomicInteger blocksInMem = new AtomicInteger(0);
 
@@ -166,6 +168,26 @@ void initRegularSync(EthereumListener.SyncState syncDoneType) {
 
         syncQueueThread = new Thread (queueProducer, "SyncQueueThread");
         syncQueueThread.start();
+
+        if (config.makeDoneByTimeout() >= 0) {
+            logger.info("Custom long sync done timeout set to {} second(s)", config.makeDoneByTimeout());
+            this.initRegularTime = LocalDateTime.now();
+            ScheduledExecutorService shortSyncAwait = Executors.newSingleThreadScheduledExecutor();
+            shortSyncAwait.scheduleAtFixedRate(() -> {
+                try {
+                    if (LocalDateTime.now().minusSeconds(config.makeDoneByTimeout()).isAfter(initRegularTime) &&
+                            getLastKnownBlockNumber() == blockchain.getBestBlock().getNumber()) {
+                        logger.info("Sync done triggered by timeout");
+                        makeSyncDone();
+                        shortSyncAwait.shutdown();
+                    } else if (syncDone) {
+                        shortSyncAwait.shutdown();
+                    }
+                } catch (Exception e) {
+                    logger.error("Unexpected", e);
+                }
+            }, 0, 10, TimeUnit.SECONDS);
+        }
     }
 
     public SyncStatus getSyncStatus() {

ethereumj-core/src/main/resources/ethereumj.conf

@@ -305,6 +305,12 @@ sync {
     # exit if we receive a block that causes state conflict
     # this option is mainly for debugging purposes
     exitOnBlockConflict = false
+
+    # Make long sync done (switch to short sync) in XX seconds
+    # if all known blocks already downloaded.
+    # Useful in private networks where auto-switch could fail.
+    # Recommended value for private networks: 60 (seconds)
+    makeDoneByTimeout = -1
 }
 
 # miner options