7311: add peertask foundation code

ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthPeers.java

@@ -17,6 +17,7 @@
 import org.hyperledger.besu.ethereum.core.BlockHeader;
 import org.hyperledger.besu.ethereum.eth.SnapProtocol;
 import org.hyperledger.besu.ethereum.eth.manager.EthPeer.DisconnectCallback;
+import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerSelector;
 import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator;
 import org.hyperledger.besu.ethereum.eth.sync.ChainHeadTracker;
 import org.hyperledger.besu.ethereum.eth.sync.SnapServerChecker;
@@ -26,6 +27,7 @@
 import org.hyperledger.besu.ethereum.forkid.ForkIdManager;
 import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec;
 import org.hyperledger.besu.ethereum.p2p.peers.Peer;
+import org.hyperledger.besu.ethereum.p2p.peers.PeerId;
 import org.hyperledger.besu.ethereum.p2p.rlpx.RlpxAgent;
 import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection;
 import org.hyperledger.besu.ethereum.p2p.rlpx.wire.messages.DisconnectMessage;
@@ -61,7 +63,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class EthPeers {
+public class EthPeers implements PeerSelector {
  private static final Logger LOG = LoggerFactory.getLogger(EthPeers.class);
  public static final Comparator<EthPeer> TOTAL_DIFFICULTY =
  Comparator.comparing((final EthPeer p) -> p.chainState().getEstimatedTotalDifficulty());
@@ -465,6 +467,22 @@ public void setTrailingPeerRequirementsSupplier(
  this.trailingPeerRequirementsSupplier = tprSupplier;
  }
 
+ // Part of the PeerSelector interface, to be split apart later
+ @Override
+ public Optional<EthPeer> getPeer(final Predicate<EthPeer> filter) {
+ return streamAvailablePeers()
+ .filter(filter)
+ .filter(EthPeer::hasAvailableRequestCapacity)
+ .filter(EthPeer::isFullyValidated)
+ .min(LEAST_TO_MOST_BUSY);
+ }
+
+ // Part of the PeerSelector interface, to be split apart later
+ @Override
+ public Optional<EthPeer> getPeerByPeerId(final PeerId peerId) {
+ return Optional.ofNullable(activeConnections.get(peerId.getId()));
+ }
+
  @FunctionalInterface
  public interface ConnectCallback {
  void onPeerConnected(EthPeer newPeer);

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/PeerSelector.java

@@ -0,0 +1,42 @@
+/*
+ * 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.Optional;
+import java.util.function.Predicate;
+
+/** Selects the EthPeers for the PeerTaskExecutor */
+public interface PeerSelector {
+
+ /**
+ * Gets a peer matching the supplied filter
+ *
+ * @param filter a Predicate\<EthPeer\> matching desirable peers
+ * @return a peer matching the supplied conditions
+ */
+ Optional<EthPeer> getPeer(final Predicate<EthPeer> filter);
+
+ /**
+ * Attempts to get the EthPeer identified by peerId
+ *
+ * @param peerId the peerId of the desired EthPeer
+ * @return An Optional\<EthPeer\> containing the EthPeer identified by peerId if present in the
+ * PeerSelector, or empty otherwise
+ */
+ Optional<EthPeer> getPeerByPeerId(PeerId peerId);
+}

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

@@ -0,0 +1,83 @@
+/*
+ * 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.rlpx.wire.MessageData;
+import org.hyperledger.besu.ethereum.p2p.rlpx.wire.SubProtocol;
+
+import java.util.function.Predicate;
+
+/**
+ * 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
+ */
+ SubProtocol getSubProtocol();
+
+ /**
+ * 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 number of times this task may be attempted against other peers
+ *
+ * @return the number of times this task may be attempted against other peers
+ */
+ default int getRetriesWithOtherPeer() {
+ return 5;
+ }
+
+ /**
+ * Gets the number of times this task may be attempted against the same peer
+ *
+ * @return the number of times this task may be attempted against the same peer
+ */
+ default int getRetriesWithSamePeer() {
+ return 5;
+ }
+
+ /**
+ * Gets a Predicate that checks if an EthPeer is suitable for this PeerTask
+ *
+ * @return a Predicate that checks if an EthPeer is suitable for this PeerTask
+ */
+ Predicate<EthPeer> getPeerRequirementFilter();
+
+ /**
+ * Checks if the supplied result is considered a partial success
+ *
+ * @return true if the supplied result is considered a partial success
+ */
+ boolean isPartialSuccess(T result);
+}