Reduce latency of cloning network sender using Arc pointers (#12103)

* Avoid cloning network sender using Arc pointers

* Removing a clone

* 100 node sweep test

* Removing a few clone operations

* reset forge test

* Removing some clones

* Removing clones

consensus/src/block_storage/sync_manager.rs

@@ -346,7 +346,7 @@ impl BlockStore {
     async fn sync_to_highest_commit_cert(
         &self,
         ledger_info: &LedgerInfoWithSignatures,
-        network: &NetworkSender,
+        network: &Arc<NetworkSender>,
     ) {
         // if the block exists between commit root and ordered root
         if self.commit_root().round() < ledger_info.commit_info().round()
@@ -404,15 +404,15 @@ impl BlockStore {
 
 /// BlockRetriever is used internally to retrieve blocks
 pub struct BlockRetriever {
-    network: NetworkSender,
+    network: Arc<NetworkSender>,
     preferred_peer: Author,
     validator_addresses: Vec<AccountAddress>,
     max_blocks_to_request: u64,
 }
 
 impl BlockRetriever {
     pub fn new(
-        network: NetworkSender,
+        network: Arc<NetworkSender>,
         preferred_peer: Author,
         validator_addresses: Vec<AccountAddress>,
         max_blocks_to_request: u64,

consensus/src/epoch_manager.rs

@@ -681,7 +681,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
         ledger_data: LedgerRecoveryData,
         onchain_consensus_config: OnChainConsensusConfig,
         epoch_state: Arc<EpochState>,
-        network_sender: NetworkSender,
+        network_sender: Arc<NetworkSender>,
     ) {
         let (recovery_manager_tx, recovery_manager_rx) = aptos_channel::new(
             QueueStyle::LIFO,
@@ -731,7 +731,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
                 self.quorum_store_to_mempool_sender.clone(),
                 self.config.mempool_txn_pull_timeout_ms,
                 self.storage.aptos_db().clone(),
-                network_sender.clone(),
+                network_sender,
                 epoch_state.verifier.clone(),
                 self.config.safety_rules.backend.clone(),
                 self.quorum_store_storage.clone(),
@@ -827,7 +827,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
         recovery_data: RecoveryData,
         epoch_state: Arc<EpochState>,
         onchain_consensus_config: OnChainConsensusConfig,
-        network_sender: NetworkSender,
+        network_sender: Arc<NetworkSender>,
         payload_client: Arc<dyn PayloadClient>,
         payload_manager: Arc<PayloadManager>,
         features: Features,
@@ -1085,7 +1085,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
                     initial_data,
                     epoch_state,
                     consensus_config,
-                    network_sender,
+                    Arc::new(network_sender),
                     payload_client,
                     payload_manager,
                     features,
@@ -1098,7 +1098,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
                     ledger_data,
                     consensus_config,
                     epoch_state,
-                    network_sender,
+                    Arc::new(network_sender),
                 )
                 .await
             },

consensus/src/network.rs

@@ -295,11 +295,12 @@ impl NetworkSender {
     /// The future is fulfilled as soon as the message is put into the mpsc channel to network
     /// internal (to provide back pressure), it does not indicate the message is delivered or sent
     /// out.
-    async fn broadcast(&mut self, msg: ConsensusMsg) {
+    async fn broadcast(&self, msg: ConsensusMsg) {
         fail_point!("consensus::send::any", |_| ());
         // Directly send the message to ourself without going through network.
         let self_msg = Event::Message(self.author, msg.clone());
-        if let Err(err) = self.self_sender.send(self_msg).await {
+        let mut self_sender = self.self_sender.clone();
+        if let Err(err) = self_sender.send(self_msg).await {
             error!("Error broadcasting to self: {:?}", err);
         }
 
@@ -369,25 +370,25 @@ impl NetworkSender {
         }
     }
 
-    pub async fn broadcast_proposal(&mut self, proposal_msg: ProposalMsg) {
+    pub async fn broadcast_proposal(&self, proposal_msg: ProposalMsg) {
         fail_point!("consensus::send::broadcast_proposal", |_| ());
         let msg = ConsensusMsg::ProposalMsg(Box::new(proposal_msg));
         self.broadcast(msg).await
     }
 
-    pub async fn broadcast_sync_info(&mut self, sync_info_msg: SyncInfo) {
+    pub async fn broadcast_sync_info(&self, sync_info_msg: SyncInfo) {
         fail_point!("consensus::send::broadcast_sync_info", |_| ());
         let msg = ConsensusMsg::SyncInfo(Box::new(sync_info_msg));
         self.broadcast(msg).await
     }
 
-    pub async fn broadcast_timeout_vote(&mut self, timeout_vote_msg: VoteMsg) {
+    pub async fn broadcast_timeout_vote(&self, timeout_vote_msg: VoteMsg) {
         fail_point!("consensus::send::broadcast_timeout_vote", |_| ());
         let msg = ConsensusMsg::VoteMsg(Box::new(timeout_vote_msg));
         self.broadcast(msg).await
     }
 
-    pub async fn broadcast_epoch_change(&mut self, epoch_change_proof: EpochChangeProof) {
+    pub async fn broadcast_epoch_change(&self, epoch_change_proof: EpochChangeProof) {
         fail_point!("consensus::send::broadcast_epoch_change", |_| ());
         let msg = ConsensusMsg::EpochChangeProof(Box::new(epoch_change_proof));
         self.broadcast(msg).await

consensus/src/quorum_store/batch_coordinator.rs

@@ -23,7 +23,7 @@ pub enum BatchCoordinatorCommand {
 
 pub struct BatchCoordinator {
     my_peer_id: PeerId,
-    network_sender: NetworkSender,
+    network_sender: Arc<NetworkSender>,
     batch_store: Arc<BatchStore>,
     max_batch_txns: u64,
     max_batch_bytes: u64,
@@ -43,7 +43,7 @@ impl BatchCoordinator {
     ) -> Self {
         Self {
             my_peer_id,
-            network_sender,
+            network_sender: Arc::new(network_sender),
             batch_store,
             max_batch_txns,
             max_batch_bytes,

consensus/src/recovery_manager.rs

@@ -27,7 +27,7 @@ use std::{mem::Discriminant, process, sync::Arc};
 /// for processing the events carrying sync info and use the info to retrieve blocks from peers
 pub struct RecoveryManager {
     epoch_state: Arc<EpochState>,
-    network: NetworkSender,
+    network: Arc<NetworkSender>,
     storage: Arc<dyn PersistentLivenessStorage>,
     state_computer: Arc<dyn StateComputer>,
     last_committed_round: Round,
@@ -38,7 +38,7 @@ pub struct RecoveryManager {
 impl RecoveryManager {
     pub fn new(
         epoch_state: Arc<EpochState>,
-        network: NetworkSender,
+        network: Arc<NetworkSender>,
         storage: Arc<dyn PersistentLivenessStorage>,
         state_computer: Arc<dyn StateComputer>,
         last_committed_round: Round,

consensus/src/round_manager.rs

@@ -188,7 +188,7 @@ pub struct RoundManager {
     proposer_election: UnequivocalProposerElection,
     proposal_generator: ProposalGenerator,
     safety_rules: Arc<Mutex<MetricsSafetyRules>>,
-    network: NetworkSender,
+    network: Arc<NetworkSender>,
     storage: Arc<dyn PersistentLivenessStorage>,
     onchain_config: OnChainConsensusConfig,
     vtxn_config: ValidatorTxnConfig,
@@ -205,7 +205,7 @@ impl RoundManager {
         proposer_election: Arc<dyn ProposerElection + Send + Sync>,
         proposal_generator: ProposalGenerator,
         safety_rules: Arc<Mutex<MetricsSafetyRules>>,
-        network: NetworkSender,
+        network: Arc<NetworkSender>,
         storage: Arc<dyn PersistentLivenessStorage>,
         onchain_config: OnChainConsensusConfig,
         buffered_proposal_tx: aptos_channel::Sender<Author, VerifiedEvent>,
@@ -295,15 +295,14 @@ impl RoundManager {
             self.log_collected_vote_stats(&new_round_event);
             self.round_state.setup_leader_timeout();
             let proposal_msg = self.generate_proposal(new_round_event).await?;
-            let mut network = self.network.clone();
             #[cfg(feature = "failpoints")]
             {
                 if self.check_whether_to_inject_reconfiguration_error() {
                     self.attempt_to_inject_reconfiguration_error(&proposal_msg)
                         .await?;
                 }
             }
-            network.broadcast_proposal(proposal_msg).await;
+            self.network.broadcast_proposal(proposal_msg).await;
             counters::PROPOSALS_COUNT.inc();
         }
         Ok(())
@@ -384,7 +383,7 @@ impl RoundManager {
     ) -> anyhow::Result<ProposalMsg> {
         // Proposal generator will ensure that at most one proposal is generated per round
         let sync_info = self.block_store.sync_info();
-        let mut sender = self.network.clone();
+        let sender = self.network.clone();
         let callback = async move {
             sender.broadcast_sync_info(sync_info).await;
         }
@@ -1168,7 +1167,6 @@ impl RoundManager {
                 .collect();
             half_peers.truncate(half_peers.len() / 2);
             self.network
-                .clone()
                 .send_proposal(proposal_msg.clone(), half_peers)
                 .await;
             Err(anyhow::anyhow!("Injected error in reconfiguration suffix"))

consensus/src/round_manager_fuzzing.rs

@@ -155,12 +155,12 @@ fn create_node_for_fuzzing() -> RoundManager {
         epoch: 1,
         verifier: storage.get_validator_set().into(),
     });
-    let network = NetworkSender::new(
+    let network = Arc::new(NetworkSender::new(
         signer.author(),
         consensus_network_client,
         self_sender,
         epoch_state.verifier.clone(),
-    );
+    ));
 
     // TODO: mock
     let block_store = build_empty_store(storage.clone(), initial_data);

consensus/src/round_manager_test.rs

@@ -243,7 +243,12 @@ impl NodeSetup {
         playground.add_node(twin_id, consensus_tx, network_reqs_rx, conn_mgr_reqs_rx);
 
         let (self_sender, self_receiver) = aptos_channels::new_test(1000);
-        let network = NetworkSender::new(author, consensus_network_client, self_sender, validators);
+        let network = Arc::new(NetworkSender::new(
+            author,
+            consensus_network_client,
+            self_sender,
+            validators,
+        ));
 
         let all_network_events = Box::new(select(network_events, self_receiver));