perf(l1): improved discovery and peer initialization (#5147)

ElFantasma · MegaRedHand · web-flow · commit 1748ddc3b24c · 2025-11-01T03:01:13.000Z
**Motivation** We are too aggressive on our DiscoveryServer and Initiator. On start we send up to 5000 Udp messages every 5 seconds and start up to 5000 Tcp connections every 3 seconds. After several tests it was found that on one hand it is not necessary to be that eager, and on the other hand it caused several issues regarding the tokio runtime (ie. it was blocked for more than 10ms in several places, causing some problems when trying to synchronize from the local machine). **Description** This PR changes the way Lookup is performed: instead of trying a lot of connections at the same time every 3 seconds, it distributes the connection attempts in time, making it more gentle with the CPU. It also lowers the attempt rate. - Our Discovery lookup is more frequent, but tries 1 node at a time. - Our Connection attempt is more frequent, but tries 1 node at a time. - We match both processes rate to avoid having both, too much pending connections, or no pending connections at all. - Also noticed that spawning some GenServers in dedicated threads was no longer necessary, and it created a lot of threads, as every `start_on_thread` creates it's own Tokio Runtime with it's thread pool. - Also removed a spawn_blocking inside a GenServer that may cause a contention issue. ** Pending (for a future PR) ** - Will create a command line argument to set up the lookup frequency or eagerness: #5148 - Will make the lookup frequency adjustable at runtime, depending on the amount of connected peers: #5149 --------- Co-authored-by: Tomás Grüner <47506558+MegaRedHand@users.noreply.github.com>
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 ## Perf
 
+### 2025-10-31
+
+- Improved discovery and peer initialization [#5147](https://github.com/lambdaclass/ethrex/pull/5147)
+
 ### 2025-10-30
 
 - Pipeline Merkleization and Execution [#5084](https://github.com/lambdaclass/ethrex/pull/5084)
diff --git a/crates/networking/p2p/discv4/peer_table.rs b/crates/networking/p2p/discv4/peer_table.rs
@@ -17,7 +17,6 @@ use std::{
     time::{Duration, Instant},
 };
 use thiserror::Error;
-use tracing::debug;
 
 const MAX_SCORE: i64 = 50;
 const MIN_SCORE: i64 = -50;
@@ -127,7 +126,7 @@ pub struct PeerTable {
 impl PeerTable {
     pub fn spawn(target_peers: usize) -> PeerTable {
         PeerTable {
-            handle: PeerTableServer::new(target_peers).start_on_thread(),
+            handle: PeerTableServer::new(target_peers).start(),
         }
     }
 
@@ -320,25 +319,20 @@ impl PeerTable {
         }
     }
 
-    /// Get all contacts available to initiate a connection
-    pub async fn get_contacts_to_initiate(
-        &mut self,
-        amount: usize,
-    ) -> Result<Vec<Contact>, PeerTableError> {
-        match self
-            .handle
-            .call(CallMessage::GetContactsToInitiate(amount))
-            .await?
-        {
-            OutMessage::Contacts(contacts) => Ok(contacts),
+    /// Provide a contact to initiate a connection
+    pub async fn get_contact_to_initiate(&mut self) -> Result<Option<Contact>, PeerTableError> {
+        match self.handle.call(CallMessage::GetContactToInitiate).await? {
+            OutMessage::Contact(contact) => Ok(Some(contact)),
+            OutMessage::NotFound => Ok(None),
             _ => unreachable!(),
         }
     }
 
-    /// Get all contacts available for lookup
-    pub async fn get_contacts_for_lookup(&mut self) -> Result<Vec<Contact>, PeerTableError> {
-        match self.handle.call(CallMessage::GetContactsForLookup).await? {
-            OutMessage::Contacts(contacts) => Ok(contacts),
+    /// Provide a contact to perform Discovery lookup
+    pub async fn get_contact_for_lookup(&mut self) -> Result<Option<Contact>, PeerTableError> {
+        match self.handle.call(CallMessage::GetContactForLookup).await? {
+            OutMessage::Contact(contact) => Ok(Some(contact)),
+            OutMessage::NotFound => Ok(None),
             _ => unreachable!(),
         }
     }
@@ -574,10 +568,7 @@ impl PeerTableServer {
         }
     }
 
-    fn get_contacts_to_initiate(&mut self, max_amount: usize) -> Vec<Contact> {
-        let mut contacts = Vec::new();
-        let mut tried_connections = 0;
-
+    fn get_contact_to_initiate(&mut self) -> Option<Contact> {
         for contact in self.contacts.values() {
             let node_id = contact.node.node_id();
             if !self.peers.contains_key(&node_id)
@@ -587,29 +578,23 @@ impl PeerTableServer {
             {
                 self.already_tried_peers.insert(node_id);
 
-                contacts.push(contact.clone());
-
-                tried_connections += 1;
-                if tried_connections >= max_amount {
-                    break;
-                }
+                return Some(contact.clone());
             }
         }
-
-        if tried_connections < max_amount {
-            debug!("Resetting list of tried peers.");
-            self.already_tried_peers.clear();
-        }
-
-        contacts
+        // No untried contact found, resetting tried peers.
+        tracing::info!("Resetting list of tried peers.");
+        self.already_tried_peers.clear();
+        None
     }
 
-    fn get_contacts_for_lookup(&mut self) -> Vec<Contact> {
+    fn get_contact_for_lookup(&mut self) -> Option<Contact> {
         self.contacts
             .values()
             .filter(|c| c.n_find_node_sent < MAX_FIND_NODE_PER_PEER && !c.disposable)
+            .collect::<Vec<_>>()
+            .choose(&mut rand::rngs::OsRng)
+            .cloned()
             .cloned()
-            .collect()
     }
 
     fn get_contacts_to_revalidate(&mut self, revalidation_interval: Duration) -> Vec<Contact> {
@@ -635,7 +620,7 @@ impl PeerTableServer {
         if sender_ip != contact.node.ip {
             return OutMessage::IpMismatch;
         }
-        OutMessage::ValidContact(contact.clone())
+        OutMessage::Contact(contact.clone())
     }
 
     fn get_closest_nodes(&mut self, node_id: H256) -> Vec<Node> {
@@ -814,8 +799,8 @@ enum CallMessage {
     PeerCountByCapabilities { capabilities: Vec<Capability> },
     TargetReached,
     TargetPeersReached,
-    GetContactsToInitiate(usize),
-    GetContactsForLookup,
+    GetContactToInitiate,
+    GetContactForLookup,
     GetContactsToRevalidate(Duration),
     GetBestPeer { capabilities: Vec<Capability> },
     GetScore { node_id: H256 },
@@ -844,7 +829,7 @@ pub enum OutMessage {
     TargetReached(bool),
     IsNew(bool),
     Nodes(Vec<Node>),
-    ValidContact(Contact),
+    Contact(Contact),
     InvalidContact,
     UnknownContact,
     IpMismatch,
@@ -890,12 +875,14 @@ impl GenServer for PeerTableServer {
             CallMessage::TargetPeersReached => CallResponse::Reply(Self::OutMsg::TargetReached(
                 self.peers.len() >= self.target_peers,
             )),
-            CallMessage::GetContactsToInitiate(amount) => CallResponse::Reply(
-                Self::OutMsg::Contacts(self.get_contacts_to_initiate(amount)),
+            CallMessage::GetContactToInitiate => CallResponse::Reply(
+                self.get_contact_to_initiate()
+                    .map_or(Self::OutMsg::NotFound, Self::OutMsg::Contact),
+            ),
+            CallMessage::GetContactForLookup => CallResponse::Reply(
+                self.get_contact_for_lookup()
+                    .map_or(Self::OutMsg::NotFound, Self::OutMsg::Contact),
             ),
-            CallMessage::GetContactsForLookup => {
-                CallResponse::Reply(Self::OutMsg::Contacts(self.get_contacts_for_lookup()))
-            }
             CallMessage::GetContactsToRevalidate(revalidation_interval) => CallResponse::Reply(
                 Self::OutMsg::Contacts(self.get_contacts_to_revalidate(revalidation_interval)),
             ),
diff --git a/crates/networking/p2p/discv4/server.rs b/crates/networking/p2p/discv4/server.rs
@@ -39,8 +39,9 @@ const REVALIDATION_INTERVAL: Duration = Duration::from_secs(12 * 60 * 60); // 12
 /// The initial interval between peer lookups, until the number of peers reaches
 /// [target_peers](DiscoverySideCarState::target_peers), or the number of
 /// contacts reaches [target_contacts](DiscoverySideCarState::target_contacts).
-const INITIAL_LOOKUP_INTERVAL: Duration = Duration::from_secs(5);
-const LOOKUP_INTERVAL: Duration = Duration::from_secs(5 * 60); // 5 minutes
+pub const INITIAL_LOOKUP_INTERVAL: Duration = Duration::from_millis(100); // 10 per second
+pub const LOOKUP_INTERVAL: Duration = Duration::from_millis(600); // 100 per minute
+const CHANGE_FIND_NODE_MESSAGE_INTERVAL: Duration = Duration::from_secs(5);
 const PRUNE_INTERVAL: Duration = Duration::from_secs(5);
 
 #[derive(Debug, thiserror::Error)]
@@ -65,6 +66,7 @@ pub enum InMessage {
     Revalidate,
     Lookup,
     Prune,
+    ChangeFindNodeMessage,
     Shutdown,
 }
 
@@ -80,6 +82,9 @@ pub struct DiscoveryServer {
     signer: SecretKey,
     udp_socket: Arc<UdpSocket>,
     peer_table: PeerTable,
+    /// The last `FindNode` message sent, cached due to message
+    /// signatures being expensive.
+    find_node_message: BytesMut,
 }
 
 impl DiscoveryServer {
@@ -100,6 +105,7 @@ impl DiscoveryServer {
             signer,
             udp_socket,
             peer_table: peer_table.clone(),
+            find_node_message: Self::random_message(&signer),
         };
 
         info!(count = bootnodes.len(), "Adding bootnodes");
@@ -111,7 +117,7 @@ impl DiscoveryServer {
             .new_contacts(bootnodes, local_node.node_id())
             .await?;
 
-        discovery_server.start_on_thread();
+        discovery_server.start();
         Ok(())
     }
 
@@ -202,6 +208,18 @@ impl DiscoveryServer {
         Ok(())
     }
 
+    /// Generate and store a FindNodeMessage with a random key. We then send the same message on Disovery lookup.
+    /// We change this message every CHANGE_FIND_NODE_MESSAGE_INTERVAL.
+    fn random_message(signer: &SecretKey) -> BytesMut {
+        let expiration: u64 = get_msg_expiration_from_seconds(EXPIRATION_SECONDS);
+        let random_priv_key = SecretKey::new(&mut OsRng);
+        let random_pub_key = public_key_from_signing_key(&random_priv_key);
+        let msg = Message::FindNode(FindNodeMessage::new(random_pub_key, expiration));
+        let mut buf = BytesMut::new();
+        msg.encode_with_header(&mut buf, signer);
+        buf
+    }
+
     async fn revalidate(&mut self) -> Result<(), DiscoveryServerError> {
         for contact in self
             .peer_table
@@ -214,17 +232,9 @@ impl DiscoveryServer {
     }
 
     async fn lookup(&mut self) -> Result<(), DiscoveryServerError> {
-        // Sending the same FindNode message to all contacts to optimize message creation and encoding
-        let expiration: u64 = get_msg_expiration_from_seconds(EXPIRATION_SECONDS);
-        let random_priv_key = SecretKey::new(&mut OsRng);
-        let random_pub_key = public_key_from_signing_key(&random_priv_key);
-        let msg = Message::FindNode(FindNodeMessage::new(random_pub_key, expiration));
-        let mut buf = BytesMut::new();
-        msg.encode_with_header(&mut buf, &self.signer);
-
-        for contact in self.peer_table.get_contacts_for_lookup().await? {
-            if self.udp_socket.send_to(&buf, &contact.node.udp_addr()).await.inspect_err(
-                |e| error!(sending = ?msg, addr = ?&contact.node.udp_addr(), err=?e, "Error sending message"),
+        if let Some(contact) = self.peer_table.get_contact_for_lookup().await? {
+            if self.udp_socket.send_to(&self.find_node_message, &contact.node.udp_addr()).await.inspect_err(
+                |e| error!(sending = "FindNode", addr = ?&contact.node.udp_addr(), err=?e, "Error sending message"),
             ).is_err() {
                 self.peer_table
                     .set_disposable(&contact.node.node_id())
@@ -454,7 +464,7 @@ impl DiscoveryServer {
                 debug!(received = message_type, to = %format!("{sender_public_key:#x}"), "IP address mismatch, skipping");
                 Err(DiscoveryServerError::InvalidContact)
             }
-            PeerTableOutMessage::ValidContact(contact) => Ok(contact),
+            PeerTableOutMessage::Contact(contact) => Ok(contact),
             _ => unreachable!(),
         }
     }
@@ -505,8 +515,12 @@ impl GenServer for DiscoveryServer {
             InMessage::Revalidate,
         );
         send_interval(PRUNE_INTERVAL, handle.clone(), InMessage::Prune);
+        send_interval(
+            CHANGE_FIND_NODE_MESSAGE_INTERVAL,
+            handle.clone(),
+            InMessage::ChangeFindNodeMessage,
+        );
         let _ = handle.clone().cast(InMessage::Lookup).await;
-
         send_message_on(handle.clone(), tokio::signal::ctrl_c(), InMessage::Shutdown);
 
         Ok(Success(self))
@@ -548,6 +562,9 @@ impl GenServer for DiscoveryServer {
                     .await
                     .inspect_err(|e| error!(err=?e, "Error Pruning peer table"));
             }
+            Self::CastMsg::ChangeFindNodeMessage => {
+                self.find_node_message = Self::random_message(&self.signer);
+            }
             Self::CastMsg::Shutdown => return CastResponse::Stop,
         }
         CastResponse::NoReply
diff --git a/crates/networking/p2p/rlpx/connection/handshake.rs b/crates/networking/p2p/rlpx/connection/handshake.rs
@@ -63,7 +63,7 @@ pub(crate) async fn perform(
     eth_version: Arc<RwLock<EthCapVersion>>,
 ) -> Result<(Established, SplitStream<Framed<TcpStream, RLPxCodec>>), PeerConnectionError> {
     let (context, node, framed) = match state {
-        ConnectionState::Initiator(Initiator { context, node, .. }) => {
+        ConnectionState::Initiator(Initiator { context, node }) => {
             let addr = SocketAddr::new(node.ip, node.tcp_port);
             let mut stream = match tcp_stream(addr).await {
                 Ok(result) => result,
diff --git a/crates/networking/p2p/rlpx/connection/server.rs b/crates/networking/p2p/rlpx/connection/server.rs
@@ -1061,14 +1061,11 @@ async fn handle_incoming_message(
         }
         Message::GetByteCodes(req) => {
             let storage_clone = state.storage.clone();
-            let response =
-                tokio::task::spawn_blocking(move || process_byte_codes_request(req, storage_clone))
-                    .await
-                    .map_err(|_| {
-                        PeerConnectionError::InternalError(
-                            "Failed to execute bytecode retrieval task".to_string(),
-                        )
-                    })??;
+            let response = process_byte_codes_request(req, storage_clone).map_err(|_| {
+                PeerConnectionError::InternalError(
+                    "Failed to execute bytecode retrieval task".to_string(),
+                )
+            })?;
             send(state, Message::ByteCodes(response)).await?
         }
         Message::GetTrieNodes(req) => {
diff --git a/crates/networking/p2p/rlpx/initiator.rs b/crates/networking/p2p/rlpx/initiator.rs
