[zaptos] move commit vote broadcast into pipeline builder

Author: Zekun Li

consensus/consensus-types/src/pipelined_block.rs

@@ -560,16 +560,6 @@ impl PipelinedBlock {
             let _ = fut.commit_ledger_fut.await;
         }
     }
-
-    pub async fn wait_for_commit_vote(&self) -> Option<CommitVote> {
-        // may be aborted (e.g. by reset)
-        if let Some(fut) = self.pipeline_futs() {
-            // this may be cancelled
-            fut.commit_vote_fut.await.ok()
-        } else {
-            None
-        }
-    }
 }
 
 #[derive(Debug, Clone, Eq, PartialEq)]

consensus/src/pipeline/decoupled_execution_utils.rs

@@ -30,7 +30,7 @@ use std::sync::{
 pub fn prepare_phases_and_buffer_manager(
     author: Author,
     safety_rules: Arc<dyn CommitSignerProvider>,
-    commit_msg_tx: NetworkSender,
+    commit_msg_tx: Arc<NetworkSender>,
     commit_msg_rx: Receiver<AccountAddress, (AccountAddress, IncomingCommitRequest)>,
     block_rx: UnboundedReceiver<OrderedBlocks>,
     sync_rx: UnboundedReceiver<ResetRequest>,
@@ -95,7 +95,7 @@ pub fn prepare_phases_and_buffer_manager(
     let (persisting_phase_request_tx, persisting_phase_request_rx) =
         create_channel::<CountedRequest<PersistingRequest>>();
     let (persisting_phase_response_tx, persisting_phase_response_rx) = create_channel();
-    let commit_msg_tx = Arc::new(commit_msg_tx);
+    let commit_msg_tx = commit_msg_tx;
 
     let persisting_phase_processor = PersistingPhase::new(commit_msg_tx.clone());
     let persisting_phase = PipelinePhase::new(

consensus/src/pipeline/execution_client.rs

@@ -208,14 +208,8 @@ impl ExecutionProxyClient {
         buffer_manager_back_pressure_enabled: bool,
         consensus_observer_config: ConsensusObserverConfig,
         consensus_publisher: Option<Arc<ConsensusPublisher>>,
+        network_sender: Arc<NetworkSender>,
     ) {
-        let network_sender = NetworkSender::new(
-            self.author,
-            self.network_sender.clone(),
-            self.self_sender.clone(),
-            epoch_state.verifier.clone(),
-        );
-
         let (reset_buffer_manager_tx, reset_buffer_manager_rx) = unbounded::<ResetRequest>();
 
         let (commit_msg_tx, commit_msg_rx) =
@@ -240,7 +234,7 @@ impl ExecutionProxyClient {
                     rand_config,
                     fast_rand_config,
                     rand_ready_block_tx,
-                    Arc::new(network_sender.clone()),
+                    network_sender.clone(),
                     self.rand_storage.clone(),
                     self.bounded_executor.clone(),
                     &self.consensus_config.rand_rb_config,
@@ -319,6 +313,12 @@ impl TExecutionClient for ExecutionProxyClient {
         rand_msg_rx: aptos_channel::Receiver<AccountAddress, IncomingRandGenRequest>,
         highest_committed_round: Round,
     ) {
+        let network_sender = Arc::new(NetworkSender::new(
+            self.author,
+            self.network_sender.clone(),
+            self.self_sender.clone(),
+            epoch_state.verifier.clone(),
+        ));
         let maybe_rand_msg_tx = self.spawn_decoupled_execution(
             maybe_consensus_key,
             commit_signer_provider,
@@ -331,6 +331,7 @@ impl TExecutionClient for ExecutionProxyClient {
             self.consensus_config.enable_pre_commit,
             self.consensus_observer_config,
             self.consensus_publisher.clone(),
+            network_sender.clone(),
         );
 
         let transaction_shuffler =
@@ -353,6 +354,7 @@ impl TExecutionClient for ExecutionProxyClient {
             randomness_enabled,
             onchain_consensus_config.clone(),
             aux_version,
+            network_sender,
         );
 
         maybe_rand_msg_tx

consensus/src/pipeline/pipeline_builder.rs

@@ -6,6 +6,7 @@ use crate::{
     block_storage::tracing::{observe_block, BlockStage},
     counters::{self, update_counters_for_block, update_counters_for_compute_result},
     monitor,
+    network::NetworkSender,
     payload_manager::TPayloadManager,
     txn_notifier::TxnNotifier,
     IntGaugeGuard,
@@ -135,6 +136,7 @@ pub struct PipelineBuilder {
     persisted_auxiliary_info_version: u8,
     rand_check_enabled: bool,
     module_cache: Arc<Mutex<Option<ValidationState<CachedStateView>>>>,
+    network_sender: Arc<NetworkSender>,
 }
 
 fn spawn_shared_fut<
@@ -257,6 +259,7 @@ impl PipelineBuilder {
         enable_pre_commit: bool,
         consensus_onchain_config: &OnChainConsensusConfig,
         persisted_auxiliary_info_version: u8,
+        network_sender: Arc<NetworkSender>,
     ) -> Self {
         let module_cache = Arc::new(Mutex::new(None));
         Self {
@@ -274,6 +277,7 @@ impl PipelineBuilder {
             persisted_auxiliary_info_version,
             rand_check_enabled: consensus_onchain_config.rand_check_enabled(),
             module_cache,
+            network_sender,
         }
     }
 
@@ -434,14 +438,15 @@ impl PipelineBuilder {
             None,
         );
         let commit_vote_fut = spawn_shared_fut(
-            Self::sign_commit_vote(
+            Self::sign_and_broadcast_commit_vote(
                 ledger_update_fut.clone(),
                 order_vote_rx,
                 order_proof_fut.clone(),
                 commit_proof_fut.clone(),
                 self.signer.clone(),
                 block.clone(),
                 self.order_vote_enabled,
+                self.network_sender.clone(),
             ),
             Some(&mut abort_handles),
         );
@@ -860,15 +865,16 @@ impl PipelineBuilder {
     }
 
     /// Precondition: 1. ledger update finishes, 2. order vote or order proof or commit proof is received
-    /// What it does: Sign the commit vote with execution result, it needs to update the timestamp for reconfig suffix blocks
-    async fn sign_commit_vote(
+    /// What it does: Sign the commit vote with execution result and broadcast, it needs to update the timestamp for reconfig suffix blocks
+    async fn sign_and_broadcast_commit_vote(
         ledger_update_fut: TaskFuture<LedgerUpdateResult>,
         order_vote_rx: oneshot::Receiver<()>,
         order_proof_fut: TaskFuture<WrappedLedgerInfo>,
         commit_proof_fut: TaskFuture<LedgerInfoWithSignatures>,
         signer: Arc<ValidatorSigner>,
         block: Arc<Block>,
         order_vote_enabled: bool,
+        network_sender: Arc<NetworkSender>,
     ) -> TaskResult<CommitVoteResult> {
         let mut tracker = Tracker::start_waiting("sign_commit_vote", &block);
         let (compute_result, _, epoch_end_timestamp) = ledger_update_fut.await?;
@@ -907,11 +913,11 @@ impl PipelineBuilder {
         let ledger_info = LedgerInfo::new(block_info, consensus_data_hash);
         info!("[Pipeline] Signed ledger info {ledger_info}");
         let signature = signer.sign(&ledger_info).expect("Signing should succeed");
-        Ok(CommitVote::new_with_signature(
-            signer.author(),
-            ledger_info,
-            signature,
-        ))
+        let commit_vote = CommitVote::new_with_signature(signer.author(), ledger_info, signature);
+        network_sender
+            .broadcast_commit_vote(commit_vote.clone())
+            .await;
+        Ok(commit_vote)
     }
 
     /// Precondition: 1. ledger update finishes, 2. parent block's phase finishes 2. order proof is received

consensus/src/pipeline/tests/buffer_manager_tests.rs

@@ -114,12 +114,12 @@ pub fn prepare_buffer_manager(
 
     let (self_loop_tx, self_loop_rx) = aptos_channels::new_unbounded_test();
     let validators = Arc::new(validators);
-    let network = NetworkSender::new(
+    let network = Arc::new(NetworkSender::new(
         author,
         consensus_network_client,
         self_loop_tx,
         validators.clone(),
-    );
+    ));
 
     let (msg_tx, msg_rx) = aptos_channel::new::<
         AccountAddress,

consensus/src/round_manager.rs

@@ -1644,12 +1644,6 @@ impl RoundManager {
                 if let Some(tx) = proposed_block.pipeline_tx().lock().as_mut() {
                     let _ = tx.order_vote_tx.take().map(|tx| tx.send(()));
                 }
-                let network = self.network.clone();
-                tokio::spawn(async move {
-                    if let Some(commit_vote) = proposed_block.wait_for_commit_vote().await {
-                        network.broadcast_commit_vote(commit_vote).await;
-                    }
-                });
             }
             ORDER_VOTE_BROADCASTED.inc();
         }

consensus/src/state_computer.rs

@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{
-    block_preparer::BlockPreparer, error::StateSyncError, monitor,
+    block_preparer::BlockPreparer, error::StateSyncError, monitor, network::NetworkSender,
     payload_manager::TPayloadManager, pipeline::pipeline_builder::PipelineBuilder,
     state_replication::StateComputer, transaction_deduper::TransactionDeduper,
     transaction_shuffler::TransactionShuffler, txn_notifier::TxnNotifier,
@@ -46,6 +46,7 @@ struct MutableState {
     is_randomness_enabled: bool,
     consensus_onchain_config: OnChainConsensusConfig,
     persisted_auxiliary_info_version: u8,
+    network_sender: Arc<NetworkSender>,
 }
 
 /// Basic communication with the Execution module;
@@ -89,6 +90,7 @@ impl ExecutionProxy {
             is_randomness_enabled,
             consensus_onchain_config,
             persisted_auxiliary_info_version,
+            network_sender,
         } = self
             .state
             .read()
@@ -115,6 +117,7 @@ impl ExecutionProxy {
             self.enable_pre_commit,
             &consensus_onchain_config,
             persisted_auxiliary_info_version,
+            network_sender,
         )
     }
 }
@@ -235,6 +238,7 @@ impl StateComputer for ExecutionProxy {
         randomness_enabled: bool,
         consensus_onchain_config: OnChainConsensusConfig,
         persisted_auxiliary_info_version: u8,
+        network_sender: Arc<NetworkSender>,
     ) {
         *self.state.write() = Some(MutableState {
             validators: epoch_state
@@ -249,6 +253,7 @@ impl StateComputer for ExecutionProxy {
             is_randomness_enabled: randomness_enabled,
             consensus_onchain_config,
             persisted_auxiliary_info_version,
+            network_sender,
         });
     }
 

consensus/src/state_replication.rs

@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{
-    error::StateSyncError, payload_manager::TPayloadManager,
+    error::StateSyncError, network::NetworkSender, payload_manager::TPayloadManager,
     transaction_deduper::TransactionDeduper, transaction_shuffler::TransactionShuffler,
 };
 use anyhow::Result;
@@ -48,6 +48,7 @@ pub trait StateComputer: Send + Sync {
         randomness_enabled: bool,
         consensus_onchain_config: OnChainConsensusConfig,
         persisted_auxiliary_info_version: u8,
+        network_sender: Arc<NetworkSender>,
     );
 
     // Reconfigure to clear epoch state at end of epoch.

consensus/src/test_utils/mock_state_computer.rs

@@ -61,6 +61,7 @@ impl StateComputer for RandomComputeResultStateComputer {
         _: bool,
         _: OnChainConsensusConfig,
         _: u8,
+        _: Arc<crate::network::NetworkSender>,
     ) {
     }