Block epilogue refactor / optimize / prep for V2 support (#17094)

Author: gelash

Cargo.lock

@@ -225,10 +225,10 @@ impl BeforeMaterializationOutput<SignatureVerifiedTransaction> for BeforeMateria
             .collect()
     }
 
-    fn for_each_resource_group_key_and_tags<F>(&self, mut callback: F) -> Result<(), PanicError>
-    where
-        F: FnMut(&StateKey, HashSet<&StructTag>) -> Result<(), PanicError>,
-    {
+    fn for_each_resource_group_key_and_tags(
+        &self,
+        callback: &mut dyn FnMut(&StateKey, HashSet<&StructTag>) -> Result<(), PanicError>,
+    ) -> Result<(), PanicError> {
         for (key, tags) in self
             .guard
             .resource_write_set()
@@ -401,29 +401,52 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
         false
     }
 
+    fn check_materialization(&self) -> Result<bool, PanicError> {
+        if let Some(output) = self.committed_output.get() {
+            if output.status().is_retry() {
+                return Err(code_invariant_error(
+                    "Committed output must not have is_retry set.",
+                ));
+            }
+            Ok(true)
+        } else {
+            if !self
+                .vm_output
+                .read()
+                .as_ref()
+                .map_or(false, |output| output.status().is_retry())
+            {
+                return Err(code_invariant_error(
+                    "Non-committed output must exist with is_retry set.",
+                ));
+            }
+            Ok(false)
+        }
+    }
+
     fn incorporate_materialized_txn_output(
         &self,
         aggregator_v1_writes: Vec<(StateKey, WriteOp)>,
         materialized_resource_write_set: Vec<(StateKey, WriteOp)>,
         materialized_events: Vec<ContractEvent>,
     ) -> Result<(), PanicError> {
-        assert!(
-            self.committed_output
-                .set(
-                    self.vm_output
-                        .write()
-                        .take()
-                        .expect("Output must be set to incorporate materialized data")
-                        .into_transaction_output_with_materialized_write_set(
-                            aggregator_v1_writes,
-                            materialized_resource_write_set,
-                            materialized_events,
-                        )?,
+        self.committed_output
+            .set(
+                self.vm_output
+                    .write()
+                    .take()
+                    .expect("Output must be set to incorporate materialized data")
+                    .into_transaction_output_with_materialized_write_set(
+                        aggregator_v1_writes,
+                        materialized_resource_write_set,
+                        materialized_events,
+                    )?,
+            )
+            .map_err(|_| {
+                code_invariant_error(
+                    "Could not combine VMOutput with the materialized resource and event data",
                 )
-                .is_ok(),
-            "Could not combine VMOutput with the materialized resource and event data"
-        );
-        Ok(())
+            })
     }
 
     fn set_txn_output_for_non_dynamic_change_set(&self) {

aptos-move/aptos-vm/src/block_executor/mod.rs

@@ -49,7 +49,6 @@ proptest = { workspace = true, optional = true }
 proptest-derive = { workspace = true, optional = true }
 rand = { workspace = true }
 rayon = { workspace = true }
-scopeguard = { workspace = true }
 
 [dev-dependencies]
 aptos-aggregator = { workspace = true, features = ["testing"] }

aptos-move/block-executor/Cargo.toml

@@ -730,6 +730,10 @@ where
         Ok(self)
     }
 
+    fn check_materialization(&self) -> Result<bool, PanicError> {
+        Ok(!self.skipped)
+    }
+
     fn legacy_sequential_materialize_agg_v1(&self, _view: &impl TAggregatorV1View<Identifier = K>) {
         // TODO[agg_v2](tests): implement this method and compare
         // against sequential execution results v. aggregator v1.
@@ -849,18 +853,18 @@ where
             .collect()
     }
 
-    fn for_each_resource_group_key_and_tags<F>(&self, mut callback: F) -> Result<(), PanicError>
-    where
-        F: FnMut(&K, HashSet<&u32>) -> Result<(), PanicError>,
-    {
+    fn for_each_resource_group_key_and_tags(
+        &self,
+        callback: &mut dyn FnMut(&K, HashSet<&u32>) -> Result<(), PanicError>,
+    ) -> Result<(), PanicError> {
         for (key, _, _, ops) in self.group_writes.iter() {
             callback(key, ops.iter().map(|(tag, _)| tag).collect())?;
         }
         Ok(())
     }
 
     fn output_approx_size(&self) -> u64 {
-        // TODO add block output limit testing
+        // TODO: add block output limit testing
         0
     }
 

aptos-move/block-executor/src/combinatorial_tests/mock_executor.rs

@@ -129,7 +129,12 @@ where
     >::new(config, executor_thread_pool, None);
 
     if block_stm_v2 {
-        block_executor.execute_transactions_parallel_v2(txn_provider, data_view, &mut guard)
+        block_executor.execute_transactions_parallel_v2(
+            txn_provider,
+            data_view,
+            &TransactionSliceMetadata::unknown(),
+            &mut guard,
+        )
     } else {
         block_executor.execute_transactions_parallel(
             txn_provider,

aptos-move/block-executor/src/combinatorial_tests/resource_tests.rs

@@ -121,6 +121,17 @@ pub static RAYON_EXECUTION_SECONDS: Lazy<Histogram> = Lazy::new(|| {
     .unwrap()
 });
 
+pub static PARALLEL_FINALIZE_SECONDS: Lazy<Histogram> = Lazy::new(|| {
+    register_histogram!(
+        // metric name
+        "aptos_finalize_parallel_execution_seconds",
+        // metric description
+        "The time spent in seconds in finalizing parallel execution",
+        time_buckets(),
+    )
+    .unwrap()
+});
+
 pub static VM_INIT_SECONDS: Lazy<Histogram> = Lazy::new(|| {
     register_histogram!(
         // metric name