[PAN-2444] Add CLI flag for setting WorldStateDownloader task cache s…

…ize (#1749)

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/SynchronizerConfiguration.java

@@ -15,6 +15,7 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import tech.pegasys.pantheon.services.tasks.CachingTaskCollection;
 import tech.pegasys.pantheon.util.uint.UInt256;
 
 import java.util.concurrent.TimeUnit;
@@ -41,6 +42,8 @@ public class SynchronizerConfiguration {
   public static final int DEFAULT_DOWNLOADER_PARALLELISM = 4;
   public static final int DEFAULT_TRANSACTIONS_PARALLELISM = 2;
   public static final int DEFAULT_COMPUTATION_PARALLELISM = 2;
+  public static final int DEFAULT_WORLD_STATE_TASK_CACHE_SIZE =
+      CachingTaskCollection.DEFAULT_CACHE_SIZE;
 
   // Fast sync config
   private final int fastSyncPivotDistance;
@@ -49,6 +52,7 @@ public class SynchronizerConfiguration {
   private final int worldStateHashCountPerRequest;
   private final int worldStateRequestParallelism;
   private final int worldStateMaxRequestsWithoutProgress;
+  private final int worldStateTaskCacheSize;
 
   // Block propagation config
   private final Range<Long> blockPropagationRange;
@@ -76,6 +80,7 @@ private SynchronizerConfiguration(
       final int worldStateRequestParallelism,
       final int worldStateMaxRequestsWithoutProgress,
       final long worldStateMinMillisBeforeStalling,
+      final int worldStateTaskCacheSize,
       final Range<Long> blockPropagationRange,
       final SyncMode syncMode,
       final long downloaderChangeTargetThresholdByHeight,
@@ -94,6 +99,7 @@ private SynchronizerConfiguration(
     this.worldStateRequestParallelism = worldStateRequestParallelism;
     this.worldStateMaxRequestsWithoutProgress = worldStateMaxRequestsWithoutProgress;
     this.worldStateMinMillisBeforeStalling = worldStateMinMillisBeforeStalling;
+    this.worldStateTaskCacheSize = worldStateTaskCacheSize;
     this.blockPropagationRange = blockPropagationRange;
     this.syncMode = syncMode;
     this.downloaderChangeTargetThresholdByHeight = downloaderChangeTargetThresholdByHeight;
@@ -203,6 +209,10 @@ public long getWorldStateMinMillisBeforeStalling() {
     return worldStateMinMillisBeforeStalling;
   }
 
+  public int getWorldStateTaskCacheSize() {
+    return worldStateTaskCacheSize;
+  }
+
   public int getMaxTrailingPeers() {
     return maxTrailingPeers;
   }
@@ -230,6 +240,7 @@ public static class Builder {
     private int worldStateMaxRequestsWithoutProgress =
         DEFAULT_WORLD_STATE_MAX_REQUESTS_WITHOUT_PROGRESS;
     private long worldStateMinMillisBeforeStalling = DEFAULT_WORLD_STATE_MIN_MILLIS_BEFORE_STALLING;
+    private int worldStateTaskCacheSize = DEFAULT_WORLD_STATE_TASK_CACHE_SIZE;
 
     public Builder fastSyncPivotDistance(final int distance) {
       fastSyncPivotDistance = distance;
@@ -327,6 +338,11 @@ public Builder worldStateMinMillisBeforeStalling(final long worldStateMinMillisB
       return this;
     }
 
+    public Builder worldStateTaskCacheSize(final int worldStateTaskCacheSize) {
+      this.worldStateTaskCacheSize = worldStateTaskCacheSize;
+      return this;
+    }
+
     public Builder maxTrailingPeers(final int maxTailingPeers) {
       this.maxTrailingPeers = maxTailingPeers;
       return this;
@@ -341,6 +357,7 @@ public SynchronizerConfiguration build() {
           worldStateRequestParallelism,
           worldStateMaxRequestsWithoutProgress,
           worldStateMinMillisBeforeStalling,
+          worldStateTaskCacheSize,
           blockPropagationRange,
           syncMode,
           downloaderChangeTargetThresholdByHeight,

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastDownloaderFactory.java

@@ -68,7 +68,9 @@ public static <C> Optional<FastSyncDownloader<C>> create(
 
     final CachingTaskCollection<NodeDataRequest> taskCollection =
         createWorldStateDownloaderTaskCollection(
-            getStateQueueDirectory(dataDirectory), metricsSystem);
+            getStateQueueDirectory(dataDirectory),
+            metricsSystem,
+            syncConfig.getWorldStateTaskCacheSize());
     final WorldStateDownloader worldStateDownloader =
         new WorldStateDownloader(
             ethContext,
@@ -116,11 +118,14 @@ private static void ensureDirectoryExists(final File dir) {
   }
 
   private static CachingTaskCollection<NodeDataRequest> createWorldStateDownloaderTaskCollection(
-      final Path dataDirectory, final MetricsSystem metricsSystem) {
+      final Path dataDirectory,
+      final MetricsSystem metricsSystem,
+      final int worldStateTaskCacheSize) {
     final CachingTaskCollection<NodeDataRequest> taskCollection =
         new CachingTaskCollection<>(
             new FlatFileTaskCollection<>(
-                dataDirectory, NodeDataRequest::serialize, NodeDataRequest::deserialize));
+                dataDirectory, NodeDataRequest::serialize, NodeDataRequest::deserialize),
+            worldStateTaskCacheSize);
 
     metricsSystem.createLongGauge(
         PantheonMetricCategory.SYNCHRONIZER,

pantheon/src/main/java/tech/pegasys/pantheon/cli/options/SynchronizerOptions.java

@@ -52,6 +52,8 @@ public class SynchronizerOptions implements CLIOptions<SynchronizerConfiguration
       "--Xsynchronizer-world-state-max-requests-without-progress";
   private static final String WORLD_STATE_MIN_MILLIS_BEFORE_STALLING_FLAG =
       "--Xsynchronizer-world-state-min-millis-before-stalling";
+  private static final String WORLD_STATE_TASK_CACHE_SIZE_FLAG =
+      "--Xsynchronizer-world-state-task-cache-size";
 
   @CommandLine.Option(
       names = BLOCK_PROPAGATION_RANGE_FLAG,
@@ -196,6 +198,16 @@ public void parseBlockPropagationRange(final String arg) {
   private long worldStateMinMillisBeforeStalling =
       SynchronizerConfiguration.DEFAULT_WORLD_STATE_MIN_MILLIS_BEFORE_STALLING;
 
+  @CommandLine.Option(
+      names = WORLD_STATE_TASK_CACHE_SIZE_FLAG,
+      hidden = true,
+      defaultValue = "1000000",
+      paramLabel = "<INTEGER>",
+      description =
+          "The max number of pending node data requests cached in-memory during fast sync world state download. (default: ${DEFAULT-VALUE})")
+  private int worldStateTaskCacheSize =
+      SynchronizerConfiguration.DEFAULT_WORLD_STATE_TASK_CACHE_SIZE;
+
   private SynchronizerOptions() {}
 
   public static SynchronizerOptions create() {
@@ -221,6 +233,7 @@ public static SynchronizerOptions fromConfig(final SynchronizerConfiguration con
     options.worldStateRequestParallelism = config.getWorldStateRequestParallelism();
     options.worldStateMaxRequestsWithoutProgress = config.getWorldStateMaxRequestsWithoutProgress();
     options.worldStateMinMillisBeforeStalling = config.getWorldStateMinMillisBeforeStalling();
+    options.worldStateTaskCacheSize = config.getWorldStateTaskCacheSize();
     return options;
   }
 
@@ -242,6 +255,7 @@ public SynchronizerConfiguration.Builder toDomainObject() {
     builder.worldStateRequestParallelism(worldStateRequestParallelism);
     builder.worldStateMaxRequestsWithoutProgress(worldStateMaxRequestsWithoutProgress);
     builder.worldStateMinMillisBeforeStalling(worldStateMinMillisBeforeStalling);
+    builder.worldStateTaskCacheSize(worldStateTaskCacheSize);
     return builder;
   }
 
@@ -277,6 +291,8 @@ public List<String> getCLIOptions() {
         WORLD_STATE_MAX_REQUESTS_WITHOUT_PROGRESS_FLAG,
         OptionParser.format(worldStateMaxRequestsWithoutProgress),
         WORLD_STATE_MIN_MILLIS_BEFORE_STALLING_FLAG,
-        OptionParser.format(worldStateMinMillisBeforeStalling));
+        OptionParser.format(worldStateMinMillisBeforeStalling),
+        WORLD_STATE_TASK_CACHE_SIZE_FLAG,
+        OptionParser.format(worldStateTaskCacheSize));
   }
 }

pantheon/src/test/java/tech/pegasys/pantheon/cli/options/SynchronizerOptionsTest.java

@@ -41,6 +41,7 @@ SynchronizerConfiguration.Builder createCustomizedDomainObject() {
             SynchronizerConfiguration.DEFAULT_WORLD_STATE_MAX_REQUESTS_WITHOUT_PROGRESS * 2)
         .worldStateMinMillisBeforeStalling(
             SynchronizerConfiguration.DEFAULT_WORLD_STATE_MIN_MILLIS_BEFORE_STALLING * 2)
+        .worldStateTaskCacheSize(SynchronizerConfiguration.DEFAULT_WORLD_STATE_TASK_CACHE_SIZE + 1)
         .blockPropagationRange(
             Range.closed(
                 SynchronizerConfiguration.DEFAULT_BLOCK_PROPAGATION_RANGE.lowerEndpoint() - 2,

services/tasks/src/main/java/tech/pegasys/pantheon/services/tasks/CachingTaskCollection.java

@@ -19,7 +19,8 @@
 import java.util.Set;
 
 public class CachingTaskCollection<T> implements TaskCollection<T> {
-  private static final int DEFAULT_CACHE_SIZE = 1_000_000;
+
+  public static final int DEFAULT_CACHE_SIZE = 1_000_000;
   private final int maxCacheSize;
 
   // The underlying collection