bug fixes

* offer -> put
  * last stage is unbound so put won't stall the pipeline
* segment size = 200
* log level demotions

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/AbstractPipelinedPeerTask.java

@@ -70,8 +70,12 @@ protected void executeTaskWithPeer(final EthPeer peer) {
  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);
- outboundQueue.offer(o);
  });
  previousInput = Optional.of(input);
  }

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

@@ -275,7 +275,7 @@ public static class Builder {
  private int downloaderHeaderRequestSize = 10;
  private int downloaderCheckpointTimeoutsPermitted = 5;
  private int downloaderChainSegmentTimeoutsPermitted = 5;
- private int downloaderChainSegmentSize = 20;
+ private int downloaderChainSegmentSize = 200;
  private long trailingPeerBlocksBehindThreshold;
  private int maxTrailingPeers = Integer.MAX_VALUE;
  private int downloaderParallelism = 2;

ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncDownloader.java

@@ -93,8 +93,6 @@ private CompletableFuture<List<Block>> importBlocksForCheckpoints(
  ethTasksTimer);
  importedBlocks = importTask.run().thenApply(PeerTaskResult::getResult);
  } else {
- // final PipelinedImportChainSegmentTask<C, Block> importTask =
- // PipelinedImportChainSegmentTask.forCheckpoints(
  final ParallelImportChainSegmentTask<C, Block> importTask =
  ParallelImportChainSegmentTask.forCheckpoints(
  protocolSchedule,
@@ -107,7 +105,6 @@ private CompletableFuture<List<Block>> importBlocksForCheckpoints(
  () -> HeaderValidationMode.DETACHED_ONLY,
  checkpointHeaders);
  importedBlocks = importTask.run().thenApply(PeerTaskResult::getResult);
- // importedBlocks = importTask.run();
  }
  return importedBlocks;
  }

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

@@ -119,7 +119,7 @@ public static <C> DownloadHeaderSequenceTask<C> endingAtHeader(
  protected CompletableFuture<List<BlockHeader>> executePeerTask(
  final Optional<EthPeer> assignedPeer) {
  LOG.debug(
- "Downloading headers from {} to {}.", startingBlockNumber, referenceHeader.getNumber() - 1);
+ "Downloading headers from {} to {}.", startingBlockNumber, referenceHeader.getNumber());
  final CompletableFuture<List<BlockHeader>> task =
  downloadHeaders(assignedPeer).thenCompose(this::processHeaders);
  return task.whenComplete(
@@ -128,8 +128,8 @@ protected CompletableFuture<List<BlockHeader>> executePeerTask(
  if (lastFilledHeaderIndex == 0) {
  LOG.debug(
  "Finished downloading headers from {} to {}.",
- startingBlockNumber,
- referenceHeader.getNumber() - 1);
+ headers[0].getNumber(),
+ headers[segmentLength - 1].getNumber());
  result.get().complete(Arrays.asList(headers));
  }
  });

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

@@ -50,13 +50,13 @@ protected Optional<List<B>> processStep(
  final List<BlockHeader> headers,
  final Optional<List<BlockHeader>> previousHeaders,
  final EthPeer peer) {
- LOG.debug(
+ LOG.trace(
  "Downloading bodies {} to {}",
  headers.get(0).getNumber(),
  headers.get(headers.size() - 1).getNumber());
  try {
  final List<B> blocks = blockHandler.downloadBlocks(headers).get();
- LOG.info(
+ LOG.debug(
  "Downloaded bodies {} to {}",
  headers.get(0).getNumber(),
  headers.get(headers.size() - 1).getNumber());

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

@@ -86,11 +86,10 @@ protected Optional<List<BlockHeader>> processStep(
  }
  headers.add(nextCheckpointHeader);
  if (headers.size() > 2) {
- LOG.trace("Received {} headers", headers.size() - 2);
- LOG.info(
+ LOG.debug(
  "Downloaded headers {} to {}",
  headers.get(1).getNumber(),
- headers.get(headers.size() - 2).getNumber());
+ headers.get(headers.size() - 1).getNumber());
  }
  return Optional.of(headers);
  }

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

@@ -130,7 +130,7 @@ protected void executeTaskWithPeer(final EthPeer peer) {
  new ParallelValidateAndImportBodiesTask<>(
  blockHandler,
  downloadBodiesTask.getOutboundQueue(),
- maxActiveChunks,
+ Integer.MAX_VALUE,
  ethContext,
  ethTasksTimer);
 

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

@@ -74,9 +74,17 @@ protected Optional<List<BlockHeader>> processStep(
  parentHeader,
  protocolContext,
  validationPolicy.getValidationModeForNextBlock())) {
+ LOG.debug(
+ "Validated Headers {} to {}",
+ headers.get(0).getNumber(),
+ headers.get(headers.size() - 1).getNumber());
  // The first header will be imported by the previous request range.
  return Optional.of(headers.subList(1, headers.size()));
  } else {
+ LOG.debug(
+ "Could not validate Headers {} to {}",
+ headers.get(0).getNumber(),
+ headers.get(headers.size() - 1).getNumber());
  // ignore the value, we only want the first exception to be there
  failExceptionally(
  new InvalidBlockException(