mirror of https://github.com/hyperledger/besu
Remove DefaultPeerSelector, make EthPeers implement PeerSelector interface, and add PeerTask.getPeerRequirementFilter Signed-off-by: Matilda Clerke <matilda.clerke@consensys.net>pull/7628/head
Matilda Clerke
2 months ago
parent
8718102277
commit
e63f4730c6
@ -1,102 +0,0 @@ |
||||
/* |
||||
* 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.peers.PeerId; |
||||
import org.hyperledger.besu.ethereum.p2p.rlpx.wire.SubProtocol; |
||||
|
||||
import java.util.Collection; |
||||
import java.util.Comparator; |
||||
import java.util.Map; |
||||
import java.util.Optional; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.function.Predicate; |
||||
import java.util.function.Supplier; |
||||
|
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
/** |
||||
* This is a simple PeerSelector implementation that can be used the default implementation in most |
||||
* situations |
||||
*/ |
||||
public class DefaultPeerSelector implements PeerSelector { |
||||
private static final Logger LOG = LoggerFactory.getLogger(DefaultPeerSelector.class); |
||||
|
||||
private final Supplier<ProtocolSpec> protocolSpecSupplier; |
||||
private final Map<PeerId, EthPeer> ethPeersByPeerId = new ConcurrentHashMap<>(); |
||||
|
||||
public DefaultPeerSelector(final Supplier<ProtocolSpec> protocolSpecSupplier) { |
||||
this.protocolSpecSupplier = protocolSpecSupplier; |
||||
} |
||||
|
||||
/** |
||||
* Gets the highest reputation peer matching the supplied 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 |
||||
*/ |
||||
private EthPeer getPeer(final Predicate<EthPeer> filter) throws NoAvailablePeerException { |
||||
LOG.trace("Finding peer from pool of {} peers", ethPeersByPeerId.size()); |
||||
return ethPeersByPeerId.values().stream() |
||||
.filter(filter) |
||||
.max(Comparator.naturalOrder()) |
||||
.orElseThrow(NoAvailablePeerException::new); |
||||
} |
||||
|
||||
@Override |
||||
public EthPeer getPeer( |
||||
final Collection<EthPeer> usedEthPeers, |
||||
final long requiredPeerHeight, |
||||
final SubProtocol requiredSubProtocol) |
||||
throws NoAvailablePeerException { |
||||
return getPeer( |
||||
(candidatePeer) -> |
||||
isPeerUnused(candidatePeer, usedEthPeers) |
||||
&& (protocolSpecSupplier.get().isPoS() |
||||
|| isPeerHeightHighEnough(candidatePeer, requiredPeerHeight)) |
||||
&& isPeerProtocolSuitable(candidatePeer, requiredSubProtocol)); |
||||
} |
||||
|
||||
@Override |
||||
public Optional<EthPeer> getPeerByPeerId(final PeerId peerId) { |
||||
return Optional.ofNullable(ethPeersByPeerId.get(peerId)); |
||||
} |
||||
|
||||
@Override |
||||
public void addPeer(final EthPeer ethPeer) { |
||||
ethPeersByPeerId.put(ethPeer.getConnection().getPeer(), ethPeer); |
||||
} |
||||
|
||||
@Override |
||||
public void removePeer(final PeerId peerId) { |
||||
ethPeersByPeerId.remove(peerId); |
||||
} |
||||
|
||||
private boolean isPeerUnused(final EthPeer ethPeer, final Collection<EthPeer> usedEthPeers) { |
||||
return !usedEthPeers.contains(ethPeer); |
||||
} |
||||
|
||||
private boolean isPeerHeightHighEnough(final EthPeer ethPeer, final long requiredHeight) { |
||||
return ethPeer.chainState().getEstimatedHeight() >= requiredHeight; |
||||
} |
||||
|
||||
private boolean isPeerProtocolSuitable(final EthPeer ethPeer, final SubProtocol protocol) { |
||||
return ethPeer.getProtocolName().equals(protocol.getName()); |
||||
} |
||||
} |
||||
@ -1,135 +0,0 @@ |
||||
/* |
||||
* 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.EthProtocol; |
||||
import org.hyperledger.besu.ethereum.eth.SnapProtocol; |
||||
import org.hyperledger.besu.ethereum.eth.manager.ChainState; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthPeer; |
||||
import org.hyperledger.besu.ethereum.eth.manager.PeerReputation; |
||||
import org.hyperledger.besu.ethereum.mainnet.ProtocolSpec; |
||||
import org.hyperledger.besu.ethereum.p2p.peers.Peer; |
||||
import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection; |
||||
import org.hyperledger.besu.ethereum.p2p.rlpx.wire.SubProtocol; |
||||
|
||||
import java.util.HashSet; |
||||
import java.util.Set; |
||||
|
||||
import org.junit.jupiter.api.Assertions; |
||||
import org.junit.jupiter.api.BeforeEach; |
||||
import org.junit.jupiter.api.Test; |
||||
import org.mockito.Mockito; |
||||
|
||||
public class DefaultPeerSelectorTest { |
||||
|
||||
public DefaultPeerSelector peerSelector; |
||||
|
||||
@BeforeEach |
||||
public void beforeTest() { |
||||
ProtocolSpec protocolSpec = Mockito.mock(ProtocolSpec.class); |
||||
Mockito.when(protocolSpec.isPoS()).thenReturn(false); |
||||
peerSelector = new DefaultPeerSelector(() -> protocolSpec); |
||||
} |
||||
|
||||
@Test |
||||
public void testGetPeer() throws NoAvailablePeerException { |
||||
EthPeer expectedPeer = createTestPeer(10, EthProtocol.get(), 5); |
||||
peerSelector.addPeer(expectedPeer); |
||||
EthPeer excludedForLowChainHeightPeer = createTestPeer(5, EthProtocol.get(), 50); |
||||
peerSelector.addPeer(excludedForLowChainHeightPeer); |
||||
EthPeer excludedForWrongProtocolPeer = createTestPeer(10, SnapProtocol.get(), 50); |
||||
peerSelector.addPeer(excludedForWrongProtocolPeer); |
||||
EthPeer excludedForLowReputationPeer = createTestPeer(10, EthProtocol.get(), 1); |
||||
peerSelector.addPeer(excludedForLowReputationPeer); |
||||
EthPeer excludedForBeingAlreadyUsedPeer = createTestPeer(10, EthProtocol.get(), 50); |
||||
peerSelector.addPeer(excludedForBeingAlreadyUsedPeer); |
||||
|
||||
Set<EthPeer> usedEthPeers = new HashSet<>(); |
||||
usedEthPeers.add(excludedForBeingAlreadyUsedPeer); |
||||
|
||||
EthPeer result = peerSelector.getPeer(usedEthPeers, 10, EthProtocol.get()); |
||||
|
||||
Assertions.assertSame(expectedPeer, result); |
||||
} |
||||
|
||||
@Test |
||||
public void testGetPeerButNoPeerMatchesFilter() { |
||||
EthPeer expectedPeer = createTestPeer(10, EthProtocol.get(), 5); |
||||
peerSelector.addPeer(expectedPeer); |
||||
EthPeer excludedForLowChainHeightPeer = createTestPeer(5, EthProtocol.get(), 50); |
||||
peerSelector.addPeer(excludedForLowChainHeightPeer); |
||||
EthPeer excludedForWrongProtocolPeer = createTestPeer(10, SnapProtocol.get(), 50); |
||||
peerSelector.addPeer(excludedForWrongProtocolPeer); |
||||
EthPeer excludedForLowReputationPeer = createTestPeer(10, EthProtocol.get(), 1); |
||||
peerSelector.addPeer(excludedForLowReputationPeer); |
||||
EthPeer excludedForBeingAlreadyUsedPeer = createTestPeer(10, EthProtocol.get(), 50); |
||||
peerSelector.addPeer(excludedForBeingAlreadyUsedPeer); |
||||
|
||||
Set<EthPeer> usedEthPeers = new HashSet<>(); |
||||
usedEthPeers.add(excludedForBeingAlreadyUsedPeer); |
||||
|
||||
Assertions.assertThrows( |
||||
NoAvailablePeerException.class, |
||||
() -> peerSelector.getPeer(usedEthPeers, 10, new MockSubProtocol())); |
||||
} |
||||
|
||||
private EthPeer createTestPeer( |
||||
final long chainHeight, final SubProtocol protocol, final int reputation) { |
||||
EthPeer ethPeer = Mockito.mock(EthPeer.class); |
||||
PeerConnection peerConnection = Mockito.mock(PeerConnection.class); |
||||
Peer peer = Mockito.mock(Peer.class); |
||||
ChainState chainState = Mockito.mock(ChainState.class); |
||||
PeerReputation peerReputation = Mockito.mock(PeerReputation.class); |
||||
|
||||
Mockito.when(ethPeer.getConnection()).thenReturn(peerConnection); |
||||
Mockito.when(peerConnection.getPeer()).thenReturn(peer); |
||||
Mockito.when(ethPeer.getProtocolName()).thenReturn(protocol.getName()); |
||||
Mockito.when(ethPeer.chainState()).thenReturn(chainState); |
||||
Mockito.when(chainState.getEstimatedHeight()).thenReturn(chainHeight); |
||||
Mockito.when(ethPeer.getReputation()).thenReturn(peerReputation); |
||||
Mockito.when(peerReputation.getScore()).thenReturn(reputation); |
||||
|
||||
Mockito.when(ethPeer.compareTo(Mockito.any(EthPeer.class))) |
||||
.thenAnswer( |
||||
(invocationOnMock) -> { |
||||
EthPeer otherPeer = invocationOnMock.getArgument(0, EthPeer.class); |
||||
return Integer.compare(reputation, otherPeer.getReputation().getScore()); |
||||
}); |
||||
return ethPeer; |
||||
} |
||||
|
||||
private static class MockSubProtocol implements SubProtocol { |
||||
|
||||
@Override |
||||
public String getName() { |
||||
return "Mock"; |
||||
} |
||||
|
||||
@Override |
||||
public int messageSpace(final int protocolVersion) { |
||||
throw new UnsupportedOperationException(); |
||||
} |
||||
|
||||
@Override |
||||
public boolean isValidMessageCode(final int protocolVersion, final int code) { |
||||
throw new UnsupportedOperationException(); |
||||
} |
||||
|
||||
@Override |
||||
public String messageName(final int protocolVersion, final int code) { |
||||
throw new UnsupportedOperationException(); |
||||
} |
||||
} |
||||
} |
||||
Loading…
Reference in new issue