[pipeline] impl Drop to ensure futures are properly gc'ed

Ends up cleaning the PipelineBlock to make sure it's only shared with Arc instead of cloning, so we can abort the pipeline upon drop.

Author: Zekun Li

config/src/config/consensus_config.rs

@@ -226,7 +226,7 @@ impl Default for ConsensusConfig {
                         min_block_time_ms_to_activate: 100,
                         min_blocks_to_activate: 4,
                         metric: ExecutionBackpressureMetric::Percentile(0.5),
-                        target_block_time_ms: 125,
+                        target_block_time_ms: 120,
                     },
                     min_calibrated_txns_per_block: 8,
                 }),
@@ -236,7 +236,7 @@ impl Default for ConsensusConfig {
                         min_block_time_ms_to_activate: 10,
                         min_blocks_to_activate: 4,
                         metric: ExecutionBackpressureMetric::Mean,
-                        target_block_time_ms: 125,
+                        target_block_time_ms: 120,
                     },
                     block_execution_overhead_ms: 10,
                     min_calibrated_block_gas_limit: 2000,

consensus/consensus-types/src/pipelined_block.rs

@@ -184,37 +184,39 @@ impl OrderedBlockWindow {
 /// A representation of a block that has been added to the execution pipeline. It might either be in ordered
 /// or in executed state. In the ordered state, the block is waiting to be executed. In the executed state,
 /// the block has been executed and the output is available.
-#[derive(Derivative, Clone)]
-#[derivative(Eq, PartialEq)]
+/// This struct is not Cloneable, use Arc to share it.
+#[derive(Derivative)]
 pub struct PipelinedBlock {
     /// Block data that cannot be regenerated.
     block: Block,
     /// A window of blocks that are needed for execution with the execution pool, EXCLUDING the current block
-    #[derivative(PartialEq = "ignore")]
     block_window: OrderedBlockWindow,
-    /// Input transactions in the order of execution
+    /// Input transactions in the order of execution. DEPRECATED stay for serialization compatibility.
     input_transactions: Vec<SignedTransaction>,
     /// The state_compute_result is calculated for all the pending blocks prior to insertion to
     /// the tree. The execution results are not persisted: they're recalculated again for the
     /// pending blocks upon restart.
-    #[derivative(PartialEq = "ignore")]
-    state_compute_result: StateComputeResult,
+    state_compute_result: Mutex<StateComputeResult>,
     randomness: OnceCell<Randomness>,
     pipeline_insertion_time: OnceCell<Instant>,
-    execution_summary: Arc<OnceCell<ExecutionSummary>>,
-    #[derivative(PartialEq = "ignore")]
-    pre_commit_fut: Arc<Mutex<Option<BoxFuture<'static, ExecutorResult<()>>>>>,
-    // pipeline related fields
-    #[derivative(PartialEq = "ignore")]
-    pipeline_futs: Arc<Mutex<Option<PipelineFutures>>>,
-    #[derivative(PartialEq = "ignore")]
-    pipeline_tx: Arc<Mutex<Option<PipelineInputTx>>>,
-    #[derivative(PartialEq = "ignore")]
-    pipeline_abort_handle: Arc<Mutex<Option<Vec<AbortHandle>>>>,
-    #[derivative(PartialEq = "ignore")]
-    block_qc: Arc<Mutex<Option<Arc<QuorumCert>>>>,
+    execution_summary: OnceCell<ExecutionSummary>,
+    pre_commit_fut: Mutex<Option<BoxFuture<'static, ExecutorResult<()>>>>,
+    /// pipeline related fields
+    pipeline_futs: Mutex<Option<PipelineFutures>>,
+    pipeline_tx: Mutex<Option<PipelineInputTx>>,
+    pipeline_abort_handle: Mutex<Option<Vec<AbortHandle>>>,
+    block_qc: Mutex<Option<Arc<QuorumCert>>>,
 }
 
+impl PartialEq for PipelinedBlock {
+    fn eq(&self, other: &Self) -> bool {
+        self.block == other.block
+            && self.input_transactions == other.input_transactions
+            && self.randomness.get() == other.randomness.get()
+    }
+}
+impl Eq for PipelinedBlock {}
+
 impl Serialize for PipelinedBlock {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
@@ -265,15 +267,13 @@ impl<'de> Deserialize<'de> for PipelinedBlock {
 
 impl PipelinedBlock {
     pub fn set_compute_result(
-        &mut self,
-        compute_result: StateComputeResult,
+        &self,
+        state_compute_result: StateComputeResult,
         execution_time: Duration,
     ) {
-        self.state_compute_result = compute_result;
-
         let mut to_commit = 0;
         let mut to_retry = 0;
-        for txn in self.state_compute_result.compute_status_for_input_txns() {
+        for txn in state_compute_result.compute_status_for_input_txns() {
             match txn {
                 TransactionStatus::Keep(_) => to_commit += 1,
                 TransactionStatus::Retry => to_retry += 1,
@@ -289,14 +289,14 @@ impl PipelinedBlock {
             to_commit,
             to_retry,
             execution_time,
-            root_hash: self.state_compute_result.root_hash(),
-            gas_used: self
-                .state_compute_result
+            root_hash: state_compute_result.root_hash(),
+            gas_used: state_compute_result
                 .execution_output
                 .block_end_info
                 .as_ref()
                 .map(|info| info.block_effective_gas_units()),
         };
+        *self.state_compute_result.lock() = state_compute_result;
 
         // We might be retrying execution, so it might have already been set.
         // Because we use this for statistics, it's ok that we drop the newer value.
@@ -321,23 +321,17 @@ impl PipelinedBlock {
         }
     }
 
-    pub fn set_execution_result(
-        mut self,
-        pipeline_execution_result: PipelineExecutionResult,
-    ) -> Self {
+    pub fn set_execution_result(&self, pipeline_execution_result: PipelineExecutionResult) {
         let PipelineExecutionResult {
-            input_txns,
+            input_txns: _,
             result,
             execution_time,
             pre_commit_fut,
         } = pipeline_execution_result;
 
-        self.input_transactions = input_txns;
-        self.pre_commit_fut = Arc::new(Mutex::new(Some(pre_commit_fut)));
+        *self.pre_commit_fut.lock() = Some(pre_commit_fut);
 
         self.set_compute_result(result, execution_time);
-
-        self
     }
 
     #[cfg(any(test, feature = "fuzzing"))]
@@ -380,6 +374,13 @@ impl Display for PipelinedBlock {
     }
 }
 
+/// Safeguard to ensure that the pipeline is aborted when the block is dropped.
+impl Drop for PipelinedBlock {
+    fn drop(&mut self) {
+        let _ = self.abort_pipeline();
+    }
+}
+
 impl PipelinedBlock {
     pub fn new(
         block: Block,
@@ -390,25 +391,28 @@ impl PipelinedBlock {
             block,
             block_window: OrderedBlockWindow::empty(),
             input_transactions,
-            state_compute_result,
+            state_compute_result: Mutex::new(state_compute_result),
             randomness: OnceCell::new(),
             pipeline_insertion_time: OnceCell::new(),
-            execution_summary: Arc::new(OnceCell::new()),
-            pre_commit_fut: Arc::new(Mutex::new(None)),
-            pipeline_futs: Arc::new(Mutex::new(None)),
-            pipeline_tx: Arc::new(Mutex::new(None)),
-            pipeline_abort_handle: Arc::new(Mutex::new(None)),
-            block_qc: Arc::new(Mutex::new(None)),
+            execution_summary: OnceCell::new(),
+            pre_commit_fut: Mutex::new(None),
+            pipeline_futs: Mutex::new(None),
+            pipeline_tx: Mutex::new(None),
+            pipeline_abort_handle: Mutex::new(None),
+            block_qc: Mutex::new(None),
         }
     }
 
+    pub fn with_block_window(self, window: OrderedBlockWindow) -> Self {
+        let mut block = self;
+        block.block_window = window;
+        block
+    }
+
     pub fn new_ordered(block: Block, window: OrderedBlockWindow) -> Self {
         let input_transactions = Vec::new();
         let state_compute_result = StateComputeResult::new_dummy();
-        Self {
-            block_window: window,
-            ..Self::new(block, input_transactions, state_compute_result)
-        }
+        Self::new(block, input_transactions, state_compute_result).with_block_window(window)
     }
 
     pub fn block(&self) -> &Block {
@@ -423,10 +427,6 @@ impl PipelinedBlock {
         self.block().id()
     }
 
-    pub fn input_transactions(&self) -> &Vec<SignedTransaction> {
-        &self.input_transactions
-    }
-
     pub fn epoch(&self) -> u64 {
         self.block.epoch()
     }
@@ -455,8 +455,8 @@ impl PipelinedBlock {
         self.block().timestamp_usecs()
     }
 
-    pub fn compute_result(&self) -> &StateComputeResult {
-        &self.state_compute_result
+    pub fn compute_result(&self) -> StateComputeResult {
+        self.state_compute_result.lock().clone()
     }
 
     pub fn randomness(&self) -> Option<&Randomness> {
@@ -468,18 +468,20 @@ impl PipelinedBlock {
     }
 
     pub fn block_info(&self) -> BlockInfo {
+        let compute_result = self.compute_result();
         self.block().gen_block_info(
-            self.compute_result().root_hash(),
-            self.compute_result().last_version_or_0(),
-            self.compute_result().epoch_state().clone(),
+            compute_result.root_hash(),
+            compute_result.last_version_or_0(),
+            compute_result.epoch_state().clone(),
         )
     }
 
     pub fn vote_proposal(&self) -> VoteProposal {
+        let compute_result = self.compute_result();
         VoteProposal::new(
-            self.compute_result().extension_proof(),
+            compute_result.extension_proof(),
             self.block.clone(),
-            self.compute_result().epoch_state().clone(),
+            compute_result.epoch_state().clone(),
             true,
         )
     }
@@ -493,15 +495,15 @@ impl PipelinedBlock {
         if self.is_reconfiguration_suffix() {
             return vec![];
         }
-        self.state_compute_result.subscribable_events().to_vec()
+        self.compute_result().subscribable_events().to_vec()
     }
 
     /// The block is suffix of a reconfiguration block if the state result carries over the epoch state
     /// from parent but has no transaction.
     pub fn is_reconfiguration_suffix(&self) -> bool {
-        self.state_compute_result.has_reconfiguration()
-            && self
-                .state_compute_result
+        let state_compute_result = self.compute_result();
+        state_compute_result.has_reconfiguration()
+            && state_compute_result
                 .compute_status_for_input_txns()
                 .is_empty()
     }
@@ -543,8 +545,8 @@ impl PipelinedBlock {
         *self.pipeline_abort_handle.lock() = Some(abort_handles);
     }
 
-    pub fn pipeline_tx(&self) -> Arc<Mutex<Option<PipelineInputTx>>> {
-        self.pipeline_tx.clone()
+    pub fn pipeline_tx(&self) -> &Mutex<Option<PipelineInputTx>> {
+        &self.pipeline_tx
     }
 
     pub fn abort_pipeline(&self) -> Option<PipelineFutures> {

consensus/src/block_storage/block_store.rs

@@ -341,7 +341,7 @@ impl BlockStore {
         // This callback is invoked synchronously with and could be used for multiple batches of blocks.
         self.execution_client
             .finalize_order(
-                &blocks_to_commit,
+                blocks_to_commit,
                 finality_proof.clone(),
                 Box::new(
                     move |committed_blocks: &[Arc<PipelinedBlock>],

consensus/src/block_storage/execution_pool/common_test.rs

@@ -25,6 +25,7 @@ pub const DEFAULT_MAX_PRUNED_BLOCKS_IN_MEM: usize = 10;
 
 /// Helper function to create a `QuorumCert` which can provide a `highest_commit_cert` via
 /// `highest_quorum_cert.into_wrapped_ledger_info()`
+#[allow(dead_code)]
 pub fn generate_qc(round: Round, parent_round: Round) -> QuorumCert {
     let num_nodes = 4;
     let (signers, validators) = random_validator_verifier(num_nodes, None, false);

consensus/src/consensus_observer/observer/consensus_observer.rs

@@ -295,7 +295,7 @@ impl ConsensusObserver {
         if let Err(error) = self
             .execution_client
             .finalize_order(
-                ordered_block.blocks(),
+                ordered_block.blocks().clone(),
                 WrappedLedgerInfo::new(VoteData::dummy(), ordered_block.ordered_proof().clone()),
                 commit_callback,
             )

consensus/src/counters.rs

@@ -1321,7 +1321,7 @@ pub fn update_counters_for_compute_result(compute_result: &StateComputeResult) {
 pub fn update_counters_for_committed_blocks(blocks_to_commit: &[Arc<PipelinedBlock>]) {
     for block in blocks_to_commit {
         update_counters_for_block(block.block());
-        update_counters_for_compute_result(block.compute_result());
+        update_counters_for_compute_result(&block.compute_result());
     }
 }
 

consensus/src/dag/adapter.rs

@@ -182,7 +182,7 @@ impl OrderedNotifier for OrderedNotifierAdapter {
         };
         NUM_ROUNDS_PER_BLOCK.observe((rounds_between + 1) as f64);
 
-        let block = PipelinedBlock::new(
+        let block = Arc::new(PipelinedBlock::new(
             Block::new_for_dag(
                 epoch,
                 round,
@@ -197,7 +197,7 @@ impl OrderedNotifier for OrderedNotifierAdapter {
             ),
             vec![],
             StateComputeResult::new_dummy(),
-        );
+        ));
         let block_info = block.block_info();
         let ledger_info_provider = self.ledger_info_provider.clone();
         let dag = self.dag.clone();