mirror of https://github.com/hyperledger/besu
Improve eth/66 support (#3616)
Currently Besu has a limited support for sending NewPooledTransactionHashes messages, and other aspect related to reduce transactions synchronization traffic, described in the Ethereum Wire Protocol version 66.
Specifically:
Besu only uses NewPooledTransactionHashes for new local transactions, while it could be extended to any transaction added to the transaction pool
Besu does not limit the sending of the full transaction messages to a small fraction of the connected peers, and sends the new transaction hashes to all the remaining peers
This PR, extends eth/66 support and does some code refactoring, to remove some reduntant code and rename some classes to identify they are related to the NewPooledTransactionHashes message.
The main changes are:
Do not have a separate tracker for transaction hashes, since for them we can reuse PeerTransactionTracker, that tracks full transactions exchange history and sending queue with a peer. So PeerPendingTransactionTracker has been removed. --tx-pool-hashes-max-size is now deprecated and has no more effect and it will be removed in a future release.
When a new peer connects, if it support eth/6[56] then we send all the transaction hashes we have in the pool, otherwise we send the full transactions.
When new transactions are added to the pool, we send full transactions to peers without eth/6[56] support, or to a small fractions of all peers, and then we send only transaction hashes to the remaining peer that support eth/6[56]. Both transactions and transaction hashes are only sent if not already exchanged with that specific peer.
Signed-off-by: Fabio Di Fabio <fabio.difabio@consensys.net>
pull/3629/head
Fabio Di Fabio
3 years ago
committed by
GitHub
parent
0d182b80ae
commit
6c179ba596
@ -1,101 +0,0 @@ |
||||
/* |
||||
* Copyright 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.transactions; |
||||
|
||||
import static java.util.Collections.emptySet; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthPeer; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.sorter.AbstractPendingTransactionsSorter; |
||||
import org.hyperledger.besu.ethereum.p2p.rlpx.wire.Capability; |
||||
|
||||
import java.util.Collection; |
||||
import java.util.Collections; |
||||
import java.util.LinkedHashMap; |
||||
import java.util.Map; |
||||
import java.util.Set; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.stream.Collectors; |
||||
|
||||
public class PeerPendingTransactionTracker implements EthPeer.DisconnectCallback { |
||||
private static final int MAX_TRACKED_SEEN_TRANSACTIONS = 100_000; |
||||
private final Map<EthPeer, Set<Hash>> seenTransactions = new ConcurrentHashMap<>(); |
||||
private final Map<EthPeer, Set<Hash>> transactionsToSend = new ConcurrentHashMap<>(); |
||||
private final AbstractPendingTransactionsSorter pendingTransactions; |
||||
|
||||
public PeerPendingTransactionTracker( |
||||
final AbstractPendingTransactionsSorter pendingTransactions) { |
||||
this.pendingTransactions = pendingTransactions; |
||||
} |
||||
|
||||
public synchronized void markTransactionsHashesAsSeen( |
||||
final EthPeer peer, final Collection<Hash> transactions) { |
||||
final Set<Hash> seenTransactionsForPeer = getOrCreateSeenTransactionsForPeer(peer); |
||||
transactions.stream().forEach(seenTransactionsForPeer::add); |
||||
} |
||||
|
||||
public synchronized void addToPeerSendQueue(final EthPeer peer, final Hash hash) { |
||||
if (!hasPeerSeenTransaction(peer, hash)) { |
||||
transactionsToSend.computeIfAbsent(peer, key -> createTransactionsSet()).add(hash); |
||||
} |
||||
} |
||||
|
||||
public Iterable<EthPeer> getEthPeersWithUnsentTransactions() { |
||||
return transactionsToSend.keySet(); |
||||
} |
||||
|
||||
public synchronized Set<Hash> claimTransactionsToSendToPeer(final EthPeer peer) { |
||||
final Set<Hash> transactionsToSend = this.transactionsToSend.remove(peer); |
||||
if (transactionsToSend != null) { |
||||
markTransactionsHashesAsSeen( |
||||
peer, |
||||
transactionsToSend.stream() |
||||
.filter(h -> pendingTransactions.getTransactionByHash(h).isPresent()) |
||||
.collect(Collectors.toSet())); |
||||
return transactionsToSend; |
||||
} else { |
||||
return emptySet(); |
||||
} |
||||
} |
||||
|
||||
public boolean isPeerSupported(final EthPeer peer, final Capability capability) { |
||||
return peer.getAgreedCapabilities().contains(capability); |
||||
} |
||||
|
||||
private Set<Hash> getOrCreateSeenTransactionsForPeer(final EthPeer peer) { |
||||
return seenTransactions.computeIfAbsent(peer, key -> createTransactionsSet()); |
||||
} |
||||
|
||||
private boolean hasPeerSeenTransaction(final EthPeer peer, final Hash hash) { |
||||
final Set<Hash> seenTransactionsForPeer = seenTransactions.get(peer); |
||||
return seenTransactionsForPeer != null && seenTransactionsForPeer.contains(hash); |
||||
} |
||||
|
||||
private <T> Set<T> createTransactionsSet() { |
||||
return Collections.newSetFromMap( |
||||
new LinkedHashMap<T, Boolean>(1 << 4, 0.75f, true) { |
||||
@Override |
||||
protected boolean removeEldestEntry(final Map.Entry<T, Boolean> eldest) { |
||||
return size() > MAX_TRACKED_SEEN_TRANSACTIONS; |
||||
} |
||||
}); |
||||
} |
||||
|
||||
@Override |
||||
public void onDisconnect(final EthPeer peer) { |
||||
seenTransactions.remove(peer); |
||||
transactionsToSend.remove(peer); |
||||
} |
||||
} |
||||
@ -1,52 +0,0 @@ |
||||
/* |
||||
* Copyright 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.transactions; |
||||
|
||||
import org.hyperledger.besu.ethereum.core.Transaction; |
||||
import org.hyperledger.besu.ethereum.eth.EthProtocol; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthContext; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool.TransactionBatchAddedListener; |
||||
|
||||
class PendingTransactionSender implements TransactionBatchAddedListener { |
||||
|
||||
private final PeerPendingTransactionTracker transactionTracker; |
||||
private final PendingTransactionsMessageSender transactionsMessageSender; |
||||
private final EthContext ethContext; |
||||
|
||||
public PendingTransactionSender( |
||||
final PeerPendingTransactionTracker transactionTracker, |
||||
final PendingTransactionsMessageSender transactionsMessageSender, |
||||
final EthContext ethContext) { |
||||
this.transactionTracker = transactionTracker; |
||||
this.transactionsMessageSender = transactionsMessageSender; |
||||
this.ethContext = ethContext; |
||||
} |
||||
|
||||
@Override |
||||
public void onTransactionsAdded(final Iterable<Transaction> transactions) { |
||||
ethContext |
||||
.getEthPeers() |
||||
.streamAvailablePeers() |
||||
.filter(peer -> transactionTracker.isPeerSupported(peer, EthProtocol.ETH65)) |
||||
.forEach( |
||||
peer -> |
||||
transactions.forEach( |
||||
transaction -> |
||||
transactionTracker.addToPeerSendQueue(peer, transaction.getHash()))); |
||||
ethContext |
||||
.getScheduler() |
||||
.scheduleSyncWorkerTask(transactionsMessageSender::sendTransactionsToPeers); |
||||
} |
||||
} |
||||
@ -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.transactions; |
||||
|
||||
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.AbstractPendingTransactionsSorter.TransactionInfo.toTransactionList; |
||||
import static org.hyperledger.besu.util.Slf4jLambdaHelper.traceLambda; |
||||
|
||||
import org.hyperledger.besu.ethereum.core.Transaction; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthContext; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthPeer; |
||||
import org.hyperledger.besu.ethereum.eth.messages.EthPV65; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool.TransactionBatchAddedListener; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.sorter.AbstractPendingTransactionsSorter; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.sorter.AbstractPendingTransactionsSorter.TransactionInfo; |
||||
|
||||
import java.util.ArrayList; |
||||
import java.util.Collections; |
||||
import java.util.List; |
||||
import java.util.Set; |
||||
|
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
public class TransactionBroadcaster implements TransactionBatchAddedListener { |
||||
private static final Logger LOG = LoggerFactory.getLogger(TransactionBroadcaster.class); |
||||
|
||||
private final AbstractPendingTransactionsSorter pendingTransactions; |
||||
private final PeerTransactionTracker transactionTracker; |
||||
private final TransactionsMessageSender transactionsMessageSender; |
||||
private final NewPooledTransactionHashesMessageSender newPooledTransactionHashesMessageSender; |
||||
private final EthContext ethContext; |
||||
private final int numPeersToSendFullTransactions; |
||||
|
||||
public TransactionBroadcaster( |
||||
final EthContext ethContext, |
||||
final AbstractPendingTransactionsSorter pendingTransactions, |
||||
final PeerTransactionTracker transactionTracker, |
||||
final TransactionsMessageSender transactionsMessageSender, |
||||
final NewPooledTransactionHashesMessageSender newPooledTransactionHashesMessageSender) { |
||||
this.pendingTransactions = pendingTransactions; |
||||
this.transactionTracker = transactionTracker; |
||||
this.transactionsMessageSender = transactionsMessageSender; |
||||
this.newPooledTransactionHashesMessageSender = newPooledTransactionHashesMessageSender; |
||||
this.ethContext = ethContext; |
||||
this.numPeersToSendFullTransactions = |
||||
(int) Math.ceil(Math.sqrt(ethContext.getEthPeers().getMaxPeers())); |
||||
} |
||||
|
||||
public void relayTransactionPoolTo(final EthPeer peer) { |
||||
Set<TransactionInfo> pendingTransactionInfo = pendingTransactions.getTransactionInfo(); |
||||
if (!pendingTransactionInfo.isEmpty()) { |
||||
if (peer.hasSupportForMessage(EthPV65.NEW_POOLED_TRANSACTION_HASHES)) { |
||||
sendTransactionHashes(toTransactionList(pendingTransactionInfo), List.of(peer)); |
||||
} else { |
||||
sendFullTransactions(toTransactionList(pendingTransactionInfo), List.of(peer)); |
||||
} |
||||
} |
||||
} |
||||
|
||||
@Override |
||||
public void onTransactionsAdded(final Iterable<Transaction> transactions) { |
||||
final int currPeerCount = ethContext.getEthPeers().peerCount(); |
||||
if (currPeerCount == 0) { |
||||
return; |
||||
} |
||||
|
||||
List<EthPeer> peersWithOnlyTransactionSupport = new ArrayList<>(currPeerCount); |
||||
List<EthPeer> peersWithTransactionHashesSupport = new ArrayList<>(currPeerCount); |
||||
|
||||
ethContext |
||||
.getEthPeers() |
||||
.streamAvailablePeers() |
||||
.forEach( |
||||
peer -> { |
||||
if (peer.hasSupportForMessage(EthPV65.NEW_POOLED_TRANSACTION_HASHES)) { |
||||
peersWithTransactionHashesSupport.add(peer); |
||||
} else { |
||||
peersWithOnlyTransactionSupport.add(peer); |
||||
} |
||||
}); |
||||
|
||||
if (peersWithOnlyTransactionSupport.size() < numPeersToSendFullTransactions) { |
||||
final int delta = |
||||
Math.min( |
||||
numPeersToSendFullTransactions - peersWithOnlyTransactionSupport.size(), |
||||
peersWithTransactionHashesSupport.size()); |
||||
|
||||
Collections.shuffle(peersWithTransactionHashesSupport); |
||||
|
||||
// move peers from the other list to reach the required size for full transaction peers
|
||||
movePeersBetweenLists( |
||||
peersWithTransactionHashesSupport, peersWithOnlyTransactionSupport, delta); |
||||
} |
||||
|
||||
traceLambda( |
||||
LOG, |
||||
"Sending full transactions to {} peers and transaction hashes to {} peers." |
||||
+ " Peers w/o eth/66 {}, peers with eth/66 {}", |
||||
peersWithOnlyTransactionSupport::size, |
||||
peersWithTransactionHashesSupport::size, |
||||
peersWithOnlyTransactionSupport::toString, |
||||
peersWithTransactionHashesSupport::toString); |
||||
|
||||
sendFullTransactions(transactions, peersWithOnlyTransactionSupport); |
||||
|
||||
sendTransactionHashes(transactions, peersWithTransactionHashesSupport); |
||||
} |
||||
|
||||
private void sendFullTransactions( |
||||
final Iterable<Transaction> transactions, final List<EthPeer> fullTransactionPeers) { |
||||
fullTransactionPeers.forEach( |
||||
peer -> { |
||||
transactions.forEach( |
||||
transaction -> transactionTracker.addToPeerSendQueue(peer, transaction)); |
||||
ethContext |
||||
.getScheduler() |
||||
.scheduleSyncWorkerTask(() -> transactionsMessageSender.sendTransactionsToPeer(peer)); |
||||
}); |
||||
} |
||||
|
||||
private void sendTransactionHashes( |
||||
final Iterable<Transaction> transactions, final List<EthPeer> transactionHashPeers) { |
||||
transactionHashPeers.stream() |
||||
.forEach( |
||||
peer -> { |
||||
transactions.forEach( |
||||
transaction -> transactionTracker.addToPeerSendQueue(peer, transaction)); |
||||
ethContext |
||||
.getScheduler() |
||||
.scheduleSyncWorkerTask( |
||||
() -> |
||||
newPooledTransactionHashesMessageSender.sendTransactionHashesToPeer( |
||||
peer)); |
||||
}); |
||||
} |
||||
|
||||
private void movePeersBetweenLists( |
||||
final List<EthPeer> sourceList, final List<EthPeer> destinationList, final int num) { |
||||
|
||||
final int stopIndex = sourceList.size() - num; |
||||
for (int i = sourceList.size() - 1; i >= stopIndex; i--) { |
||||
destinationList.add(sourceList.remove(i)); |
||||
} |
||||
} |
||||
} |
||||
@ -1,49 +0,0 @@ |
||||
/* |
||||
* Copyright 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.transactions; |
||||
|
||||
import org.hyperledger.besu.ethereum.core.Transaction; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthContext; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool.TransactionBatchAddedListener; |
||||
|
||||
class TransactionSender implements TransactionBatchAddedListener { |
||||
|
||||
private final PeerTransactionTracker transactionTracker; |
||||
private final TransactionsMessageSender transactionsMessageSender; |
||||
private final EthContext ethContext; |
||||
|
||||
public TransactionSender( |
||||
final PeerTransactionTracker transactionTracker, |
||||
final TransactionsMessageSender transactionsMessageSender, |
||||
final EthContext ethContext) { |
||||
this.transactionTracker = transactionTracker; |
||||
this.transactionsMessageSender = transactionsMessageSender; |
||||
this.ethContext = ethContext; |
||||
} |
||||
|
||||
@Override |
||||
public void onTransactionsAdded(final Iterable<Transaction> transactions) { |
||||
ethContext |
||||
.getEthPeers() |
||||
.streamAvailablePeers() |
||||
.forEach( |
||||
peer -> |
||||
transactions.forEach( |
||||
transaction -> transactionTracker.addToPeerSendQueue(peer, transaction))); |
||||
ethContext |
||||
.getScheduler() |
||||
.scheduleSyncWorkerTask(transactionsMessageSender::sendTransactionsToPeers); |
||||
} |
||||
} |
||||
@ -1,124 +0,0 @@ |
||||
/* |
||||
* Copyright 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.transactions; |
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
import static org.mockito.ArgumentMatchers.any; |
||||
import static org.mockito.Mockito.mock; |
||||
import static org.mockito.Mockito.when; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.core.BlockDataGenerator; |
||||
import org.hyperledger.besu.ethereum.core.Transaction; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthPeer; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.sorter.AbstractPendingTransactionsSorter; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.sorter.BaseFeePendingTransactionsSorter; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.sorter.GasPricePendingTransactionsSorter; |
||||
|
||||
import java.util.Arrays; |
||||
import java.util.Collection; |
||||
import java.util.Optional; |
||||
|
||||
import com.google.common.collect.ImmutableSet; |
||||
import org.junit.Before; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.junit.runners.Parameterized; |
||||
|
||||
@RunWith(Parameterized.class) |
||||
public class PeerPendingTransactionTrackerTest { |
||||
|
||||
@Parameterized.Parameter public AbstractPendingTransactionsSorter pendingTransactions; |
||||
|
||||
private final EthPeer ethPeer1 = mock(EthPeer.class); |
||||
private final EthPeer ethPeer2 = mock(EthPeer.class); |
||||
private final BlockDataGenerator generator = new BlockDataGenerator(); |
||||
private PeerPendingTransactionTracker tracker; |
||||
private final Hash hash1 = generator.transaction().getHash(); |
||||
private final Hash hash2 = generator.transaction().getHash(); |
||||
private final Hash hash3 = generator.transaction().getHash(); |
||||
|
||||
@Parameterized.Parameters |
||||
public static Collection<Object[]> data() { |
||||
return Arrays.asList( |
||||
new Object[][] { |
||||
{mock(GasPricePendingTransactionsSorter.class)}, |
||||
{mock(BaseFeePendingTransactionsSorter.class)} |
||||
}); |
||||
} |
||||
|
||||
@Before |
||||
public void setUp() { |
||||
tracker = new PeerPendingTransactionTracker(pendingTransactions); |
||||
Transaction tx = mock(Transaction.class); |
||||
when(pendingTransactions.getTransactionByHash(any())).thenReturn(Optional.of(tx)); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldTrackTransactionsToSendToPeer() { |
||||
tracker.addToPeerSendQueue(ethPeer1, hash1); |
||||
tracker.addToPeerSendQueue(ethPeer1, hash2); |
||||
tracker.addToPeerSendQueue(ethPeer2, hash3); |
||||
|
||||
assertThat(tracker.getEthPeersWithUnsentTransactions()).containsOnly(ethPeer1, ethPeer2); |
||||
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer1)).containsOnly(hash1, hash2); |
||||
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer2)).containsOnly(hash3); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldExcludeAlreadySeenTransactionsFromTransactionsToSend() { |
||||
tracker.markTransactionsHashesAsSeen(ethPeer1, ImmutableSet.of(hash2)); |
||||
|
||||
tracker.addToPeerSendQueue(ethPeer1, hash1); |
||||
tracker.addToPeerSendQueue(ethPeer1, hash2); |
||||
tracker.addToPeerSendQueue(ethPeer2, hash3); |
||||
|
||||
assertThat(tracker.getEthPeersWithUnsentTransactions()).containsOnly(ethPeer1, ethPeer2); |
||||
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer1)).containsOnly(hash1); |
||||
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer2)).containsOnly(hash3); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldExcludeAlreadySeenTransactionsAsACollectionFromTransactionsToSend() { |
||||
tracker.markTransactionsHashesAsSeen(ethPeer1, ImmutableSet.of(hash1, hash2)); |
||||
|
||||
tracker.addToPeerSendQueue(ethPeer1, hash1); |
||||
tracker.addToPeerSendQueue(ethPeer1, hash2); |
||||
tracker.addToPeerSendQueue(ethPeer2, hash3); |
||||
|
||||
assertThat(tracker.getEthPeersWithUnsentTransactions()).containsOnly(ethPeer2); |
||||
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer1)).isEmpty(); |
||||
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer2)).containsOnly(hash3); |
||||
} |
||||
|
||||
@Test |
||||
public void shouldClearDataWhenPeerDisconnects() { |
||||
tracker.markTransactionsHashesAsSeen(ethPeer1, ImmutableSet.of(hash3)); |
||||
|
||||
tracker.addToPeerSendQueue(ethPeer1, hash2); |
||||
tracker.addToPeerSendQueue(ethPeer2, hash3); |
||||
|
||||
tracker.onDisconnect(ethPeer1); |
||||
|
||||
assertThat(tracker.getEthPeersWithUnsentTransactions()).containsOnly(ethPeer2); |
||||
|
||||
// Should have cleared data that ethPeer1 has already seen transaction1
|
||||
tracker.addToPeerSendQueue(ethPeer1, hash1); |
||||
|
||||
assertThat(tracker.getEthPeersWithUnsentTransactions()).containsOnly(ethPeer1, ethPeer2); |
||||
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer1)).containsOnly(hash1); |
||||
assertThat(tracker.claimTransactionsToSendToPeer(ethPeer2)).containsOnly(hash3); |
||||
} |
||||
} |
||||
@ -1,67 +0,0 @@ |
||||
/* |
||||
* Copyright 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. |
||||
* |
||||
* SPDX-License-Identifier: Apache-2.0 |
||||
*/ |
||||
package org.hyperledger.besu.ethereum.eth.transactions; |
||||
|
||||
import static org.mockito.Mockito.mock; |
||||
import static org.mockito.Mockito.never; |
||||
import static org.mockito.Mockito.times; |
||||
import static org.mockito.Mockito.verify; |
||||
import static org.mockito.Mockito.when; |
||||
|
||||
import org.hyperledger.besu.datatypes.Hash; |
||||
import org.hyperledger.besu.ethereum.core.Transaction; |
||||
import org.hyperledger.besu.ethereum.eth.EthProtocol; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthContext; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthPeer; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthPeers; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthScheduler; |
||||
|
||||
import java.util.Arrays; |
||||
import java.util.Collections; |
||||
|
||||
import org.apache.tuweni.bytes.Bytes32; |
||||
import org.junit.Test; |
||||
|
||||
public class PendingTransactionsSenderTest { |
||||
|
||||
@Test |
||||
public void testSendEth65PeersOnly() { |
||||
PeerPendingTransactionTracker peerPendingTransactionTracker = |
||||
mock(PeerPendingTransactionTracker.class); |
||||
|
||||
PendingTransactionsMessageSender pendingTransactionsMessageSender = |
||||
mock(PendingTransactionsMessageSender.class); |
||||
EthContext ethContext = mock(EthContext.class); |
||||
EthScheduler ethScheduler = mock(EthScheduler.class); |
||||
when(ethContext.getScheduler()).thenReturn(ethScheduler); |
||||
PendingTransactionSender sender = |
||||
new PendingTransactionSender( |
||||
peerPendingTransactionTracker, pendingTransactionsMessageSender, ethContext); |
||||
|
||||
EthPeer peer1 = mock(EthPeer.class); |
||||
EthPeer peer2 = mock(EthPeer.class); |
||||
Transaction tx = mock(Transaction.class); |
||||
Hash hash = Hash.wrap(Bytes32.random()); |
||||
when(tx.getHash()).thenReturn(hash); |
||||
EthPeers ethPeers = mock(EthPeers.class); |
||||
when(ethContext.getEthPeers()).thenReturn(ethPeers); |
||||
when(ethPeers.streamAvailablePeers()).thenReturn(Arrays.asList(peer1, peer2).stream()); |
||||
when(peerPendingTransactionTracker.isPeerSupported(peer1, EthProtocol.ETH65)).thenReturn(true); |
||||
when(peerPendingTransactionTracker.isPeerSupported(peer2, EthProtocol.ETH65)).thenReturn(false); |
||||
sender.onTransactionsAdded(Collections.singleton(tx)); |
||||
verify(peerPendingTransactionTracker, times(1)).addToPeerSendQueue(peer1, hash); |
||||
verify(peerPendingTransactionTracker, never()).addToPeerSendQueue(peer2, hash); |
||||
} |
||||
} |
||||
@ -0,0 +1,272 @@ |
||||
/* |
||||
* 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.transactions; |
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat; |
||||
import static org.hyperledger.besu.ethereum.eth.transactions.sorter.AbstractPendingTransactionsSorter.TransactionInfo.toTransactionList; |
||||
import static org.mockito.ArgumentMatchers.any; |
||||
import static org.mockito.ArgumentMatchers.eq; |
||||
import static org.mockito.Mockito.doNothing; |
||||
import static org.mockito.Mockito.mock; |
||||
import static org.mockito.Mockito.times; |
||||
import static org.mockito.Mockito.verify; |
||||
import static org.mockito.Mockito.verifyNoInteractions; |
||||
import static org.mockito.Mockito.when; |
||||
|
||||
import org.hyperledger.besu.ethereum.core.BlockDataGenerator; |
||||
import org.hyperledger.besu.ethereum.core.Transaction; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthContext; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthPeer; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthPeers; |
||||
import org.hyperledger.besu.ethereum.eth.manager.EthScheduler; |
||||
import org.hyperledger.besu.ethereum.eth.messages.EthPV65; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.sorter.AbstractPendingTransactionsSorter; |
||||
import org.hyperledger.besu.ethereum.eth.transactions.sorter.AbstractPendingTransactionsSorter.TransactionInfo; |
||||
|
||||
import java.time.Instant; |
||||
import java.util.ArrayList; |
||||
import java.util.Collection; |
||||
import java.util.List; |
||||
import java.util.Set; |
||||
import java.util.stream.Collectors; |
||||
import java.util.stream.IntStream; |
||||
import java.util.stream.Stream; |
||||
|
||||
import org.junit.Before; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.mockito.ArgumentCaptor; |
||||
import org.mockito.Mock; |
||||
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
||||
@RunWith(MockitoJUnitRunner.class) |
||||
public class TransactionBroadcasterTest { |
||||
|
||||
@Mock private EthContext ethContext; |
||||
@Mock private EthPeers ethPeers; |
||||
@Mock private EthScheduler ethScheduler; |
||||
@Mock private AbstractPendingTransactionsSorter pendingTransactions; |
||||
@Mock private PeerTransactionTracker transactionTracker; |
||||
@Mock private TransactionsMessageSender transactionsMessageSender; |
||||
@Mock private NewPooledTransactionHashesMessageSender newPooledTransactionHashesMessageSender; |
||||
|
||||
private final EthPeer ethPeerNoEth66 = mock(EthPeer.class); |
||||
private final EthPeer ethPeerWithEth66 = mock(EthPeer.class); |
||||
private final EthPeer ethPeerNoEth66_2 = mock(EthPeer.class); |
||||
private final EthPeer ethPeerWithEth66_2 = mock(EthPeer.class); |
||||
private final EthPeer ethPeerWithEth66_3 = mock(EthPeer.class); |
||||
private final BlockDataGenerator generator = new BlockDataGenerator(); |
||||
|
||||
private TransactionBroadcaster txBroadcaster; |
||||
private ArgumentCaptor<Runnable> sendTaskCapture; |
||||
|
||||
@Before |
||||
public void setUp() { |
||||
when(ethPeerNoEth66.hasSupportForMessage(EthPV65.NEW_POOLED_TRANSACTION_HASHES)) |
||||
.thenReturn(Boolean.FALSE); |
||||
when(ethPeerNoEth66_2.hasSupportForMessage(EthPV65.NEW_POOLED_TRANSACTION_HASHES)) |
||||
.thenReturn(Boolean.FALSE); |
||||
when(ethPeerWithEth66.hasSupportForMessage(EthPV65.NEW_POOLED_TRANSACTION_HASHES)) |
||||
.thenReturn(Boolean.TRUE); |
||||
when(ethPeerWithEth66_2.hasSupportForMessage(EthPV65.NEW_POOLED_TRANSACTION_HASHES)) |
||||
.thenReturn(Boolean.TRUE); |
||||
when(ethPeerWithEth66_3.hasSupportForMessage(EthPV65.NEW_POOLED_TRANSACTION_HASHES)) |
||||
.thenReturn(Boolean.TRUE); |
||||
|
||||
sendTaskCapture = ArgumentCaptor.forClass(Runnable.class); |
||||
doNothing().when(ethScheduler).scheduleSyncWorkerTask(sendTaskCapture.capture()); |
||||
|
||||
when(ethPeers.getMaxPeers()).thenReturn(4); |
||||
|
||||
when(ethContext.getEthPeers()).thenReturn(ethPeers); |
||||
when(ethContext.getScheduler()).thenReturn(ethScheduler); |
||||
|
||||
txBroadcaster = |
||||
new TransactionBroadcaster( |
||||
ethContext, |
||||
pendingTransactions, |
||||
transactionTracker, |
||||
transactionsMessageSender, |
||||
newPooledTransactionHashesMessageSender); |
||||
} |
||||
|
||||
@Test |
||||
public void doNotRelayTransactionsWhenPoolIsEmpty() { |
||||
setupTransactionPool(0, 0); |
||||
|
||||
txBroadcaster.relayTransactionPoolTo(ethPeerNoEth66); |
||||
txBroadcaster.relayTransactionPoolTo(ethPeerWithEth66); |
||||
|
||||
verifyNothingSent(); |
||||
} |
||||
|
||||
@Test |
||||
public void relayFullTransactionsFromPoolWhenPeerDoesNotSupportEth66() { |
||||
List<Transaction> txs = toTransactionList(setupTransactionPool(1, 1)); |
||||
|
||||
txBroadcaster.relayTransactionPoolTo(ethPeerNoEth66); |
||||
|
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerNoEth66, txs); |
||||
|
||||
sendTaskCapture.getValue().run(); |
||||
|
||||
verify(transactionsMessageSender).sendTransactionsToPeer(ethPeerNoEth66); |
||||
verifyNoInteractions(newPooledTransactionHashesMessageSender); |
||||
} |
||||
|
||||
@Test |
||||
public void relayTransactionHashesFromPoolWhenPeerSupportEth66() { |
||||
List<Transaction> txs = toTransactionList(setupTransactionPool(1, 1)); |
||||
|
||||
txBroadcaster.relayTransactionPoolTo(ethPeerWithEth66); |
||||
|
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerWithEth66, txs); |
||||
|
||||
sendTaskCapture.getValue().run(); |
||||
|
||||
verify(newPooledTransactionHashesMessageSender).sendTransactionHashesToPeer(ethPeerWithEth66); |
||||
verifyNoInteractions(transactionsMessageSender); |
||||
} |
||||
|
||||
@Test |
||||
public void onTransactionsAddedWithNoPeersDoesNothing() { |
||||
when(ethPeers.peerCount()).thenReturn(0); |
||||
|
||||
txBroadcaster.onTransactionsAdded(toTransactionList(setupTransactionPool(1, 1))); |
||||
|
||||
verifyNothingSent(); |
||||
} |
||||
|
||||
@Test |
||||
public void onTransactionsAddedWithOnlyNonEth66PeersSendFullTransactions() { |
||||
when(ethPeers.peerCount()).thenReturn(2); |
||||
when(ethPeers.streamAvailablePeers()).thenReturn(Stream.of(ethPeerNoEth66, ethPeerNoEth66_2)); |
||||
|
||||
List<Transaction> txs = toTransactionList(setupTransactionPool(1, 1)); |
||||
|
||||
txBroadcaster.onTransactionsAdded(txs); |
||||
|
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerNoEth66, txs); |
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerNoEth66_2, txs); |
||||
|
||||
sendTaskCapture.getAllValues().forEach(Runnable::run); |
||||
|
||||
verify(transactionsMessageSender).sendTransactionsToPeer(ethPeerNoEth66); |
||||
verify(transactionsMessageSender).sendTransactionsToPeer(ethPeerNoEth66_2); |
||||
verifyNoInteractions(newPooledTransactionHashesMessageSender); |
||||
} |
||||
|
||||
@Test |
||||
public void onTransactionsAddedWithOnlyFewEth66PeersSendFullTransactions() { |
||||
when(ethPeers.peerCount()).thenReturn(2); |
||||
when(ethPeers.streamAvailablePeers()) |
||||
.thenReturn(Stream.of(ethPeerWithEth66, ethPeerWithEth66_2)); |
||||
|
||||
List<Transaction> txs = toTransactionList(setupTransactionPool(1, 1)); |
||||
|
||||
txBroadcaster.onTransactionsAdded(txs); |
||||
|
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerWithEth66, txs); |
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerWithEth66_2, txs); |
||||
|
||||
sendTaskCapture.getAllValues().forEach(Runnable::run); |
||||
|
||||
verify(transactionsMessageSender, times(2)).sendTransactionsToPeer(any(EthPeer.class)); |
||||
verifyNoInteractions(newPooledTransactionHashesMessageSender); |
||||
} |
||||
|
||||
@Test |
||||
public void onTransactionsAddedWithOnlyEth66PeersSendFullTransactionsAndTransactionHashes() { |
||||
when(ethPeers.peerCount()).thenReturn(3); |
||||
when(ethPeers.streamAvailablePeers()) |
||||
.thenReturn(Stream.of(ethPeerWithEth66, ethPeerWithEth66_2, ethPeerWithEth66_3)); |
||||
|
||||
List<Transaction> txs = toTransactionList(setupTransactionPool(1, 1)); |
||||
|
||||
txBroadcaster.onTransactionsAdded(txs); |
||||
|
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerWithEth66, txs); |
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerWithEth66_2, txs); |
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerWithEth66_3, txs); |
||||
|
||||
sendTaskCapture.getAllValues().forEach(Runnable::run); |
||||
|
||||
verify(transactionsMessageSender, times(2)).sendTransactionsToPeer(any(EthPeer.class)); |
||||
verify(newPooledTransactionHashesMessageSender).sendTransactionHashesToPeer(any(EthPeer.class)); |
||||
} |
||||
|
||||
@Test |
||||
public void onTransactionsAddedWithMixedPeersSendFullTransactionsAndTransactionHashes() { |
||||
List<EthPeer> eth66Peers = List.of(ethPeerWithEth66, ethPeerWithEth66_2); |
||||
|
||||
when(ethPeers.peerCount()).thenReturn(3); |
||||
when(ethPeers.streamAvailablePeers()) |
||||
.thenReturn(Stream.concat(eth66Peers.stream(), Stream.of(ethPeerNoEth66))); |
||||
|
||||
List<Transaction> txs = toTransactionList(setupTransactionPool(1, 1)); |
||||
|
||||
txBroadcaster.onTransactionsAdded(txs); |
||||
|
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerWithEth66, txs); |
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerWithEth66_2, txs); |
||||
verifyTransactionAddedToPeerSendingQueue(ethPeerNoEth66, txs); |
||||
|
||||
sendTaskCapture.getAllValues().forEach(Runnable::run); |
||||
|
||||
ArgumentCaptor<EthPeer> capPeerFullTransactions = ArgumentCaptor.forClass(EthPeer.class); |
||||
verify(transactionsMessageSender, times(2)) |
||||
.sendTransactionsToPeer(capPeerFullTransactions.capture()); |
||||
List<EthPeer> fullTransactionPeers = new ArrayList<>(capPeerFullTransactions.getAllValues()); |
||||
assertThat(fullTransactionPeers.remove(ethPeerNoEth66)).isTrue(); |
||||
assertThat(fullTransactionPeers).hasSize(1).first().isIn(eth66Peers); |
||||
|
||||
ArgumentCaptor<EthPeer> capPeerTransactionHashes = ArgumentCaptor.forClass(EthPeer.class); |
||||
verify(newPooledTransactionHashesMessageSender) |
||||
.sendTransactionHashesToPeer(capPeerTransactionHashes.capture()); |
||||
assertThat(capPeerTransactionHashes.getValue()).isIn(eth66Peers); |
||||
} |
||||
|
||||
private void verifyNothingSent() { |
||||
verifyNoInteractions( |
||||
transactionTracker, transactionsMessageSender, newPooledTransactionHashesMessageSender); |
||||
} |
||||
|
||||
private Set<TransactionInfo> setupTransactionPool( |
||||
final int numLocalTransactions, final int numRemoteTransactions) { |
||||
Set<TransactionInfo> txInfo = createTransactionInfoList(numLocalTransactions, true); |
||||
txInfo.addAll(createTransactionInfoList(numRemoteTransactions, false)); |
||||
|
||||
when(pendingTransactions.getTransactionInfo()).thenReturn(txInfo); |
||||
|
||||
return txInfo; |
||||
} |
||||
|
||||
private Set<TransactionInfo> createTransactionInfoList(final int num, final boolean local) { |
||||
return IntStream.range(0, num) |
||||
.mapToObj(unused -> generator.transaction()) |
||||
.map(tx -> new TransactionInfo(tx, local, Instant.now())) |
||||
.collect(Collectors.toSet()); |
||||
} |
||||
|
||||
private void verifyTransactionAddedToPeerSendingQueue( |
||||
final EthPeer peer, final Collection<Transaction> transactions) { |
||||
|
||||
ArgumentCaptor<Transaction> trackedTransactions = ArgumentCaptor.forClass(Transaction.class); |
||||
verify(transactionTracker, times(transactions.size())) |
||||
.addToPeerSendQueue(eq(peer), trackedTransactions.capture()); |
||||
assertThat(trackedTransactions.getAllValues()) |
||||
.containsExactlyInAnyOrderElementsOf(transactions); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue