Parallel downloader should stop on puts if requested. (#927)

* move to an offer() instead of a put() on the downloader so that when
  the task is stopped the put will see that it is done and not wait
  forever.
* remove peer based focus of the task. Let subtasks pick their peers.
Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractPipelinedTask.java

@@ -12,8 +12,6 @@
  */
 package tech.pegasys.pantheon.ethereum.eth.manager.task;
 
-import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
-import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
 import tech.pegasys.pantheon.metrics.LabelledMetric;
 import tech.pegasys.pantheon.metrics.OperationTimer;
 
@@ -28,61 +26,69 @@ import java.util.concurrent.atomic.AtomicReference;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
-public abstract class AbstractPipelinedPeerTask<I, O> extends AbstractPeerTask<List<O>> {
+public abstract class AbstractPipelinedTask<I, O> extends AbstractEthTask<List<O>> {
   private static final Logger LOG = LogManager.getLogger();
 
   static final int TIMEOUT_MS = 1000;
 
-  private BlockingQueue<I> inboundQueue;
-  private BlockingQueue<O> outboundQueue;
-  private List<O> results;
+  private final BlockingQueue<I> inboundQueue;
+  private final BlockingQueue<O> outboundQueue;
+  private final List<O> results;
 
   private boolean shuttingDown = false;
-  private AtomicReference<Throwable> processingException = new AtomicReference<>(null);
+  private final AtomicReference<Throwable> processingException = new AtomicReference<>(null);
 
-  protected AbstractPipelinedPeerTask(
+  protected AbstractPipelinedTask(
       final BlockingQueue<I> inboundQueue,
       final int outboundBacklogSize,
-      final EthContext ethContext,
       final LabelledMetric<OperationTimer> ethTasksTimer) {
-    super(ethContext, ethTasksTimer);
+    super(ethTasksTimer);
     this.inboundQueue = inboundQueue;
     outboundQueue = new LinkedBlockingQueue<>(outboundBacklogSize);
     results = new ArrayList<>();
   }
 
   @Override
-  protected void executeTaskWithPeer(final EthPeer peer) {
+  protected void executeTask() {
     Optional<I> previousInput = Optional.empty();
-    while (!isDone() && processingException.get() == null) {
-      if (shuttingDown && inboundQueue.isEmpty()) {
-        break;
-      }
-      final I input;
-      try {
-        input = inboundQueue.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS);
-        if (input == null) {
-          // timed out waiting for a result
+    try {
+      while (!isDone() && processingException.get() == null) {
+        if (shuttingDown && inboundQueue.isEmpty()) {
+          break;
+        }
+        final I input;
+        try {
+          input = inboundQueue.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+          if (input == null) {
+            // timed out waiting for a result
+            continue;
+          }
+        } catch (final InterruptedException e) {
+          // this is expected
           continue;
         }
-      } catch (final InterruptedException e) {
-        // this is expected
-        continue;
+        final Optional<O> output = processStep(input, previousInput);
+        output.ifPresent(
+            o -> {
+              while (!isDone()) {
+                try {
+                  if (outboundQueue.offer(o, 1, TimeUnit.SECONDS)) {
+                    results.add(o);
+                    break;
+                  }
+                } catch (final InterruptedException e) {
+                  processingException.compareAndSet(null, e);
+                  break;
+                }
+              }
+            });
+        previousInput = Optional.of(input);
       }
-      final Optional<O> output = processStep(input, previousInput, peer);
-      output.ifPresent(
-          o -> {
-            try {
-              outboundQueue.put(o);
-            } catch (final InterruptedException e) {
-              processingException.compareAndSet(null, e);
-            }
-            results.add(o);
-          });
-      previousInput = Optional.of(input);
+    } catch (final RuntimeException e) {
+      processingException.compareAndSet(null, e);
     }
     if (processingException.get() == null) {
-      result.get().complete(new PeerTaskResult<>(peer, results));
+      result.get().complete(results);
     } else {
       result.get().completeExceptionally(processingException.get());
     }
@@ -105,5 +111,5 @@ public abstract class AbstractPipelinedPeerTask<I, O> extends AbstractPeerTask<L
     cancel();
   }
 
-  protected abstract Optional<O> processStep(I input, Optional<I> previousInput, EthPeer peer);
+  protected abstract Optional<O> processStep(I input, Optional<I> previousInput);
 }

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelDownloadBodiesTask.java

@@ -13,9 +13,7 @@
 package tech.pegasys.pantheon.ethereum.eth.sync.tasks;
 
 import tech.pegasys.pantheon.ethereum.core.BlockHeader;
-import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
-import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
-import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPipelinedPeerTask;
+import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPipelinedTask;
 import tech.pegasys.pantheon.ethereum.eth.sync.BlockHandler;
 import tech.pegasys.pantheon.metrics.LabelledMetric;
 import tech.pegasys.pantheon.metrics.OperationTimer;
@@ -29,7 +27,7 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 public class ParallelDownloadBodiesTask<B>
-    extends AbstractPipelinedPeerTask<List<BlockHeader>, List<B>> {
+    extends AbstractPipelinedTask<List<BlockHeader>, List<B>> {
   private static final Logger LOG = LogManager.getLogger();
 
   private final BlockHandler<B> blockHandler;
@@ -38,18 +36,15 @@ public class ParallelDownloadBodiesTask<B>
       final BlockHandler<B> blockHandler,
       final BlockingQueue<List<BlockHeader>> inboundQueue,
       final int outboundBacklogSize,
-      final EthContext ethContext,
       final LabelledMetric<OperationTimer> ethTasksTimer) {
-    super(inboundQueue, outboundBacklogSize, ethContext, ethTasksTimer);
+    super(inboundQueue, outboundBacklogSize, ethTasksTimer);
 
     this.blockHandler = blockHandler;
   }
 
   @Override
   protected Optional<List<B>> processStep(
-      final List<BlockHeader> headers,
-      final Optional<List<BlockHeader>> previousHeaders,
-      final EthPeer peer) {
+      final List<BlockHeader> headers, final Optional<List<BlockHeader>> previousHeaders) {
     LOG.trace(
         "Downloading bodies {} to {}",
         headers.get(0).getNumber(),

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelDownloadHeadersTask.java

@@ -15,8 +15,7 @@ package tech.pegasys.pantheon.ethereum.eth.sync.tasks;
 import tech.pegasys.pantheon.ethereum.ProtocolContext;
 import tech.pegasys.pantheon.ethereum.core.BlockHeader;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
-import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
-import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPipelinedPeerTask;
+import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPipelinedTask;
 import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
 import tech.pegasys.pantheon.metrics.LabelledMetric;
 import tech.pegasys.pantheon.metrics.OperationTimer;
@@ -32,11 +31,12 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 public class ParallelDownloadHeadersTask<C>
-    extends AbstractPipelinedPeerTask<BlockHeader, List<BlockHeader>> {
+    extends AbstractPipelinedTask<BlockHeader, List<BlockHeader>> {
   private static final Logger LOG = LogManager.getLogger();
 
   private final ProtocolSchedule<C> protocolSchedule;
   private final ProtocolContext<C> protocolContext;
+  private final EthContext ethContext;
 
   ParallelDownloadHeadersTask(
       final BlockingQueue<BlockHeader> inboundQueue,
@@ -45,17 +45,17 @@ public class ParallelDownloadHeadersTask<C>
       final ProtocolContext<C> protocolContext,
       final EthContext ethContext,
       final LabelledMetric<OperationTimer> ethTasksTimer) {
-    super(inboundQueue, outboundBacklogSize, ethContext, ethTasksTimer);
+    super(inboundQueue, outboundBacklogSize, ethTasksTimer);
 
     this.protocolSchedule = protocolSchedule;
     this.protocolContext = protocolContext;
+    this.ethContext = ethContext;
   }
 
   @Override
   protected Optional<List<BlockHeader>> processStep(
       final BlockHeader nextCheckpointHeader,
-      final Optional<BlockHeader> previousCheckpointHeader,
-      final EthPeer peer) {
+      final Optional<BlockHeader> previousCheckpointHeader) {
     if (!previousCheckpointHeader.isPresent()) {
       return Optional.empty();
     }
@@ -73,7 +73,6 @@ public class ParallelDownloadHeadersTask<C>
             nextCheckpointHeader,
             segmentLength,
             ethTasksTimer);
-    downloadTask.assignPeer(peer);
     final CompletableFuture<List<BlockHeader>> headerFuture = executeSubTask(downloadTask::run);
 
     final List<BlockHeader> headers = Lists.newArrayList(previousCheckpointHeader.get());

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelImportChainSegmentTask.java

@@ -17,7 +17,6 @@ import tech.pegasys.pantheon.ethereum.core.BlockHeader;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
 import tech.pegasys.pantheon.ethereum.eth.manager.EthScheduler;
 import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractEthTask;
-import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPeerTask;
 import tech.pegasys.pantheon.ethereum.eth.sync.BlockHandler;
 import tech.pegasys.pantheon.ethereum.eth.sync.ValidationPolicy;
 import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
@@ -120,21 +119,15 @@ public class ParallelImportChainSegmentTask<C, B> extends AbstractEthTask<List<B
               maxActiveChunks,
               protocolSchedule,
               protocolContext,
-              ethContext,
               ethTasksTimer);
       final ParallelDownloadBodiesTask<B> downloadBodiesTask =
           new ParallelDownloadBodiesTask<>(
-              blockHandler,
-              validateHeadersTask.getOutboundQueue(),
-              maxActiveChunks,
-              ethContext,
-              ethTasksTimer);
+              blockHandler, validateHeadersTask.getOutboundQueue(), maxActiveChunks, ethTasksTimer);
       final ParallelValidateAndImportBodiesTask<B> validateAndImportBodiesTask =
           new ParallelValidateAndImportBodiesTask<>(
               blockHandler,
               downloadBodiesTask.getOutboundQueue(),
               Integer.MAX_VALUE,
-              ethContext,
               ethTasksTimer);
 
       // Start the pipeline.
@@ -148,15 +141,15 @@ public class ParallelImportChainSegmentTask<C, B> extends AbstractEthTask<List<B
       final CompletableFuture<?> downloadBodiesFuture =
           scheduler.scheduleServiceTask(downloadBodiesTask);
       registerSubTask(downloadBodiesFuture);
-      final CompletableFuture<AbstractPeerTask.PeerTaskResult<List<List<B>>>> validateBodiesFuture =
+      final CompletableFuture<List<List<B>>> validateBodiesFuture =
           scheduler.scheduleServiceTask(validateAndImportBodiesTask);
       registerSubTask(validateBodiesFuture);
 
       // Hook in pipeline completion signaling.
       downloadHeadersTask.shutdown();
-      downloadHeaderFuture.thenRun(() -> validateHeadersTask.shutdown());
-      validateHeaderFuture.thenRun(() -> downloadBodiesTask.shutdown());
-      downloadBodiesFuture.thenRun(() -> validateAndImportBodiesTask.shutdown());
+      downloadHeaderFuture.thenRun(validateHeadersTask::shutdown);
+      validateHeaderFuture.thenRun(downloadBodiesTask::shutdown);
+      downloadBodiesFuture.thenRun(validateAndImportBodiesTask::shutdown);
 
       final BiConsumer<? super Object, ? super Throwable> cancelOnException =
           (s, e) -> {
@@ -179,7 +172,7 @@ public class ParallelImportChainSegmentTask<C, B> extends AbstractEthTask<List<B
             } else if (r != null) {
               try {
                 final List<B> importedBlocks =
-                    validateBodiesFuture.get().getResult().stream()
+                    validateBodiesFuture.get().stream()
                         .flatMap(Collection::stream)
                         .collect(Collectors.toList());
                 result.get().complete(importedBlocks);

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelValidateAndImportBodiesTask.java

@@ -12,9 +12,7 @@
  */
 package tech.pegasys.pantheon.ethereum.eth.sync.tasks;
 
-import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
-import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
-import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPipelinedPeerTask;
+import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPipelinedTask;
 import tech.pegasys.pantheon.ethereum.eth.sync.BlockHandler;
 import tech.pegasys.pantheon.metrics.LabelledMetric;
 import tech.pegasys.pantheon.metrics.OperationTimer;
@@ -29,7 +27,7 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 public class ParallelValidateAndImportBodiesTask<B>
-    extends AbstractPipelinedPeerTask<List<B>, List<B>> {
+    extends AbstractPipelinedTask<List<B>, List<B>> {
   private static final Logger LOG = LogManager.getLogger();
 
   private final BlockHandler<B> blockHandler;
@@ -38,16 +36,15 @@ public class ParallelValidateAndImportBodiesTask<B>
       final BlockHandler<B> blockHandler,
       final BlockingQueue<List<B>> inboundQueue,
       final int outboundBacklogSize,
-      final EthContext ethContext,
       final LabelledMetric<OperationTimer> ethTasksTimer) {
-    super(inboundQueue, outboundBacklogSize, ethContext, ethTasksTimer);
+    super(inboundQueue, outboundBacklogSize, ethTasksTimer);
 
     this.blockHandler = blockHandler;
   }
 
   @Override
   protected Optional<List<B>> processStep(
-      final List<B> blocks, final Optional<List<B>> previousBlocks, final EthPeer peer) {
+      final List<B> blocks, final Optional<List<B>> previousBlocks) {
     final long firstBlock = blockHandler.extractBlockNumber(blocks.get(0));
     final long lastBlock = blockHandler.extractBlockNumber(blocks.get(blocks.size() - 1));
     LOG.debug("Starting import of chain segment {} to {}", firstBlock, lastBlock);

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelValidateHeadersTask.java

@@ -14,9 +14,7 @@ package tech.pegasys.pantheon.ethereum.eth.sync.tasks;
 
 import tech.pegasys.pantheon.ethereum.ProtocolContext;
 import tech.pegasys.pantheon.ethereum.core.BlockHeader;
-import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
-import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
-import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPipelinedPeerTask;
+import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPipelinedTask;
 import tech.pegasys.pantheon.ethereum.eth.sync.ValidationPolicy;
 import tech.pegasys.pantheon.ethereum.eth.sync.tasks.exceptions.InvalidBlockException;
 import tech.pegasys.pantheon.ethereum.mainnet.BlockHeaderValidator;
@@ -33,7 +31,7 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 public class ParallelValidateHeadersTask<C>
-    extends AbstractPipelinedPeerTask<List<BlockHeader>, List<BlockHeader>> {
+    extends AbstractPipelinedTask<List<BlockHeader>, List<BlockHeader>> {
   private static final Logger LOG = LogManager.getLogger();
 
   private final ProtocolSchedule<C> protocolSchedule;
@@ -46,9 +44,8 @@ public class ParallelValidateHeadersTask<C>
       final int outboundBacklogSize,
       final ProtocolSchedule<C> protocolSchedule,
       final ProtocolContext<C> protocolContext,
-      final EthContext ethContext,
       final LabelledMetric<OperationTimer> ethTasksTimer) {
-    super(inboundQueue, outboundBacklogSize, ethContext, ethTasksTimer);
+    super(inboundQueue, outboundBacklogSize, ethTasksTimer);
 
     this.protocolSchedule = protocolSchedule;
     this.protocolContext = protocolContext;
@@ -57,9 +54,7 @@ public class ParallelValidateHeadersTask<C>
 
   @Override
   protected Optional<List<BlockHeader>> processStep(
-      final List<BlockHeader> headers,
-      final Optional<List<BlockHeader>> previousHeaders,
-      final EthPeer peer) {
+      final List<BlockHeader> headers, final Optional<List<BlockHeader>> previousHeaders) {
     LOG.debug(
         "Validating Headers {} to {}",
         headers.get(0).getNumber(),