Makes StateComputeResult an combination of immutable stage outputs

consensus/consensus-types/src/pipelined_block.rs

@@ -288,7 +288,7 @@ impl PipelinedBlock {
     pub fn block_info(&self) -> BlockInfo {
         self.block().gen_block_info(
             self.compute_result().root_hash(),
-            self.compute_result().version(),
+            self.compute_result().last_version_or_0(),
             self.compute_result().epoch_state().clone(),
         )
     }

consensus/src/block_storage/block_store.rs

@@ -177,7 +177,7 @@ impl BlockStore {
             root_metadata.accu_hash,
         );
 
-        let result = StateComputeResult::new_empty(Arc::new(
+        let result = StateComputeResult::new_dummy_with_accumulator(Arc::new(
             InMemoryTransactionAccumulator::new(
                 root_metadata.frozen_root_hashes,
                 root_metadata.num_leaves,

consensus/src/block_storage/block_store_test.rs

@@ -290,7 +290,7 @@ async fn test_insert_vote() {
             VoteData::new(
                 block.block().gen_block_info(
                     block.compute_result().root_hash(),
-                    block.compute_result().version(),
+                    block.compute_result().last_version_or_0(),
                     block.compute_result().epoch_state().clone(),
                 ),
                 block.quorum_cert().certified_block().clone(),
@@ -320,7 +320,7 @@ async fn test_insert_vote() {
         VoteData::new(
             block.block().gen_block_info(
                 block.compute_result().root_hash(),
-                block.compute_result().version(),
+                block.compute_result().last_version_or_0(),
                 block.compute_result().epoch_state().clone(),
             ),
             block.quorum_cert().certified_block().clone(),

consensus/src/counters.rs

@@ -1249,7 +1249,7 @@ pub fn update_counters_for_committed_blocks(blocks_to_commit: &[Arc<PipelinedBlo
             .observe(block.block().payload().map_or(0, |payload| payload.size()) as f64);
         COMMITTED_BLOCKS_COUNT.inc();
         LAST_COMMITTED_ROUND.set(block.round() as i64);
-        LAST_COMMITTED_VERSION.set(block.compute_result().num_leaves() as i64);
+        LAST_COMMITTED_VERSION.set(block.compute_result().last_version_or_0() as i64);
 
         let failed_rounds = block
             .block()

consensus/src/pipeline/tests/test_utils.rs

@@ -101,7 +101,7 @@ pub fn prepare_executed_blocks_with_ledger_info(
     let li = LedgerInfo::new(
         proposals.last().unwrap().block().gen_block_info(
             compute_result.root_hash(),
-            compute_result.version(),
+            compute_result.last_version_or_0(),
             compute_result.epoch_state().clone(),
         ),
         consensus_hash,

consensus/src/state_computer.rs

@@ -144,10 +144,9 @@ impl ExecutionProxy {
             Box::pin(async move {
                 pre_commit_notifier
                     .send(Box::pin(async move {
-                        let txns = state_compute_result
-                            .ledger_update_output
-                            .transactions
-                            .clone();
+                        let _timer = counters::OP_COUNTERS.timer("pre_commit_notify");
+
+                        let txns = state_compute_result.transactions_to_commit().to_vec();
                         let subscribable_events =
                             state_compute_result.subscribable_events().to_vec();
                         if let Err(e) = monitor!(

execution/executor-benchmark/src/ledger_update_stage.rs

@@ -5,7 +5,6 @@
 use crate::pipeline::{CommitBlockMessage, LedgerUpdateMessage};
 use aptos_executor::block_executor::{BlockExecutor, TransactionBlockExecutor};
 use aptos_executor_types::BlockExecutorTrait;
-use aptos_types::transaction::Version;
 use std::sync::{mpsc, Arc};
 
 pub enum CommitProcessing {
@@ -18,7 +17,9 @@ pub enum CommitProcessing {
 pub struct LedgerUpdateStage<V> {
     executor: Arc<BlockExecutor<V>>,
     commit_processing: CommitProcessing,
-    version: Version,
+    allow_aborts: bool,
+    allow_discards: bool,
+    allow_retries: bool,
 }
 
 impl<V> LedgerUpdateStage<V>
@@ -28,52 +29,62 @@ where
     pub fn new(
         executor: Arc<BlockExecutor<V>>,
         commit_processing: CommitProcessing,
-        version: Version,
+        allow_aborts: bool,
+        allow_discards: bool,
+        allow_retries: bool,
     ) -> Self {
         Self {
             executor,
-            version,
             commit_processing,
+            allow_aborts,
+            allow_discards,
+            allow_retries,
         }
     }
 
     pub fn ledger_update(&mut self, ledger_update_message: LedgerUpdateMessage) {
         // let ledger_update_start_time = Instant::now();
         let LedgerUpdateMessage {
+            first_block_start_time,
             current_block_start_time,
-            execution_time,
             partition_time,
+            execution_time,
             block_id,
             parent_block_id,
+            num_input_txns,
             state_checkpoint_output,
-            first_block_start_time,
         } = ledger_update_message;
 
         let output = self
             .executor
-            .ledger_update(block_id, parent_block_id, state_checkpoint_output)
+            .ledger_update(block_id, parent_block_id, state_checkpoint_output.clone())
             .unwrap();
-
-        self.version += output.transactions_to_commit_len() as Version;
+        output.execution_output.check_aborts_discards_retries(
+            self.allow_aborts,
+            self.allow_discards,
+            self.allow_retries,
+        );
+        if !self.allow_retries {
+            assert_eq!(output.num_transactions_to_commit(), num_input_txns + 1);
+        }
 
         match &self.commit_processing {
             CommitProcessing::SendToQueue(commit_sender) => {
                 let msg = CommitBlockMessage {
                     block_id,
-                    root_hash: output.root_hash(),
                     first_block_start_time,
                     current_block_start_time,
                     partition_time,
                     execution_time,
-                    num_txns: output.transactions_to_commit_len(),
+                    output,
                 };
                 commit_sender.send(msg).unwrap();
             },
             CommitProcessing::ExecuteInline => {
                 let ledger_info_with_sigs = super::transaction_committer::gen_li_with_sigs(
                     block_id,
                     output.root_hash(),
-                    self.version,
+                    output.expect_last_version(),
                 );
                 self.executor.pre_commit_block(block_id).unwrap();
                 self.executor.commit_ledger(ledger_info_with_sigs).unwrap();

execution/executor-benchmark/src/lib.rs

@@ -165,10 +165,9 @@ pub fn run_benchmark<V>(
         ((0..pipeline_config.num_generator_workers).map(|_| transaction_generator_creator.create_transaction_generator()).collect::<Vec<_>>(), phase)
     });
 
-    let version = db.reader.expect_synced_version();
-
+    let start_version = db.reader.expect_synced_version();
     let (pipeline, block_sender) =
-        Pipeline::new(executor, version, &pipeline_config, Some(num_blocks));
+        Pipeline::new(executor, start_version, &pipeline_config, Some(num_blocks));
 
     let mut num_accounts_to_load = num_main_signer_accounts;
     if let Some(mix) = &transaction_mix {
@@ -235,7 +234,8 @@ pub fn run_benchmark<V>(
     );
 
     if !pipeline_config.skip_commit {
-        let num_txns = db.reader.expect_synced_version() - version - num_blocks_created as u64;
+        let num_txns =
+            db.reader.expect_synced_version() - start_version - num_blocks_created as u64;
         overall_measuring.print_end("Overall", num_txns);
 
         if verify_sequence_numbers {
@@ -259,10 +259,10 @@ fn init_workload<V>(
 where
     V: TransactionBlockExecutor + 'static,
 {
-    let version = db.reader.expect_synced_version();
+    let start_version = db.reader.expect_synced_version();
     let (pipeline, block_sender) = Pipeline::<V>::new(
         BlockExecutor::new(db.clone()),
-        version,
+        start_version,
         pipeline_config,
         None,
     );