7311: add peertask foundation code

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/InvalidPeerTaskResponseException.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * 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.manager.peertask;
+
+public class InvalidPeerTaskResponseException extends Exception {
+
+ public InvalidPeerTaskResponseException() {
+ super();
+ }
+
+ public InvalidPeerTaskResponseException(final Throwable cause) {
+ super(cause);
+ }
+}

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/NoAvailablePeerException.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * 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.manager.peertask;
+
+public class NoAvailablePeerException extends Exception {}

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerManager.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * 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.manager.peertask;
+
+import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
+import org.hyperledger.besu.ethereum.p2p.peers.PeerId;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Predicate;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** "Manages" the EthPeers for the PeerTaskExecutor */
+public class PeerManager {
+ private static final Logger LOG = LoggerFactory.getLogger(PeerManager.class);
+
+ // use a synchronized map to ensure the map is never modified by multiple threads at once
+ private final Map<PeerId, EthPeer> ethPeersByPeerId =
+ Collections.synchronizedMap(new HashMap<>());
+
+ /**
+ * Gets the highest reputation peer matching the supplies filter
+ *
+ * @param filter a filter to match prospective peers with
+ * @return the highest reputation peer matching the supplies filter
+ * @throws NoAvailablePeerException If there are no suitable peers
+ */
+ public EthPeer getPeer(final Predicate<EthPeer> filter) throws NoAvailablePeerException {
+ LOG.trace("Getting peer from pool of {} peers", ethPeersByPeerId.size());
+ return ethPeersByPeerId.values().stream()
+ .filter(filter)
+ .max(Comparator.naturalOrder())
+ .orElseThrow(NoAvailablePeerException::new);
+ }
+
+ public Optional<EthPeer> getPeerByPeerId(final PeerId peerId) {
+ return Optional.ofNullable(ethPeersByPeerId.get(peerId));
+ }
+
+ public void addPeer(final EthPeer ethPeer) {
+ ethPeersByPeerId.put(ethPeer.getConnection().getPeer(), ethPeer);
+ }
+
+ public void removePeer(final PeerId peerId) {
+ ethPeersByPeerId.remove(peerId);
+ }
+}

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTask.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * 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.manager.peertask;
+
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
+
+import java.util.Collection;
+
+/**
+ * Represents a task to be executed on an EthPeer by the PeerTaskExecutor
+ *
+ * @param <T> The type of the result of this PeerTask
+ */
+public interface PeerTask<T> {
+ /**
+ * Returns the SubProtocol used for this PeerTask
+ *
+ * @return the SubProtocol used for this PeerTask
+ */
+ String getSubProtocol();
+
+ /**
+ * Gets the minimum required block number for a peer to have to successfully execute this task
+ *
+ * @return the minimum required block number for a peer to have to successfully execute this task
+ */
+ long getRequiredBlockNumber();
+
+ /**
+ * Gets the request data to send to the EthPeer
+ *
+ * @return the request data to send to the EthPeer
+ */
+ MessageData getRequestMessage();
+
+ /**
+ * Parses the MessageData response from the EthPeer
+ *
+ * @param messageData the response MessageData to be parsed
+ * @return a T built from the response MessageData
+ * @throws InvalidPeerTaskResponseException if the response messageData is invalid
+ */
+ T parseResponse(MessageData messageData) throws InvalidPeerTaskResponseException;
+
+ /**
+ * Gets the Collection of behaviors this task is expected to exhibit in the PeetTaskExecutor
+ *
+ * @return the Collection of behaviors this task is expected to exhibit in the PeetTaskExecutor
+ */
+ Collection<PeerTaskBehavior> getPeerTaskBehaviors();
+}

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskBehavior.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * 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.manager.peertask;
+
+public enum PeerTaskBehavior {
+ RETRY_WITH_SAME_PEER,
+ RETRY_WITH_OTHER_PEERS
+}

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java

