mirror of https://github.com/hyperledger/besu
[NC-2138] Extract out generic parts of Downloader (#659)
Separate the management of sync target and actual import from the rest of the Downloader logic in preparation for introducing a fast sync chain downloader. Signed-off-by: Adrian Sutton <adrian.sutton@consensys.net>pull/2/head
Adrian Sutton
6 years ago
committed by
GitHub
parent
de9e7f40fd
commit
3249bbfbd3
@ -0,0 +1,123 @@ |
||||
/* |
||||
* 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; |
||||
|
||||
import tech.pegasys.pantheon.ethereum.ProtocolContext; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer; |
||||
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.eth.sync.tasks.WaitForPeerTask; |
||||
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule; |
||||
import tech.pegasys.pantheon.metrics.LabelledMetric; |
||||
import tech.pegasys.pantheon.metrics.OperationTimer; |
||||
|
||||
import java.time.Duration; |
||||
import java.util.Optional; |
||||
import java.util.concurrent.CompletableFuture; |
||||
|
||||
import org.apache.logging.log4j.LogManager; |
||||
import org.apache.logging.log4j.Logger; |
||||
|
||||
public abstract class SyncTargetManager<C> { |
||||
|
||||
private static final Logger LOG = LogManager.getLogger(); |
||||
private volatile long syncTargetDisconnectListenerId; |
||||
private volatile boolean syncTargetDisconnected = false; |
||||
private final SynchronizerConfiguration config; |
||||
private final ProtocolSchedule<C> protocolSchedule; |
||||
private final ProtocolContext<C> protocolContext; |
||||
private final EthContext ethContext; |
||||
private final SyncState syncState; |
||||
private final LabelledMetric<OperationTimer> ethTasksTimer; |
||||
|
||||
public SyncTargetManager( |
||||
final SynchronizerConfiguration config, |
||||
final ProtocolSchedule<C> protocolSchedule, |
||||
final ProtocolContext<C> protocolContext, |
||||
final EthContext ethContext, |
||||
final SyncState syncState, |
||||
final LabelledMetric<OperationTimer> ethTasksTimer) { |
||||
this.config = config; |
||||
this.protocolSchedule = protocolSchedule; |
||||
this.protocolContext = protocolContext; |
||||
this.ethContext = ethContext; |
||||
this.syncState = syncState; |
||||
this.ethTasksTimer = ethTasksTimer; |
||||
} |
||||
|
||||
public CompletableFuture<SyncTarget> findSyncTarget() { |
||||
final Optional<SyncTarget> maybeSyncTarget = syncState.syncTarget(); |
||||
if (maybeSyncTarget.isPresent()) { |
||||
// Nothing to do
|
||||
return CompletableFuture.completedFuture(maybeSyncTarget.get()); |
||||
} |
||||
|
||||
final Optional<EthPeer> maybeBestPeer = selectBestAvailableSyncTarget(); |
||||
if (maybeBestPeer.isPresent()) { |
||||
final EthPeer bestPeer = maybeBestPeer.get(); |
||||
return DetermineCommonAncestorTask.create( |
||||
protocolSchedule, |
||||
protocolContext, |
||||
ethContext, |
||||
bestPeer, |
||||
config.downloaderHeaderRequestSize(), |
||||
ethTasksTimer) |
||||
.run() |
||||
.handle((r, t) -> r) |
||||
.thenCompose( |
||||
(target) -> { |
||||
if (target == null) { |
||||
return waitForPeerAndThenSetSyncTarget(); |
||||
} |
||||
final SyncTarget syncTarget = syncState.setSyncTarget(bestPeer, target); |
||||
LOG.info( |
||||
"Found common ancestor with peer {} at block {}", bestPeer, target.getNumber()); |
||||
syncTargetDisconnectListenerId = |
||||
bestPeer.subscribeDisconnect(this::onSyncTargetPeerDisconnect); |
||||
return CompletableFuture.completedFuture(syncTarget); |
||||
}); |
||||
} else { |
||||
return waitForPeerAndThenSetSyncTarget(); |
||||
} |
||||
} |
||||
|
||||
protected abstract Optional<EthPeer> selectBestAvailableSyncTarget(); |
||||
|
||||
private CompletableFuture<SyncTarget> waitForPeerAndThenSetSyncTarget() { |
||||
return waitForNewPeer().handle((r, t) -> r).thenCompose((r) -> findSyncTarget()); |
||||
} |
||||
|
||||
private CompletableFuture<?> waitForNewPeer() { |
||||
return ethContext |
||||
.getScheduler() |
||||
.timeout(WaitForPeerTask.create(ethContext, ethTasksTimer), Duration.ofSeconds(5)); |
||||
} |
||||
|
||||
private void onSyncTargetPeerDisconnect(final EthPeer ethPeer) { |
||||
LOG.info("Sync target disconnected: {}", ethPeer); |
||||
syncTargetDisconnected = true; |
||||
} |
||||
|
||||
public boolean isSyncTargetDisconnected() { |
||||
return syncTargetDisconnected; |
||||
} |
||||
|
||||
public void clearSyncTarget(final SyncTarget syncTarget) { |
||||
syncTarget.peer().unsubscribeDisconnect(syncTargetDisconnectListenerId); |
||||
syncTargetDisconnected = false; |
||||
} |
||||
|
||||
public abstract boolean shouldSwitchSyncTarget(final SyncTarget currentTarget); |
||||
} |
||||
@ -0,0 +1,105 @@ |
||||
/* |
||||
* 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.BlockHeader; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.AbstractPeerTask.PeerTaskResult; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.ChainDownloader; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.SynchronizerConfiguration; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.state.SyncState; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.ImportBlocksTask; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.tasks.PipelinedImportChainSegmentTask; |
||||
import tech.pegasys.pantheon.ethereum.mainnet.ProtocolSchedule; |
||||
import tech.pegasys.pantheon.metrics.LabelledMetric; |
||||
import tech.pegasys.pantheon.metrics.OperationTimer; |
||||
|
||||
import java.util.Deque; |
||||
import java.util.List; |
||||
import java.util.concurrent.CompletableFuture; |
||||
|
||||
import com.google.common.annotations.VisibleForTesting; |
||||
import com.google.common.collect.Lists; |
||||
|
||||
public class FullSyncDownloader<C> { |
||||
private final ChainDownloader<C> chainDownloader; |
||||
private final SynchronizerConfiguration config; |
||||
private final ProtocolSchedule<C> protocolSchedule; |
||||
private final ProtocolContext<C> protocolContext; |
||||
private final EthContext ethContext; |
||||
private final LabelledMetric<OperationTimer> ethTasksTimer; |
||||
|
||||
public FullSyncDownloader( |
||||
final SynchronizerConfiguration config, |
||||
final ProtocolSchedule<C> protocolSchedule, |
||||
final ProtocolContext<C> protocolContext, |
||||
final EthContext ethContext, |
||||
final SyncState syncState, |
||||
final LabelledMetric<OperationTimer> ethTasksTimer) { |
||||
this.config = config; |
||||
this.protocolSchedule = protocolSchedule; |
||||
this.protocolContext = protocolContext; |
||||
this.ethContext = ethContext; |
||||
this.ethTasksTimer = ethTasksTimer; |
||||
chainDownloader = |
||||
new ChainDownloader<>( |
||||
config, |
||||
protocolSchedule, |
||||
protocolContext, |
||||
ethContext, |
||||
syncState, |
||||
ethTasksTimer, |
||||
new FullSyncTargetManager<>( |
||||
config, protocolSchedule, protocolContext, ethContext, syncState, ethTasksTimer), |
||||
this::importBlocksForCheckpoints); |
||||
} |
||||
|
||||
public void start() { |
||||
chainDownloader.start(); |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
CompletableFuture<?> getCurrentTask() { |
||||
return chainDownloader.getCurrentTask(); |
||||
} |
||||
|
||||
private CompletableFuture<List<Block>> importBlocksForCheckpoints( |
||||
final Deque<BlockHeader> checkpointHeaders) { |
||||
final CompletableFuture<List<Block>> importedBlocks; |
||||
if (checkpointHeaders.size() < 2) { |
||||
// Download blocks without constraining the end block
|
||||
final ImportBlocksTask<C> importTask = |
||||
ImportBlocksTask.fromHeader( |
||||
protocolSchedule, |
||||
protocolContext, |
||||
ethContext, |
||||
checkpointHeaders.getFirst(), |
||||
config.downloaderChainSegmentSize(), |
||||
ethTasksTimer); |
||||
importedBlocks = importTask.run().thenApply(PeerTaskResult::getResult); |
||||
} else { |
||||
final PipelinedImportChainSegmentTask<C> importTask = |
||||
PipelinedImportChainSegmentTask.forCheckpoints( |
||||
protocolSchedule, |
||||
protocolContext, |
||||
ethContext, |
||||
config.downloaderParallelism(), |
||||
ethTasksTimer, |
||||
Lists.newArrayList(checkpointHeaders)); |
||||
importedBlocks = importTask.run(); |
||||
} |
||||
return importedBlocks; |
||||
} |
||||
} |
||||
@ -0,0 +1,107 @@ |
||||
/* |
||||
* 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.eth.manager.ChainState; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.EthContext; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.EthPeer; |
||||
import tech.pegasys.pantheon.ethereum.eth.manager.EthPeers; |
||||
import tech.pegasys.pantheon.ethereum.eth.sync.SyncTargetManager; |
||||
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.util.uint.UInt256; |
||||
|
||||
import java.util.Optional; |
||||
|
||||
import org.apache.logging.log4j.LogManager; |
||||
import org.apache.logging.log4j.Logger; |
||||
|
||||
class FullSyncTargetManager<C> extends SyncTargetManager<C> { |
||||
|
||||
private static final Logger LOG = LogManager.getLogger(); |
||||
private final SynchronizerConfiguration config; |
||||
private final EthContext ethContext; |
||||
private final SyncState syncState; |
||||
|
||||
FullSyncTargetManager( |
||||
final SynchronizerConfiguration config, |
||||
final ProtocolSchedule<C> protocolSchedule, |
||||
final ProtocolContext<C> protocolContext, |
||||
final EthContext ethContext, |
||||
final SyncState syncState, |
||||
final LabelledMetric<OperationTimer> ethTasksTimer) { |
||||
super(config, protocolSchedule, protocolContext, ethContext, syncState, ethTasksTimer); |
||||
this.config = config; |
||||
this.ethContext = ethContext; |
||||
this.syncState = syncState; |
||||
} |
||||
|
||||
@Override |
||||
protected Optional<EthPeer> selectBestAvailableSyncTarget() { |
||||
final Optional<EthPeer> maybeBestPeer = ethContext.getEthPeers().bestPeer(); |
||||
if (!maybeBestPeer.isPresent()) { |
||||
LOG.info("No sync target, wait for peers."); |
||||
return Optional.empty(); |
||||
} else { |
||||
final EthPeer bestPeer = maybeBestPeer.get(); |
||||
final long peerHeight = bestPeer.chainState().getEstimatedHeight(); |
||||
final UInt256 peerTd = bestPeer.chainState().getBestBlock().getTotalDifficulty(); |
||||
if (peerTd.compareTo(syncState.chainHeadTotalDifficulty()) <= 0 |
||||
&& peerHeight <= syncState.chainHeadNumber()) { |
||||
// We're caught up to our best peer, try again when a new peer connects
|
||||
LOG.debug("Caught up to best peer: " + bestPeer.chainState().getEstimatedHeight()); |
||||
return Optional.empty(); |
||||
} |
||||
return maybeBestPeer; |
||||
} |
||||
} |
||||
|
||||
@Override |
||||
public boolean shouldSwitchSyncTarget(final SyncTarget currentTarget) { |
||||
final EthPeer currentPeer = currentTarget.peer(); |
||||
final ChainState currentPeerChainState = currentPeer.chainState(); |
||||
final Optional<EthPeer> maybeBestPeer = ethContext.getEthPeers().bestPeer(); |
||||
|
||||
return maybeBestPeer |
||||
.map( |
||||
bestPeer -> { |
||||
if (EthPeers.BEST_CHAIN.compare(bestPeer, currentPeer) <= 0) { |
||||
// Our current target is better or equal to the best peer
|
||||
return false; |
||||
} |
||||
// Require some threshold to be exceeded before switching targets to keep some
|
||||
// stability
|
||||
// when multiple peers are in range of each other
|
||||
final ChainState bestPeerChainState = bestPeer.chainState(); |
||||
final long heightDifference = |
||||
bestPeerChainState.getEstimatedHeight() |
||||
- currentPeerChainState.getEstimatedHeight(); |
||||
if (heightDifference == 0 && bestPeerChainState.getEstimatedHeight() == 0) { |
||||
// Only check td if we don't have a height metric
|
||||
final UInt256 tdDifference = |
||||
bestPeerChainState |
||||
.getBestBlock() |
||||
.getTotalDifficulty() |
||||
.minus(currentPeerChainState.getBestBlock().getTotalDifficulty()); |
||||
return tdDifference.compareTo(config.downloaderChangeTargetThresholdByTd()) > 0; |
||||
} |
||||
return heightDifference > config.downloaderChangeTargetThresholdByHeight(); |
||||
}) |
||||
.orElse(false); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue