[loader] Global caches for execution (#15192)

This adds support for cross-block environment and module caches

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

@@ -2,13 +2,16 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use anyhow::{bail, format_err, Result};
-use aptos_block_executor::txn_commit_hook::NoOpTransactionCommitHook;
+use aptos_block_executor::{
+    code_cache_global_manager::AptosModuleCacheManager, txn_commit_hook::NoOpTransactionCommitHook,
+};
 use aptos_gas_profiling::{GasProfiler, TransactionGasLog};
 use aptos_rest_client::Client;
 use aptos_types::{
     account_address::AccountAddress,
-    block_executor::config::{
-        BlockExecutorConfig, BlockExecutorConfigFromOnchain, BlockExecutorLocalConfig,
+    block_executor::{
+        config::{BlockExecutorConfig, BlockExecutorConfigFromOnchain, BlockExecutorLocalConfig},
+        execution_state::TransactionSliceMetadata,
     },
     contract_event::ContractEvent,
     state_store::TStateView,
@@ -431,14 +434,12 @@ fn execute_block_no_limit(
     BlockAptosVM::execute_block::<_, NoOpTransactionCommitHook<AptosTransactionOutput, VMStatus>>(
         sig_verified_txns,
         state_view,
+        &AptosModuleCacheManager::new(),
         BlockExecutorConfig {
-            local: BlockExecutorLocalConfig {
-                concurrency_level,
-                allow_fallback: true,
-                discard_failed_blocks: false,
-            },
+            local: BlockExecutorLocalConfig::default_with_concurrency_level(concurrency_level),
             onchain: BlockExecutorConfigFromOnchain::new_no_block_limit(),
         },
+        TransactionSliceMetadata::unknown(),
         None,
     )
     .map(BlockOutput::into_transaction_outputs_forced)

aptos-move/aptos-transaction-benchmarks/src/transaction_bench_state.rs

@@ -3,7 +3,9 @@
 
 use crate::transactions;
 use aptos_bitvec::BitVec;
-use aptos_block_executor::txn_commit_hook::NoOpTransactionCommitHook;
+use aptos_block_executor::{
+    code_cache_global_manager::AptosModuleCacheManager, txn_commit_hook::NoOpTransactionCommitHook,
+};
 use aptos_block_partitioner::{
     v2::config::PartitionerV2Config, BlockPartitioner, PartitionerConfig,
 };
@@ -16,6 +18,7 @@ use aptos_language_e2e_tests::{
 use aptos_types::{
     block_executor::{
         config::{BlockExecutorConfig, BlockExecutorConfigFromOnchain},
+        execution_state::TransactionSliceMetadata,
         partitioner::PartitionedTransactions,
     },
     block_metadata::BlockMetadata,
@@ -218,7 +221,9 @@ where
         >(
             transactions,
             self.state_view.as_ref(),
+            &AptosModuleCacheManager::new(),
             BlockExecutorConfig::new_maybe_block_limit(1, maybe_block_gas_limit),
+            TransactionSliceMetadata::unknown(),
             None,
         )
         .expect("VM should not fail to start")
@@ -266,10 +271,12 @@ where
         >(
             transactions,
             self.state_view.as_ref(),
+            &AptosModuleCacheManager::new(),
             BlockExecutorConfig::new_maybe_block_limit(
                 concurrency_level_per_shard,
                 maybe_block_gas_limit,
             ),
+            TransactionSliceMetadata::unknown(),
             None,
         )
         .expect("VM should not fail to start")
@@ -285,21 +292,21 @@ where
         partitioned_txns: Option<PartitionedTransactions>,
         run_par: bool,
         run_seq: bool,
-        conurrency_level_per_shard: usize,
+        concurrency_level_per_shard: usize,
         maybe_block_gas_limit: Option<u64>,
     ) -> (usize, usize) {
         let (output, par_tps) = if run_par {
             println!("Parallel execution starts...");
             let (output, tps) = if self.is_shareded() {
                 self.execute_benchmark_sharded(
                     partitioned_txns.unwrap(),
-                    conurrency_level_per_shard,
+                    concurrency_level_per_shard,
                     maybe_block_gas_limit,
                 )
             } else {
                 self.execute_benchmark_parallel(
                     &transactions,
-                    conurrency_level_per_shard,
+                    concurrency_level_per_shard,
                     maybe_block_gas_limit,
                 )
             };

aptos-move/aptos-transactional-test-harness/src/aptos_test_harness.rs

@@ -14,11 +14,9 @@ use aptos_language_e2e_tests::data_store::{FakeDataStore, GENESIS_CHANGE_SET_HEA
 use aptos_resource_viewer::{AnnotatedMoveValue, AptosValueAnnotator};
 use aptos_types::{
     account_config::{aptos_test_root_address, AccountResource, CoinStoreResource},
-    block_executor::config::BlockExecutorConfigFromOnchain,
     block_metadata::BlockMetadata,
     chain_id::ChainId,
     contract_event::ContractEvent,
-    on_chain_config::BlockGasLimitType,
     state_store::{state_key::StateKey, table::TableHandle, TStateView},
     transaction::{
         signature_verified_transaction::into_signature_verified_block,
@@ -517,14 +515,8 @@ impl<'a> AptosTestAdapter<'a> {
     fn run_transaction(&mut self, txn: Transaction) -> Result<TransactionOutput> {
         let txn_block = vec![txn];
         let sig_verified_block = into_signature_verified_block(txn_block);
-        let onchain_config = BlockExecutorConfigFromOnchain {
-            // TODO fetch values from state?
-            // Or should we just use execute_block_no_limit ?
-            block_gas_limit_type: BlockGasLimitType::Limit(30000),
-        };
-        let (mut outputs, _) = AptosVMBlockExecutor::new()
-            .execute_block(&sig_verified_block, &self.storage.clone(), onchain_config)?
-            .into_inner();
+        let mut outputs = AptosVMBlockExecutor::new()
+            .execute_block_no_limit(&sig_verified_block, &self.storage.clone())?;
 
         assert_eq!(outputs.len(), 1);
 

aptos-move/aptos-vm-profiling/src/bins/run_aptos_p2p.rs

@@ -48,9 +48,9 @@ fn main() -> Result<()> {
         })
         .collect();
 
-    let res = AptosVMBlockExecutor::new().execute_block_no_limit(&txns, &state_store)?;
+    let outputs = AptosVMBlockExecutor::new().execute_block_no_limit(&txns, &state_store)?;
     for i in 0..NUM_TXNS {
-        assert!(res[i as usize].status().status().unwrap().is_success());
+        assert!(outputs[i as usize].status().status().unwrap().is_success());
     }
 
     Ok(())

aptos-move/aptos-vm-types/src/module_and_script_storage/state_view_adapter.rs

@@ -3,22 +3,32 @@
 
 use crate::module_and_script_storage::module_storage::AptosModuleStorage;
 use ambassador::Delegate;
-use aptos_types::state_store::{state_key::StateKey, state_value::StateValueMetadata, StateView};
+use aptos_types::{
+    error::PanicError,
+    state_store::{state_key::StateKey, state_value::StateValueMetadata, StateView, TStateView},
+    vm::modules::AptosModuleExtension,
+};
 use bytes::Bytes;
 use move_binary_format::{
     errors::{PartialVMResult, VMResult},
     file_format::CompiledScript,
     CompiledModule,
 };
-use move_core_types::{account_address::AccountAddress, identifier::IdentStr, metadata::Metadata};
+use move_core_types::{
+    account_address::AccountAddress, identifier::IdentStr, language_storage::ModuleId,
+    metadata::Metadata,
+};
 use move_vm_runtime::{
     ambassador_impl_CodeStorage, ambassador_impl_ModuleStorage,
     ambassador_impl_WithRuntimeEnvironment, AsUnsyncCodeStorage, BorrowedOrOwned, CodeStorage,
     Module, ModuleStorage, RuntimeEnvironment, Script, UnsyncCodeStorage, UnsyncModuleStorage,
     WithRuntimeEnvironment,
 };
-use move_vm_types::{code::ModuleBytesStorage, module_storage_error};
-use std::sync::Arc;
+use move_vm_types::{
+    code::{ModuleBytesStorage, ModuleCode},
+    module_storage_error,
+};
+use std::{ops::Deref, sync::Arc};
 
 /// Avoids orphan rule to implement [ModuleBytesStorage] for [StateView].
 struct StateViewAdapter<'s, S> {
@@ -38,6 +48,14 @@ impl<'s, S: StateView> ModuleBytesStorage for StateViewAdapter<'s, S> {
     }
 }
 
+impl<'s, S: StateView> Deref for StateViewAdapter<'s, S> {
+    type Target = S;
+
+    fn deref(&self) -> &Self::Target {
+        &self.state_view
+    }
+}
+
 /// A (not thread-safe) implementation of code storage on top of a state view. It is never built
 /// directly by clients - only via [AsAptosCodeStorage] trait. Can be used to resolve both modules
 /// and cached scripts.
@@ -72,6 +90,57 @@ impl<'s, S: StateView, E: WithRuntimeEnvironment> AptosCodeStorageAdapter<'s, S,
         let storage = adapter.into_unsync_code_storage(runtime_environment);
         Self { storage }
     }
+
+    /// Drains cached verified modules from the code storage, transforming them into format used by
+    /// global caches.
+    pub fn into_verified_module_code_iter(
+        self,
+    ) -> Result<
+        impl Iterator<
+            Item = (
+                ModuleId,
+                Arc<ModuleCode<CompiledModule, Module, AptosModuleExtension>>,
+            ),
+        >,
+        PanicError,
+    > {
+        let (state_view, verified_modules_iter) = self
+            .storage
+            .into_module_storage()
+            .unpack_into_verified_modules_iter();
+
+        Ok(verified_modules_iter
+            .map(|(key, verified_code)| {
+                // We have cached the module previously, so we must be able to find it in storage.
+                let extension = state_view
+                    .get_state_value(&StateKey::module_id(&key))
+                    .map_err(|err| {
+                        let msg = format!(
+                            "Failed to retrieve module {}::{} from storage {:?}",
+                            key.address(),
+                            key.name(),
+                            err
+                        );
+                        PanicError::CodeInvariantError(msg)
+                    })?
+                    .map_or_else(
+                        || {
+                            let msg = format!(
+                                "Module {}::{} should exist, but it does not anymore",
+                                key.address(),
+                                key.name()
+                            );
+                            Err(PanicError::CodeInvariantError(msg))
+                        },
+                        |state_value| Ok(AptosModuleExtension::new(state_value)),
+                    )?;
+
+                let module = ModuleCode::from_arced_verified(verified_code, Arc::new(extension));
+                Ok((key, Arc::new(module)))
+            })
+            .collect::<Result<Vec<_>, PanicError>>()?
+            .into_iter())
+    }
 }
 
 impl<'s, S: StateView, E: WithRuntimeEnvironment> AptosModuleStorage

aptos-move/aptos-vm/Cargo.toml

@@ -63,7 +63,7 @@ serde = { workspace = true }
 aptos-aggregator = { workspace = true, features = ["testing"] }
 aptos-block-executor = { workspace = true, features = ["testing"] }
 aptos-language-e2e-tests = { workspace = true }
-aptos-types = { workspace = true, features = ["fuzzing"] }
+aptos-types = { workspace = true, features = ["fuzzing", "testing"] }
 claims = { workspace = true }
 move-vm-types = { workspace = true, features = ["testing"] }
 proptest = { workspace = true }

aptos-move/aptos-vm/src/aptos_vm.rs

@@ -27,7 +27,9 @@ use crate::{
     VMBlockExecutor, VMValidator,
 };
 use anyhow::anyhow;
-use aptos_block_executor::txn_commit_hook::NoOpTransactionCommitHook;
+use aptos_block_executor::{
+    code_cache_global_manager::AptosModuleCacheManager, txn_commit_hook::NoOpTransactionCommitHook,
+};
 use aptos_crypto::HashValue;
 use aptos_framework::{
     natives::{code::PublishRequest, randomness::RandomnessContext},
@@ -43,7 +45,11 @@ use aptos_types::state_store::StateViewId;
 use aptos_types::{
     account_config::{self, new_block_event_key, AccountResource},
     block_executor::{
-        config::{BlockExecutorConfig, BlockExecutorConfigFromOnchain, BlockExecutorLocalConfig},
+        config::{
+            BlockExecutorConfig, BlockExecutorConfigFromOnchain, BlockExecutorLocalConfig,
+            BlockExecutorModuleCacheLocalConfig,
+        },
+        execution_state::TransactionSliceMetadata,
         partitioner::PartitionedTransactions,
     },
     block_metadata::BlockMetadata,
@@ -2773,29 +2779,25 @@ impl AptosVM {
 /// Transaction execution: AptosVM
 /// Executing conflicts: in the input order, via BlockSTM,
 /// State: BlockSTM-provided MVHashMap-based view with caching
-pub struct AptosVMBlockExecutor;
+pub struct AptosVMBlockExecutor {
+    /// Manages module cache and execution environment of this block executor. Users of executor
+    /// must use manager's API to ensure the correct state of caches.
+    module_cache_manager: AptosModuleCacheManager,
+}
 
-// Executor external API
 impl VMBlockExecutor for AptosVMBlockExecutor {
-    // NOTE: At the moment there are no persistent caches that live past the end of a block (that's
-    // why AptosVMBlockExecutor has no state)
-    // There are some cache invalidation issues around transactions publishing code that need to be
-    // sorted out before that's possible.
-
     fn new() -> Self {
-        Self
+        Self {
+            module_cache_manager: AptosModuleCacheManager::new(),
+        }
     }
 
-    /// Execute a block of `transactions`. The output vector will have the exact same length as the
-    /// input vector. The discarded transactions will be marked as `TransactionStatus::Discard` and
-    /// have an empty `WriteSet`. Also `state_view` is immutable, and does not have interior
-    /// mutability. Writes to be applied to the data view are encoded in the write set part of a
-    /// transaction output.
     fn execute_block(
         &self,
         transactions: &[SignatureVerifiedTransaction],
         state_view: &(impl StateView + Sync),
         onchain_config: BlockExecutorConfigFromOnchain,
+        transaction_slice_metadata: TransactionSliceMetadata,
     ) -> Result<BlockOutput<TransactionOutput>, VMStatus> {
         fail_point!("move_adapter::execute_block", |_| {
             Err(VMStatus::error(
@@ -2817,14 +2819,17 @@ impl VMBlockExecutor for AptosVMBlockExecutor {
         >(
             transactions,
             state_view,
+            &self.module_cache_manager,
             BlockExecutorConfig {
                 local: BlockExecutorLocalConfig {
                     concurrency_level: AptosVM::get_concurrency_level(),
                     allow_fallback: true,
                     discard_failed_blocks: AptosVM::get_discard_failed_blocks(),
+                    module_cache_config: BlockExecutorModuleCacheLocalConfig::default(),
                 },
                 onchain: onchain_config,
             },
+            transaction_slice_metadata,
             None,
         );
         if ret.is_ok() {