@@ -0,0 +1,157 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * 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.manager.peertask;
+
+import org.hyperledger.besu.ethereum.eth.manager.EthPeer;
+import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec;
+import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData;
+import org.hyperledger.besu.metrics.BesuMetricCategory;
+import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.plugin.services.metrics.LabelledMetric;
+import org.hyperledger.besu.plugin.services.metrics.OperationTimer;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
+/** Manages the execution of PeerTasks, respecting their PeerTaskBehavior */
+public class PeerTaskExecutor {
+ private static final long[] WAIT_TIME_BEFORE_RETRY = {0, 20000, 5000};
+
+ private final PeerManager peerManager;
+ private final PeerTaskRequestSender requestSender;
+ private final Supplier<ProtocolSpec> protocolSpecSupplier;
+ private final LabelledMetric<OperationTimer> requestTimer;
+
+ public PeerTaskExecutor(
+ final PeerManager peerManager,
+ final PeerTaskRequestSender requestSender,
+ final Supplier<ProtocolSpec> protocolSpecSupplier,
+ final MetricsSystem metricsSystem) {
+ this.peerManager = peerManager;
+ this.requestSender = requestSender;
+ this.protocolSpecSupplier = protocolSpecSupplier;
+ requestTimer =
+ metricsSystem.createLabelledTimer(
+ BesuMetricCategory.PEERS, "Peer Task Executor Request Time", "", "Task Class Name");
+ }
+
+ public <T> PeerTaskExecutorResult<T> execute(final PeerTask<T> peerTask) {
+ PeerTaskExecutorResult<T> executorResult;
+ int triesRemaining =
+ peerTask.getPeerTaskBehaviors().contains(PeerTaskBehavior.RETRY_WITH_OTHER_PEERS) ? 3 : 1;
+ final Collection<EthPeer> usedEthPeers = new ArrayList<>();
+ do {
+ EthPeer peer;
+ try {
+ peer =
+ peerManager.getPeer(
+ (candidatePeer) ->
+ isPeerUnused(candidatePeer, usedEthPeers)
+ && (protocolSpecSupplier.get().isPoS()
+ || isPeerHeightHighEnough(
+ candidatePeer, peerTask.getRequiredBlockNumber()))
+ && isPeerProtocolSuitable(candidatePeer, peerTask.getSubProtocol()));
+ usedEthPeers.add(peer);
+ executorResult = executeAgainstPeer(peerTask, peer);
+ } catch (NoAvailablePeerException e) {
+ executorResult =
+ new PeerTaskExecutorResult<>(null, PeerTaskExecutorResponseCode.NO_PEER_AVAILABLE);
+ }
+ } while (--triesRemaining > 0
+ && executorResult.getResponseCode() != PeerTaskExecutorResponseCode.SUCCESS);
+
+ return executorResult;
+ }
+
+ public <T> CompletableFuture<PeerTaskExecutorResult<T>> executeAsync(final PeerTask<T> peerTask) {
+ return CompletableFuture.supplyAsync(() -> execute(peerTask));
+ }
+
+ public <T> PeerTaskExecutorResult<T> executeAgainstPeer(
+ final PeerTask<T> peerTask, final EthPeer peer) {
+ MessageData requestMessageData = peerTask.getRequestMessage();
+ PeerTaskExecutorResult<T> executorResult;
+ int triesRemaining =
+ peerTask.getPeerTaskBehaviors().contains(PeerTaskBehavior.RETRY_WITH_SAME_PEER) ? 3 : 1;
+ do {
+ try {
+
+ MessageData responseMessageData;
+ try (final OperationTimer.TimingContext timingContext =
+ requestTimer.labels(peerTask.getClass().getSimpleName()).startTimer()) {
+ responseMessageData =
+ requestSender.sendRequest(peerTask.getSubProtocol(), requestMessageData, peer);
+ }
+ T result = peerTask.parseResponse(responseMessageData);
+ peer.recordUsefulResponse();
+ executorResult = new PeerTaskExecutorResult<>(result, PeerTaskExecutorResponseCode.SUCCESS);
+
+ } catch (PeerConnection.PeerNotConnected e) {
+ executorResult =
+ new PeerTaskExecutorResult<>(null, PeerTaskExecutorResponseCode.PEER_DISCONNECTED);
+
+ } catch (InterruptedException | TimeoutException e) {
+ peer.recordRequestTimeout(requestMessageData.getCode());
+ executorResult = new PeerTaskExecutorResult<>(null, PeerTaskExecutorResponseCode.TIMEOUT);
+
+ } catch (InvalidPeerTaskResponseException e) {
+ peer.recordUselessResponse(e.getMessage());
+ executorResult =
+ new PeerTaskExecutorResult<>(null, PeerTaskExecutorResponseCode.INVALID_RESPONSE);
+
+ } catch (ExecutionException e) {
+ executorResult =
+ new PeerTaskExecutorResult<>(null, PeerTaskExecutorResponseCode.INTERNAL_SERVER_ERROR);
+ }
+ } while (--triesRemaining > 0
+ && executorResult.getResponseCode() != PeerTaskExecutorResponseCode.SUCCESS
+ && executorResult.getResponseCode() != PeerTaskExecutorResponseCode.PEER_DISCONNECTED
+ && sleepBetweenRetries(WAIT_TIME_BEFORE_RETRY[triesRemaining]));
+
+ return executorResult;
+ }
+
+ public <T> CompletableFuture<PeerTaskExecutorResult<T>> executeAgainstPeerAsync(
+ final PeerTask<T> peerTask, final EthPeer peer) {
+ return CompletableFuture.supplyAsync(() -> executeAgainstPeer(peerTask, peer));
+ }
+
+ private boolean sleepBetweenRetries(final long sleepTime) {
+ try {
+ Thread.sleep(sleepTime);
+ return true;
+ } catch (InterruptedException e) {
+ return false;
+ }
+ }
+
+ private static boolean isPeerUnused(
+ final EthPeer ethPeer, final Collection<EthPeer> usedEthPeers) {
+ return !usedEthPeers.contains(ethPeer);
+ }
+
+ private static boolean isPeerHeightHighEnough(final EthPeer ethPeer, final long requiredHeight) {
+ return ethPeer.chainState().getEstimatedHeight() >= requiredHeight;
+ }
+
+ private static boolean isPeerProtocolSuitable(final EthPeer ethPeer, final String protocol) {
+ return ethPeer.getProtocolName().equals(protocol);
+ }
+}

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutorResponseCode.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright contributors to Hyperledger Besu.
+ *
+ * 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.manager.peertask;
+
+public enum PeerTaskExecutorResponseCode {
+ SUCCESS,
+ NO_PEER_AVAILABLE,
+ PEER_DISCONNECTED,
+ INTERNAL_SERVER_ERROR,
+ TIMEOUT,
+ INVALID_RESPONSE
+}