mirror of https://github.com/hyperledger/besu
Pipeline based full sync (#1291)
Introduce a pipeline based full sync process. Currently toggled off but can be enabled via a --X option. Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>pull/2/head
Adrian Sutton
6 years ago
committed by
GitHub
parent
fbf5db7828
commit
a4b473ad6f
@ -0,0 +1,33 @@ |
||||
/* |
||||
* Copyright 2019 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.ethereum.eth.sync.fullsync; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.core.Block; |
||||
import tech.pegasys.pantheon.ethereum.core.Transaction; |
||||
|
||||
import java.util.List; |
||||
import java.util.function.Function; |
||||
import java.util.stream.Stream; |
||||
|
||||
public class ExtractTxSignaturesTask implements Function<List<Block>, Stream<Block>> { |
||||
|
||||
@Override |
||||
public Stream<Block> apply(final List<Block> blocks) { |
||||
return blocks.stream().map(this::extractSignatures); |
||||
} |
||||
|
||||
private Block extractSignatures(final Block block) { |
||||
block.getBody().getTransactions().forEach(Transaction::getSender); |
||||
return block; |
||||
} |
||||
} |
||||
@ -0,0 +1,50 @@ |
||||
/* |
||||
* Copyright 2019 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.ethereum.eth.sync.fullsync; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.ProtocolContext; |
||||
import tech.pegasys.pantheon.ethereum.core.Block; |
||||
import tech.pegasys.pantheon.ethereum.core.BlockImporter; |
||||
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 java.util.function.Consumer; |
||||
|
||||
import org.apache.logging.log4j.LogManager; |
||||
import org.apache.logging.log4j.Logger; |
||||
|
||||
public class FullImportBlockStep<C> implements Consumer<Block> { |
||||
private static final Logger LOG = LogManager.getLogger(); |
||||
private final ProtocolSchedule<C> protocolSchedule; |
||||
private final ProtocolContext<C> protocolContext; |
||||
|
||||
public FullImportBlockStep( |
||||
final ProtocolSchedule<C> protocolSchedule, final ProtocolContext<C> protocolContext) { |
||||
this.protocolSchedule = protocolSchedule; |
||||
this.protocolContext = protocolContext; |
||||
} |
||||
|
||||
@Override |
||||
public void accept(final Block block) { |
||||
final long blockNumber = block.getHeader().getNumber(); |
||||
final BlockImporter<C> importer = |
||||
protocolSchedule.getByBlockNumber(blockNumber).getBlockImporter(); |
||||
if (!importer.importBlock(protocolContext, block, HeaderValidationMode.SKIP_DETACHED)) { |
||||
throw new InvalidBlockException("Failed to import block", blockNumber, block.getHash()); |
||||
} |
||||
if (blockNumber % 200 == 0) { |
||||
LOG.info("Import reached block {}", blockNumber); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,119 @@ |
||||
/* |
||||
* Copyright 2019 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.ethereum.eth.sync.fullsync; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.ProtocolContext; |
||||
import tech.pegasys.pantheon.ethereum.core.BlockHeader; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.CheckpointHeaderFetcher; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.CheckpointHeaderValidationStep; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.CheckpointRangeSource; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.DownloadBodiesStep; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.DownloadHeadersStep; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.DownloadPipelineFactory; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.ValidationPolicy; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncTarget; |
||||
import tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode; |
||||
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule; |
||||
import tech.pegasys.pantheon.metrics.MetricCategory; |
||||
import tech.pegasys.pantheon.metrics.MetricsSystem; |
||||
import tech.pegasys.pantheon.services.pipeline.Pipeline; |
||||
import tech.pegasys.pantheon.services.pipeline.PipelineBuilder; |
||||
|
||||
import java.util.Optional; |
||||
|
||||
public class FullSyncDownloadPipelineFactory<C> implements DownloadPipelineFactory { |
||||
|
||||
private final SynchronizerConfiguration syncConfig; |
||||
private final ProtocolSchedule<C> protocolSchedule; |
||||
private final ProtocolContext<C> protocolContext; |
||||
private final EthContext ethContext; |
||||
private final MetricsSystem metricsSystem; |
||||
private final ValidationPolicy detachedValidationPolicy = |
||||
() -> HeaderValidationMode.DETACHED_ONLY; |
||||
private final BetterSyncTargetEvaluator betterSyncTargetEvaluator; |
||||
|
||||
public FullSyncDownloadPipelineFactory( |
||||
final SynchronizerConfiguration syncConfig, |
||||
final ProtocolSchedule<C> protocolSchedule, |
||||
final ProtocolContext<C> protocolContext, |
||||
final EthContext ethContext, |
||||
final MetricsSystem metricsSystem) { |
||||
this.syncConfig = syncConfig; |
||||
this.protocolSchedule = protocolSchedule; |
||||
this.protocolContext = protocolContext; |
||||
this.ethContext = ethContext; |
||||
this.metricsSystem = metricsSystem; |
||||
betterSyncTargetEvaluator = new BetterSyncTargetEvaluator(syncConfig, ethContext.getEthPeers()); |
||||
} |
||||
|
||||
@Override |
||||
public Pipeline<?> createDownloadPipelineForSyncTarget(final SyncTarget target) { |
||||
final int downloaderParallelism = syncConfig.downloaderParallelism(); |
||||
final int headerRequestSize = syncConfig.downloaderHeaderRequestSize(); |
||||
final int singleHeaderBufferSize = headerRequestSize * downloaderParallelism; |
||||
final CheckpointRangeSource checkpointRangeSource = |
||||
new CheckpointRangeSource( |
||||
new CheckpointHeaderFetcher( |
||||
syncConfig, protocolSchedule, ethContext, Optional.empty(), metricsSystem), |
||||
this::shouldContinueDownloadingFromPeer, |
||||
ethContext.getScheduler(), |
||||
target.peer(), |
||||
target.commonAncestor(), |
||||
syncConfig.downloaderCheckpointTimeoutsPermitted()); |
||||
final DownloadHeadersStep<C> downloadHeadersStep = |
||||
new DownloadHeadersStep<>( |
||||
protocolSchedule, |
||||
protocolContext, |
||||
ethContext, |
||||
detachedValidationPolicy, |
||||
headerRequestSize, |
||||
metricsSystem); |
||||
final CheckpointHeaderValidationStep<C> validateHeadersJoinUpStep = |
||||
new CheckpointHeaderValidationStep<>( |
||||
protocolSchedule, protocolContext, detachedValidationPolicy); |
||||
final DownloadBodiesStep<C> downloadBodiesStep = |
||||
new DownloadBodiesStep<>(protocolSchedule, ethContext, metricsSystem); |
||||
final ExtractTxSignaturesTask extractTxSignaturesTask = new ExtractTxSignaturesTask(); |
||||
final FullImportBlockStep<C> importBlockStep = |
||||
new FullImportBlockStep<>(protocolSchedule, protocolContext); |
||||
|
||||
return PipelineBuilder.createPipelineFrom( |
||||
"fetchCheckpoints", |
||||
checkpointRangeSource, |
||||
downloaderParallelism, |
||||
metricsSystem.createLabelledCounter( |
||||
MetricCategory.SYNCHRONIZER, |
||||
"chain_download_pipeline_processed_total", |
||||
"Number of entries process by each chain download pipeline stage", |
||||
"step", |
||||
"action")) |
||||
.thenProcessAsyncOrdered("downloadHeaders", downloadHeadersStep, downloaderParallelism) |
||||
.thenFlatMap("validateHeadersJoin", validateHeadersJoinUpStep, singleHeaderBufferSize) |
||||
.inBatches(headerRequestSize) |
||||
.thenProcessAsyncOrdered("downloadBodies", downloadBodiesStep, downloaderParallelism) |
||||
.thenFlatMap("extractTxSignatures", extractTxSignaturesTask, singleHeaderBufferSize) |
||||
.andFinishWith("importBlock", importBlockStep); |
||||
} |
||||
|
||||
private boolean shouldContinueDownloadingFromPeer( |
||||
final EthPeer peer, final BlockHeader lastCheckpointHeader) { |
||||
final boolean caughtUpToPeer = |
||||
peer.chainState().getEstimatedHeight() <= lastCheckpointHeader.getNumber(); |
||||
return !peer.isDisconnected() |
||||
&& !caughtUpToPeer |
||||
&& !betterSyncTargetEvaluator.shouldSwitchSyncTarget(peer); |
||||
} |
||||
} |
||||
@ -0,0 +1,73 @@ |
||||
/* |
||||
* Copyright 2019 ConsenSys AG. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
||||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
||||
* specific language governing permissions and limitations under the License. |
||||
*/ |
||||
package tech.pegasys.pantheon.ethereum.eth.sync.fullsync; |
||||
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
||||
import static org.mockito.ArgumentMatchers.anyLong; |
||||
import static org.mockito.Mockito.verify; |
||||
import static org.mockito.Mockito.when; |
||||
import static tech.pegasys.pantheon.ethereum.mainnet.HeaderValidationMode.SKIP_DETACHED; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.ProtocolContext; |
||||
import tech.pegasys.pantheon.ethereum.core.Block; |
||||
import tech.pegasys.pantheon.ethereum.core.BlockDataGenerator; |
||||
import tech.pegasys.pantheon.ethereum.core.BlockImporter; |
||||
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 org.junit.Before; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.mockito.Mock; |
||||
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
||||
@RunWith(MockitoJUnitRunner.class) |
||||
public class FullImportBlockStepTest { |
||||
|
||||
@Mock private ProtocolSchedule<Void> protocolSchedule; |
||||
@Mock private ProtocolSpec<Void> protocolSpec; |
||||
@Mock private ProtocolContext<Void> protocolContext; |
||||
@Mock private BlockImporter<Void> blockImporter; |
||||
private final BlockDataGenerator gen = new BlockDataGenerator(); |
||||
|
||||
private FullImportBlockStep<Void> importBlocksStep; |
||||
|
||||
@Before |
||||
public void setUp() { |
||||
when(protocolSchedule.getByBlockNumber(anyLong())).thenReturn(protocolSpec); |
||||
when(protocolSpec.getBlockImporter()).thenReturn(blockImporter); |
||||
|
||||
importBlocksStep = new FullImportBlockStep<>(protocolSchedule, protocolContext); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldImportBlock() { |
||||
final Block block = gen.block(); |
||||
|
||||
when(blockImporter.importBlock(protocolContext, block, SKIP_DETACHED)).thenReturn(true); |
||||
importBlocksStep.accept(block); |
||||
|
||||
verify(protocolSchedule).getByBlockNumber(block.getHeader().getNumber()); |
||||
verify(blockImporter).importBlock(protocolContext, block, SKIP_DETACHED); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldThrowExceptionWhenValidationFails() { |
||||
final Block block = gen.block(); |
||||
|
||||
when(blockImporter.importBlock(protocolContext, block, SKIP_DETACHED)).thenReturn(false); |
||||
assertThatThrownBy(() -> importBlocksStep.accept(block)) |
||||
.isInstanceOf(InvalidBlockException.class); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue