[MINOR] Move metric instantiation (#896)

* Instantiate individual metrics where they're used

* Cache prometheus metrics to allow "duplicate" creation of metrics

Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>
pull/2/head
mbaxter 6 years ago committed by Adrian Sutton
parent 0bbc2abab8
commit d436d5f0bd
  1. 18
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractEthTask.java
  2. 7
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractGetHeadersFromPeerTask.java
  3. 9
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractPeerRequestTask.java
  4. 8
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractPeerTask.java
  5. 7
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractPipelinedTask.java
  6. 15
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractRetryingPeerTask.java
  7. 17
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetBlockFromPeerTask.java
  8. 11
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetBodiesFromPeerTask.java
  9. 27
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetHeadersFromPeerByHashTask.java
  10. 23
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetHeadersFromPeerByNumberTask.java
  11. 11
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetNodeDataFromPeerTask.java
  12. 11
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetReceiptsFromPeerTask.java
  13. 12
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/WaitForPeerTask.java
  14. 15
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/manager/task/WaitForPeersTask.java
  15. 15
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/peervalidation/DaoForkPeerValidator.java
  16. 15
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/BlockPropagationManager.java
  17. 13
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/ChainDownloader.java
  18. 15
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/ChainHeadTracker.java
  19. 11
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/CheckpointHeaderManager.java
  20. 35
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/DefaultSynchronizer.java
  21. 12
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/FastSynchronizer.java
  22. 13
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/SyncTargetManager.java
  23. 22
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncActions.java
  24. 13
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncBlockHandler.java
  25. 32
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncChainDownloader.java
  26. 7
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncCheckpointHeaderManager.java
  27. 13
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncTargetManager.java
  28. 11
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/PivotBlockRetriever.java
  29. 13
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncBlockHandler.java
  30. 25
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncDownloader.java
  31. 7
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncTargetManager.java
  32. 19
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/CompleteBlocksTask.java
  33. 15
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/DetermineCommonAncestorTask.java
  34. 19
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/DownloadHeaderSequenceTask.java
  35. 19
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/GetReceiptsForHeadersTask.java
  36. 19
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ImportBlocksTask.java
  37. 7
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelDownloadBodiesTask.java
  38. 11
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelDownloadHeadersTask.java
  39. 7
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelExtractTxSignaturesTask.java
  40. 29
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelImportChainSegmentTask.java
  41. 7
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelValidateAndImportBodiesTask.java
  42. 7
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ParallelValidateHeadersTask.java
  43. 27
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/PersistBlockTask.java
  44. 19
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/PipelinedImportChainSegmentTask.java
  45. 15
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/RetryingGetHeaderFromPeerByNumberTask.java
  46. 11
      ethereum/eth/src/main/java/tech/pegasys/pantheon/ethereum/eth/sync/worldstate/WorldStateDownloader.java
  47. 2
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/MockEthTask.java
  48. 6
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/ethtaskutils/AbstractMessageTaskTest.java
  49. 2
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/AbstractEthTaskTest.java
  50. 2
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetBlockFromPeerTaskTest.java
  51. 3
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetBodiesFromPeerTaskTest.java
  52. 4
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetHeadersFromPeerByHashTaskTest.java
  53. 15
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetHeadersFromPeerByNumberTaskTest.java
  54. 2
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetNodeDataFromPeerTaskTest.java
  55. 2
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/GetReceiptsFromPeerTaskTest.java
  56. 12
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/WaitForPeerTaskTest.java
  57. 14
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/manager/task/WaitForPeersTaskTest.java
  58. 10
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/peervalidation/DaoForkPeerValidatorTest.java
  59. 12
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/BlockPropagationManagerTest.java
  60. 2
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/ChainHeadTrackerTest.java
  61. 9
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/CheckpointHeaderManagerTest.java
  62. 9
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncActionsTest.java
  63. 9
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncBlockHandlerTest.java
  64. 3
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncChainDownloaderTest.java
  65. 9
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/FastSyncCheckpointHeaderManagerTest.java
  66. 9
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fastsync/PivotBlockRetrieverTest.java
  67. 9
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncDownloaderTest.java
  68. 2
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/fullsync/FullSyncTargetManagerTest.java
  69. 6
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/CompleteBlocksTaskTest.java
  70. 8
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/DetermineCommonAncestorTaskParameterizedTest.java
  71. 18
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/DetermineCommonAncestorTaskTest.java
  72. 6
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/DownloadHeaderSequenceTaskTest.java
  73. 3
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/GetReceiptsForHeadersTaskTest.java
  74. 3
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/ImportBlocksTaskTest.java
  75. 28
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/PersistBlockTaskTest.java
  76. 16
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/PipelinedImportChainSegmentTaskTest.java
  77. 2
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/tasks/RetryingGetHeaderFromPeerByNumberTaskTest.java
  78. 1
      ethereum/eth/src/test/java/tech/pegasys/pantheon/ethereum/eth/sync/worldstate/WorldStateDownloaderTest.java
  79. 56
      metrics/src/main/java/tech/pegasys/pantheon/metrics/prometheus/PrometheusMetricsSystem.java
  80. 35
      metrics/src/test/java/tech/pegasys/pantheon/metrics/prometheus/PrometheusMetricsSystemTest.java
  81. 11
      pantheon/src/main/java/tech/pegasys/pantheon/controller/MainnetPantheonController.java

@ -16,6 +16,8 @@ import static tech.pegasys.pantheon.util.FutureUtils.completedExceptionally;
import tech.pegasys.pantheon.ethereum.eth.manager.EthScheduler;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.MetricCategory;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.OperationTimer;
import java.util.Collection;
@ -29,16 +31,16 @@ import java.util.function.Supplier;
public abstract class AbstractEthTask<T> implements EthTask<T> {
protected double taskTimeInSec = -1.0D;
protected final LabelledMetric<OperationTimer> ethTasksTimer;
protected final OperationTimer taskTimer;
private double taskTimeInSec = -1.0D;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final OperationTimer taskTimer;
protected final AtomicReference<CompletableFuture<T>> result = new AtomicReference<>();
protected volatile Collection<CompletableFuture<?>> subTaskFutures =
new ConcurrentLinkedDeque<>();
private volatile Collection<CompletableFuture<?>> subTaskFutures = new ConcurrentLinkedDeque<>();
/** @param ethTasksTimer The metrics timer to use to time the duration of the task. */
protected AbstractEthTask(final LabelledMetric<OperationTimer> ethTasksTimer) {
this.ethTasksTimer = ethTasksTimer;
protected AbstractEthTask(final MetricsSystem metricsSystem) {
ethTasksTimer =
metricsSystem.createLabelledTimer(
MetricCategory.SYNCHRONIZER, "task", "Internal processing tasks", "taskName");
taskTimer = ethTasksTimer.labels(getClass().getSimpleName());
}

@ -21,8 +21,7 @@ import tech.pegasys.pantheon.ethereum.eth.messages.BlockHeadersMessage;
import tech.pegasys.pantheon.ethereum.eth.messages.EthPV62;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.ArrayList;
import java.util.Collections;
@ -51,8 +50,8 @@ public abstract class AbstractGetHeadersFromPeerTask
final int count,
final int skip,
final boolean reverse,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, EthPV62.GET_BLOCK_HEADERS, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(ethContext, EthPV62.GET_BLOCK_HEADERS, metricsSystem);
checkArgument(count > 0);
this.protocolSchedule = protocolSchedule;
this.count = count;

@ -20,8 +20,7 @@ import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.ethereum.rlp.RLPException;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.ExceptionUtils;
import java.time.Duration;
@ -37,10 +36,8 @@ public abstract class AbstractPeerRequestTask<R> extends AbstractPeerTask<R> {
private volatile ResponseStream responseStream;
protected AbstractPeerRequestTask(
final EthContext ethContext,
final int requestCode,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, ethTasksTimer);
final EthContext ethContext, final int requestCode, final MetricsSystem metricsSystem) {
super(ethContext, metricsSystem);
this.requestCode = requestCode;
}

@ -18,8 +18,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.exceptions.NoAvailablePeersExc
import tech.pegasys.pantheon.ethereum.eth.manager.exceptions.PeerDisconnectedException;
import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPeerTask.PeerTaskResult;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.Optional;
@ -27,9 +26,8 @@ public abstract class AbstractPeerTask<R> extends AbstractEthTask<PeerTaskResult
protected Optional<EthPeer> assignedPeer = Optional.empty();
protected final EthContext ethContext;
protected AbstractPeerTask(
final EthContext ethContext, final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethTasksTimer);
protected AbstractPeerTask(final EthContext ethContext, final MetricsSystem metricsSystem) {
super(metricsSystem);
this.ethContext = ethContext;
}

@ -12,8 +12,7 @@
*/
package tech.pegasys.pantheon.ethereum.eth.manager.task;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.ArrayList;
import java.util.List;
@ -41,8 +40,8 @@ public abstract class AbstractPipelinedTask<I, O> extends AbstractEthTask<List<O
protected AbstractPipelinedTask(
final BlockingQueue<I> inboundQueue,
final int outboundBacklogSize,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethTasksTimer);
final MetricsSystem metricsSystem) {
super(metricsSystem);
this.inboundQueue = inboundQueue;
outboundQueue = new LinkedBlockingQueue<>(outboundBacklogSize);
results = new ArrayList<>();

@ -18,8 +18,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.exceptions.MaxRetriesReachedEx
import tech.pegasys.pantheon.ethereum.eth.manager.exceptions.NoAvailablePeersException;
import tech.pegasys.pantheon.ethereum.eth.manager.exceptions.PeerBreachedProtocolException;
import tech.pegasys.pantheon.ethereum.eth.manager.exceptions.PeerDisconnectedException;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.ExceptionUtils;
import java.time.Duration;
@ -44,24 +43,26 @@ public abstract class AbstractRetryingPeerTask<T> extends AbstractEthTask<T> {
private final EthContext ethContext;
private final int maxRetries;
private final Predicate<T> isEmptyResponse;
private final MetricsSystem metricsSystem;
private int retryCount = 0;
private Optional<EthPeer> assignedPeer = Optional.empty();
/**
* @param ethContext The context of the current Eth network we are attached to.
* @param maxRetries Maximum number of retries to accept before completing exceptionally.
* @param ethTasksTimer The metrics timer to use to time the duration of the task.
* @param isEmptyResponse Test if the response received was empty.
* @param metricsSystem The metrics system used to measure task.
*/
public AbstractRetryingPeerTask(
final EthContext ethContext,
final int maxRetries,
final LabelledMetric<OperationTimer> ethTasksTimer,
final Predicate<T> isEmptyResponse) {
super(ethTasksTimer);
final Predicate<T> isEmptyResponse,
final MetricsSystem metricsSystem) {
super(metricsSystem);
this.ethContext = ethContext;
this.maxRetries = maxRetries;
this.isEmptyResponse = isEmptyResponse;
this.metricsSystem = metricsSystem;
}
public void assignPeer(final EthPeer peer) {
@ -108,7 +109,7 @@ public abstract class AbstractRetryingPeerTask<T> extends AbstractEthTask<T> {
if (cause instanceof NoAvailablePeersException) {
LOG.info("No peers available, wait for peer.");
// Wait for new peer to connect
final WaitForPeerTask waitTask = WaitForPeerTask.create(ethContext, ethTasksTimer);
final WaitForPeerTask waitTask = WaitForPeerTask.create(ethContext, metricsSystem);
executeSubTask(
() ->
ethContext

@ -22,8 +22,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
import tech.pegasys.pantheon.ethereum.eth.manager.exceptions.IncompleteResultsException;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@ -37,13 +36,15 @@ public class GetBlockFromPeerTask extends AbstractPeerTask<Block> {
private final ProtocolSchedule<?> protocolSchedule;
private final Hash hash;
private final MetricsSystem metricsSystem;
protected GetBlockFromPeerTask(
final ProtocolSchedule<?> protocolSchedule,
final EthContext ethContext,
final Hash hash,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(ethContext, metricsSystem);
this.metricsSystem = metricsSystem;
this.protocolSchedule = protocolSchedule;
this.hash = hash;
}
@ -52,8 +53,8 @@ public class GetBlockFromPeerTask extends AbstractPeerTask<Block> {
final ProtocolSchedule<?> protocolSchedule,
final EthContext ethContext,
final Hash hash,
final LabelledMetric<OperationTimer> ethTasksTimer) {
return new GetBlockFromPeerTask(protocolSchedule, ethContext, hash, ethTasksTimer);
final MetricsSystem metricsSystem) {
return new GetBlockFromPeerTask(protocolSchedule, ethContext, hash, metricsSystem);
}
@Override
@ -80,7 +81,7 @@ public class GetBlockFromPeerTask extends AbstractPeerTask<Block> {
return executeSubTask(
() ->
GetHeadersFromPeerByHashTask.forSingleHash(
protocolSchedule, ethContext, hash, ethTasksTimer)
protocolSchedule, ethContext, hash, metricsSystem)
.assignPeer(peer)
.run());
}
@ -95,7 +96,7 @@ public class GetBlockFromPeerTask extends AbstractPeerTask<Block> {
() -> {
final GetBodiesFromPeerTask<?> task =
GetBodiesFromPeerTask.forHeaders(
protocolSchedule, ethContext, headerResult.getResult(), ethTasksTimer);
protocolSchedule, ethContext, headerResult.getResult(), metricsSystem);
task.assignPeer(headerResult.getPeer());
return task.run();
});

@ -28,8 +28,7 @@ import tech.pegasys.pantheon.ethereum.mainnet.BodyValidation;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.bytes.Bytes32;
import java.util.ArrayList;
@ -60,8 +59,8 @@ public class GetBodiesFromPeerTask<C> extends AbstractPeerRequestTask<List<Block
final ProtocolSchedule<C> protocolSchedule,
final EthContext ethContext,
final List<BlockHeader> headers,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, EthPV62.GET_BLOCK_BODIES, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(ethContext, EthPV62.GET_BLOCK_BODIES, metricsSystem);
checkArgument(headers.size() > 0);
this.protocolSchedule = protocolSchedule;
@ -78,8 +77,8 @@ public class GetBodiesFromPeerTask<C> extends AbstractPeerRequestTask<List<Block
final ProtocolSchedule<C> protocolSchedule,
final EthContext ethContext,
final List<BlockHeader> headers,
final LabelledMetric<OperationTimer> ethTasksTimer) {
return new GetBodiesFromPeerTask<>(protocolSchedule, ethContext, headers, ethTasksTimer);
final MetricsSystem metricsSystem) {
return new GetBodiesFromPeerTask<>(protocolSchedule, ethContext, headers, metricsSystem);
}
@Override

@ -21,8 +21,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
import tech.pegasys.pantheon.ethereum.eth.manager.RequestManager.ResponseStream;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import com.google.common.annotations.VisibleForTesting;
import org.apache.logging.log4j.LogManager;
@ -43,7 +42,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
final int count,
final int skip,
final boolean reverse,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
super(
protocolSchedule,
ethContext,
@ -51,7 +50,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
count,
skip,
reverse,
ethTasksTimer);
metricsSystem);
checkNotNull(referenceHash);
this.referenceHash = referenceHash;
}
@ -62,7 +61,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
final Hash firstHash,
final long firstBlockNumber,
final int segmentLength,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByHashTask(
protocolSchedule,
ethContext,
@ -71,7 +70,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
segmentLength,
0,
false,
ethTasksTimer);
metricsSystem);
}
public static AbstractGetHeadersFromPeerTask startingAtHash(
@ -81,7 +80,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
final long firstBlockNumber,
final int segmentLength,
final int skip,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByHashTask(
protocolSchedule,
ethContext,
@ -90,7 +89,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
segmentLength,
skip,
false,
ethTasksTimer);
metricsSystem);
}
public static AbstractGetHeadersFromPeerTask endingAtHash(
@ -99,7 +98,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
final Hash lastHash,
final long lastBlockNumber,
final int segmentLength,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByHashTask(
protocolSchedule,
ethContext,
@ -108,7 +107,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
segmentLength,
0,
true,
ethTasksTimer);
metricsSystem);
}
public static AbstractGetHeadersFromPeerTask endingAtHash(
@ -118,7 +117,7 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
final long lastBlockNumber,
final int segmentLength,
final int skip,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByHashTask(
protocolSchedule,
ethContext,
@ -127,16 +126,16 @@ public class GetHeadersFromPeerByHashTask extends AbstractGetHeadersFromPeerTask
segmentLength,
skip,
true,
ethTasksTimer);
metricsSystem);
}
public static AbstractGetHeadersFromPeerTask forSingleHash(
final ProtocolSchedule<?> protocolSchedule,
final EthContext ethContext,
final Hash hash,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByHashTask(
protocolSchedule, ethContext, hash, 0, 1, 0, false, ethTasksTimer);
protocolSchedule, ethContext, hash, 0, 1, 0, false, metricsSystem);
}
@Override

