[move] Benchmarking historical transactions

Cargo.toml

@@ -19,6 +19,7 @@ members = [
     "aptos-move/aptos-memory-usage-tracker",
     "aptos-move/aptos-native-interface",
     "aptos-move/aptos-release-builder",
+    "aptos-move/aptos-replay-benchmark",
     "aptos-move/aptos-resource-viewer",
     "aptos-move/aptos-sdk-builder",
     "aptos-move/aptos-transaction-benchmarks",
@@ -419,6 +420,7 @@ aptos-push-metrics = { path = "crates/aptos-push-metrics" }
 aptos-rate-limiter = { path = "crates/aptos-rate-limiter" }
 aptos-release-builder = { path = "aptos-move/aptos-release-builder" }
 aptos-reliable-broadcast = { path = "crates/reliable-broadcast" }
+aptos-replay-benchmark = { path = "aptos-move/aptos-replay-benchmark" }
 aptos-resource-viewer = { path = "aptos-move/aptos-resource-viewer" }
 aptos-rest-client = { path = "crates/aptos-rest-client" }
 aptos-retrier = { path = "crates/aptos-retrier" }

aptos-move/aptos-debugger/src/aptos_debugger.rs

@@ -1,12 +1,8 @@
 // Copyright © Aptos Foundation
 // SPDX-License-Identifier: Apache-2.0
 
-use anyhow::{bail, format_err, Result};
-use aptos_block_executor::{
-    code_cache_global_manager::AptosModuleCacheManager,
-    txn_commit_hook::NoOpTransactionCommitHook,
-    txn_provider::{default::DefaultTxnProvider, TxnProvider},
-};
+use anyhow::{bail, format_err};
+use aptos_block_executor::txn_provider::{default::DefaultTxnProvider, TxnProvider};
 use aptos_gas_profiling::{GasProfiler, TransactionGasLog};
 use aptos_rest_client::Client;
 use aptos_types::{
@@ -28,9 +24,7 @@ use aptos_validator_interface::{
     AptosValidatorInterface, DBDebuggerInterface, DebuggerStateView, RestDebuggerInterface,
 };
 use aptos_vm::{
-    block_executor::{AptosTransactionOutput, AptosVMBlockExecutorWrapper},
-    data_cache::AsMoveResolver,
-    AptosVM,
+    aptos_vm::AptosVMBlockExecutor, data_cache::AsMoveResolver, AptosVM, VMBlockExecutor,
 };
 use aptos_vm_environment::environment::AptosEnvironment;
 use aptos_vm_logging::log_schema::AdapterLogSchema;
@@ -47,23 +41,31 @@ impl AptosDebugger {
         Self { debugger }
     }
 
-    pub fn rest_client(rest_client: Client) -> Result<Self> {
+    pub fn rest_client(rest_client: Client) -> anyhow::Result<Self> {
         Ok(Self::new(Arc::new(RestDebuggerInterface::new(rest_client))))
     }
 
-    pub fn db<P: AsRef<Path> + Clone>(db_root_path: P) -> Result<Self> {
+    pub fn db<P: AsRef<Path> + Clone>(db_root_path: P) -> anyhow::Result<Self> {
         Ok(Self::new(Arc::new(DBDebuggerInterface::open(
             db_root_path,
         )?)))
     }
 
+    pub async fn get_committed_transactions(
+        &self,
+        begin: Version,
+        limit: u64,
+    ) -> anyhow::Result<(Vec<Transaction>, Vec<TransactionInfo>)> {
+        self.debugger.get_committed_transactions(begin, limit).await
+    }
+
     pub fn execute_transactions_at_version(
         &self,
         version: Version,
         txns: Vec<Transaction>,
         repeat_execution_times: u64,
         concurrency_levels: &[usize],
-    ) -> Result<Vec<TransactionOutput>> {
+    ) -> anyhow::Result<Vec<TransactionOutput>> {
         let sig_verified_txns: Vec<SignatureVerifiedTransaction> =
             txns.into_iter().map(|x| x.into()).collect::<Vec<_>>();
         let txn_provider = DefaultTxnProvider::new(sig_verified_txns);
@@ -114,7 +116,7 @@ impl AptosDebugger {
         &self,
         version: Version,
         txn: SignedTransaction,
-    ) -> Result<(VMStatus, VMOutput, TransactionGasLog)> {
+    ) -> anyhow::Result<(VMStatus, VMOutput, TransactionGasLog)> {
         let state_view = DebuggerStateView::new(self.debugger.clone(), version);
         let log_context = AdapterLogSchema::new(state_view.id(), 0);
         let txn = txn
@@ -166,11 +168,8 @@ impl AptosDebugger {
         use_same_block_boundaries: bool,
         repeat_execution_times: u64,
         concurrency_levels: &[usize],
-    ) -> Result<Vec<TransactionOutput>> {
-        let (txns, txn_infos) = self
-            .debugger
-            .get_committed_transactions(begin, limit)
-            .await?;
+    ) -> anyhow::Result<Vec<TransactionOutput>> {
+        let (txns, txn_infos) = self.get_committed_transactions(begin, limit).await?;
 
         if use_same_block_boundaries {
             // when going block by block, no need to worry about epoch boundaries
@@ -238,7 +237,7 @@ impl AptosDebugger {
         txns: Vec<Transaction>,
         repeat_execution_times: u64,
         concurrency_levels: &[usize],
-    ) -> Result<Vec<TransactionOutput>> {
+    ) -> anyhow::Result<Vec<TransactionOutput>> {
         let results = self.execute_transactions_at_version(
             begin,
             txns,
@@ -268,7 +267,7 @@ impl AptosDebugger {
         repeat_execution_times: u64,
         concurrency_levels: &[usize],
         mut txn_infos: Vec<TransactionInfo>,
-    ) -> Result<Vec<TransactionOutput>> {
+    ) -> anyhow::Result<Vec<TransactionOutput>> {
         let mut ret = vec![];
         while limit != 0 {
             println!(
@@ -301,7 +300,7 @@ impl AptosDebugger {
         txns: Vec<Transaction>,
         repeat_execution_times: u64,
         concurrency_levels: &[usize],
-    ) -> Result<Vec<TransactionOutput>> {
+    ) -> anyhow::Result<Vec<TransactionOutput>> {
         let mut ret = vec![];
         let mut cur = vec![];
         let mut cur_version = begin;
@@ -336,7 +335,7 @@ impl AptosDebugger {
         &self,
         account: AccountAddress,
         seq: u64,
-    ) -> Result<Option<Version>> {
+    ) -> anyhow::Result<Option<Version>> {
         self.debugger
             .get_version_by_account_sequence(account, seq)
             .await
@@ -345,7 +344,7 @@ impl AptosDebugger {
     pub async fn get_committed_transaction_at_version(
         &self,
         version: Version,
-    ) -> Result<(Transaction, TransactionInfo)> {
+    ) -> anyhow::Result<(Transaction, TransactionInfo)> {
         let (mut txns, mut info) = self.debugger.get_committed_transactions(version, 1).await?;
 
         let txn = txns.pop().expect("there must be exactly 1 txn in the vec");
@@ -434,20 +433,16 @@ fn execute_block_no_limit(
     state_view: &DebuggerStateView,
     concurrency_level: usize,
 ) -> Result<Vec<TransactionOutput>, VMStatus> {
-    AptosVMBlockExecutorWrapper::execute_block::<
-        _,
-        NoOpTransactionCommitHook<AptosTransactionOutput, VMStatus>,
-        DefaultTxnProvider<SignatureVerifiedTransaction>,
-    >(
-        txn_provider,
-        state_view,
-        &AptosModuleCacheManager::new(),
-        BlockExecutorConfig {
-            local: BlockExecutorLocalConfig::default_with_concurrency_level(concurrency_level),
-            onchain: BlockExecutorConfigFromOnchain::new_no_block_limit(),
-        },
-        TransactionSliceMetadata::unknown(),
-        None,
-    )
-    .map(BlockOutput::into_transaction_outputs_forced)
+    let executor = AptosVMBlockExecutor::new();
+    executor
+        .execute_block_with_config(
+            txn_provider,
+            state_view,
+            BlockExecutorConfig {
+                local: BlockExecutorLocalConfig::default_with_concurrency_level(concurrency_level),
+                onchain: BlockExecutorConfigFromOnchain::new_no_block_limit(),
+            },
+            TransactionSliceMetadata::unknown(),
+        )
+        .map(BlockOutput::into_transaction_outputs_forced)
 }

aptos-move/aptos-replay-benchmark/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "aptos-replay-benchmark"
+version = "0.1.0"
+description = "A tool to replay and locally benchmark on-chain transactions."
+
+# Workspace inherited keys
+authors = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+repository = { workspace = true }
+rust-version = { workspace = true }
+
+[dependencies]
+anyhow = { workspace = true }
+aptos-block-executor = { workspace = true }
+aptos-logger = { workspace = true }
+aptos-move-debugger = { workspace = true }
+aptos-push-metrics = { workspace = true }
+aptos-rest-client = { workspace = true }
+aptos-types = { workspace = true }
+aptos-vm = { workspace = true }
+bcs = { workspace = true }
+clap = { workspace = true }
+claims = { workspace = true }
+serde = { workspace = true }
+parking_lot = { workspace = true }
+tokio = { workspace = true }
+url = { workspace = true }

aptos-move/aptos-replay-benchmark/src/block.rs

@@ -0,0 +1,136 @@
+// Copyright (c) Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{
+    comparison::Comparison,
+    state_view::{ReadSet, ReadSetCapturingStateView},
+    workload::Workload,
+};
+use aptos_types::{
+    block_executor::config::{
+        BlockExecutorConfig, BlockExecutorConfigFromOnchain, BlockExecutorLocalConfig,
+        BlockExecutorModuleCacheLocalConfig,
+    },
+    state_store::{state_key::StateKey, state_value::StateValue, StateView},
+    transaction::{TransactionOutput, Version},
+};
+use aptos_vm::{aptos_vm::AptosVMBlockExecutor, VMBlockExecutor};
+use std::collections::HashMap;
+
+/// Config used by benchmarking blocks.
+fn block_execution_config(concurrency_level: usize) -> BlockExecutorConfig {
+    BlockExecutorConfig {
+        local: BlockExecutorLocalConfig {
+            concurrency_level,
+            allow_fallback: true,
+            discard_failed_blocks: false,
+            module_cache_config: BlockExecutorModuleCacheLocalConfig::default(),
+        },
+        // For replay, there is no block limit.
+        onchain: BlockExecutorConfigFromOnchain::new_no_block_limit(),
+    }
+}
+
+pub struct Block {
+    inputs: ReadSet,
+    workload: Workload,
+    comparisons: Vec<Comparison>,
+}
+
+impl Block {
+    pub(crate) fn new(
+        workload: Workload,
+        state_view: &(impl StateView + Sync),
+        state_override: HashMap<StateKey, StateValue>,
+    ) -> Self {
+        let onchain_outputs = if state_override.is_empty() {
+            None
+        } else {
+            // Execute transactions with on-chain configs.
+            let onchain_outputs =
+                execute_workload(&AptosVMBlockExecutor::new(), &workload, state_view, 1);
+
+            // Check on-chain outputs do not modify state we override. If so, benchmarking results may
+            // not be correct.
+            let begin = workload.first_version();
+            for (idx, output) in onchain_outputs.iter().enumerate() {
+                for (state_key, _) in output.write_set() {
+                    if state_override.contains_key(state_key) {
+                        println!(
+                            "Transaction {} writes to overridden state value for {:?}",
+                            begin + idx as Version,
+                            state_key
+                        );
+                    }
+                }
+            }
+            Some(onchain_outputs)
+        };
+
+        // Execute transactions, recording all reads.
+        let state_view = ReadSetCapturingStateView::new(state_view, state_override);
+        let outputs = execute_workload(&AptosVMBlockExecutor::new(), &workload, &state_view, 1);
+        let inputs = state_view.into_read_set();
+
+        let comparisons = if let Some(onchain_outputs) = onchain_outputs {
+            let mut comparisons = Vec::with_capacity(onchain_outputs.len());
+            for (left, right) in onchain_outputs.into_iter().zip(outputs) {
+                let comparison = Comparison::diff(left, right);
+                comparisons.push(comparison);
+            }
+            comparisons
+        } else {
+            vec![]
+        };
+
+        Self {
+            inputs,
+            workload,
+            comparisons,
+        }
+    }
+
+    /// Prints the difference in transaction outputs when running with overrides.
+    pub fn print_diffs(&self) {
+        let begin = self.workload.first_version();
+
+        for (idx, comparison) in self.comparisons.iter().enumerate() {
+            if !comparison.is_ok() {
+                println!(
+                    "Transaction {} diff:\n {}\n",
+                    begin + idx as Version,
+                    comparison
+                );
+            }
+        }
+    }
+
+    /// Executes the workload for benchmarking.
+    #[inline(always)]
+    pub(crate) fn run(&self, executor: &AptosVMBlockExecutor, concurrency_level: usize) {
+        execute_workload(executor, &self.workload, &self.inputs, concurrency_level);
+    }
+}
+
+#[inline(always)]
+fn execute_workload(
+    executor: &AptosVMBlockExecutor,
+    workload: &Workload,
+    state_view: &(impl StateView + Sync),
+    concurrency_level: usize,
+) -> Vec<TransactionOutput> {
+    executor
+        .execute_block_with_config(
+            workload.txn_provider(),
+            state_view,
+            block_execution_config(concurrency_level),
+            workload.transaction_slice_metadata(),
+        )
+        .unwrap_or_else(|err| {
+            panic!(
+                "Block execution should not fail, but returned an error: {:?}",
+                err
+            )
+        })
+        .into_transaction_outputs_forced()
+}