perf: use blocking tasks for some heavy inbound operations

Signed-off-by: ljedrz <[email protected]>
Mona22vs · May 23, 2023 · 3d3ae96 · 3d3ae96
1 parent 02ba5c2
commit 3d3ae96
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 8 deletions.
diff --git a/node/router/src/inbound.rs b/node/router/src/inbound.rs
@@ -29,6 +29,7 @@ use snarkvm::prelude::{Block, EpochChallenge, Header, Network, ProverSolution, T
 
 use anyhow::{bail, ensure, Result};
 use std::{net::SocketAddr, time::Instant};
+use tokio::task::spawn_blocking;
 
 #[async_trait]
 pub trait Inbound<N: Network>: Reading + Outbound<N> {
@@ -115,7 +116,8 @@ pub trait Inbound<N: Network>: Reading + Outbound<N> {
                     bail!("Block request from '{peer_ip}' has an excessive range ({start_height}..{end_height})")
                 }
 
-                match self.block_request(peer_ip, message) {
+                let node = self.clone();
+                match spawn_blocking(move || node.block_request(peer_ip, message)).await? {
                     true => Ok(()),
                     false => bail!("Peer '{peer_ip}' sent an invalid block request"),
                 }
@@ -150,7 +152,8 @@ pub trait Inbound<N: Network>: Reading + Outbound<N> {
                 }
 
                 // Process the block response.
-                match self.block_response(peer_ip, blocks.0) {
+                let node = self.clone();
+                match spawn_blocking(move || node.block_response(peer_ip, blocks.0)).await? {
                     true => Ok(()),
                     false => bail!("Peer '{peer_ip}' sent an invalid block response"),
                 }

diff --git a/node/src/beacon/router.rs b/node/src/beacon/router.rs
@@ -20,6 +20,7 @@ use snarkos_node_tcp::{Connection, ConnectionSide, Tcp};
 use snarkvm::prelude::{error, EpochChallenge, Header};
 
 use std::{io, net::SocketAddr};
+use tokio::task::spawn_blocking;
 
 impl<N: Network, C: ConsensusStorage<N>> P2P for Beacon<N, C> {
     /// Returns a reference to the TCP instance.
@@ -234,9 +235,17 @@ impl<N: Network, C: ConsensusStorage<N>> Inbound<N> for Beacon<N, C> {
         solution: ProverSolution<N>,
     ) -> bool {
         // Add the unconfirmed solution to the memory pool.
-        if let Err(error) = self.consensus.add_unconfirmed_solution(&solution) {
-            trace!("[UnconfirmedSolution] {error}");
-            return true; // Maintain the connection.
+        let node = self.clone();
+        match spawn_blocking(move || node.consensus.add_unconfirmed_solution(&solution)).await {
+            Ok(Err(error)) => {
+                trace!("[UnconfirmedSolution] {error}");
+                return true; // Maintain the connection.
+            }
+            Err(error) => {
+                trace!("[UnconfirmedSolution] {error}");
+                return true; // Maintain the connection.
+            }
+            _ => {}
         }
         let message = Message::UnconfirmedSolution(serialized);
         // Propagate the "UnconfirmedSolution" to the connected beacons.

diff --git a/node/src/validator/router.rs b/node/src/validator/router.rs
@@ -29,6 +29,7 @@ use snarkos_node_tcp::{Connection, ConnectionSide, Tcp};
 use snarkvm::prelude::{error, EpochChallenge, Network, Transaction};
 
 use std::{io, net::SocketAddr, time::Duration};
+use tokio::task::spawn_blocking;
 
 impl<N: Network, C: ConsensusStorage<N>> P2P for Validator<N, C> {
     /// Returns a reference to the TCP instance.
@@ -226,9 +227,17 @@ impl<N: Network, C: ConsensusStorage<N>> Inbound<N> for Validator<N, C> {
         solution: ProverSolution<N>,
     ) -> bool {
         // Add the unconfirmed solution to the memory pool.
-        if let Err(error) = self.consensus.add_unconfirmed_solution(&solution) {
-            trace!("[UnconfirmedSolution] {error}");
-            return true; // Maintain the connection.
+        let node = self.clone();
+        match spawn_blocking(move || node.consensus.add_unconfirmed_solution(&solution)).await {
+            Ok(Err(error)) => {
+                trace!("[UnconfirmedSolution] {error}");
+                return true; // Maintain the connection.
+            }
+            Err(error) => {
+                trace!("[UnconfirmedSolution] {error}");
+                return true; // Maintain the connection.
+            }
+            _ => {}
         }
         let message = Message::UnconfirmedSolution(serialized);
         // Propagate the "UnconfirmedSolution" to the connected beacons.