Display World State Sync Progress in Logs (#1645)

* Display World State Sync Progress in Logs

- compute estimated world state completion
- display estimated world state completion in the logs
- `CompleteTaskStep` now have access to a `LongSupplier` to retireve the number of pending requests
- use a `RunnableCounter` to trigger displaying every 1000 requests completed
- only show a new log when the estimation changes
- added test to check the estimation computation

* report using raw ratio rather than percentage

* resolve PR discussion

* rename getTotal to get

Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/worldstate/CompleteTaskStep.java

@@ -17,23 +17,39 @@ import tech.pegasys.pantheon.ethereum.worldstate.WorldStateStorage;
 import tech.pegasys.pantheon.metrics.Counter;
 import tech.pegasys.pantheon.metrics.MetricsSystem;
 import tech.pegasys.pantheon.metrics.PantheonMetricCategory;
+import tech.pegasys.pantheon.metrics.RunnableCounter;
 import tech.pegasys.pantheon.services.tasks.Task;
 
-public class CompleteTaskStep {
+import java.text.DecimalFormat;
+import java.util.function.LongSupplier;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
+public class CompleteTaskStep {
+  private static final Logger LOG = LogManager.getLogger();
+  private static final int DISPLAY_PROGRESS_STEP = 100000;
   private final WorldStateStorage worldStateStorage;
-  private final Counter completedRequestsCounter;
+  private final RunnableCounter completedRequestsCounter;
   private final Counter retriedRequestsCounter;
+  private final LongSupplier worldStatePendingRequestsCurrentSupplier;
+  private final DecimalFormat doubleFormatter = new DecimalFormat("#.##");
+  private double estimatedWorldStateCompletion;
 
   public CompleteTaskStep(
-      final WorldStateStorage worldStateStorage, final MetricsSystem metricsSystem) {
+      final WorldStateStorage worldStateStorage,
+      final MetricsSystem metricsSystem,
+      final LongSupplier worldStatePendingRequestsCurrentSupplier) {
     this.worldStateStorage = worldStateStorage;
-
+    this.worldStatePendingRequestsCurrentSupplier = worldStatePendingRequestsCurrentSupplier;
     completedRequestsCounter =
-        metricsSystem.createCounter(
+        new RunnableCounter(
-            PantheonMetricCategory.SYNCHRONIZER,
+            metricsSystem.createCounter(
-            "world_state_completed_requests_total",
+                PantheonMetricCategory.SYNCHRONIZER,
-            "Total number of node data requests completed as part of fast sync world state download");
+                "world_state_completed_requests_total",
+                "Total number of node data requests completed as part of fast sync world state download"),
+            this::displayWorldStateSyncProgress,
+            DISPLAY_PROGRESS_STEP);
     retriedRequestsCounter =
         metricsSystem.createCounter(
             PantheonMetricCategory.SYNCHRONIZER,
@@ -59,6 +75,21 @@ public class CompleteTaskStep {
     }
   }
 
+  private void displayWorldStateSyncProgress() {
+    LOG.info(
+        "Downloaded {} world state nodes. At least {} nodes remaining. Estimated World State completion: {} %.",
+        completedRequestsCounter.get(),
+        worldStatePendingRequestsCurrentSupplier.getAsLong(),
+        doubleFormatter.format(computeWorldStateSyncProgress() * 100.0));
+  }
+
+  public double computeWorldStateSyncProgress() {
+    final double pendingRequests = worldStatePendingRequestsCurrentSupplier.getAsLong();
+    final double completedRequests = completedRequestsCounter.get();
+    estimatedWorldStateCompletion = completedRequests / (completedRequests + pendingRequests);
+    return this.estimatedWorldStateCompletion;
+  }
+
   private void enqueueChildren(
       final Task<NodeDataRequest> task,
       final BlockHeader blockHeader,

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/worldstate/WorldStateDownloader.java

@@ -128,7 +128,8 @@ public class WorldStateDownloader {
               .loadLocalDataStep(new LoadLocalDataStep(worldStateStorage, metricsSystem))
               .requestDataStep(new RequestDataStep(ethContext, metricsSystem))
               .persistDataStep(new PersistDataStep(worldStateStorage))
-              .completeTaskStep(new CompleteTaskStep(worldStateStorage, metricsSystem))
+              .completeTaskStep(
+                  new CompleteTaskStep(worldStateStorage, metricsSystem, taskCollection::size))
               .downloadState(newDownloadState)
               .pivotBlockHeader(header)
               .metricsSystem(metricsSystem)

ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/worldstate/CompleteTaskStepTest.java

@@ -38,7 +38,7 @@ public class CompleteTaskStepTest {
       new BlockHeaderTestFixture().stateRoot(ROOT_HASH).buildHeader();
 
   private final CompleteTaskStep completeTaskStep =
-      new CompleteTaskStep(worldStateStorage, new NoOpMetricsSystem());
+      new CompleteTaskStep(worldStateStorage, new NoOpMetricsSystem(), () -> 3);
 
   @SuppressWarnings("unchecked")
   private final ArgumentCaptor<Stream<NodeDataRequest>> streamCaptor =
@@ -59,14 +59,7 @@ public class CompleteTaskStepTest {
   @Test
   public void shouldEnqueueChildrenAndMarkCompleteWhenTaskHasData() {
     // Use an arbitrary but actually valid trie node to get children from.
-    final Hash hash =
+    final StubTask task = validTask();
-        Hash.fromHexString("0x601a7b0d0267209790cf4c4d9e0cab11b26c537e2ade006412f48b070010e847");
-    final BytesValue data =
-        BytesValue.fromHexString(
-            "0xf85180808080a05ac6993e3fbca0bfbd30173396dd5c2412657fae0bad92e401d17b2aa9a3698f80808080a012f96a0812be538c302416dc6e8df19ce18f1cc7b06a3c7a16831d766c87a9b580808080808080");
-    final StubTask task = new StubTask(NodeDataRequest.createAccountDataRequest(hash));
-    task.getData().setData(data);
-
     completeTaskStep.markAsCompleteOrFailed(blockHeader, downloadState, task);
 
     assertThat(task.isCompleted()).isTrue();
@@ -79,4 +72,22 @@ public class CompleteTaskStepTest {
 
     verify(downloadState).checkCompletion(worldStateStorage, blockHeader);
   }
+
+  @Test
+  public void shouldComputeWorldStateProgress() {
+    completeTaskStep.markAsCompleteOrFailed(blockHeader, downloadState, validTask());
+    // One task has been completed and there are 2 pending requests, progress should be 1/4 (25%)
+    assertThat(completeTaskStep.computeWorldStateSyncProgress()).isEqualTo(1.0 / 4.0);
+  }
+
+  private StubTask validTask() {
+    final Hash hash =
+        Hash.fromHexString("0x601a7b0d0267209790cf4c4d9e0cab11b26c537e2ade006412f48b070010e847");
+    final BytesValue data =
+        BytesValue.fromHexString(
+            "0xf85180808080a05ac6993e3fbca0bfbd30173396dd5c2412657fae0bad92e401d17b2aa9a3698f80808080a012f96a0812be538c302416dc6e8df19ce18f1cc7b06a3c7a16831d766c87a9b580808080808080");
+    final StubTask task = new StubTask(NodeDataRequest.createAccountDataRequest(hash));
+    task.getData().setData(data);
+    return task;
+  }
 }

metrics/core/src/main/java/tech/pegasys/pantheon/metrics/RunnableCounter.java

@@ -12,7 +12,7 @@
  */
 package tech.pegasys.pantheon.metrics;
 
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 /** Counter that triggers a specific task each time a step is hit. */
 public class RunnableCounter implements Counter {
@@ -20,13 +20,13 @@ public class RunnableCounter implements Counter {
   private final Counter backedCounter;
   private final Runnable task;
   private final int step;
-  private AtomicInteger stepCounter;
+  private AtomicLong stepCounter;
 
   public RunnableCounter(final Counter backedCounter, final Runnable task, final int step) {
     this.backedCounter = backedCounter;
     this.task = task;
     this.step = step;
-    this.stepCounter = new AtomicInteger(0);
+    this.stepCounter = new AtomicLong(0);
   }
 
   /**
@@ -46,11 +46,13 @@ public class RunnableCounter implements Counter {
    */
   @Override
   public void inc(final long amount) {
-    stepCounter.addAndGet((int) amount);
     backedCounter.inc(amount);
-    if (stepCounter.get() == step) {
+    if (stepCounter.addAndGet(amount) % step == 0) {
       task.run();
-      stepCounter = new AtomicInteger(0);
     }
   }
+
+  public long get() {
+    return stepCounter.get();
+  }
 }