mirror of https://github.com/hyperledger/besu
Improving backwards sync (#3638)
Refactor Backwards sync to use a rocks db. Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>pull/3720/head
Jiri Peinlich
3 years ago
committed by
Karim TAAM
parent
6c8602d435
commit
918b5359b1
@ -1,172 +0,0 @@ |
||||
/* |
||||
* Copyright Hyperledger Besu Contributors. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.sync.backwardsync; |
||||
|
||||
import static org.slf4j.LoggerFactory.getLogger; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.ProtocolContext; |
||||
import org.hyperledger.besu.ethereum.core.Block; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthContext; |
||||
import org.hyperledger.besu.ethereum.eth.manager.task.AbstractPeerTask; |
||||
import org.hyperledger.besu.ethereum.eth.manager.task.RetryingGetBlockFromPeersTask; |
||||
import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule; |
||||
import org.hyperledger.besu.plugin.services.MetricsSystem; |
||||
|
||||
import java.time.Duration; |
||||
import java.util.ArrayDeque; |
||||
import java.util.ArrayList; |
||||
import java.util.Collections; |
||||
import java.util.List; |
||||
import java.util.Optional; |
||||
import java.util.Queue; |
||||
import java.util.concurrent.CompletableFuture; |
||||
import java.util.function.Function; |
||||
import javax.annotation.concurrent.GuardedBy; |
||||
import javax.annotation.concurrent.ThreadSafe; |
||||
|
||||
import org.slf4j.Logger; |
||||
|
||||
@ThreadSafe |
||||
public class BackwardSyncLookupService { |
||||
private static final Logger LOG = getLogger(BackwardSyncLookupService.class); |
||||
private static final int MAX_RETRIES = 100; |
||||
public static final int UNUSED = -1; |
||||
|
||||
@GuardedBy("this") |
||||
private final Queue<Hash> hashes = new ArrayDeque<>(); |
||||
|
||||
@GuardedBy("this") |
||||
boolean running = false; |
||||
|
||||
private final ProtocolSchedule protocolSchedule; |
||||
private final EthContext ethContext; |
||||
private final MetricsSystem metricsSystem; |
||||
private List<Block> results = new ArrayList<>(); |
||||
private final ProtocolContext protocolContext; |
||||
|
||||
public BackwardSyncLookupService( |
||||
final ProtocolSchedule protocolSchedule, |
||||
final EthContext ethContext, |
||||
final MetricsSystem metricsSystem, |
||||
final ProtocolContext protocolContext) { |
||||
this.protocolSchedule = protocolSchedule; |
||||
this.ethContext = ethContext; |
||||
this.metricsSystem = metricsSystem; |
||||
this.protocolContext = protocolContext; |
||||
} |
||||
|
||||
public CompletableFuture<List<Block>> lookup(final Hash newBlockhash) { |
||||
synchronized (this) { |
||||
hashes.add(newBlockhash); |
||||
if (running) { |
||||
LOG.info( |
||||
"some other future is already running and will process our hash {} when time comes...", |
||||
newBlockhash.toHexString()); |
||||
return CompletableFuture.completedFuture(Collections.emptyList()); |
||||
} |
||||
running = true; |
||||
} |
||||
return findBlocksWithRetries() |
||||
.handle( |
||||
(blocks, throwable) -> { |
||||
synchronized (this) { |
||||
running = false; |
||||
} |
||||
if (throwable != null) { |
||||
throw new BackwardSyncException(throwable); |
||||
} |
||||
return blocks; |
||||
}); |
||||
} |
||||
|
||||
private CompletableFuture<List<Block>> findBlocksWithRetries() { |
||||
|
||||
CompletableFuture<List<Block>> f = tryToFindBlocks(); |
||||
for (int i = 0; i < MAX_RETRIES; i++) { |
||||
f = |
||||
f.thenApply(CompletableFuture::completedFuture) |
||||
.exceptionally( |
||||
ex -> { |
||||
synchronized (this) { |
||||
if (!results.isEmpty()) { |
||||
List<Block> copy = new ArrayList<>(results); |
||||
results = new ArrayList<>(); |
||||
return CompletableFuture.completedFuture(copy); |
||||
} |
||||
} |
||||
LOG.error( |
||||
"Failed to fetch blocks because {} Current peers: {}. Waiting for few seconds ...", |
||||
ex.getMessage(), |
||||
ethContext.getEthPeers().peerCount()); |
||||
return ethContext |
||||
.getScheduler() |
||||
.scheduleFutureTask(this::tryToFindBlocks, Duration.ofSeconds(5)); |
||||
}) |
||||
.thenCompose(Function.identity()); |
||||
} |
||||
return f.thenApply(this::rememberResults).thenCompose(this::possibleNextHash); |
||||
} |
||||
|
||||
private CompletableFuture<List<Block>> tryToFindBlocks() { |
||||
return CompletableFuture.supplyAsync(this::getNextHash) |
||||
.thenCompose(this::tryToFindBlock) |
||||
.thenApply(this::rememberResult) |
||||
.thenCompose(this::possibleNextHash); |
||||
} |
||||
|
||||
private CompletableFuture<List<Block>> possibleNextHash(final List<Block> blocks) { |
||||
synchronized (this) { |
||||
hashes.poll(); |
||||
if (hashes.isEmpty()) { |
||||
results = new ArrayList<>(); |
||||
running = false; |
||||
return CompletableFuture.completedFuture(blocks); |
||||
} |
||||
} |
||||
return tryToFindBlocks(); |
||||
} |
||||
|
||||
private List<Block> rememberResult(final Block block) { |
||||
this.results.add(block); |
||||
return results; |
||||
} |
||||
|
||||
private List<Block> rememberResults(final List<Block> blocks) { |
||||
this.results.addAll(blocks); |
||||
return results; |
||||
} |
||||
|
||||
private synchronized Hash getNextHash() { |
||||
return hashes.peek(); |
||||
} |
||||
|
||||
private CompletableFuture<Block> tryToFindBlock(final Hash targetHash) { |
||||
|
||||
final RetryingGetBlockFromPeersTask getBlockTask = |
||||
RetryingGetBlockFromPeersTask.create( |
||||
protocolContext, |
||||
protocolSchedule, |
||||
ethContext, |
||||
metricsSystem, |
||||
ethContext.getEthPeers().getMaxPeers(), |
||||
Optional.of(targetHash), |
||||
UNUSED); |
||||
return ethContext |
||||
.getScheduler() |
||||
.scheduleSyncWorkerTask(getBlockTask::run) |
||||
.thenApply(AbstractPeerTask.PeerTaskResult::getResult); |
||||
} |
||||
} |
||||
@ -1,193 +0,0 @@ |
||||
/* |
||||
* Copyright Hyperledger Besu Contributors. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.sync.backwardsync; |
||||
|
||||
import static org.hyperledger.besu.util.Slf4jLambdaHelper.debugLambda; |
||||
import static org.hyperledger.besu.util.Slf4jLambdaHelper.infoLambda; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.core.BlockHeader; |
||||
import org.hyperledger.besu.ethereum.eth.manager.task.GetHeadersFromPeerByHashTask; |
||||
import org.hyperledger.besu.ethereum.eth.manager.task.RetryingGetHeadersEndingAtFromPeerByHashTask; |
||||
|
||||
import java.time.Duration; |
||||
import java.util.List; |
||||
import java.util.Optional; |
||||
import java.util.concurrent.CompletableFuture; |
||||
import java.util.function.Function; |
||||
|
||||
import com.google.common.annotations.VisibleForTesting; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
public class BackwardSyncPhase extends BackwardSyncTask { |
||||
private static final Logger LOG = LoggerFactory.getLogger(BackwardSyncPhase.class); |
||||
|
||||
public BackwardSyncPhase(final BackwardSyncContext context, final BackwardChain backwardChain) { |
||||
super(context, backwardChain); |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected CompletableFuture<Void> waitForTTD() { |
||||
if (context.isOnTTD()) { |
||||
return CompletableFuture.completedFuture(null); |
||||
} |
||||
LOG.debug("Did not reach TTD yet, falling asleep..."); |
||||
return context |
||||
.getEthContext() |
||||
.getScheduler() |
||||
.scheduleFutureTask(this::waitForTTD, Duration.ofSeconds(5)); |
||||
} |
||||
|
||||
@Override |
||||
public CompletableFuture<Void> executeStep() { |
||||
return CompletableFuture.supplyAsync(this::waitForTTD) |
||||
.thenCompose(Function.identity()) |
||||
.thenApply(this::earliestUnprocessedHash) |
||||
.thenCompose(this::requestHeaders) |
||||
.thenApply(this::saveHeaders) |
||||
.thenApply(this::possibleMerge) |
||||
.thenCompose(this::possiblyMoreBackwardSteps); |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected Hash earliestUnprocessedHash(final Void unused) { |
||||
BlockHeader firstHeader = |
||||
backwardChain |
||||
.getFirstAncestorHeader() |
||||
.orElseThrow( |
||||
() -> |
||||
new BackwardSyncException( |
||||
"No unprocessed hashes during backward sync. that is probably a bug.", |
||||
true)); |
||||
Hash parentHash = firstHeader.getParentHash(); |
||||
debugLambda( |
||||
LOG, |
||||
"First unprocessed hash for current pivot is {} expected on height {}", |
||||
parentHash::toHexString, |
||||
() -> firstHeader.getNumber() - 1); |
||||
return parentHash; |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected CompletableFuture<BlockHeader> requestHeader(final Hash hash) { |
||||
debugLambda(LOG, "Requesting header for hash {}", hash::toHexString); |
||||
return GetHeadersFromPeerByHashTask.forSingleHash( |
||||
context.getProtocolSchedule(), |
||||
context.getEthContext(), |
||||
hash, |
||||
context.getProtocolContext().getBlockchain().getChainHead().getHeight(), |
||||
context.getMetricsSystem()) |
||||
.run() |
||||
.thenApply( |
||||
peerResult -> { |
||||
final List<BlockHeader> result = peerResult.getResult(); |
||||
if (result.isEmpty()) { |
||||
throw new BackwardSyncException( |
||||
"Did not receive a header for hash {}" + hash.toHexString(), true); |
||||
} |
||||
BlockHeader blockHeader = result.get(0); |
||||
debugLambda( |
||||
LOG, |
||||
"Got header {} with height {}", |
||||
() -> blockHeader.getHash().toHexString(), |
||||
blockHeader::getNumber); |
||||
return blockHeader; |
||||
}); |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected CompletableFuture<List<BlockHeader>> requestHeaders(final Hash hash) { |
||||
debugLambda(LOG, "Requesting header for hash {}", hash::toHexString); |
||||
final RetryingGetHeadersEndingAtFromPeerByHashTask |
||||
retryingGetHeadersEndingAtFromPeerByHashTask = |
||||
RetryingGetHeadersEndingAtFromPeerByHashTask.endingAtHash( |
||||
context.getProtocolSchedule(), |
||||
context.getEthContext(), |
||||
hash, |
||||
context.getProtocolContext().getBlockchain().getChainHead().getHeight(), |
||||
BackwardSyncContext.BATCH_SIZE, |
||||
context.getMetricsSystem()); |
||||
return context |
||||
.getEthContext() |
||||
.getScheduler() |
||||
.scheduleSyncWorkerTask(retryingGetHeadersEndingAtFromPeerByHashTask::run) |
||||
.thenApply( |
||||
blockHeaders -> { |
||||
if (blockHeaders.isEmpty()) { |
||||
throw new BackwardSyncException( |
||||
"Did not receive a header for hash {}" + hash.toHexString(), true); |
||||
} |
||||
debugLambda( |
||||
LOG, |
||||
"Got headers {} -> {}", |
||||
blockHeaders.get(0)::getNumber, |
||||
blockHeaders.get(blockHeaders.size() - 1)::getNumber); |
||||
return blockHeaders; |
||||
}); |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected Void saveHeader(final BlockHeader blockHeader) { |
||||
backwardChain.prependAncestorsHeader(blockHeader); |
||||
context.putCurrentChainToHeight(blockHeader.getNumber(), backwardChain); |
||||
return null; |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected Void saveHeaders(final List<BlockHeader> blockHeaders) { |
||||
for (BlockHeader blockHeader : blockHeaders) { |
||||
saveHeader(blockHeader); |
||||
} |
||||
infoLambda( |
||||
LOG, |
||||
"Saved headers {} -> {}", |
||||
() -> blockHeaders.get(0).getNumber(), |
||||
() -> blockHeaders.get(blockHeaders.size() - 1).getNumber()); |
||||
return null; |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected BlockHeader possibleMerge(final Void unused) { |
||||
Optional<BackwardChain> maybeHistoricalBackwardChain = |
||||
context.findCorrectChainFromPivot( |
||||
backwardChain.getFirstAncestorHeader().orElseThrow().getNumber() - 1); |
||||
maybeHistoricalBackwardChain.ifPresent(backwardChain::prependChain); |
||||
return backwardChain.getFirstAncestorHeader().orElseThrow(); |
||||
} |
||||
|
||||
// if the previous header is not present yet, we need to go deeper
|
||||
@VisibleForTesting |
||||
protected CompletableFuture<Void> possiblyMoreBackwardSteps(final BlockHeader blockHeader) { |
||||
CompletableFuture<Void> completableFuture = new CompletableFuture<>(); |
||||
if (context.getProtocolContext().getBlockchain().contains(blockHeader.getHash())) { |
||||
LOG.info("Backward Phase finished."); |
||||
completableFuture.complete(null); |
||||
return completableFuture; |
||||
} |
||||
if (context.getProtocolContext().getBlockchain().getChainHead().getHeight() |
||||
> blockHeader.getNumber() - 1) { |
||||
LOG.warn( |
||||
"Backward sync is following unknown branch {} ({}) and reached bellow previous head {}({})", |
||||
blockHeader.getNumber(), |
||||
blockHeader.getHash(), |
||||
context.getProtocolContext().getBlockchain().getChainHead().getHeight(), |
||||
context.getProtocolContext().getBlockchain().getChainHead().getHash().toHexString()); |
||||
} |
||||
LOG.debug("Backward sync did not reach a know block, need to go deeper"); |
||||
completableFuture.complete(null); |
||||
return completableFuture.thenCompose(this::executeAsync); |
||||
} |
||||
} |
||||
@ -0,0 +1,112 @@ |
||||
/* |
||||
* Copyright Hyperledger Besu Contributors. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.sync.backwardsync; |
||||
|
||||
import static org.hyperledger.besu.util.Slf4jLambdaHelper.debugLambda; |
||||
import static org.hyperledger.besu.util.Slf4jLambdaHelper.infoLambda; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.core.BlockHeader; |
||||
import org.hyperledger.besu.ethereum.eth.manager.task.RetryingGetHeadersEndingAtFromPeerByHashTask; |
||||
|
||||
import java.util.List; |
||||
import java.util.Optional; |
||||
import java.util.concurrent.CompletableFuture; |
||||
|
||||
import com.google.common.annotations.VisibleForTesting; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
public class BackwardSyncStep { |
||||
private static final Logger LOG = LoggerFactory.getLogger(BackwardSyncStep.class); |
||||
private final BackwardSyncContext context; |
||||
private final BackwardChain backwardChain; |
||||
|
||||
public BackwardSyncStep(final BackwardSyncContext context, final BackwardChain backwardChain) { |
||||
this.context = context; |
||||
this.backwardChain = backwardChain; |
||||
} |
||||
|
||||
public CompletableFuture<Void> executeAsync(final BlockHeader firstHeader) { |
||||
return CompletableFuture.supplyAsync(() -> firstHeader) |
||||
.thenApply(this::possibleRestoreOldNodes) |
||||
.thenCompose(this::requestHeaders) |
||||
.thenApply(this::saveHeaders) |
||||
.thenCompose(context::executeNextStep); |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected Hash possibleRestoreOldNodes(final BlockHeader firstAncestor) { |
||||
Hash lastHash = firstAncestor.getParentHash(); |
||||
Optional<BlockHeader> iterator = backwardChain.getHeader(lastHash); |
||||
while (iterator.isPresent()) { |
||||
backwardChain.prependAncestorsHeader(iterator.get()); |
||||
lastHash = iterator.get().getParentHash(); |
||||
iterator = backwardChain.getHeader(lastHash); |
||||
} |
||||
return lastHash; |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected CompletableFuture<List<BlockHeader>> requestHeaders(final Hash hash) { |
||||
debugLambda(LOG, "Requesting header for hash {}", hash::toHexString); |
||||
final RetryingGetHeadersEndingAtFromPeerByHashTask |
||||
retryingGetHeadersEndingAtFromPeerByHashTask = |
||||
RetryingGetHeadersEndingAtFromPeerByHashTask.endingAtHash( |
||||
context.getProtocolSchedule(), |
||||
context.getEthContext(), |
||||
hash, |
||||
context.getProtocolContext().getBlockchain().getChainHead().getHeight(), |
||||
context.getBatchSize(), |
||||
context.getMetricsSystem()); |
||||
return context |
||||
.getEthContext() |
||||
.getScheduler() |
||||
.scheduleSyncWorkerTask(retryingGetHeadersEndingAtFromPeerByHashTask::run) |
||||
.thenApply( |
||||
blockHeaders -> { |
||||
if (blockHeaders.isEmpty()) { |
||||
throw new BackwardSyncException( |
||||
"Did not receive a header for hash {}" + hash.toHexString(), true); |
||||
} |
||||
debugLambda( |
||||
LOG, |
||||
"Got headers {} -> {}", |
||||
blockHeaders.get(0)::getNumber, |
||||
blockHeaders.get(blockHeaders.size() - 1)::getNumber); |
||||
return blockHeaders; |
||||
}); |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected Void saveHeader(final BlockHeader blockHeader) { |
||||
backwardChain.prependAncestorsHeader(blockHeader); |
||||
return null; |
||||
} |
||||
|
||||
@VisibleForTesting |
||||
protected Void saveHeaders(final List<BlockHeader> blockHeaders) { |
||||
for (BlockHeader blockHeader : blockHeaders) { |
||||
saveHeader(blockHeader); |
||||
} |
||||
infoLambda( |
||||
LOG, |
||||
"Saved headers {} -> {} (head: {})", |
||||
() -> blockHeaders.get(0).getNumber(), |
||||
() -> blockHeaders.get(blockHeaders.size() - 1).getNumber(), |
||||
() -> context.getProtocolContext().getBlockchain().getChainHead().getHeight()); |
||||
return null; |
||||
} |
||||
} |
||||
@ -1,58 +0,0 @@ |
||||
/* |
||||
* Copyright Hyperledger Besu Contributors. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.sync.backwardsync; |
||||
|
||||
import static org.slf4j.LoggerFactory.getLogger; |
||||
|
||||
import java.util.Optional; |
||||
import java.util.concurrent.CompletableFuture; |
||||
|
||||
import org.slf4j.Logger; |
||||
|
||||
public abstract class BackwardSyncTask { |
||||
protected BackwardSyncContext context; |
||||
protected BackwardChain backwardChain; |
||||
private static final Logger LOG = getLogger(BackwardSyncTask.class); |
||||
|
||||
protected BackwardSyncTask(final BackwardSyncContext context, final BackwardChain backwardChain) { |
||||
this.context = context; |
||||
this.backwardChain = backwardChain; |
||||
} |
||||
|
||||
CompletableFuture<Void> executeAsync(final Void unused) { |
||||
Optional<BackwardChain> currentChain = context.getCurrentChain(); |
||||
if (currentChain.isPresent()) { |
||||
if (!backwardChain.equals(currentChain.get())) { |
||||
LOG.debug( |
||||
"The pivot changed, we should stop current flow, some new flow is waiting to take over..."); |
||||
return CompletableFuture.completedFuture(null); |
||||
} |
||||
if (backwardChain.getFirstAncestorHeader().isEmpty()) { |
||||
LOG.info("The Backwards sync is already finished..."); |
||||
return CompletableFuture.completedFuture(null); |
||||
} |
||||
return executeStep(); |
||||
|
||||
} else { |
||||
CompletableFuture<Void> result = new CompletableFuture<>(); |
||||
result.completeExceptionally( |
||||
new BackwardSyncException( |
||||
"No pivot... that is weird and should not have happened. This method should have been called after the pivot was set...")); |
||||
return result; |
||||
} |
||||
} |
||||
|
||||
abstract CompletableFuture<Void> executeStep(); |
||||
} |
||||
@ -0,0 +1,34 @@ |
||||
/* |
||||
* |
||||
* * Copyright Hyperledger Besu Contributors. |
||||
* * |
||||
* * 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. |
||||
* * |
||||
* * SPDX-License-Identifier: Apache-2.0 |
||||
* |
||||
*/ |
||||
|
||||
package org.hyperledger.besu.ethereum.eth.sync.backwardsync; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
|
||||
import org.apache.tuweni.bytes.Bytes32; |
||||
|
||||
public class HashConvertor implements ValueConvertor<Hash> { |
||||
@Override |
||||
public Hash fromBytes(final byte[] bytes) { |
||||
return Hash.wrap(Bytes32.wrap(bytes)); |
||||
} |
||||
|
||||
@Override |
||||
public byte[] toBytes(final Hash value) { |
||||
return value.toArrayUnsafe(); |
||||
} |
||||
} |
||||
@ -0,0 +1,74 @@ |
||||
/* |
||||
* |
||||
* * Copyright Hyperledger Besu Contributors. |
||||
* * |
||||
* * 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. |
||||
* * |
||||
* * SPDX-License-Identifier: Apache-2.0 |
||||
* |
||||
*/ |
||||
|
||||
package org.hyperledger.besu.ethereum.eth.sync.backwardsync; |
||||
|
||||
import static org.slf4j.LoggerFactory.getLogger; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.core.Block; |
||||
import org.hyperledger.besu.ethereum.eth.manager.task.AbstractPeerTask; |
||||
import org.hyperledger.besu.ethereum.eth.manager.task.RetryingGetBlockFromPeersTask; |
||||
|
||||
import java.util.Optional; |
||||
import java.util.concurrent.CompletableFuture; |
||||
|
||||
import org.slf4j.Logger; |
||||
|
||||
public class SyncStepStep { |
||||
private static final Logger LOG = getLogger(SyncStepStep.class); |
||||
|
||||
public static final int UNUSED = -1; |
||||
private final BackwardSyncContext context; |
||||
private final BackwardChain backwardChain; |
||||
|
||||
public SyncStepStep(final BackwardSyncContext context, final BackwardChain backwardChain) { |
||||
this.context = context; |
||||
this.backwardChain = backwardChain; |
||||
} |
||||
|
||||
public CompletableFuture<Void> executeAsync(final Hash hash) { |
||||
return CompletableFuture.supplyAsync(() -> hash) |
||||
.thenCompose(this::requestBlock) |
||||
.thenApply(this::saveBlock) |
||||
.thenCompose(context::executeNextStep); |
||||
} |
||||
|
||||
private CompletableFuture<Block> requestBlock(final Hash targetHash) { |
||||
final RetryingGetBlockFromPeersTask getBlockTask = |
||||
RetryingGetBlockFromPeersTask.create( |
||||
context.getProtocolContext(), |
||||
context.getProtocolSchedule(), |
||||
context.getEthContext(), |
||||
context.getMetricsSystem(), |
||||
context.getEthContext().getEthPeers().getMaxPeers(), |
||||
Optional.of(targetHash), |
||||
UNUSED); |
||||
return context |
||||
.getEthContext() |
||||
.getScheduler() |
||||
.scheduleSyncWorkerTask(getBlockTask::run) |
||||
.thenApply(AbstractPeerTask.PeerTaskResult::getResult); |
||||
} |
||||
|
||||
private Void saveBlock(final Block block) { |
||||
LOG.debug( |
||||
"Appending block {}({})", block.getHeader().getNumber(), block.getHash().toHexString()); |
||||
backwardChain.appendTrustedBlock(block); |
||||
return null; |
||||
} |
||||
} |
||||
@ -1,132 +0,0 @@ |
||||
/* |
||||
* Copyright Hyperledger Besu Contributors. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
|
||||
package org.hyperledger.besu.ethereum.eth.sync.backwardsync; |
||||
|
||||
import static org.assertj.core.api.AssertionsForClassTypes.assertThat; |
||||
import static org.hyperledger.besu.ethereum.core.InMemoryKeyValueStorageProvider.createInMemoryBlockchain; |
||||
import static org.mockito.ArgumentMatchers.anyLong; |
||||
import static org.mockito.Mockito.spy; |
||||
import static org.mockito.Mockito.when; |
||||
|
||||
import org.hyperledger.besu.config.StubGenesisConfigOptions; |
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.ProtocolContext; |
||||
import org.hyperledger.besu.ethereum.chain.MutableBlockchain; |
||||
import org.hyperledger.besu.ethereum.core.Block; |
||||
import org.hyperledger.besu.ethereum.core.BlockDataGenerator; |
||||
import org.hyperledger.besu.ethereum.core.TransactionReceipt; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthContext; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManager; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthProtocolManagerTestUtil; |
||||
import org.hyperledger.besu.ethereum.eth.manager.RespondingEthPeer; |
||||
import org.hyperledger.besu.ethereum.mainnet.MainnetProtocolSchedule; |
||||
import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule; |
||||
import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec; |
||||
import org.hyperledger.besu.plugin.services.MetricsSystem; |
||||
|
||||
import java.lang.reflect.Field; |
||||
import java.lang.reflect.Modifier; |
||||
import java.util.List; |
||||
import java.util.concurrent.CompletableFuture; |
||||
import javax.annotation.Nonnull; |
||||
|
||||
import org.junit.Before; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.mockito.Answers; |
||||
import org.mockito.Mock; |
||||
import org.mockito.Spy; |
||||
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
||||
@RunWith(MockitoJUnitRunner.class) |
||||
public class BackwardSyncLookupServiceTest { |
||||
public static final int REMOTE_HEIGHT = 50; |
||||
private static final BlockDataGenerator blockDataGenerator = new BlockDataGenerator(); |
||||
|
||||
@Spy |
||||
private final ProtocolSchedule protocolSchedule = |
||||
MainnetProtocolSchedule.fromConfig(new StubGenesisConfigOptions()); |
||||
|
||||
@Spy private final ProtocolSpec mockProtocolSpec = protocolSchedule.getByBlockNumber(0L); |
||||
private MutableBlockchain remoteBlockchain; |
||||
private RespondingEthPeer peer; |
||||
|
||||
@Mock(answer = Answers.RETURNS_DEEP_STUBS) |
||||
private MetricsSystem metricsSystem; |
||||
|
||||
@Mock private ProtocolContext protocolContext; |
||||
|
||||
private BackwardSyncLookupService backwardSyncLookupService; |
||||
|
||||
@Before |
||||
public void setup() throws NoSuchFieldException, IllegalAccessException { |
||||
when(protocolSchedule.getByBlockNumber(anyLong())).thenReturn(mockProtocolSpec); |
||||
Block genesisBlock = blockDataGenerator.genesisBlock(); |
||||
remoteBlockchain = createInMemoryBlockchain(genesisBlock); |
||||
final Field max_retries = BackwardSyncLookupService.class.getDeclaredField("MAX_RETRIES"); |
||||
|
||||
Field modifiersField = Field.class.getDeclaredField("modifiers"); |
||||
modifiersField.setAccessible(true); |
||||
modifiersField.setInt(max_retries, max_retries.getModifiers() & ~Modifier.FINAL); |
||||
|
||||
max_retries.setAccessible(true); |
||||
max_retries.set(null, 1); |
||||
|
||||
for (int i = 1; i <= REMOTE_HEIGHT; i++) { |
||||
final BlockDataGenerator.BlockOptions options = |
||||
new BlockDataGenerator.BlockOptions() |
||||
.setBlockNumber(i) |
||||
.setParentHash(remoteBlockchain.getBlockHashByNumber(i - 1).orElseThrow()); |
||||
final Block block = blockDataGenerator.block(options); |
||||
final List<TransactionReceipt> receipts = blockDataGenerator.receipts(block); |
||||
|
||||
remoteBlockchain.appendBlock(block, receipts); |
||||
} |
||||
EthProtocolManager ethProtocolManager = EthProtocolManagerTestUtil.create(); |
||||
|
||||
peer = EthProtocolManagerTestUtil.createPeer(ethProtocolManager); |
||||
EthContext ethContext = ethProtocolManager.ethContext(); |
||||
|
||||
backwardSyncLookupService = |
||||
spy( |
||||
new BackwardSyncLookupService( |
||||
protocolSchedule, ethContext, metricsSystem, protocolContext)); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldFindABlockWhenResponding() throws Exception { |
||||
final Hash hash = getBlockByNumber(23).getHash(); |
||||
|
||||
final CompletableFuture<List<Block>> future = backwardSyncLookupService.lookup(hash); |
||||
|
||||
respondUntilFutureIsDone(future); |
||||
|
||||
final List<Block> blocks = future.get(); |
||||
assertThat(blocks.get(0)).isEqualTo(getBlockByNumber(23)); |
||||
} |
||||
|
||||
private void respondUntilFutureIsDone(final CompletableFuture<?> future) { |
||||
final RespondingEthPeer.Responder responder = |
||||
RespondingEthPeer.blockchainResponder(remoteBlockchain); |
||||
|
||||
peer.respondWhileOtherThreadsWork(responder, () -> !future.isDone()); |
||||
} |
||||
|
||||
@Nonnull |
||||
private Block getBlockByNumber(final int number) { |
||||
return remoteBlockchain.getBlockByNumber(number).orElseThrow(); |
||||
} |
||||
} |
||||
@ -1,111 +0,0 @@ |
||||
/* |
||||
* Copyright Hyperledger Besu Contributors. |
||||
* |
||||
* 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.sync.backwardsync; |
||||
|
||||
import static org.assertj.core.api.AssertionsForClassTypes.assertThat; |
||||
import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy; |
||||
import static org.mockito.Mockito.when; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.core.Block; |
||||
import org.hyperledger.besu.ethereum.core.BlockHeader; |
||||
import org.hyperledger.besu.ethereum.mainnet.MainnetBlockHeaderFunctions; |
||||
import org.hyperledger.besu.services.kvstore.InMemoryKeyValueStorage; |
||||
|
||||
import java.util.List; |
||||
import java.util.Optional; |
||||
import java.util.concurrent.CompletableFuture; |
||||
import javax.annotation.Nonnull; |
||||
|
||||
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 BackwardSyncTaskTest { |
||||
|
||||
public static final int HEIGHT = 20_000; |
||||
|
||||
@Mock private BackwardSyncContext context; |
||||
private List<Block> blocks; |
||||
|
||||
GenericKeyValueStorageFacade<Hash, BlockHeader> headersStorage; |
||||
GenericKeyValueStorageFacade<Hash, Block> blocksStorage; |
||||
|
||||
@Before |
||||
public void initContextAndChain() { |
||||
blocks = ChainForTestCreator.prepareChain(2, HEIGHT); |
||||
headersStorage = |
||||
new GenericKeyValueStorageFacade<>( |
||||
Hash::toArrayUnsafe, |
||||
new BlocksHeadersConvertor(new MainnetBlockHeaderFunctions()), |
||||
new InMemoryKeyValueStorage()); |
||||
blocksStorage = |
||||
new GenericKeyValueStorageFacade<>( |
||||
Hash::toArrayUnsafe, |
||||
new BlocksConvertor(new MainnetBlockHeaderFunctions()), |
||||
new InMemoryKeyValueStorage()); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldFailWhenPivotNotSetInContext() { |
||||
when(context.getCurrentChain()).thenReturn(Optional.empty()); |
||||
BackwardSyncTask step = createBackwardSyncTask(); |
||||
CompletableFuture<Void> completableFuture = step.executeAsync(null); |
||||
assertThatThrownBy(completableFuture::get) |
||||
.getCause() |
||||
.isInstanceOf(BackwardSyncException.class) |
||||
.hasMessageContaining("No pivot"); |
||||
} |
||||
|
||||
@Nonnull |
||||
private BackwardSyncTask createBackwardSyncTask() { |
||||
final BackwardChain backwardChain = |
||||
new BackwardChain(headersStorage, blocksStorage, blocks.get(1)); |
||||
return createBackwardSyncTask(backwardChain); |
||||
} |
||||
|
||||
@Nonnull |
||||
private BackwardSyncTask createBackwardSyncTask(final BackwardChain backwardChain) { |
||||
return new BackwardSyncTask(context, backwardChain) { |
||||
@Override |
||||
CompletableFuture<Void> executeStep() { |
||||
return CompletableFuture.completedFuture(null); |
||||
} |
||||
}; |
||||
} |
||||
|
||||
@Test |
||||
public void shouldFinishImmediatelyFailWhenPivotIsDifferent() { |
||||
final BackwardChain backwardChain = |
||||
new BackwardChain(headersStorage, blocksStorage, blocks.get(0)); |
||||
when(context.getCurrentChain()).thenReturn(Optional.of(backwardChain)); |
||||
BackwardSyncTask step = createBackwardSyncTask(); |
||||
CompletableFuture<Void> completableFuture = step.executeAsync(null); |
||||
assertThat(completableFuture.isDone()).isTrue(); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldExecuteWhenPivotIsCorrect() { |
||||
final BackwardChain backwardChain = |
||||
new BackwardChain(headersStorage, blocksStorage, blocks.get(1)); |
||||
BackwardSyncTask step = createBackwardSyncTask(); |
||||
when(context.getCurrentChain()).thenReturn(Optional.of(backwardChain)); |
||||
CompletableFuture<Void> completableFuture = step.executeAsync(null); |
||||
assertThat(completableFuture.isDone()).isTrue(); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue