feat(net): send full transactions to fraction of all peers (paradigmx…

…yz#272) * refactor(net): use shared objects on a per peer basis * feat(net): send full transactions to fraction of all peers Co-authored-by: Georgios Konstantopoulos <[email protected]>
zakisaad · Dec 2, 2022 · 5300c83 · 5300c83
1 parent debc871
commit 5300c83
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 6 deletions.
diff --git a/crates/net/network/src/network.rs b/crates/net/network/src/network.rs
@@ -7,7 +7,7 @@ use crate::{
 };
 use parking_lot::Mutex;
 use reth_eth_wire::{NewBlock, NewPooledTransactionHashes, SharedTransactions};
-use reth_primitives::{PeerId, TransactionSigned, H256};
+use reth_primitives::{PeerId, TransactionSigned, TxHash, H256};
 use std::{
     net::SocketAddr,
     sync::{
@@ -130,6 +130,14 @@ impl NetworkHandle {
         self.send_message(NetworkHandleMessage::EthRequest { peer_id, request })
     }
 
+    /// Send transactions hashes to the peer.
+    pub fn send_transactions_hashes(&self, peer_id: PeerId, msg: Vec<TxHash>) {
+        self.send_message(NetworkHandleMessage::SendPooledTransactionHashes {
+            peer_id,
+            msg: NewPooledTransactionHashes(msg),
+        })
+    }
+
     /// Send full transactions to the peer
     pub fn send_transactions(&self, peer_id: PeerId, msg: Vec<Arc<TransactionSigned>>) {
         self.send_message(NetworkHandleMessage::SendTransaction {

diff --git a/crates/net/network/src/transactions.rs b/crates/net/network/src/transactions.rs
@@ -201,16 +201,29 @@ where
     ) -> PropagatedTransactions {
         let mut propagated = PropagatedTransactions::default();
 
-        for (peer_id, peer) in self.peers.iter_mut() {
+        // send full transactions to a fraction fo the connected peers (square root of the total
+        // number of connected peers)
+        let max_num_full = (self.peers.len() as f64).sqrt() as usize + 1;
+
+        // Note: Assuming ~random~ order due to random state of the peers map hasher
+        for (idx, (peer_id, peer)) in self.peers.iter_mut().enumerate() {
             let (hashes, full): (Vec<_>, Vec<_>) =
                 txs.iter().filter(|(hash, _)| peer.transactions.insert(*hash)).cloned().unzip();
 
             if !full.is_empty() {
-                // TODO select peer for full or hash
-                self.network.send_transactions(*peer_id, full);
+                if idx > max_num_full {
+                    for hash in &hashes {
+                        propagated.0.entry(*hash).or_default().push(PropagateKind::Hash(*peer_id));
+                    }
+                    // send hashes of transactions
+                    self.network.send_transactions_hashes(*peer_id, hashes);
+                } else {
+                    // send full transactions
+                    self.network.send_transactions(*peer_id, full);
 
-                for hash in hashes {
-                    propagated.0.entry(hash).or_default().push(PropagateKind::Full(*peer_id));
+                    for hash in hashes {
+                        propagated.0.entry(hash).or_default().push(PropagateKind::Full(*peer_id));
+                    }
                 }
             }
         }