@ -18,8 +18,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
import tech.pegasys.pantheon.ethereum.eth.manager.RequestManager.ResponseStream;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import com.google.common.annotations.VisibleForTesting;
import org.apache.logging.log4j.LogManager;
@ -39,8 +38,8 @@ public class GetHeadersFromPeerByNumberTask extends AbstractGetHeadersFromPeerTa
final int count,
final int skip,
final boolean reverse,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(protocolSchedule, ethContext, blockNumber, count, skip, reverse, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(protocolSchedule, ethContext, blockNumber, count, skip, reverse, metricsSystem);
this.blockNumber = blockNumber;
}
@ -49,9 +48,9 @@ public class GetHeadersFromPeerByNumberTask extends AbstractGetHeadersFromPeerTa
final EthContext ethContext,
final long firstBlockNumber,
final int segmentLength,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByNumberTask(
protocolSchedule, ethContext, firstBlockNumber, segmentLength, 0, false, ethTasksTimer);
protocolSchedule, ethContext, firstBlockNumber, segmentLength, 0, false, metricsSystem);
}
public static AbstractGetHeadersFromPeerTask endingAtNumber(
@ -59,9 +58,9 @@ public class GetHeadersFromPeerByNumberTask extends AbstractGetHeadersFromPeerTa
final EthContext ethContext,
final long lastlockNumber,
final int segmentLength,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByNumberTask(
protocolSchedule, ethContext, lastlockNumber, segmentLength, 0, true, ethTasksTimer);
protocolSchedule, ethContext, lastlockNumber, segmentLength, 0, true, metricsSystem);
}
public static AbstractGetHeadersFromPeerTask endingAtNumber(
@ -70,18 +69,18 @@ public class GetHeadersFromPeerByNumberTask extends AbstractGetHeadersFromPeerTa
final long lastlockNumber,
final int segmentLength,
final int skip,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByNumberTask(
protocolSchedule, ethContext, lastlockNumber, segmentLength, skip, true, ethTasksTimer);
protocolSchedule, ethContext, lastlockNumber, segmentLength, skip, true, metricsSystem);
}
public static AbstractGetHeadersFromPeerTask forSingleNumber(
final ProtocolSchedule<?> protocolSchedule,
final EthContext ethContext,
final long blockNumber,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetHeadersFromPeerByNumberTask(
protocolSchedule, ethContext, blockNumber, 1, 0, false, ethTasksTimer);
protocolSchedule, ethContext, blockNumber, 1, 0, false, metricsSystem);
}
@Override

@ -22,8 +22,7 @@ import tech.pegasys.pantheon.ethereum.eth.messages.EthPV63;
import tech.pegasys.pantheon.ethereum.eth.messages.NodeDataMessage;
import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.bytes.BytesValue;
import java.util.Collection;
@ -46,16 +45,16 @@ public class GetNodeDataFromPeerTask extends AbstractPeerRequestTask<Map<Hash, B
private GetNodeDataFromPeerTask(
final EthContext ethContext,
final Collection<Hash> hashes,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, EthPV63.GET_NODE_DATA, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(ethContext, EthPV63.GET_NODE_DATA, metricsSystem);
this.hashes = new HashSet<>(hashes);
}
public static GetNodeDataFromPeerTask forHashes(
final EthContext ethContext,
final Collection<Hash> hashes,
final LabelledMetric<OperationTimer> ethTasksTimer) {
return new GetNodeDataFromPeerTask(ethContext, hashes, ethTasksTimer);
final MetricsSystem metricsSystem) {
return new GetNodeDataFromPeerTask(ethContext, hashes, metricsSystem);
}
@Override

@ -26,8 +26,7 @@ import tech.pegasys.pantheon.ethereum.eth.messages.EthPV63;
import tech.pegasys.pantheon.ethereum.eth.messages.ReceiptsMessage;
import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.ArrayList;
import java.util.Collection;
@ -49,8 +48,8 @@ public class GetReceiptsFromPeerTask
private GetReceiptsFromPeerTask(
final EthContext ethContext,
final Collection<BlockHeader> blockHeaders,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, EthPV63.GET_RECEIPTS, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(ethContext, EthPV63.GET_RECEIPTS, metricsSystem);
this.blockHeaders = blockHeaders;
blockHeaders.forEach(
header ->
@ -62,9 +61,9 @@ public class GetReceiptsFromPeerTask
public static GetReceiptsFromPeerTask forHeaders(
final EthContext ethContext,
final Collection<BlockHeader> blockHeaders,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new GetReceiptsFromPeerTask(ethContext, blockHeaders, ethTasksTimer);
return new GetReceiptsFromPeerTask(ethContext, blockHeaders, metricsSystem);
}
@Override

@ -14,8 +14,7 @@ package tech.pegasys.pantheon.ethereum.eth.manager.task;
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
import tech.pegasys.pantheon.ethereum.eth.manager.EthPeers;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -27,15 +26,14 @@ public class WaitForPeerTask extends AbstractEthTask<Void> {
private final EthContext ethContext;
private volatile Long peerListenerId;
private WaitForPeerTask(
final EthContext ethContext, final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethTasksTimer);
private WaitForPeerTask(final EthContext ethContext, final MetricsSystem metricsSystem) {
super(metricsSystem);
this.ethContext = ethContext;
}
public static WaitForPeerTask create(
final EthContext ethContext, final LabelledMetric<OperationTimer> ethTasksTimer) {
return new WaitForPeerTask(ethContext, ethTasksTimer);
final EthContext ethContext, final MetricsSystem metricsSystem) {
return new WaitForPeerTask(ethContext, metricsSystem);
}
@Override

@ -14,8 +14,7 @@ package tech.pegasys.pantheon.ethereum.eth.manager.task;
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
import tech.pegasys.pantheon.ethereum.eth.manager.EthPeers;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -29,19 +28,15 @@ public class WaitForPeersTask extends AbstractEthTask<Void> {
private volatile Long peerListenerId;
private WaitForPeersTask(
final EthContext ethContext,
final int targetPeerCount,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethTasksTimer);
final EthContext ethContext, final int targetPeerCount, final MetricsSystem metricsSystem) {
super(metricsSystem);
this.targetPeerCount = targetPeerCount;
this.ethContext = ethContext;
}
public static WaitForPeersTask create(
final EthContext ethContext,
final int targetPeerCount,
final LabelledMetric<OperationTimer> ethTasksTimer) {
return new WaitForPeersTask(ethContext, targetPeerCount, ethTasksTimer);
final EthContext ethContext, final int targetPeerCount, final MetricsSystem metricsSystem) {
return new WaitForPeersTask(ethContext, targetPeerCount, metricsSystem);
}
@Override

@ -21,8 +21,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPeerTask;
import tech.pegasys.pantheon.ethereum.eth.manager.task.GetHeadersFromPeerByNumberTask;
import tech.pegasys.pantheon.ethereum.mainnet.MainnetBlockHeaderValidator;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.time.Duration;
import java.util.List;
@ -37,7 +36,7 @@ public class DaoForkPeerValidator implements PeerValidator {
private final EthContext ethContext;
private final ProtocolSchedule<?> protocolSchedule;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private final long daoBlockNumber;
// Wait for peer's chainhead to advance some distance beyond daoBlockNumber before validating
@ -46,13 +45,13 @@ public class DaoForkPeerValidator implements PeerValidator {
public DaoForkPeerValidator(
final EthContext ethContext,
final ProtocolSchedule<?> protocolSchedule,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final long daoBlockNumber,
final long chainHeightEstimationBuffer) {
checkArgument(chainHeightEstimationBuffer >= 0);
this.ethContext = ethContext;
this.protocolSchedule = protocolSchedule;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
this.daoBlockNumber = daoBlockNumber;
this.chainHeightEstimationBuffer = chainHeightEstimationBuffer;
}
@ -60,12 +59,12 @@ public class DaoForkPeerValidator implements PeerValidator {
public DaoForkPeerValidator(
final EthContext ethContext,
final ProtocolSchedule<?> protocolSchedule,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final long daoBlockNumber) {
this(
ethContext,
protocolSchedule,
ethTasksTimer,
metricsSystem,
daoBlockNumber,
DEFAULT_CHAIN_HEIGHT_ESTIMATION_BUFFER);
}
@ -74,7 +73,7 @@ public class DaoForkPeerValidator implements PeerValidator {
public CompletableFuture<Boolean> validatePeer(final EthPeer ethPeer) {
AbstractPeerTask<List<BlockHeader>> getHeaderTask =
GetHeadersFromPeerByNumberTask.forSingleNumber(
protocolSchedule, ethContext, daoBlockNumber, ethTasksTimer)
protocolSchedule, ethContext, daoBlockNumber, metricsSystem)
.setTimeout(Duration.ofSeconds(20))
.assignPeer(ethPeer);
return getHeaderTask

@ -37,8 +37,7 @@ import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.ethereum.rlp.RLPException;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.uint.UInt256;
import java.util.ArrayList;
@ -65,7 +64,7 @@ public class BlockPropagationManager<C> {
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final SyncState syncState;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private final BlockBroadcaster blockBroadcaster;
private final AtomicBoolean started = new AtomicBoolean(false);
@ -81,13 +80,13 @@ public class BlockPropagationManager<C> {
final EthContext ethContext,
final SyncState syncState,
final PendingBlocks pendingBlocks,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final BlockBroadcaster blockBroadcaster) {
this.config = config;
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
this.blockBroadcaster = blockBroadcaster;
this.syncState = syncState;
this.pendingBlocks = pendingBlocks;
@ -156,7 +155,7 @@ public class BlockPropagationManager<C> {
protocolContext,
readyForImport,
HeaderValidationMode.FULL,
ethTasksTimer);
metricsSystem);
ethContext
.getScheduler()
.scheduleSyncWorkerTask(importBlocksTask)
@ -258,7 +257,7 @@ public class BlockPropagationManager<C> {
private CompletableFuture<Block> processAnnouncedBlock(
final EthPeer peer, final NewBlockHash newBlock) {
final AbstractPeerTask<Block> getBlockTask =
GetBlockFromPeerTask.create(protocolSchedule, ethContext, newBlock.hash(), ethTasksTimer)
GetBlockFromPeerTask.create(protocolSchedule, ethContext, newBlock.hash(), metricsSystem)
.assignPeer(peer);
return getBlockTask.run().thenCompose((r) -> importOrSavePendingBlock(r.getResult()));
@ -298,7 +297,7 @@ public class BlockPropagationManager<C> {
// Import block
final PersistBlockTask<C> importTask =
PersistBlockTask.create(
protocolSchedule, protocolContext, block, HeaderValidationMode.FULL, ethTasksTimer);
protocolSchedule, protocolContext, block, HeaderValidationMode.FULL, metricsSystem);
return ethContext
.getScheduler()
.scheduleSyncWorkerTask(importTask::run)

@ -24,8 +24,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncTarget;
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.exceptions.InvalidBlockException;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.ExceptionUtils;
import java.time.Duration;
@ -49,7 +48,7 @@ public class ChainDownloader<C> {
private final SyncTargetManager<C> syncTargetManager;
private final CheckpointHeaderManager<C> checkpointHeaderManager;
private final BlockImportTaskFactory blockImportTaskFactory;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private final CompletableFuture<Void> downloadFuture = new CompletableFuture<>();
private int chainSegmentTimeouts = 0;
@ -61,11 +60,11 @@ public class ChainDownloader<C> {
final SynchronizerConfiguration config,
final EthContext ethContext,
final SyncState syncState,
final LabelledMetric<OperationTimer> ethTasksTimer,
final SyncTargetManager<C> syncTargetManager,
final CheckpointHeaderManager<C> checkpointHeaderManager,
final BlockImportTaskFactory blockImportTaskFactory) {
this.ethTasksTimer = ethTasksTimer;
final BlockImportTaskFactory blockImportTaskFactory,
final MetricsSystem metricsSystem) {
this.metricsSystem = metricsSystem;
this.config = config;
this.ethContext = ethContext;
@ -136,7 +135,7 @@ public class ChainDownloader<C> {
}
private CompletableFuture<?> waitForPeers() {
return WaitForPeersTask.create(ethContext, 1, ethTasksTimer).run();
return WaitForPeersTask.create(ethContext, 1, metricsSystem).run();
}
private CompletableFuture<Void> checkSyncTarget() {

@ -23,8 +23,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.EthPeers.ConnectCallback;
import tech.pegasys.pantheon.ethereum.eth.manager.task.GetHeadersFromPeerByHashTask;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import org.apache.logging.log4j.Logger;
@ -35,17 +34,17 @@ public class ChainHeadTracker implements ConnectCallback {
private final EthContext ethContext;
private final ProtocolSchedule<?> protocolSchedule;
private final TrailingPeerLimiter trailingPeerLimiter;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
public ChainHeadTracker(
final EthContext ethContext,
final ProtocolSchedule<?> protocolSchedule,
final TrailingPeerLimiter trailingPeerLimiter,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
this.ethContext = ethContext;
this.protocolSchedule = protocolSchedule;
this.trailingPeerLimiter = trailingPeerLimiter;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
}
public static void trackChainHeadForPeers(
@ -53,7 +52,7 @@ public class ChainHeadTracker implements ConnectCallback {
final ProtocolSchedule<?> protocolSchedule,
final Blockchain blockchain,
final SynchronizerConfiguration syncConfiguration,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
final TrailingPeerLimiter trailingPeerLimiter =
new TrailingPeerLimiter(
ethContext.getEthPeers(),
@ -61,7 +60,7 @@ public class ChainHeadTracker implements ConnectCallback {
syncConfiguration.trailingPeerBlocksBehindThreshold(),
syncConfiguration.maxTrailingPeers());
final ChainHeadTracker tracker =
new ChainHeadTracker(ethContext, protocolSchedule, trailingPeerLimiter, ethTasksTimer);
new ChainHeadTracker(ethContext, protocolSchedule, trailingPeerLimiter, metricsSystem);
ethContext.getEthPeers().subscribeConnect(tracker);
blockchain.observeBlockAdded(trailingPeerLimiter);
}
@ -73,7 +72,7 @@ public class ChainHeadTracker implements ConnectCallback {
protocolSchedule,
ethContext,
Hash.wrap(peer.chainState().getBestBlock().getHash()),
ethTasksTimer)
metricsSystem)
.assignPeer(peer)
.run()
.whenComplete(

@ -23,8 +23,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.GetHeadersFromPeerByHashT
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncTarget;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.ExceptionUtils;
import java.util.ArrayList;
@ -48,7 +47,7 @@ public class CheckpointHeaderManager<C> {
private final EthContext ethContext;
private final SyncState syncState;
private final ProtocolSchedule<C> protocolSchedule;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private int checkpointTimeouts = 0;
@ -58,13 +57,13 @@ public class CheckpointHeaderManager<C> {
final EthContext ethContext,
final SyncState syncState,
final ProtocolSchedule<C> protocolSchedule,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
this.config = config;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.syncState = syncState;
this.protocolSchedule = protocolSchedule;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
}
public CompletableFuture<List<BlockHeader>> pullCheckpointHeaders(final SyncTarget syncTarget) {
@ -130,7 +129,7 @@ public class CheckpointHeaderManager<C> {
// + 1 because lastHeader will be returned as well.
additionalHeaderCount + 1,
skip,
ethTasksTimer)
metricsSystem)
.assignPeer(syncTarget.peer())
.run()
.thenApply(PeerTaskResult::getResult);

@ -23,10 +23,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.state.PendingBlocks;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.worldstate.WorldStateStorage;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.MetricCategory;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.util.ExceptionUtils;
import java.nio.file.Path;
@ -56,8 +53,7 @@ public class DefaultSynchronizer<C> implements Synchronizer {
final EthContext ethContext,
final SyncState syncState,
final Path dataDirectory,
final MetricsSystem metricsSystem,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
this.syncConfig = syncConfig;
this.ethContext = ethContext;
this.syncState = syncState;
@ -70,15 +66,15 @@ public class DefaultSynchronizer<C> implements Synchronizer {
ethContext,
syncState,
new PendingBlocks(),
ethTasksTimer,
metricsSystem,
new BlockBroadcaster(ethContext));
ChainHeadTracker.trackChainHeadForPeers(
ethContext, protocolSchedule, protocolContext.getBlockchain(), syncConfig, ethTasksTimer);
ethContext, protocolSchedule, protocolContext.getBlockchain(), syncConfig, metricsSystem);
this.fullSyncDownloader =
new FullSyncDownloader<>(
syncConfig, protocolSchedule, protocolContext, ethContext, syncState, ethTasksTimer);
syncConfig, protocolSchedule, protocolContext, ethContext, syncState, metricsSystem);
fastSynchronizer =
FastSynchronizer.create(
@ -89,32 +85,9 @@ public class DefaultSynchronizer<C> implements Synchronizer {
metricsSystem,
ethContext,
worldStateStorage,
ethTasksTimer,
syncState);
}
public DefaultSynchronizer(
final SynchronizerConfiguration syncConfig,
final ProtocolSchedule<C> protocolSchedule,
final ProtocolContext<C> protocolContext,
final WorldStateStorage worldStateStorage,
final EthContext ethContext,
final SyncState syncState,
final Path dataDirectory,
final MetricsSystem metricsSystem) {
this(
syncConfig,
protocolSchedule,
protocolContext,
worldStateStorage,
ethContext,
syncState,
dataDirectory,
metricsSystem,
metricsSystem.createLabelledTimer(
MetricCategory.SYNCHRONIZER, "task", "Internal processing tasks", "taskName"));
}
@Override
public void start() {
if (started.compareAndSet(false, true)) {

@ -25,10 +25,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.worldstate.WorldStateDownloader;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ScheduleBasedBlockHashFunction;
import tech.pegasys.pantheon.ethereum.worldstate.WorldStateStorage;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.MetricCategory;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.services.queue.RocksDbTaskQueue;
import tech.pegasys.pantheon.services.queue.TaskQueue;
@ -73,7 +70,6 @@ class FastSynchronizer<C> {
final MetricsSystem metricsSystem,
final EthContext ethContext,
final WorldStateStorage worldStateStorage,
final LabelledMetric<OperationTimer> ethTasksTimer,
final SyncState syncState) {
if (syncConfig.syncMode() != SyncMode.FAST) {
return Optional.empty();
@ -102,7 +98,6 @@ class FastSynchronizer<C> {
syncConfig.getWorldStateHashCountPerRequest(),
syncConfig.getWorldStateRequestParallelism(),
syncConfig.getWorldStateRequestMaxRetries(),
ethTasksTimer,
metricsSystem);
final FastSyncDownloader<C> fastSyncDownloader =
new FastSyncDownloader<>(
@ -112,12 +107,7 @@ class FastSynchronizer<C> {
protocolContext,
ethContext,
syncState,
ethTasksTimer,
metricsSystem.createLabelledCounter(
MetricCategory.SYNCHRONIZER,
"fast_sync_validation_mode",
"Number of blocks validated using light vs full validation during fast sync",
"validationMode")),
metricsSystem),
worldStateDownloader,
fastSyncStateStorage);
return Optional.of(

@ -22,8 +22,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncTarget;
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.DetermineCommonAncestorTask;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.time.Duration;
import java.util.Optional;
@ -42,7 +41,7 @@ public abstract class SyncTargetManager<C> {
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final SyncState syncState;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
public SyncTargetManager(
final SynchronizerConfiguration config,
@ -50,13 +49,13 @@ public abstract class SyncTargetManager<C> {
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final SyncState syncState,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
this.config = config;
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.syncState = syncState;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
}
public CompletableFuture<SyncTarget> findSyncTarget() {
@ -78,7 +77,7 @@ public abstract class SyncTargetManager<C> {
ethContext,
bestPeer,
config.downloaderHeaderRequestSize(),
ethTasksTimer)
metricsSystem)
.run()
.handle(
(result, error) -> {
@ -125,7 +124,7 @@ public abstract class SyncTargetManager<C> {
private CompletableFuture<?> waitForNewPeer() {
return ethContext
.getScheduler()
.timeout(WaitForPeerTask.create(ethContext, ethTasksTimer), Duration.ofSeconds(5));
.timeout(WaitForPeerTask.create(ethContext, metricsSystem), Duration.ofSeconds(5));
}
private void onSyncTargetPeerDisconnect(final EthPeer ethPeer) {

@ -25,9 +25,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.WaitForPeersTask;
import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.Counter;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.ExceptionUtils;
import java.time.Duration;
@ -45,8 +43,7 @@ public class FastSyncActions<C> {
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final SyncState syncState;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final LabelledMetric<Counter> fastSyncValidationCounter;
private final MetricsSystem metricsSystem;
public FastSyncActions(
final SynchronizerConfiguration syncConfig,
@ -54,15 +51,13 @@ public class FastSyncActions<C> {
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final SyncState syncState,
final LabelledMetric<OperationTimer> ethTasksTimer,
final LabelledMetric<Counter> fastSyncValidationCounter) {
final MetricsSystem metricsSystem) {
this.syncConfig = syncConfig;
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.syncState = syncState;
this.ethTasksTimer = ethTasksTimer;
this.fastSyncValidationCounter = fastSyncValidationCounter;
this.metricsSystem = metricsSystem;
}
public CompletableFuture<FastSyncState> waitForSuitablePeers(final FastSyncState fastSyncState) {
@ -72,7 +67,7 @@ public class FastSyncActions<C> {
final WaitForPeersTask waitForPeersTask =
WaitForPeersTask.create(
ethContext, syncConfig.getFastSyncMinimumPeerCount(), ethTasksTimer);
ethContext, syncConfig.getFastSyncMinimumPeerCount(), metricsSystem);
final EthScheduler scheduler = ethContext.getScheduler();
return exceptionallyCompose(
@ -99,7 +94,7 @@ public class FastSyncActions<C> {
private CompletableFuture<Void> waitForAnyPeer() {
final CompletableFuture<Void> waitForPeerResult =
ethContext.getScheduler().timeout(WaitForPeersTask.create(ethContext, 1, ethTasksTimer));
ethContext.getScheduler().timeout(WaitForPeersTask.create(ethContext, 1, metricsSystem));
return exceptionallyCompose(
waitForPeerResult,
throwable -> {
@ -152,7 +147,7 @@ public class FastSyncActions<C> {
return new PivotBlockRetriever<>(
protocolSchedule,
ethContext,
ethTasksTimer,
metricsSystem,
currentState.getPivotBlockNumber().getAsLong())
.downloadPivotBlockHeader();
}
@ -165,8 +160,7 @@ public class FastSyncActions<C> {
protocolContext,
ethContext,
syncState,
ethTasksTimer,
fastSyncValidationCounter,
metricsSystem,
currentState.getPivotBlockHeader().get());
return downloader.start().thenApply(ignore -> currentState);
}

@ -28,8 +28,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.tasks.GetReceiptsForHeadersTask;
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.exceptions.InvalidBlockException;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.Map;
@ -45,19 +44,19 @@ public class FastSyncBlockHandler<C> implements BlockHandler<BlockWithReceipts>
private final ProtocolSchedule<C> protocolSchedule;
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private final ValidationPolicy validationPolicy;
public FastSyncBlockHandler(
final ProtocolSchedule<C> protocolSchedule,
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final ValidationPolicy validationPolicy) {
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
this.validationPolicy = validationPolicy;
}
@ -69,13 +68,13 @@ public class FastSyncBlockHandler<C> implements BlockHandler<BlockWithReceipts>
}
private CompletableFuture<List<Block>> downloadBodies(final List<BlockHeader> headers) {
return CompleteBlocksTask.forHeaders(protocolSchedule, ethContext, headers, ethTasksTimer)
return CompleteBlocksTask.forHeaders(protocolSchedule, ethContext, headers, metricsSystem)
.run();
}
private CompletableFuture<Map<BlockHeader, List<TransactionReceipt>>> downloadReceipts(
final List<BlockHeader> headers) {
return GetReceiptsForHeadersTask.forHeaders(ethContext, headers, ethTasksTimer).run();
return GetReceiptsForHeadersTask.forHeaders(ethContext, headers, metricsSystem).run();
}
private List<BlockWithReceipts> combineBlocksAndReceipts(

@ -26,7 +26,8 @@ import tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.Counter;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricCategory;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@ -38,7 +39,7 @@ public class FastSyncChainDownloader<C> {
private final ProtocolSchedule<C> protocolSchedule;
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private final LabelledMetric<Counter> fastSyncValidationCounter;
FastSyncChainDownloader(
@ -47,28 +48,26 @@ public class FastSyncChainDownloader<C> {
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final SyncState syncState,
final LabelledMetric<OperationTimer> ethTasksTimer,
final LabelledMetric<Counter> fastSyncValidationCounter,
final MetricsSystem metricsSystem,
final BlockHeader pivotBlockHeader) {
this.config = config;
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.ethTasksTimer = ethTasksTimer;
this.fastSyncValidationCounter = fastSyncValidationCounter;
this.metricsSystem = metricsSystem;
chainDownloader =
new ChainDownloader<>(
config,
ethContext,
syncState,
ethTasksTimer,
new FastSyncTargetManager<>(
config,
protocolSchedule,
protocolContext,
ethContext,
syncState,
ethTasksTimer,
metricsSystem,
pivotBlockHeader),
new FastSyncCheckpointHeaderManager<>(
config,
@ -76,9 +75,16 @@ public class FastSyncChainDownloader<C> {
ethContext,
syncState,
protocolSchedule,
ethTasksTimer,
metricsSystem,
pivotBlockHeader),
this::importBlocksForCheckpoints);
this::importBlocksForCheckpoints,
metricsSystem);
this.fastSyncValidationCounter =
metricsSystem.createLabelledCounter(
MetricCategory.SYNCHRONIZER,
"fast_sync_validation_mode",
"Number of blocks validated using light vs full validation during fast sync",
"validationMode");
}
public CompletableFuture<Void> start() {
@ -109,15 +115,15 @@ public class FastSyncChainDownloader<C> {
protocolContext,
ethContext,
config.downloaderParallelism(),
ethTasksTimer,
new FastSyncBlockHandler<>(
protocolSchedule,
protocolContext,
ethContext,
ethTasksTimer,
metricsSystem,
attachedValidationPolicy),
detatchedValidationPolicy,
checkpointHeaders);
checkpointHeaders,
metricsSystem);
return importTask
.run()
.thenApply(

@ -20,8 +20,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncTarget;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@ -40,9 +39,9 @@ class FastSyncCheckpointHeaderManager<C> extends CheckpointHeaderManager<C> {
final EthContext ethContext,
final SyncState syncState,
final ProtocolSchedule<C> protocolSchedule,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final BlockHeader pivotBlockHeader) {
super(config, protocolContext, ethContext, syncState, protocolSchedule, ethTasksTimer);
super(config, protocolContext, ethContext, syncState, protocolSchedule, metricsSystem);
this.config = config;
this.pivotBlockHeader = pivotBlockHeader;
}

@ -26,8 +26,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncTarget;
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.RetryingGetHeaderFromPeerByNumberTask;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.Optional;
@ -42,7 +41,7 @@ class FastSyncTargetManager<C> extends SyncTargetManager<C> {
private final ProtocolSchedule<C> protocolSchedule;
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private final BlockHeader pivotBlockHeader;
public FastSyncTargetManager(
@ -51,13 +50,13 @@ class FastSyncTargetManager<C> extends SyncTargetManager<C> {
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final SyncState syncState,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final BlockHeader pivotBlockHeader) {
super(config, protocolSchedule, protocolContext, ethContext, syncState, ethTasksTimer);
super(config, protocolSchedule, protocolContext, ethContext, syncState, metricsSystem);
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
this.pivotBlockHeader = pivotBlockHeader;
}
@ -83,7 +82,7 @@ class FastSyncTargetManager<C> extends SyncTargetManager<C> {
RetryingGetHeaderFromPeerByNumberTask.forSingleNumber(
protocolSchedule,
ethContext,
ethTasksTimer,
metricsSystem,
pivotBlockHeader.getNumber(),
MAX_PIVOT_BLOCK_RETRIES);
task.assignPeer(bestPeer);

@ -17,8 +17,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.RetryingGetHeaderFromPeerByNumberTask;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.Collection;
import java.util.List;
@ -38,7 +37,7 @@ public class PivotBlockRetriever<C> {
static final int MAX_PIVOT_BLOCK_RETRIES = 3;
private final long pivotBlockNumber;
private final EthContext ethContext;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private final ProtocolSchedule<C> protocolSchedule;
private final Map<BlockHeader, AtomicInteger> confirmationsByBlockNumber =
new ConcurrentHashMap<>();
@ -49,11 +48,11 @@ public class PivotBlockRetriever<C> {
public PivotBlockRetriever(
final ProtocolSchedule<C> protocolSchedule,
final EthContext ethContext,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final long pivotBlockNumber) {
this.pivotBlockNumber = pivotBlockNumber;
this.ethContext = ethContext;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
this.protocolSchedule = protocolSchedule;
}
@ -97,7 +96,7 @@ public class PivotBlockRetriever<C> {
private RetryingGetHeaderFromPeerByNumberTask createGetHeaderTask(final EthPeer peer) {
final RetryingGetHeaderFromPeerByNumberTask task =
RetryingGetHeaderFromPeerByNumberTask.forSingleNumber(
protocolSchedule, ethContext, ethTasksTimer, pivotBlockNumber, MAX_PIVOT_BLOCK_RETRIES);
protocolSchedule, ethContext, metricsSystem, pivotBlockNumber, MAX_PIVOT_BLOCK_RETRIES);
task.assignPeer(peer);
return task;
}

@ -23,8 +23,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.tasks.CompleteBlocksTask;
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.PersistBlockTask;
import tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.ArrayList;
import java.util.List;
@ -39,17 +38,17 @@ public class FullSyncBlockHandler<C> implements BlockHandler<Block> {
private final ProtocolSchedule<C> protocolSchedule;
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
public FullSyncBlockHandler(
final ProtocolSchedule<C> protocolSchedule,
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
}
@Override
@ -63,13 +62,13 @@ public class FullSyncBlockHandler<C> implements BlockHandler<Block> {
protocolContext,
blocks,
HeaderValidationMode.SKIP_DETACHED,
ethTasksTimer)
metricsSystem)
.get();
}
@Override
public CompletableFuture<List<Block>> downloadBlocks(final List<BlockHeader> headers) {
return CompleteBlocksTask.forHeaders(protocolSchedule, ethContext, headers, ethTasksTimer)
return CompleteBlocksTask.forHeaders(protocolSchedule, ethContext, headers, metricsSystem)
.run();
}

@ -25,8 +25,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.tasks.ImportBlocksTask;
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.ParallelImportChainSegmentTask;
import tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@ -42,7 +41,7 @@ public class FullSyncDownloader<C> {
private final ProtocolSchedule<C> protocolSchedule;
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
public FullSyncDownloader(
final SynchronizerConfiguration config,
@ -50,23 +49,23 @@ public class FullSyncDownloader<C> {
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final SyncState syncState,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
this.config = config;
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
chainDownloader =
new ChainDownloader<>(
config,
ethContext,
syncState,
ethTasksTimer,
new FullSyncTargetManager<>(
config, protocolSchedule, protocolContext, ethContext, syncState, ethTasksTimer),
config, protocolSchedule, protocolContext, ethContext, syncState, metricsSystem),
new CheckpointHeaderManager<>(
config, protocolContext, ethContext, syncState, protocolSchedule, ethTasksTimer),
this::importBlocksForCheckpoints);
config, protocolContext, ethContext, syncState, protocolSchedule, metricsSystem),
this::importBlocksForCheckpoints,
metricsSystem);
}
public void start() {
@ -90,7 +89,7 @@ public class FullSyncDownloader<C> {
ethContext,
checkpointHeaders.get(0),
config.downloaderChainSegmentSize(),
ethTasksTimer);
metricsSystem);
importedBlocks = importTask.run().thenApply(PeerTaskResult::getResult);
} else {
final ParallelImportChainSegmentTask<C, Block> importTask =
@ -99,11 +98,11 @@ public class FullSyncDownloader<C> {
protocolContext,
ethContext,
config.downloaderParallelism(),
ethTasksTimer,
new FullSyncBlockHandler<>(
protocolSchedule, protocolContext, ethContext, ethTasksTimer),
protocolSchedule, protocolContext, ethContext, metricsSystem),
() -> HeaderValidationMode.DETACHED_ONLY,
checkpointHeaders);
checkpointHeaders,
metricsSystem);
importedBlocks = importTask.run();
}
return importedBlocks;

@ -26,8 +26,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncTarget;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.uint.UInt256;
import java.util.Optional;
@ -50,8 +49,8 @@ class FullSyncTargetManager<C> extends SyncTargetManager<C> {
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final SyncState syncState,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(config, protocolSchedule, protocolContext, ethContext, syncState, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(config, protocolSchedule, protocolContext, ethContext, syncState, metricsSystem);
this.config = config;
this.protocolContext = protocolContext;
this.ethContext = ethContext;

@ -27,8 +27,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPeerTask.PeerTask
import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractRetryingPeerTask;
import tech.pegasys.pantheon.ethereum.eth.manager.task.GetBodiesFromPeerTask;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.Collection;
import java.util.List;
@ -55,17 +54,19 @@ public class CompleteBlocksTask<C> extends AbstractRetryingPeerTask<List<Block>>
private final List<BlockHeader> headers;
private final Map<Long, Block> blocks;
private final MetricsSystem metricsSystem;
private CompleteBlocksTask(
final ProtocolSchedule<C> protocolSchedule,
final EthContext ethContext,
final List<BlockHeader> headers,
final int maxRetries,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, maxRetries, ethTasksTimer, Collection::isEmpty);
final MetricsSystem metricsSystem) {
super(ethContext, maxRetries, Collection::isEmpty, metricsSystem);
checkArgument(headers.size() > 0, "Must supply a non-empty headers list");
this.protocolSchedule = protocolSchedule;
this.ethContext = ethContext;
this.metricsSystem = metricsSystem;
this.headers = headers;
this.blocks =
@ -84,18 +85,18 @@ public class CompleteBlocksTask<C> extends AbstractRetryingPeerTask<List<Block>>
final EthContext ethContext,
final List<BlockHeader> headers,
final int maxRetries,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new CompleteBlocksTask<>(
protocolSchedule, ethContext, headers, maxRetries, ethTasksTimer);
protocolSchedule, ethContext, headers, maxRetries, metricsSystem);
}
public static <C> CompleteBlocksTask<C> forHeaders(
final ProtocolSchedule<C> protocolSchedule,
final EthContext ethContext,
final List<BlockHeader> headers,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new CompleteBlocksTask<>(
protocolSchedule, ethContext, headers, DEFAULT_RETRIES, ethTasksTimer);
protocolSchedule, ethContext, headers, DEFAULT_RETRIES, metricsSystem);
}
@Override
@ -116,7 +117,7 @@ public class CompleteBlocksTask<C> extends AbstractRetryingPeerTask<List<Block>>
() -> {
final GetBodiesFromPeerTask<C> task =
GetBodiesFromPeerTask.forHeaders(
protocolSchedule, ethContext, incompleteHeaders, ethTasksTimer);
protocolSchedule, ethContext, incompleteHeaders, metricsSystem);
assignedPeer.ifPresent(task::assignPeer);
return task.run().thenApply(PeerTaskResult::getResult);
});

@ -21,8 +21,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPeerTask;
import tech.pegasys.pantheon.ethereum.eth.manager.task.GetHeadersFromPeerByNumberTask;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.util.BlockchainUtil;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.OptionalInt;
@ -39,6 +38,7 @@ public class DetermineCommonAncestorTask<C> extends AbstractEthTask<BlockHeader>
private final ProtocolContext<C> protocolContext;
private final EthPeer peer;
private final int headerRequestSize;
private final MetricsSystem metricsSystem;
private long maximumPossibleCommonAncestorNumber;
private long minimumPossibleCommonAncestorNumber;
@ -51,13 +51,14 @@ public class DetermineCommonAncestorTask<C> extends AbstractEthTask<BlockHeader>
final EthContext ethContext,
final EthPeer peer,
final int headerRequestSize,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethTasksTimer);
final MetricsSystem metricsSystem) {
super(metricsSystem);
this.protocolSchedule = protocolSchedule;
this.ethContext = ethContext;
this.protocolContext = protocolContext;
this.peer = peer;
this.headerRequestSize = headerRequestSize;
this.metricsSystem = metricsSystem;
maximumPossibleCommonAncestorNumber = protocolContext.getBlockchain().getChainHeadBlockNumber();
minimumPossibleCommonAncestorNumber = BlockHeader.GENESIS_BLOCK_NUMBER;
@ -71,9 +72,9 @@ public class DetermineCommonAncestorTask<C> extends AbstractEthTask<BlockHeader>
final EthContext ethContext,
final EthPeer peer,
final int headerRequestSize,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new DetermineCommonAncestorTask<>(
protocolSchedule, protocolContext, ethContext, peer, headerRequestSize, ethTasksTimer);
protocolSchedule, protocolContext, ethContext, peer, headerRequestSize, metricsSystem);
}
@Override
@ -119,7 +120,7 @@ public class DetermineCommonAncestorTask<C> extends AbstractEthTask<BlockHeader>
maximumPossibleCommonAncestorNumber,
count,
skipInterval,
ethTasksTimer)
metricsSystem)
.assignPeer(peer)
.run());
}

@ -29,8 +29,7 @@ import tech.pegasys.pantheon.ethereum.mainnet.BlockHeaderValidator;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.Arrays;
import java.util.Collection;
@ -60,6 +59,7 @@ public class DownloadHeaderSequenceTask<C> extends AbstractRetryingPeerTask<List
private final BlockHeader referenceHeader;
private final int segmentLength;
private final long startingBlockNumber;
private final MetricsSystem metricsSystem;
private int lastFilledHeaderIndex;
@ -70,13 +70,14 @@ public class DownloadHeaderSequenceTask<C> extends AbstractRetryingPeerTask<List
final BlockHeader referenceHeader,
final int segmentLength,
final int maxRetries,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, maxRetries, ethTasksTimer, Collection::isEmpty);
final MetricsSystem metricsSystem) {
super(ethContext, maxRetries, Collection::isEmpty, metricsSystem);
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.referenceHeader = referenceHeader;
this.segmentLength = segmentLength;
this.metricsSystem = metricsSystem;
startingBlockNumber = referenceHeader.getNumber() - segmentLength;
headers = new BlockHeader[segmentLength];
@ -90,7 +91,7 @@ public class DownloadHeaderSequenceTask<C> extends AbstractRetryingPeerTask<List
final BlockHeader referenceHeader,
final int segmentLength,
final int maxRetries,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new DownloadHeaderSequenceTask<>(
protocolSchedule,
protocolContext,
@ -98,7 +99,7 @@ public class DownloadHeaderSequenceTask<C> extends AbstractRetryingPeerTask<List
referenceHeader,
segmentLength,
maxRetries,
ethTasksTimer);
metricsSystem);
}
public static <C> DownloadHeaderSequenceTask<C> endingAtHeader(
@ -107,7 +108,7 @@ public class DownloadHeaderSequenceTask<C> extends AbstractRetryingPeerTask<List
final EthContext ethContext,
final BlockHeader referenceHeader,
final int segmentLength,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new DownloadHeaderSequenceTask<>(
protocolSchedule,
protocolContext,
@ -115,7 +116,7 @@ public class DownloadHeaderSequenceTask<C> extends AbstractRetryingPeerTask<List
referenceHeader,
segmentLength,
DEFAULT_RETRIES,
ethTasksTimer);
metricsSystem);
}
@Override
@ -157,7 +158,7 @@ public class DownloadHeaderSequenceTask<C> extends AbstractRetryingPeerTask<List
referenceHash,
referenceHeaderForNextRequest.getNumber(),
count + 1,
ethTasksTimer);
metricsSystem);
assignedPeer.ifPresent(headersTask::assignPeer);
return headersTask.run();
});

@ -24,8 +24,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer;
import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPeerTask.PeerTaskResult;
import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractRetryingPeerTask;
import tech.pegasys.pantheon.ethereum.eth.manager.task.GetReceiptsFromPeerTask;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.HashMap;
import java.util.List;
@ -47,15 +46,17 @@ public class GetReceiptsForHeadersTask
private final List<BlockHeader> headers;
private final Map<BlockHeader, List<TransactionReceipt>> receipts;
private final MetricsSystem metricsSystem;
private GetReceiptsForHeadersTask(
final EthContext ethContext,
final List<BlockHeader> headers,
final int maxRetries,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, maxRetries, ethTasksTimer, Map::isEmpty);
final MetricsSystem metricsSystem) {
super(ethContext, maxRetries, Map::isEmpty, metricsSystem);
checkArgument(headers.size() > 0, "Must supply a non-empty headers list");
this.ethContext = ethContext;
this.metricsSystem = metricsSystem;
this.headers = headers;
this.receipts = new HashMap<>();
@ -66,15 +67,15 @@ public class GetReceiptsForHeadersTask
final EthContext ethContext,
final List<BlockHeader> headers,
final int maxRetries,
final LabelledMetric<OperationTimer> ethTasksTimer) {
return new GetReceiptsForHeadersTask(ethContext, headers, maxRetries, ethTasksTimer);
final MetricsSystem metricsSystem) {
return new GetReceiptsForHeadersTask(ethContext, headers, maxRetries, metricsSystem);
}
public static GetReceiptsForHeadersTask forHeaders(
final EthContext ethContext,
final List<BlockHeader> headers,
final LabelledMetric<OperationTimer> ethTasksTimer) {
return new GetReceiptsForHeadersTask(ethContext, headers, DEFAULT_RETRIES, ethTasksTimer);
final MetricsSystem metricsSystem) {
return new GetReceiptsForHeadersTask(ethContext, headers, DEFAULT_RETRIES, metricsSystem);
}
private void completeEmptyReceipts(final List<BlockHeader> headers) {
@ -102,7 +103,7 @@ public class GetReceiptsForHeadersTask
return executeSubTask(
() -> {
final GetReceiptsFromPeerTask task =
GetReceiptsFromPeerTask.forHeaders(ethContext, incompleteHeaders, ethTasksTimer);
GetReceiptsFromPeerTask.forHeaders(ethContext, incompleteHeaders, metricsSystem);
assignedPeer.ifPresent(task::assignPeer);
return task.run().thenApply(PeerTaskResult::getResult);
});

@ -22,8 +22,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.GetHeadersFromPeerByHashT
import tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.Collections;
import java.util.List;
@ -47,6 +46,7 @@ public class ImportBlocksTask<C> extends AbstractPeerTask<List<Block>> {
private final BlockHeader referenceHeader;
private final int maxBlocks;
private final MetricsSystem metricsSystem;
private EthPeer peer;
protected ImportBlocksTask(
@ -55,12 +55,13 @@ public class ImportBlocksTask<C> extends AbstractPeerTask<List<Block>> {
final EthContext ethContext,
final BlockHeader referenceHeader,
final int maxBlocks,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethContext, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(ethContext, metricsSystem);
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.referenceHeader = referenceHeader;
this.maxBlocks = maxBlocks;
this.metricsSystem = metricsSystem;
this.startNumber = referenceHeader.getNumber();
}
@ -71,9 +72,9 @@ public class ImportBlocksTask<C> extends AbstractPeerTask<List<Block>> {
final EthContext ethContext,
final BlockHeader previousHeader,
final int maxBlocks,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new ImportBlocksTask<>(
protocolSchedule, protocolContext, ethContext, previousHeader, maxBlocks, ethTasksTimer);
protocolSchedule, protocolContext, ethContext, previousHeader, maxBlocks, metricsSystem);
}
@Override
@ -103,7 +104,7 @@ public class ImportBlocksTask<C> extends AbstractPeerTask<List<Block>> {
referenceHeader.getHash(),
referenceHeader.getNumber(),
maxBlocks,
ethTasksTimer)
metricsSystem)
.assignPeer(peer);
return executeSubTask(task::run);
}
@ -115,7 +116,7 @@ public class ImportBlocksTask<C> extends AbstractPeerTask<List<Block>> {
}
final CompleteBlocksTask<C> task =
CompleteBlocksTask.forHeaders(
protocolSchedule, ethContext, headers.getResult(), ethTasksTimer);
protocolSchedule, ethContext, headers.getResult(), metricsSystem);
task.assignPeer(peer);
return executeSubTask(() -> ethContext.getScheduler().timeout(task));
}
@ -130,7 +131,7 @@ public class ImportBlocksTask<C> extends AbstractPeerTask<List<Block>> {
}
final Supplier<CompletableFuture<List<Block>>> task =
PersistBlockTask.forSequentialBlocks(
protocolSchedule, protocolContext, blocks, HeaderValidationMode.FULL, ethTasksTimer);
protocolSchedule, protocolContext, blocks, HeaderValidationMode.FULL, metricsSystem);
return executeWorkerSubTask(ethContext.getScheduler(), task);
}
}

@ -15,8 +15,7 @@ package tech.pegasys.pantheon.ethereum.eth.sync.tasks;
import tech.pegasys.pantheon.ethereum.core.BlockHeader;
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;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.Optional;
@ -36,8 +35,8 @@ public class ParallelDownloadBodiesTask<B>
final BlockHandler<B> blockHandler,
final BlockingQueue<List<BlockHeader>> inboundQueue,
final int outboundBacklogSize,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(inboundQueue, outboundBacklogSize, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(inboundQueue, outboundBacklogSize, metricsSystem);
this.blockHandler = blockHandler;
}

@ -17,8 +17,7 @@ import tech.pegasys.pantheon.ethereum.core.BlockHeader;
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
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;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.Optional;
@ -37,6 +36,7 @@ public class ParallelDownloadHeadersTask<C>
private final ProtocolSchedule<C> protocolSchedule;
private final ProtocolContext<C> protocolContext;
private final EthContext ethContext;
private final MetricsSystem metricsSystem;
ParallelDownloadHeadersTask(
final BlockingQueue<BlockHeader> inboundQueue,
@ -44,12 +44,13 @@ public class ParallelDownloadHeadersTask<C>
final ProtocolSchedule<C> protocolSchedule,
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(inboundQueue, outboundBacklogSize, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(inboundQueue, outboundBacklogSize, metricsSystem);
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
this.metricsSystem = metricsSystem;
}
@Override
@ -72,7 +73,7 @@ public class ParallelDownloadHeadersTask<C>
ethContext,
nextCheckpointHeader,
segmentLength,
ethTasksTimer);
metricsSystem);
final CompletableFuture<List<BlockHeader>> headerFuture = executeSubTask(downloadTask::run);
final List<BlockHeader> headers = Lists.newArrayList(previousCheckpointHeader.get());

@ -14,8 +14,7 @@ package tech.pegasys.pantheon.ethereum.eth.sync.tasks;
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;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.Optional;
@ -34,8 +33,8 @@ class ParallelExtractTxSignaturesTask<B> extends AbstractPipelinedTask<List<B>,
final BlockHandler<B> blockHandler,
final BlockingQueue<List<B>> inboundQueue,
final int outboundBacklogSize,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(inboundQueue, outboundBacklogSize, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(inboundQueue, outboundBacklogSize, metricsSystem);
this.blockHandler = blockHandler;
}

@ -20,8 +20,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractEthTask;
import tech.pegasys.pantheon.ethereum.eth.sync.BlockHandler;
import tech.pegasys.pantheon.ethereum.eth.sync.ValidationPolicy;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.Collection;
import java.util.List;
@ -49,6 +48,7 @@ public class ParallelImportChainSegmentTask<C, B> extends AbstractEthTask<List<B
private final BlockHandler<B> blockHandler;
private final ValidationPolicy validationPolicy;
private final MetricsSystem metricsSystem;
private ParallelImportChainSegmentTask(
final ProtocolSchedule<C> protocolSchedule,
@ -56,14 +56,15 @@ public class ParallelImportChainSegmentTask<C, B> extends AbstractEthTask<List<B
final EthContext ethContext,
final int maxActiveChunks,
final List<BlockHeader> checkpointHeaders,
final LabelledMetric<OperationTimer> ethTasksTimer,
final BlockHandler<B> blockHandler,
final ValidationPolicy validationPolicy) {
super(ethTasksTimer);
final ValidationPolicy validationPolicy,
final MetricsSystem metricsSystem) {
super(metricsSystem);
this.ethContext = ethContext;
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.maxActiveChunks = maxActiveChunks;
this.metricsSystem = metricsSystem;
if (checkpointHeaders.size() > 1) {
this.firstHeaderNumber = checkpointHeaders.get(0).getNumber();
@ -83,19 +84,19 @@ public class ParallelImportChainSegmentTask<C, B> extends AbstractEthTask<List<B
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final int maxActiveChunks,
final LabelledMetric<OperationTimer> ethTasksTimer,
final BlockHandler<B> blockHandler,
final ValidationPolicy validationPolicy,
final List<BlockHeader> checkpointHeaders) {
final List<BlockHeader> checkpointHeaders,
final MetricsSystem metricsSystem) {
return new ParallelImportChainSegmentTask<>(
protocolSchedule,
protocolContext,
ethContext,
maxActiveChunks,
checkpointHeaders,
ethTasksTimer,
blockHandler,
validationPolicy);
validationPolicy,
metricsSystem);
}
@Override
@ -111,7 +112,7 @@ public class ParallelImportChainSegmentTask<C, B> extends AbstractEthTask<List<B
protocolSchedule,
protocolContext,
ethContext,
ethTasksTimer);
metricsSystem);
final ParallelValidateHeadersTask<C> validateHeadersTask =
new ParallelValidateHeadersTask<>(
validationPolicy,
@ -119,19 +120,19 @@ public class ParallelImportChainSegmentTask<C, B> extends AbstractEthTask<List<B
maxActiveChunks,
protocolSchedule,
protocolContext,
ethTasksTimer);
metricsSystem);
final ParallelDownloadBodiesTask<B> downloadBodiesTask =
new ParallelDownloadBodiesTask<>(
blockHandler, validateHeadersTask.getOutboundQueue(), maxActiveChunks, ethTasksTimer);
blockHandler, validateHeadersTask.getOutboundQueue(), maxActiveChunks, metricsSystem);
final ParallelExtractTxSignaturesTask<B> extractTxSignaturesTask =
new ParallelExtractTxSignaturesTask<>(
blockHandler, downloadBodiesTask.getOutboundQueue(), maxActiveChunks, ethTasksTimer);
blockHandler, downloadBodiesTask.getOutboundQueue(), maxActiveChunks, metricsSystem);
final ParallelValidateAndImportBodiesTask<B> validateAndImportBodiesTask =
new ParallelValidateAndImportBodiesTask<>(
blockHandler,
extractTxSignaturesTask.getOutboundQueue(),
Integer.MAX_VALUE,
ethTasksTimer);
metricsSystem);
// Start the pipeline.
final EthScheduler scheduler = ethContext.getScheduler();

@ -14,8 +14,7 @@ package tech.pegasys.pantheon.ethereum.eth.sync.tasks;
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;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.Optional;
@ -36,8 +35,8 @@ public class ParallelValidateAndImportBodiesTask<B>
final BlockHandler<B> blockHandler,
final BlockingQueue<List<B>> inboundQueue,
final int outboundBacklogSize,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(inboundQueue, outboundBacklogSize, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(inboundQueue, outboundBacklogSize, metricsSystem);
this.blockHandler = blockHandler;
}

@ -20,8 +20,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.tasks.exceptions.InvalidBlockExce
import tech.pegasys.pantheon.ethereum.mainnet.BlockHeaderValidator;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.List;
import java.util.Optional;
@ -44,8 +43,8 @@ public class ParallelValidateHeadersTask<C>
final int outboundBacklogSize,
final ProtocolSchedule<C> protocolSchedule,
final ProtocolContext<C> protocolContext,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(inboundQueue, outboundBacklogSize, ethTasksTimer);
final MetricsSystem metricsSystem) {
super(inboundQueue, outboundBacklogSize, metricsSystem);
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;

@ -22,8 +22,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.tasks.exceptions.InvalidBlockExce
import tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.ArrayList;
import java.util.List;
@ -42,8 +41,8 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
final ProtocolContext<C> protocolContext,
final Block block,
final HeaderValidationMode headerValidationMode,
final LabelledMetric<OperationTimer> ethTasksTimer) {
super(ethTasksTimer);
final MetricsSystem metricsSystem) {
super(metricsSystem);
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.block = block;
@ -55,9 +54,9 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
final ProtocolContext<C> protocolContext,
final Block block,
final HeaderValidationMode headerValidationMode,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return new PersistBlockTask<>(
protocolSchedule, protocolContext, block, headerValidationMode, ethTasksTimer);
protocolSchedule, protocolContext, block, headerValidationMode, metricsSystem);
}
public static <C> Supplier<CompletableFuture<List<Block>>> forSequentialBlocks(
@ -65,7 +64,7 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
final ProtocolContext<C> protocolContext,
final List<Block> blocks,
final HeaderValidationMode headerValidationMode,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
checkArgument(blocks.size() > 0);
return () -> {
final List<Block> successfulImports = new ArrayList<>();
@ -79,7 +78,7 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
block,
successfulImports,
headerValidationMode,
ethTasksTimer);
metricsSystem);
continue;
}
future =
@ -91,7 +90,7 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
block,
successfulImports,
headerValidationMode,
ethTasksTimer));
metricsSystem));
}
return future.thenApply(r -> successfulImports);
};
@ -103,9 +102,9 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
final Block block,
final List<Block> list,
final HeaderValidationMode headerValidationMode,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
return PersistBlockTask.create(
protocolSchedule, protocolContext, block, headerValidationMode, ethTasksTimer)
protocolSchedule, protocolContext, block, headerValidationMode, metricsSystem)
.run()
.whenComplete(
(r, t) -> {
@ -120,7 +119,7 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
final ProtocolContext<C> protocolContext,
final List<Block> blocks,
final HeaderValidationMode headerValidationMode,
final LabelledMetric<OperationTimer> ethTasksTimer) {
final MetricsSystem metricsSystem) {
checkArgument(blocks.size() > 0);
return () -> {
final CompletableFuture<List<Block>> finalResult = new CompletableFuture<>();
@ -130,7 +129,7 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
if (future == null) {
future =
PersistBlockTask.create(
protocolSchedule, protocolContext, block, headerValidationMode, ethTasksTimer)
protocolSchedule, protocolContext, block, headerValidationMode, metricsSystem)
.run();
continue;
}
@ -147,7 +146,7 @@ public class PersistBlockTask<C> extends AbstractEthTask<Block> {
protocolContext,
block,
headerValidationMode,
ethTasksTimer)
metricsSystem)
.run();
});
}

