7311: Move GetReceipts to services worker for parallelism

Signed-off-by: Matilda Clerke <matilda.clerke@consensys.net>

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthScheduler.java

@@ -145,7 +145,7 @@ public class EthScheduler {
     servicesExecutor.execute(command);
   }
 
-  public <T> CompletableFuture<Void> scheduleServiceTask(final Runnable task) {
+  public CompletableFuture<Void> scheduleServiceTask(final Runnable task) {
     return CompletableFuture.runAsync(task, servicesExecutor);
   }
 
@@ -156,6 +156,19 @@ public class EthScheduler {
     return serviceFuture;
   }
 
+  public <T> CompletableFuture<T> scheduleServiceTask(final Supplier<CompletableFuture<T>> future) {
+    final CompletableFuture<T> promise = new CompletableFuture<>();
+    final Future<?> workerFuture = servicesExecutor.submit(() -> propagateResult(future, promise));
+    // If returned promise is cancelled, cancel the worker future
+    promise.whenComplete(
+        (r, t) -> {
+          if (t instanceof CancellationException) {
+            workerFuture.cancel(false);
+          }
+        });
+    return promise;
+  }
+
   public CompletableFuture<Void> startPipeline(final Pipeline<?> pipeline) {
     final CompletableFuture<Void> pipelineFuture = pipeline.start(servicesExecutor);
     pendingFutures.add(pipelineFuture);

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/fastsync/DownloadReceiptsStep.java

@@ -37,6 +37,9 @@ import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 public class DownloadReceiptsStep
     implements Function<List<Block>, CompletableFuture<List<BlockWithReceipts>>> {
 
@@ -62,7 +65,7 @@ public class DownloadReceiptsStep
     if (synchronizerConfiguration.isPeerTaskSystemEnabled()) {
       return ethContext
           .getScheduler()
-          .scheduleSyncWorkerTask(
+          .scheduleServiceTask(
               () -> {
                 Map<BlockHeader, List<TransactionReceipt>> getReceipts = new HashMap<>();
                 do {