@ -22,8 +22,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.tasks.exceptions.InvalidBlockExce
import tech.pegasys.pantheon.ethereum.mainnet.BlockHeaderValidator;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.util.ExceptionUtils;
import java.util.ArrayList;
@ -52,6 +51,7 @@ public class PipelinedImportChainSegmentTask<C, B> extends AbstractEthTask<List<
private final int chunksInTotal;
private final BlockHandler<B> blockHandler;
private final ValidationPolicy validationPolicy;
private final MetricsSystem metricsSystem;
private int chunksIssued;
private int chunksCompleted;
private final int maxActiveChunks;
@ -69,10 +69,10 @@ public class PipelinedImportChainSegmentTask<C, B> extends AbstractEthTask<List<
final EthContext ethContext,
final int maxActiveChunks,
final List<BlockHeader> checkpointHeaders,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final BlockHandler<B> blockHandler,
final ValidationPolicy validationPolicy) {
super(ethTasksTimer);
super(metricsSystem);
this.protocolSchedule = protocolSchedule;
this.protocolContext = protocolContext;
this.ethContext = ethContext;
@ -80,6 +80,7 @@ public class PipelinedImportChainSegmentTask<C, B> extends AbstractEthTask<List<
this.chunksInTotal = checkpointHeaders.size() - 1;
this.blockHandler = blockHandler;
this.validationPolicy = validationPolicy;
this.metricsSystem = metricsSystem;
this.chunksIssued = 0;
this.chunksCompleted = 0;
this.maxActiveChunks = maxActiveChunks;
@ -90,7 +91,7 @@ public class PipelinedImportChainSegmentTask<C, B> extends AbstractEthTask<List<
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final int maxActiveChunks,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final BlockHandler<B> blockHandler,
final ValidationPolicy validationPolicy,
final BlockHeader... checkpointHeaders) {
@ -99,7 +100,7 @@ public class PipelinedImportChainSegmentTask<C, B> extends AbstractEthTask<List<
protocolContext,
ethContext,
maxActiveChunks,
ethTasksTimer,
metricsSystem,
blockHandler,
validationPolicy,
Arrays.asList(checkpointHeaders));
@ -110,7 +111,7 @@ public class PipelinedImportChainSegmentTask<C, B> extends AbstractEthTask<List<
final ProtocolContext<C> protocolContext,
final EthContext ethContext,
final int maxActiveChunks,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final BlockHandler<B> blockHandler,
final ValidationPolicy validationPolicy,
final List<BlockHeader> checkpointHeaders) {
@ -120,7 +121,7 @@ public class PipelinedImportChainSegmentTask<C, B> extends AbstractEthTask<List<
ethContext,
maxActiveChunks,
checkpointHeaders,
ethTasksTimer,
metricsSystem,
blockHandler,
validationPolicy);
}
@ -216,7 +217,7 @@ public class PipelinedImportChainSegmentTask<C, B> extends AbstractEthTask<List<
ethContext,
lastChunkHeader,
segmentLength,
ethTasksTimer);
metricsSystem);
return executeSubTask(task::run)
.thenApply(
headers -> {

@ -19,8 +19,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractGetHeadersFromPee
import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractRetryingPeerTask;
import tech.pegasys.pantheon.ethereum.eth.manager.task.GetHeadersFromPeerByNumberTask;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import java.util.Collection;
import java.util.List;
@ -32,27 +31,29 @@ public class RetryingGetHeaderFromPeerByNumberTask
private final ProtocolSchedule<?> protocolSchedule;
private final EthContext ethContext;
private final long pivotBlockNumber;
private final MetricsSystem metricsSystem;
private RetryingGetHeaderFromPeerByNumberTask(
final ProtocolSchedule<?> protocolSchedule,
final EthContext ethContext,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final long pivotBlockNumber,
final int maxRetries) {
super(ethContext, maxRetries, ethTasksTimer, Collection::isEmpty);
super(ethContext, maxRetries, Collection::isEmpty, metricsSystem);
this.protocolSchedule = protocolSchedule;
this.ethContext = ethContext;
this.pivotBlockNumber = pivotBlockNumber;
this.metricsSystem = metricsSystem;
}
public static RetryingGetHeaderFromPeerByNumberTask forSingleNumber(
final ProtocolSchedule<?> protocolSchedule,
final EthContext ethContext,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem,
final long pivotBlockNumber,
final int maxRetries) {
return new RetryingGetHeaderFromPeerByNumberTask(
protocolSchedule, ethContext, ethTasksTimer, pivotBlockNumber, maxRetries);
protocolSchedule, ethContext, metricsSystem, pivotBlockNumber, maxRetries);
}
@Override
@ -60,7 +61,7 @@ public class RetryingGetHeaderFromPeerByNumberTask
final Optional<EthPeer> assignedPeer) {
final AbstractGetHeadersFromPeerTask getHeadersTask =
GetHeadersFromPeerByNumberTask.forSingleNumber(
protocolSchedule, ethContext, pivotBlockNumber, ethTasksTimer);
protocolSchedule, ethContext, pivotBlockNumber, metricsSystem);
assignedPeer.ifPresent(getHeadersTask::assignPeer);
return executeSubTask(getHeadersTask::run)
.thenApply(

@ -25,10 +25,8 @@ import tech.pegasys.pantheon.ethereum.eth.manager.task.WaitForPeerTask;
import tech.pegasys.pantheon.ethereum.worldstate.WorldStateStorage;
import tech.pegasys.pantheon.ethereum.worldstate.WorldStateStorage.Updater;
import tech.pegasys.pantheon.metrics.Counter;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.MetricCategory;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.services.queue.TaskQueue;
import tech.pegasys.pantheon.services.queue.TaskQueue.Task;
import tech.pegasys.pantheon.util.ExceptionUtils;
@ -71,7 +69,7 @@ public class WorldStateDownloader {
private final int maxNodeRequestRetries;
private final Set<EthTask<?>> outstandingRequests =
Collections.newSetFromMap(new ConcurrentHashMap<>());
private final LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem;
private final WorldStateStorage worldStateStorage;
private final AtomicBoolean sendingRequests = new AtomicBoolean(false);
private volatile CompletableFuture<Void> future;
@ -86,7 +84,6 @@ public class WorldStateDownloader {
final int hashCountPerRequest,
final int maxOutstandingRequests,
final int maxNodeRequestRetries,
final LabelledMetric<OperationTimer> ethTasksTimer,
final MetricsSystem metricsSystem) {
this.ethContext = ethContext;
this.worldStateStorage = worldStateStorage;
@ -94,7 +91,7 @@ public class WorldStateDownloader {
this.hashCountPerRequest = hashCountPerRequest;
this.maxOutstandingRequests = maxOutstandingRequests;
this.maxNodeRequestRetries = maxNodeRequestRetries;
this.ethTasksTimer = ethTasksTimer;
this.metricsSystem = metricsSystem;
metricsSystem.createLongGauge(
MetricCategory.SYNCHRONIZER,
"world_state_pending_requests_current",
@ -236,7 +233,7 @@ public class WorldStateDownloader {
private CompletableFuture<?> waitForNewPeer() {
return ethContext
.getScheduler()
.timeout(WaitForPeerTask.create(ethContext, ethTasksTimer), Duration.ofSeconds(5));
.timeout(WaitForPeerTask.create(ethContext, metricsSystem), Duration.ofSeconds(5));
}
private CompletableFuture<AbstractPeerTask<Map<Hash, BytesValue>>> sendAndProcessRequests(
@ -250,7 +247,7 @@ public class WorldStateDownloader {
.distinct()
.collect(Collectors.toList());
final AbstractPeerTask<Map<Hash, BytesValue>> ethTask =
GetNodeDataFromPeerTask.forHashes(ethContext, hashes, ethTasksTimer).assignPeer(peer);
GetNodeDataFromPeerTask.forHashes(ethContext, hashes, metricsSystem).assignPeer(peer);
outstandingRequests.add(ethTask);
return ethTask
.run()

@ -23,7 +23,7 @@ public class MockEthTask extends AbstractEthTask<Object> {
private CountDownLatch countdown;
MockEthTask(final int count) {
super(NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
super(new NoOpMetricsSystem());
countdown = new CountDownLatch(count);
}

@ -24,8 +24,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.RespondingEthPeer;
import tech.pegasys.pantheon.ethereum.eth.manager.RespondingEthPeer.Responder;
import tech.pegasys.pantheon.ethereum.eth.manager.task.EthTask;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.concurrent.CompletableFuture;
@ -45,7 +44,7 @@ public abstract class AbstractMessageTaskTest<T, R> {
protected static Blockchain blockchain;
protected static ProtocolSchedule<Void> protocolSchedule;
protected static ProtocolContext<Void> protocolContext;
protected static LabelledMetric<OperationTimer> ethTasksTimer;
protected static MetricsSystem metricsSystem = new NoOpMetricsSystem();
protected EthProtocolManager ethProtocolManager;
protected EthContext ethContext;
protected AtomicBoolean peersDoTimeout;
@ -58,7 +57,6 @@ public abstract class AbstractMessageTaskTest<T, R> {
blockchain = blockchainSetupUtil.getBlockchain();
protocolSchedule = blockchainSetupUtil.getProtocolSchedule();
protocolContext = blockchainSetupUtil.getProtocolContext();
ethTasksTimer = NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
assert (blockchainSetupUtil.getMaxBlockNumber() >= 20L);
}

@ -83,7 +83,7 @@ public class AbstractEthTaskTest {
private final List<CompletableFuture<?>> subtasks;
private EthTaskWithMultipleSubtasks(final List<CompletableFuture<?>> subtasks) {
super(NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
super(new NoOpMetricsSystem());
this.subtasks = subtasks;
}

@ -47,7 +47,7 @@ public class GetBlockFromPeerTaskTest
@Override
protected EthTask<PeerTaskResult<Block>> createTask(final Block requestedData) {
return GetBlockFromPeerTask.create(
protocolSchedule, ethContext, requestedData.getHash(), ethTasksTimer);
protocolSchedule, ethContext, requestedData.getHash(), metricsSystem);
}
@Override

@ -19,7 +19,6 @@ import tech.pegasys.pantheon.ethereum.core.BlockBody;
import tech.pegasys.pantheon.ethereum.core.BlockHeader;
import tech.pegasys.pantheon.ethereum.eth.manager.ethtaskutils.PeerMessageTaskTest;
import tech.pegasys.pantheon.ethereum.eth.manager.task.AbstractPeerTask.PeerTaskResult;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.ArrayList;
import java.util.List;
@ -43,7 +42,7 @@ public class GetBodiesFromPeerTaskTest extends PeerMessageTaskTest<List<Block>>
final List<BlockHeader> headersToComplete =
requestedData.stream().map(Block::getHeader).collect(Collectors.toList());
return GetBodiesFromPeerTask.forHeaders(
protocolSchedule, ethContext, headersToComplete, NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
protocolSchedule, ethContext, headersToComplete, metricsSystem);
}
@Override

@ -61,7 +61,7 @@ public class GetHeadersFromPeerByHashTaskTest extends PeerMessageTaskTest<List<B
firstHeader.getHash(),
firstHeader.getNumber(),
requestedData.size(),
ethTasksTimer);
metricsSystem);
}
@Test
@ -109,7 +109,7 @@ public class GetHeadersFromPeerByHashTaskTest extends PeerMessageTaskTest<List<B
count,
skip,
reverse,
ethTasksTimer);
metricsSystem);
final AtomicReference<PeerTaskResult<List<BlockHeader>>> actualResult = new AtomicReference<>();
final AtomicBoolean done = new AtomicBoolean(false);
final CompletableFuture<PeerTaskResult<List<BlockHeader>>> future = task.run();

@ -18,7 +18,6 @@ import tech.pegasys.pantheon.ethereum.core.BlockHeader;
import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManagerTestUtil;
import tech.pegasys.pantheon.ethereum.eth.manager.RespondingEthPeer;
import tech.pegasys.pantheon.ethereum.eth.manager.ethtaskutils.PeerMessageTaskTest;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.ArrayList;
import java.util.List;
@ -55,11 +54,7 @@ public class GetHeadersFromPeerByNumberTaskTest extends PeerMessageTaskTest<List
final List<BlockHeader> requestedData) {
final BlockHeader firstHeader = requestedData.get(0);
return GetHeadersFromPeerByNumberTask.startingAtNumber(
protocolSchedule,
ethContext,
firstHeader.getNumber(),
requestedData.size(),
NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
protocolSchedule, ethContext, firstHeader.getNumber(), requestedData.size(), metricsSystem);
}
@Test
@ -100,13 +95,7 @@ public class GetHeadersFromPeerByNumberTaskTest extends PeerMessageTaskTest<List
// Execute task and wait for response
final AbstractGetHeadersFromPeerTask task =
new GetHeadersFromPeerByNumberTask(
protocolSchedule,
ethContext,
startNumber,
count,
skip,
reverse,
NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
protocolSchedule, ethContext, startNumber, count, skip, reverse, metricsSystem);
final AtomicReference<AbstractPeerTask.PeerTaskResult<List<BlockHeader>>> actualResult =
new AtomicReference<>();
final AtomicBoolean done = new AtomicBoolean(false);

@ -46,7 +46,7 @@ public class GetNodeDataFromPeerTaskTest extends PeerMessageTaskTest<Map<Hash, B
protected EthTask<PeerTaskResult<Map<Hash, BytesValue>>> createTask(
final Map<Hash, BytesValue> requestedData) {
final List<Hash> hashes = Lists.newArrayList(requestedData.keySet());
return GetNodeDataFromPeerTask.forHashes(ethContext, hashes, ethTasksTimer);
return GetNodeDataFromPeerTask.forHashes(ethContext, hashes, metricsSystem);
}
@Override

@ -41,7 +41,7 @@ public class GetReceiptsFromPeerTaskTest
@Override
protected EthTask<PeerTaskResult<Map<BlockHeader, List<TransactionReceipt>>>> createTask(
final Map<BlockHeader, List<TransactionReceipt>> requestedData) {
return GetReceiptsFromPeerTask.forHeaders(ethContext, requestedData.keySet(), ethTasksTimer);
return GetReceiptsFromPeerTask.forHeaders(ethContext, requestedData.keySet(), metricsSystem);
}
@Override

@ -17,8 +17,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManager;
import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManagerTestUtil;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.concurrent.CompletableFuture;
@ -31,20 +30,19 @@ import org.junit.Test;
public class WaitForPeerTaskTest {
private EthProtocolManager ethProtocolManager;
private EthContext ethContext;
private LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();;
@Before
public void setupTest() {
ethProtocolManager = EthProtocolManagerTestUtil.create();
ethContext = ethProtocolManager.ethContext();
ethTasksTimer = NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
}
@Test
public void completesWhenPeerConnects() throws ExecutionException, InterruptedException {
// Execute task and wait for response
final AtomicBoolean successful = new AtomicBoolean(false);
final EthTask<Void> task = WaitForPeerTask.create(ethContext, ethTasksTimer);
final EthTask<Void> task = WaitForPeerTask.create(ethContext, metricsSystem);
final CompletableFuture<Void> future = task.run();
future.whenComplete(
(result, error) -> {
@ -59,7 +57,7 @@ public class WaitForPeerTaskTest {
@Test
public void doesNotCompleteWhenNoPeerConnects() throws ExecutionException, InterruptedException {
final AtomicBoolean successful = new AtomicBoolean(false);
final EthTask<Void> task = WaitForPeerTask.create(ethContext, ethTasksTimer);
final EthTask<Void> task = WaitForPeerTask.create(ethContext, metricsSystem);
final CompletableFuture<Void> future = task.run();
future.whenComplete(
(result, error) -> {
@ -74,7 +72,7 @@ public class WaitForPeerTaskTest {
@Test
public void cancel() throws ExecutionException, InterruptedException {
// Execute task
final EthTask<Void> task = WaitForPeerTask.create(ethContext, ethTasksTimer);
final EthTask<Void> task = WaitForPeerTask.create(ethContext, metricsSystem);
final CompletableFuture<Void> future = task.run();
assertThat(future.isDone()).isFalse();

@ -17,8 +17,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext;
import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManager;
import tech.pegasys.pantheon.ethereum.eth.manager.EthProtocolManagerTestUtil;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.concurrent.CompletableFuture;
@ -31,20 +30,19 @@ import org.junit.Test;
public class WaitForPeersTaskTest {
private EthProtocolManager ethProtocolManager;
private EthContext ethContext;
private LabelledMetric<OperationTimer> ethTasksTimer;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();;
@Before
public void setupTest() {
ethProtocolManager = EthProtocolManagerTestUtil.create();
ethContext = ethProtocolManager.ethContext();
ethTasksTimer = NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
}
@Test
public void completesWhenPeersConnects() throws ExecutionException, InterruptedException {
// Execute task and wait for response
final AtomicBoolean successful = new AtomicBoolean(false);
final EthTask<Void> task = WaitForPeersTask.create(ethContext, 2, ethTasksTimer);
final EthTask<Void> task = WaitForPeersTask.create(ethContext, 2, metricsSystem);
final CompletableFuture<Void> future = task.run();
future.whenComplete(
(result, error) -> {
@ -60,7 +58,7 @@ public class WaitForPeersTaskTest {
@Test
public void doesNotCompleteWhenNoPeerConnects() throws ExecutionException, InterruptedException {
final AtomicBoolean successful = new AtomicBoolean(false);
final EthTask<Void> task = WaitForPeersTask.create(ethContext, 2, ethTasksTimer);
final EthTask<Void> task = WaitForPeersTask.create(ethContext, 2, metricsSystem);
final CompletableFuture<Void> future = task.run();
future.whenComplete(
(result, error) -> {
@ -76,7 +74,7 @@ public class WaitForPeersTaskTest {
public void doesNotCompleteWhenSomePeersConnects()
throws ExecutionException, InterruptedException {
final AtomicBoolean successful = new AtomicBoolean(false);
final EthTask<Void> task = WaitForPeersTask.create(ethContext, 2, ethTasksTimer);
final EthTask<Void> task = WaitForPeersTask.create(ethContext, 2, metricsSystem);
final CompletableFuture<Void> future = task.run();
future.whenComplete(
(result, error) -> {
@ -92,7 +90,7 @@ public class WaitForPeersTaskTest {
@Test
public void cancel() throws ExecutionException, InterruptedException {
// Execute task
final EthTask<Void> task = WaitForPeersTask.create(ethContext, 2, ethTasksTimer);
final EthTask<Void> task = WaitForPeersTask.create(ethContext, 2, metricsSystem);
final CompletableFuture<Void> future = task.run();
assertThat(future.isDone()).isFalse();

@ -56,7 +56,7 @@ public class DaoForkPeerValidatorTest {
new DaoForkPeerValidator(
ethProtocolManager.ethContext(),
MainnetProtocolSchedule.create(),
NoOpMetricsSystem.NO_OP_LABELLED_TIMER,
new NoOpMetricsSystem(),
daoBlockNumber,
0);
@ -88,7 +88,7 @@ public class DaoForkPeerValidatorTest {
new DaoForkPeerValidator(
ethProtocolManager.ethContext(),
MainnetProtocolSchedule.create(),
NoOpMetricsSystem.NO_OP_LABELLED_TIMER,
new NoOpMetricsSystem(),
daoBlockNumber,
0);
@ -116,7 +116,7 @@ public class DaoForkPeerValidatorTest {
new DaoForkPeerValidator(
ethProtocolManager.ethContext(),
MainnetProtocolSchedule.create(),
NoOpMetricsSystem.NO_OP_LABELLED_TIMER,
new NoOpMetricsSystem(),
daoBlockNumber,
0);
@ -145,7 +145,7 @@ public class DaoForkPeerValidatorTest {
new DaoForkPeerValidator(
ethProtocolManager.ethContext(),
MainnetProtocolSchedule.create(),
NoOpMetricsSystem.NO_OP_LABELLED_TIMER,
new NoOpMetricsSystem(),
daoBlockNumber,
0);
@ -184,7 +184,7 @@ public class DaoForkPeerValidatorTest {
new DaoForkPeerValidator(
ethProtocolManager.ethContext(),
MainnetProtocolSchedule.create(),
NoOpMetricsSystem.NO_OP_LABELLED_TIMER,
new NoOpMetricsSystem(),
daoBlockNumber,
buffer);

@ -41,8 +41,7 @@ import tech.pegasys.pantheon.ethereum.eth.messages.NewBlockMessage;
import tech.pegasys.pantheon.ethereum.eth.sync.state.PendingBlocks;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import tech.pegasys.pantheon.util.uint.UInt256;
@ -68,8 +67,7 @@ public class BlockPropagationManagerTest {
private SynchronizerConfiguration syncConfig;
private final PendingBlocks pendingBlocks = new PendingBlocks();
private SyncState syncState;
private final LabelledMetric<OperationTimer> ethTasksTimer =
NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
@BeforeClass
public static void setupSuite() {
@ -100,7 +98,7 @@ public class BlockPropagationManagerTest {
ethProtocolManager.ethContext(),
syncState,
pendingBlocks,
ethTasksTimer,
metricsSystem,
blockBroadcaster);
}
@ -474,7 +472,7 @@ public class BlockPropagationManagerTest {
ethProtocolManager.ethContext(),
syncState,
pendingBlocks,
ethTasksTimer,
metricsSystem,
blockBroadcaster);
final BlockDataGenerator gen = new BlockDataGenerator();
@ -555,7 +553,7 @@ public class BlockPropagationManagerTest {
ethContext,
syncState,
pendingBlocks,
ethTasksTimer,
metricsSystem,
blockBroadcaster);
blockchainUtil.importFirstBlocks(2);

@ -54,7 +54,7 @@ public class ChainHeadTrackerTest {
ethProtocolManager.ethContext(),
protocolSchedule,
trailingPeerLimiter,
NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
new NoOpMetricsSystem());
@Test
public void shouldRequestHeaderChainHeadWhenNewPeerConnects() {

@ -21,7 +21,6 @@ import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.when;
import static tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
import tech.pegasys.pantheon.ethereum.ProtocolContext;
import tech.pegasys.pantheon.ethereum.chain.MutableBlockchain;
@ -42,8 +41,8 @@ import tech.pegasys.pantheon.ethereum.mainnet.MainnetProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.PeerConnection.PeerNotConnected;
import tech.pegasys.pantheon.ethereum.worldstate.WorldStateArchive;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.concurrent.atomic.AtomicBoolean;
@ -66,7 +65,7 @@ public class CheckpointHeaderManagerTest {
private final AtomicBoolean timeout = new AtomicBoolean(false);
private final EthContext ethContext = EthContextTestUtil.createTestEthContext(timeout::get);
private final SyncState syncState = new SyncState(blockchain, ethContext.getEthPeers());
private final LabelledMetric<OperationTimer> ethTasksTimer = NO_OP_LABELLED_TIMER;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
private final EthPeer syncTargetPeer = mock(EthPeer.class);
private final RequestManager requestManager = new RequestManager(syncTargetPeer);
private SyncTarget syncTarget;
@ -82,7 +81,7 @@ public class CheckpointHeaderManagerTest {
ethContext,
syncState,
PROTOCOL_SCHEDULE,
ethTasksTimer);
metricsSystem);
@Before
public void setUp() {

@ -19,8 +19,6 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static tech.pegasys.pantheon.ethereum.eth.sync.fastsync.FastSyncError.CHAIN_TOO_SHORT;
import static tech.pegasys.pantheon.ethereum.eth.sync.fastsync.FastSyncState.EMPTY_SYNC_STATE;
import static tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem.NO_OP_LABELLED_COUNTER;
import static tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
import tech.pegasys.pantheon.ethereum.ProtocolContext;
import tech.pegasys.pantheon.ethereum.chain.MutableBlockchain;
@ -37,8 +35,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.SyncMode;
import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import tech.pegasys.pantheon.util.uint.UInt256;
import java.util.concurrent.CompletableFuture;
@ -57,7 +54,6 @@ public class FastSyncActionsTest {
.build();
private final FastSyncStateStorage fastSyncStateStorage = mock(FastSyncStateStorage.class);
private final LabelledMetric<OperationTimer> ethTasksTimer = NO_OP_LABELLED_TIMER;
private final AtomicInteger timeoutCount = new AtomicInteger(0);
private FastSyncActions<Void> fastSyncActions;
private EthProtocolManager ethProtocolManager;
@ -83,8 +79,7 @@ public class FastSyncActionsTest {
protocolContext,
ethContext,
new SyncState(blockchain, ethContext.getEthPeers()),
ethTasksTimer,
NO_OP_LABELLED_COUNTER);
new NoOpMetricsSystem());
}
@Test

@ -23,7 +23,6 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyZeroInteractions;
import static org.mockito.Mockito.when;
import static tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode.LIGHT_SKIP_DETACHED;
import static tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
import tech.pegasys.pantheon.ethereum.ProtocolContext;
import tech.pegasys.pantheon.ethereum.chain.MutableBlockchain;
@ -41,8 +40,8 @@ import tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSpec;
import tech.pegasys.pantheon.ethereum.worldstate.WorldStateArchive;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@ -81,12 +80,12 @@ public class FastSyncBlockHandlerTest {
new EthPeers(PROTOCOL_NAME),
new EthMessages(),
new DeterministicEthScheduler());
private final LabelledMetric<OperationTimer> ethTasksTimer = NO_OP_LABELLED_TIMER;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
private final ValidationPolicy validationPolicy = mock(ValidationPolicy.class);
private final FastSyncBlockHandler<Void> blockHandler =
new FastSyncBlockHandler<>(
protocolSchedule, protocolContext, ethContext, ethTasksTimer, validationPolicy);
protocolSchedule, protocolContext, ethContext, metricsSystem, validationPolicy);
@Before
public void setUp() {

@ -82,8 +82,7 @@ public class FastSyncChainDownloaderTest {
protocolContext,
ethContext,
syncState,
NoOpMetricsSystem.NO_OP_LABELLED_TIMER,
NoOpMetricsSystem.NO_OP_LABELLED_COUNTER,
new NoOpMetricsSystem(),
otherBlockchain.getBlockHeader(pivotBlockNumber).get());
}

@ -32,8 +32,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncTarget;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.List;
@ -55,7 +54,7 @@ public class FastSyncCheckpointHeaderManagerTest {
protected MutableBlockchain localBlockchain;
private BlockchainSetupUtil<Void> otherBlockchainSetup;
protected Blockchain otherBlockchain;
private LabelledMetric<OperationTimer> ethTasksTimer;
MetricsSystem metricsSystem = new NoOpMetricsSystem();;
private BlockHeader pivotBlockHeader;
private FastSyncCheckpointHeaderManager<Void> checkpointHeaderManager;
private RespondingEthPeer peer;
@ -75,8 +74,6 @@ public class FastSyncCheckpointHeaderManagerTest {
ethContext = ethProtocolManager.ethContext();
syncState = new SyncState(protocolContext.getBlockchain(), ethContext.getEthPeers());
ethTasksTimer = NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
otherBlockchainSetup.importFirstBlocks(30);
pivotBlockHeader = block(17);
@ -93,7 +90,7 @@ public class FastSyncCheckpointHeaderManagerTest {
ethContext,
syncState,
protocolSchedule,
ethTasksTimer,
metricsSystem,
pivotBlockHeader);
}

@ -15,7 +15,6 @@ package tech.pegasys.pantheon.ethereum.eth.sync.fastsync;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
import tech.pegasys.pantheon.ethereum.ProtocolContext;
import tech.pegasys.pantheon.ethereum.chain.Blockchain;
@ -27,8 +26,8 @@ import tech.pegasys.pantheon.ethereum.eth.manager.RespondingEthPeer;
import tech.pegasys.pantheon.ethereum.eth.manager.RespondingEthPeer.Responder;
import tech.pegasys.pantheon.ethereum.eth.manager.ethtaskutils.BlockchainSetupUtil;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import tech.pegasys.pantheon.util.bytes.BytesValue;
import java.util.Optional;
@ -45,7 +44,7 @@ public class PivotBlockRetrieverTest {
private ProtocolContext<Void> protocolContext;
private final LabelledMetric<OperationTimer> ethTasksTimer = NO_OP_LABELLED_TIMER;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
private final AtomicBoolean timeout = new AtomicBoolean(false);
private EthProtocolManager ethProtocolManager;
private MutableBlockchain blockchain;
@ -63,7 +62,7 @@ public class PivotBlockRetrieverTest {
blockchain, blockchainSetupUtil.getWorldArchive(), timeout::get);
pivotBlockRetriever =
new PivotBlockRetriever<>(
protocolSchedule, ethProtocolManager.ethContext(), ethTasksTimer, PIVOT_BLOCK_NUMBER);
protocolSchedule, ethProtocolManager.ethContext(), metricsSystem, PIVOT_BLOCK_NUMBER);
}
@Test

@ -43,8 +43,7 @@ import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import tech.pegasys.pantheon.util.uint.UInt256;
@ -71,7 +70,7 @@ public class FullSyncDownloaderTest {
protected MutableBlockchain localBlockchain;
private BlockchainSetupUtil<Void> otherBlockchainSetup;
protected Blockchain otherBlockchain;
private LabelledMetric<OperationTimer> ethTashsTimer;
private MetricsSystem metricsSystem = new NoOpMetricsSystem();
@Before
public void setupTest() {
@ -91,13 +90,11 @@ public class FullSyncDownloaderTest {
new EthScheduler(1, 1, 1, new NoOpMetricsSystem()));
ethContext = ethProtocolManager.ethContext();
syncState = new SyncState(protocolContext.getBlockchain(), ethContext.getEthPeers());
ethTashsTimer = NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
}
private FullSyncDownloader<?> downloader(final SynchronizerConfiguration syncConfig) {
return new FullSyncDownloader<>(
syncConfig, protocolSchedule, protocolContext, ethContext, syncState, ethTashsTimer);
syncConfig, protocolSchedule, protocolContext, ethContext, syncState, metricsSystem);
}
private FullSyncDownloader<?> downloader() {

@ -79,7 +79,7 @@ public class FullSyncTargetManagerTest {
protocolContext,
ethContext,
syncState,
NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
new NoOpMetricsSystem());
}
@Test

@ -48,11 +48,7 @@ public class CompleteBlocksTaskTest extends RetryingMessageTaskTest<List<Block>>
final List<BlockHeader> headersToComplete =
requestedData.stream().map(Block::getHeader).collect(Collectors.toList());
return CompleteBlocksTask.forHeaders(
protocolSchedule,
ethContext,
headersToComplete,
maxRetries,
NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
protocolSchedule, ethContext, headersToComplete, maxRetries, new NoOpMetricsSystem());
}
@Test

@ -32,8 +32,7 @@ import tech.pegasys.pantheon.ethereum.mainnet.MainnetBlockHashFunction;
import tech.pegasys.pantheon.ethereum.mainnet.MainnetProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.worldstate.WorldStateArchive;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import tech.pegasys.pantheon.util.uint.UInt256;
@ -56,8 +55,7 @@ import org.junit.runners.Parameterized.Parameters;
public class DetermineCommonAncestorTaskParameterizedTest {
private final ProtocolSchedule<Void> protocolSchedule = MainnetProtocolSchedule.create();
private static final BlockDataGenerator blockDataGenerator = new BlockDataGenerator();
private final LabelledMetric<OperationTimer> ethTasksTimer =
NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
private static Block genesisBlock;
private static MutableBlockchain localBlockchain;
@ -162,7 +160,7 @@ public class DetermineCommonAncestorTaskParameterizedTest {
ethContext,
respondingEthPeer.getEthPeer(),
headerRequestSize,
ethTasksTimer);
metricsSystem);
final CompletableFuture<BlockHeader> future = task.run();
respondingEthPeer.respondWhile(responder, () -> !future.isDone());

@ -42,8 +42,7 @@ import tech.pegasys.pantheon.ethereum.mainnet.MainnetProtocolSchedule;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.wire.messages.DisconnectMessage.DisconnectReason;
import tech.pegasys.pantheon.ethereum.worldstate.WorldStateArchive;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import tech.pegasys.pantheon.util.ExceptionUtils;
import tech.pegasys.pantheon.util.uint.UInt256;
@ -60,8 +59,7 @@ public class DetermineCommonAncestorTaskTest {
private final ProtocolSchedule<Void> protocolSchedule = MainnetProtocolSchedule.create();
private final BlockDataGenerator blockDataGenerator = new BlockDataGenerator();
private final LabelledMetric<OperationTimer> ethTasksTimer =
NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
private MutableBlockchain localBlockchain;
private final int defaultHeaderRequestSize = 10;
Block genesisBlock;
@ -117,7 +115,7 @@ public class DetermineCommonAncestorTaskTest {
ethContext,
respondingEthPeer.getEthPeer(),
defaultHeaderRequestSize,
ethTasksTimer);
metricsSystem);
final CompletableFuture<BlockHeader> future = task.run();
respondingEthPeer.respondWhile(responder, () -> !future.isDone());
@ -154,7 +152,7 @@ public class DetermineCommonAncestorTaskTest {
ethContext,
respondingEthPeer.getEthPeer(),
defaultHeaderRequestSize,
ethTasksTimer);
metricsSystem);
// Execute task and wait for response
final AtomicReference<Throwable> failure = new AtomicReference<>();
@ -225,7 +223,7 @@ public class DetermineCommonAncestorTaskTest {
ethContext,
respondingEthPeer.getEthPeer(),
defaultHeaderRequestSize,
ethTasksTimer);
metricsSystem);
final DetermineCommonAncestorTask<Void> spy = spy(task);
// Execute task
@ -282,7 +280,7 @@ public class DetermineCommonAncestorTaskTest {
ethContext,
respondingEthPeer.getEthPeer(),
defaultHeaderRequestSize,
ethTasksTimer);
metricsSystem);
final DetermineCommonAncestorTask<Void> spy = spy(task);
// Execute task
@ -354,7 +352,7 @@ public class DetermineCommonAncestorTaskTest {
ethContext,
respondingEthPeer.getEthPeer(),
defaultHeaderRequestSize,
ethTasksTimer);
metricsSystem);
final DetermineCommonAncestorTask<Void> spy = spy(task);
// Execute task
@ -386,7 +384,7 @@ public class DetermineCommonAncestorTaskTest {
ethContext,
peer,
defaultHeaderRequestSize,
ethTasksTimer);
metricsSystem);
final CompletableFuture<BlockHeader> result = task.run();
assertThat(result).isCompletedWithValue(genesisBlock.getHeader());

@ -59,7 +59,7 @@ public class DownloadHeaderSequenceTaskTest extends RetryingMessageTaskTest<List
referenceHeader,
requestedData.size(),
maxRetries,
ethTasksTimer);
metricsSystem);
}
@Test
@ -77,7 +77,7 @@ public class DownloadHeaderSequenceTaskTest extends RetryingMessageTaskTest<List
referenceHeader,
10,
maxRetries,
ethTasksTimer);
metricsSystem);
final CompletableFuture<List<BlockHeader>> future = task.run();
// Respond with only the reference header
@ -106,7 +106,7 @@ public class DownloadHeaderSequenceTaskTest extends RetryingMessageTaskTest<List
referenceHeader,
10,
maxRetries,
ethTasksTimer);
metricsSystem);
final CompletableFuture<List<BlockHeader>> future = task.run();
// Filter response to include only reference header and previous header

@ -21,7 +21,6 @@ import tech.pegasys.pantheon.ethereum.core.BlockHeaderTestFixture;
import tech.pegasys.pantheon.ethereum.core.TransactionReceipt;
import tech.pegasys.pantheon.ethereum.eth.manager.ethtaskutils.RetryingMessageTaskTest;
import tech.pegasys.pantheon.ethereum.eth.manager.task.EthTask;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.ArrayList;
import java.util.HashMap;
@ -50,7 +49,7 @@ public class GetReceiptsForHeadersTaskTest
final Map<BlockHeader, List<TransactionReceipt>> requestedData) {
final List<BlockHeader> headersToComplete = new ArrayList<>(requestedData.keySet());
return GetReceiptsForHeadersTask.forHeaders(
ethContext, headersToComplete, maxRetries, NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
ethContext, headersToComplete, maxRetries, metricsSystem);
}
@Test

@ -32,7 +32,6 @@ import tech.pegasys.pantheon.ethereum.eth.messages.BlockHeadersMessage;
import tech.pegasys.pantheon.ethereum.eth.messages.EthPV62;
import tech.pegasys.pantheon.ethereum.p2p.api.MessageData;
import tech.pegasys.pantheon.ethereum.p2p.wire.Capability;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.ArrayList;
import java.util.List;
@ -78,7 +77,7 @@ public class ImportBlocksTaskTest
ethContext,
firstBlock.getHeader(),
requestedData.size(),
NoOpMetricsSystem.NO_OP_LABELLED_TIMER);
metricsSystem);
}
@Override

@ -24,8 +24,7 @@ import tech.pegasys.pantheon.ethereum.eth.manager.ethtaskutils.BlockchainSetupUt
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.exceptions.InvalidBlockException;
import tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode;
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.OperationTimer;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem;
import java.util.Arrays;
@ -44,8 +43,7 @@ public class PersistBlockTaskTest {
private ProtocolSchedule<Void> protocolSchedule;
private ProtocolContext<Void> protocolContext;
private MutableBlockchain blockchain;
private final LabelledMetric<OperationTimer> ethTasksTimer =
NoOpMetricsSystem.NO_OP_LABELLED_TIMER;
private final MetricsSystem metricsSystem = new NoOpMetricsSystem();
@Before
public void setup() {
@ -66,7 +64,7 @@ public class PersistBlockTaskTest {
// Create task
final PersistBlockTask<Void> task =
PersistBlockTask.create(
protocolSchedule, protocolContext, nextBlock, HeaderValidationMode.FULL, ethTasksTimer);
protocolSchedule, protocolContext, nextBlock, HeaderValidationMode.FULL, metricsSystem);
final CompletableFuture<Block> result = task.run();
Awaitility.await().atMost(30, SECONDS).until(result::isDone);
@ -88,7 +86,7 @@ public class PersistBlockTaskTest {
// Create task
final PersistBlockTask<Void> task =
PersistBlockTask.create(
protocolSchedule, protocolContext, nextBlock, HeaderValidationMode.FULL, ethTasksTimer);
protocolSchedule, protocolContext, nextBlock, HeaderValidationMode.FULL, metricsSystem);
final CompletableFuture<Block> result = task.run();
Awaitility.await().atMost(30, SECONDS).until(result::isDone);
@ -116,7 +114,7 @@ public class PersistBlockTaskTest {
protocolContext,
nextBlocks,
HeaderValidationMode.FULL,
ethTasksTimer)
metricsSystem)
.get();
Awaitility.await().atMost(30, SECONDS).until(task::isDone);
@ -146,7 +144,7 @@ public class PersistBlockTaskTest {
protocolContext,
nextBlocks,
HeaderValidationMode.FULL,
ethTasksTimer)
metricsSystem)
.get();
Awaitility.await().atMost(30, SECONDS).until(task::isDone);
@ -175,7 +173,7 @@ public class PersistBlockTaskTest {
protocolContext,
nextBlocks,
HeaderValidationMode.FULL,
ethTasksTimer)
metricsSystem)
.get();
Awaitility.await().atMost(30, SECONDS).until(task::isDone);
@ -204,7 +202,7 @@ public class PersistBlockTaskTest {
protocolContext,
nextBlocks,
HeaderValidationMode.FULL,
ethTasksTimer)
metricsSystem)
.get();
Awaitility.await().atMost(30, SECONDS).until(task::isDone);
@ -236,7 +234,7 @@ public class PersistBlockTaskTest {
protocolContext,
nextBlocks,
HeaderValidationMode.FULL,
ethTasksTimer)
metricsSystem)
.get();
Awaitility.await().atMost(30, SECONDS).until(task::isDone);
@ -265,7 +263,7 @@ public class PersistBlockTaskTest {
protocolContext,
nextBlocks,
HeaderValidationMode.FULL,
ethTasksTimer)
metricsSystem)
.get();
Awaitility.await().atMost(30, SECONDS).until(task::isDone);
@ -296,7 +294,7 @@ public class PersistBlockTaskTest {
protocolContext,
nextBlocks,
HeaderValidationMode.FULL,
ethTasksTimer)
metricsSystem)
.get();
Awaitility.await().atMost(30, SECONDS).until(task::isDone);
@ -319,7 +317,7 @@ public class PersistBlockTaskTest {
// Create task
final PersistBlockTask<Void> task =
PersistBlockTask.create(
protocolSchedule, protocolContext, nextBlock, HeaderValidationMode.FULL, ethTasksTimer);
protocolSchedule, protocolContext, nextBlock, HeaderValidationMode.FULL, metricsSystem);
task.cancel();
final CompletableFuture<Block> result = task.run();
@ -339,7 +337,7 @@ public class PersistBlockTaskTest {
// Create task
final PersistBlockTask<Void> task =
PersistBlockTask.create(
protocolSchedule, protocolContext, nextBlock, HeaderValidationMode.FULL, ethTasksTimer);
protocolSchedule, protocolContext, nextBlock, HeaderValidationMode.FULL, metricsSystem);
final PersistBlockTask<Void> taskSpy = Mockito.spy(task);
Mockito.doNothing().when(taskSpy).executeTaskTimed();

@ -80,7 +80,7 @@ public class PipelinedImportChainSegmentTaskTest
protocolContext,
blocks,
HeaderValidationMode.SKIP_DETACHED,
ethTasksTimer)
metricsSystem)
.get();
}
@ -100,7 +100,7 @@ public class PipelinedImportChainSegmentTaskTest
modifiedContext,
ethContext,
1,
ethTasksTimer,
metricsSystem,
createBlockHandler(),
DETACHED_ONLY_VALIDATION_POLICY,
previousBlock.getHeader(),
@ -136,7 +136,7 @@ public class PipelinedImportChainSegmentTaskTest
modifiedContext,
ethContext,
1,
ethTasksTimer,
metricsSystem,
createBlockHandler(),
DETACHED_ONLY_VALIDATION_POLICY,
firstBlock.getHeader(),
@ -188,7 +188,7 @@ public class PipelinedImportChainSegmentTaskTest
modifiedContext,
ethContext,
1,
ethTasksTimer,
metricsSystem,
createBlockHandler(),
DETACHED_ONLY_VALIDATION_POLICY,
fakeFirstBlock.getHeader(),
@ -244,7 +244,7 @@ public class PipelinedImportChainSegmentTaskTest
modifiedContext,
ethContext,
1,
ethTasksTimer,
metricsSystem,
createBlockHandler(),
DETACHED_ONLY_VALIDATION_POLICY,
checkpointHeaders);
@ -307,7 +307,7 @@ public class PipelinedImportChainSegmentTaskTest
modifiedContext,
ethContext,
2,
ethTasksTimer,
metricsSystem,
createBlockHandler(),
DETACHED_ONLY_VALIDATION_POLICY,
checkpointHeaders);
@ -374,7 +374,7 @@ public class PipelinedImportChainSegmentTaskTest
modifiedContext,
ethContext,
3,
ethTasksTimer,
metricsSystem,
createBlockHandler(),
DETACHED_ONLY_VALIDATION_POLICY,
checkpointHeaders);
@ -432,7 +432,7 @@ public class PipelinedImportChainSegmentTaskTest
}
private FullSyncBlockHandler<Void> createBlockHandler() {
return new FullSyncBlockHandler<>(protocolSchedule, protocolContext, ethContext, ethTasksTimer);
return new FullSyncBlockHandler<>(protocolSchedule, protocolContext, ethContext, metricsSystem);
}
private static class CountingResponder implements Responder {

@ -36,7 +36,7 @@ public class RetryingGetHeaderFromPeerByNumberTaskTest
@Override
protected EthTask<List<BlockHeader>> createTask(final List<BlockHeader> requestedData) {
return RetryingGetHeaderFromPeerByNumberTask.forSingleNumber(
protocolSchedule, ethContext, ethTasksTimer, PIVOT_BLOCK_NUMBER, maxRetries);
protocolSchedule, ethContext, metricsSystem, PIVOT_BLOCK_NUMBER, maxRetries);
}
@Test

@ -863,7 +863,6 @@ public class WorldStateDownloaderTest {
config.getWorldStateHashCountPerRequest(),
config.getWorldStateRequestParallelism(),
config.getWorldStateRequestMaxRetries(),
NoOpMetricsSystem.NO_OP_LABELLED_TIMER,
new NoOpMetricsSystem());
}

@ -49,6 +49,11 @@ public class PrometheusMetricsSystem implements MetricsSystem {
private static final String PANTHEON_PREFIX = "pantheon_";
private final Map<MetricCategory, Collection<Collector>> collectors = new ConcurrentHashMap<>();
private final CollectorRegistry registry = new CollectorRegistry(true);
private final Map<String, LabelledMetric<tech.pegasys.pantheon.metrics.Counter>> cachedCounters =
new ConcurrentHashMap<>();
private final Map<String, LabelledMetric<tech.pegasys.pantheon.metrics.OperationTimer>>
cachedTimers = new ConcurrentHashMap<>();
private final Map<String, Collector> cachedGauges = new ConcurrentHashMap<>();
PrometheusMetricsSystem() {}
@ -73,12 +78,14 @@ public class PrometheusMetricsSystem implements MetricsSystem {
final String name,
final String help,
final String... labelNames) {
final Counter counter =
Counter.build(convertToPrometheusName(category, name), help)
.labelNames(labelNames)
.create();
addCollector(category, counter);
return new PrometheusCounter(counter);
final String metricName = convertToPrometheusName(category, name);
return cachedCounters.computeIfAbsent(
metricName,
(k) -> {
final Counter counter = Counter.build(metricName, help).labelNames(labelNames).create();
addCollector(category, counter);
return new PrometheusCounter(counter);
});
}
@Override
@ -87,18 +94,23 @@ public class PrometheusMetricsSystem implements MetricsSystem {
final String name,
final String help,
final String... labelNames) {
final Summary summary =
Summary.build(convertToPrometheusName(category, name), help)
.quantile(0.2, 0.02)
.quantile(0.5, 0.05)
.quantile(0.8, 0.02)
.quantile(0.95, 0.005)
.quantile(0.99, 0.001)
.quantile(1.0, 0)
.labelNames(labelNames)
.create();
addCollector(category, summary);
return new PrometheusTimer(summary);
final String metricName = convertToPrometheusName(category, name);
return cachedTimers.computeIfAbsent(
metricName,
(k) -> {
final Summary summary =
Summary.build(metricName, help)
.quantile(0.2, 0.02)
.quantile(0.5, 0.05)
.quantile(0.8, 0.02)
.quantile(0.95, 0.005)
.quantile(0.99, 0.001)
.quantile(1.0, 0)
.labelNames(labelNames)
.create();
addCollector(category, summary);
return new PrometheusTimer(summary);
});
}
@Override
@ -108,7 +120,13 @@ public class PrometheusMetricsSystem implements MetricsSystem {
final String help,
final Supplier<Double> valueSupplier) {
final String metricName = convertToPrometheusName(category, name);
addCollector(category, new CurrentValueCollector(metricName, help, valueSupplier));
cachedGauges.computeIfAbsent(
metricName,
(k) -> {
Collector collector = new CurrentValueCollector(metricName, help, valueSupplier);
addCollector(category, collector);
return collector;
});
}
private void addCollector(final MetricCategory category, final Collector metric) {

@ -53,6 +53,23 @@ public class PrometheusMetricsSystemTest {
.containsExactly(new Observation(PEERS, "connected", 2d, emptyList()));
}
@Test
public void shouldHandleDuplicateCounterCreation() {
final LabelledMetric<Counter> counter1 =
metricsSystem.createLabelledCounter(PEERS, "connected", "Some help string");
final LabelledMetric<Counter> counter2 =
metricsSystem.createLabelledCounter(PEERS, "connected", "Some help string");
assertThat(counter1).isEqualTo(counter2);
counter1.labels().inc();
assertThat(metricsSystem.getMetrics())
.containsExactly(new Observation(PEERS, "connected", 1d, emptyList()));
counter2.labels().inc();
assertThat(metricsSystem.getMetrics())
.containsExactly(new Observation(PEERS, "connected", 2d, emptyList()));
}
@Test
public void shouldCreateSeparateObservationsForEachCounterLabelValue() {
final LabelledMetric<Counter> counter =
@ -101,6 +118,15 @@ public class PrometheusMetricsSystemTest {
new Observation(RPC, "request", null, singletonList("count")));
}
@Test
public void shouldHandleDuplicateTimerCreation() {
final LabelledMetric<OperationTimer> timer1 =
metricsSystem.createLabelledTimer(RPC, "request", "Some help");
final LabelledMetric<OperationTimer> timer2 =
metricsSystem.createLabelledTimer(RPC, "request", "Some help");
assertThat(timer1).isEqualTo(timer2);
}
@Test
public void shouldCreateObservationsFromTimerWithLabels() {
final LabelledMetric<OperationTimer> timer =
@ -128,4 +154,13 @@ public class PrometheusMetricsSystemTest {
assertThat(metricsSystem.getMetrics())
.containsExactlyInAnyOrder(new Observation(JVM, "myValue", 7d, emptyList()));
}
@Test
public void shouldHandleDuplicateGaugeDeclaration() {
metricsSystem.createGauge(JVM, "myValue", "Help", () -> 7d);
metricsSystem.createGauge(JVM, "myValue", "Help", () -> 7d);
assertThat(metricsSystem.getMetrics())
.containsExactlyInAnyOrder(new Observation(JVM, "myValue", 7d, emptyList()));
}
}

@ -40,10 +40,7 @@ import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule;
import tech.pegasys.pantheon.ethereum.p2p.api.ProtocolManager;
import tech.pegasys.pantheon.ethereum.p2p.config.SubProtocolConfiguration;
import tech.pegasys.pantheon.ethereum.storage.StorageProvider;
import tech.pegasys.pantheon.metrics.LabelledMetric;
import tech.pegasys.pantheon.metrics.MetricCategory;
import tech.pegasys.pantheon.metrics.MetricsSystem;
import tech.pegasys.pantheon.metrics.OperationTimer;
import java.io.IOException;
import java.nio.file.Path;
@ -123,9 +120,6 @@ public class MainnetPantheonController implements PantheonController<Void> {
metricsSystem);
final SyncState syncState =
new SyncState(blockchain, ethProtocolManager.ethContext().getEthPeers());
final LabelledMetric<OperationTimer> ethTasksTimer =
metricsSystem.createLabelledTimer(
MetricCategory.SYNCHRONIZER, "task", "Internal processing tasks", "taskName");
final Synchronizer synchronizer =
new DefaultSynchronizer<>(
syncConfig,
@ -135,8 +129,7 @@ public class MainnetPantheonController implements PantheonController<Void> {
ethProtocolManager.ethContext(),
syncState,
dataDirectory,
metricsSystem,
ethTasksTimer);
metricsSystem);
OptionalLong daoBlock = genesisConfig.getConfigOptions().getDaoForkBlock();
if (daoBlock.isPresent()) {
@ -144,7 +137,7 @@ public class MainnetPantheonController implements PantheonController<Void> {
EthContext ethContext = ethProtocolManager.ethContext();
DaoForkPeerValidator daoForkPeerValidator =
new DaoForkPeerValidator(
ethContext, protocolSchedule, ethTasksTimer, daoBlock.getAsLong());
ethContext, protocolSchedule, metricsSystem, daoBlock.getAsLong());
PeerValidatorRunner.runValidator(ethContext, daoForkPeerValidator);
}

Loading…
Cancel
Save