Filter committed transactions. Up until round-1 in prepare phase, and…

… round-1 in execution phase.

This requires using blocking txn provider to quickly provide shuffled txns in the execution phase.

Various improvements to avoid cloning transactions.

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

@@ -3,9 +3,7 @@
 
 use anyhow::{bail, format_err, Result};
 use aptos_block_executor::{
-    code_cache_global_manager::AptosModuleCacheManager,
-    txn_commit_hook::NoOpTransactionCommitHook,
-    txn_provider::{default::DefaultTxnProvider, TxnProvider},
+    code_cache_global_manager::AptosModuleCacheManager, txn_commit_hook::NoOpTransactionCommitHook,
 };
 use aptos_gas_profiling::{GasProfiler, TransactionGasLog};
 use aptos_rest_client::Client;
@@ -22,6 +20,7 @@ use aptos_types::{
         SignedTransaction, Transaction, TransactionInfo, TransactionOutput, TransactionPayload,
         Version,
     },
+    txn_provider::{default::DefaultTxnProvider, TxnProvider},
     vm_status::VMStatus,
 };
 use aptos_validator_interface::{
@@ -430,14 +429,14 @@ fn is_reconfiguration(vm_output: &TransactionOutput) -> bool {
 }
 
 fn execute_block_no_limit(
-    txn_provider: &DefaultTxnProvider<SignatureVerifiedTransaction>,
+    txn_provider: &dyn TxnProvider<SignatureVerifiedTransaction>,
     state_view: &DebuggerStateView,
     concurrency_level: usize,
 ) -> Result<Vec<TransactionOutput>, VMStatus> {
     BlockAptosVM::execute_block::<
         _,
         NoOpTransactionCommitHook<AptosTransactionOutput, VMStatus>,
-        DefaultTxnProvider<SignatureVerifiedTransaction>,
+        dyn TxnProvider<SignatureVerifiedTransaction>,
     >(
         txn_provider,
         state_view,

aptos-move/aptos-e2e-comparison-testing/Cargo.toml

@@ -12,7 +12,6 @@ default-run = "aptos-comparison-testing"
 
 [dependencies]
 anyhow = { workspace = true }
-aptos-block-executor = { workspace = true }
 aptos-framework = { workspace = true }
 aptos-language-e2e-tests = { workspace = true }
 aptos-rest-client = { workspace = true }

aptos-move/aptos-e2e-comparison-testing/src/data_collection.rs

@@ -6,7 +6,6 @@ use crate::{
     CompilationCache, DataManager, IndexWriter, PackageInfo, TxnIndex,
 };
 use anyhow::{format_err, Result};
-use aptos_block_executor::txn_provider::default::DefaultTxnProvider;
 use aptos_framework::natives::code::PackageMetadata;
 use aptos_rest_client::Client;
 use aptos_types::{
@@ -15,6 +14,7 @@ use aptos_types::{
         signature_verified_transaction::SignatureVerifiedTransaction, Transaction,
         TransactionOutput, Version,
     },
+    txn_provider::default::DefaultTxnProvider,
     write_set::TOTAL_SUPPLY_STATE_KEY,
 };
 use aptos_validator_interface::{AptosValidatorInterface, FilterCondition, RestDebuggerInterface};

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

@@ -4,9 +4,7 @@
 use crate::transactions;
 use aptos_bitvec::BitVec;
 use aptos_block_executor::{
-    code_cache_global_manager::AptosModuleCacheManager,
-    txn_commit_hook::NoOpTransactionCommitHook,
-    txn_provider::{default::DefaultTxnProvider, TxnProvider},
+    code_cache_global_manager::AptosModuleCacheManager, txn_commit_hook::NoOpTransactionCommitHook,
 };
 use aptos_block_partitioner::{
     v2::config::PartitionerV2Config, BlockPartitioner, PartitionerConfig,
@@ -32,6 +30,7 @@ use aptos_types::{
         },
         ExecutionStatus, Transaction, TransactionOutput, TransactionStatus,
     },
+    txn_provider::{default::DefaultTxnProvider, TxnProvider},
     vm_status::VMStatus,
 };
 use aptos_vm::{
@@ -220,7 +219,7 @@ where
         let output = BlockAptosVM::execute_block::<
             _,
             NoOpTransactionCommitHook<AptosTransactionOutput, VMStatus>,
-            DefaultTxnProvider<SignatureVerifiedTransaction>,
+            dyn TxnProvider<SignatureVerifiedTransaction>,
         >(
             txn_provider,
             self.state_view.as_ref(),
@@ -271,7 +270,7 @@ where
         let output = BlockAptosVM::execute_block::<
             _,
             NoOpTransactionCommitHook<AptosTransactionOutput, VMStatus>,
-            DefaultTxnProvider<SignatureVerifiedTransaction>,
+            dyn TxnProvider<SignatureVerifiedTransaction>,
         >(
             txn_provider,
             self.state_view.as_ref(),

aptos-move/aptos-transactional-test-harness/Cargo.toml

@@ -15,7 +15,6 @@ rust-version = { workspace = true }
 [dependencies]
 anyhow = { workspace = true }
 aptos-api-types = { workspace = true }
-aptos-block-executor = { workspace = true }
 aptos-cached-packages = { workspace = true }
 aptos-crypto = { workspace = true }
 aptos-framework = { workspace = true }

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

@@ -4,7 +4,6 @@
 
 use anyhow::{bail, format_err, Result};
 use aptos_api_types::AsConverter;
-use aptos_block_executor::txn_provider::default::DefaultTxnProvider;
 use aptos_crypto::{
     ed25519::{Ed25519PrivateKey, Ed25519PublicKey},
     hash::HashValue,
@@ -24,6 +23,7 @@ use aptos_types::{
         EntryFunction as TransactionEntryFunction, ExecutionStatus, RawTransaction,
         Script as TransactionScript, Transaction, TransactionOutput, TransactionStatus,
     },
+    txn_provider::default::DefaultTxnProvider,
     AptosCoinType,
 };
 use aptos_vm::{aptos_vm::AptosVMBlockExecutor, VMBlockExecutor};

aptos-move/aptos-vm-profiling/Cargo.toml

@@ -17,7 +17,6 @@ glob = { workspace = true }
 once_cell = { workspace = true }
 smallvec = { workspace = true }
 
-aptos-block-executor = { workspace = true }
 aptos-cached-packages = { workspace = true }
 aptos-gas-schedule = { workspace = true }
 aptos-language-e2e-tests = { workspace = true }
@@ -36,7 +35,7 @@ move-vm-types = { workspace = true }
 
 [[bin]]
 name = "main"
-path  = "src/main.rs"
+path = "src/main.rs"
 
 [[bin]]
 name = "run-move"

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

@@ -2,10 +2,10 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use anyhow::Result;
-use aptos_block_executor::txn_provider::default::DefaultTxnProvider;
 use aptos_language_e2e_tests::{account::AccountData, data_store::FakeDataStore};
 use aptos_types::{
     transaction::{signature_verified_transaction::SignatureVerifiedTransaction, Transaction},
+    txn_provider::default::DefaultTxnProvider,
     write_set::WriteSet,
 };
 use aptos_vm::{aptos_vm::AptosVMBlockExecutor, VMBlockExecutor};

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

@@ -28,9 +28,7 @@ use crate::{
 };
 use anyhow::anyhow;
 use aptos_block_executor::{
-    code_cache_global_manager::AptosModuleCacheManager,
-    txn_commit_hook::NoOpTransactionCommitHook,
-    txn_provider::{default::DefaultTxnProvider, TxnProvider},
+    code_cache_global_manager::AptosModuleCacheManager, txn_commit_hook::NoOpTransactionCommitHook,
 };
 use aptos_crypto::HashValue;
 use aptos_framework::{
@@ -73,6 +71,7 @@ use aptos_types::{
         TransactionAuxiliaryData, TransactionOutput, TransactionPayload, TransactionStatus,
         VMValidatorResult, ViewFunctionOutput, WriteSetPayload,
     },
+    txn_provider::TxnProvider,
     vm_status::{AbortLocation, StatusCode, VMStatus},
 };
 use aptos_utils::aptos_try;
@@ -2801,7 +2800,7 @@ impl VMBlockExecutor for AptosVMBlockExecutor {
 
     fn execute_block(
         &self,
-        txn_provider: &DefaultTxnProvider<SignatureVerifiedTransaction>,
+        txn_provider: &dyn TxnProvider<SignatureVerifiedTransaction>,
         state_view: &(impl StateView + Sync),
         onchain_config: BlockExecutorConfigFromOnchain,
         transaction_slice_metadata: TransactionSliceMetadata,
@@ -2823,7 +2822,7 @@ impl VMBlockExecutor for AptosVMBlockExecutor {
         let ret = BlockAptosVM::execute_block::<
             _,
             NoOpTransactionCommitHook<AptosTransactionOutput, VMStatus>,
-            DefaultTxnProvider<SignatureVerifiedTransaction>,
+            dyn TxnProvider<SignatureVerifiedTransaction>,
         >(
             txn_provider,
             state_view,

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

@@ -14,7 +14,7 @@ use aptos_aggregator::{
 use aptos_block_executor::{
     code_cache_global_manager::AptosModuleCacheManager, errors::BlockExecutionError,
     executor::BlockExecutor, task::TransactionOutput as BlockExecutorTransactionOutput,
-    txn_commit_hook::TransactionCommitHook, txn_provider::TxnProvider, types::InputOutputKey,
+    txn_commit_hook::TransactionCommitHook, types::InputOutputKey,
 };
 use aptos_infallible::Mutex;
 use aptos_types::{
@@ -28,6 +28,7 @@ use aptos_types::{
         signature_verified_transaction::SignatureVerifiedTransaction, BlockOutput,
         TransactionOutput, TransactionStatus,
     },
+    txn_provider::TxnProvider,
     write_set::WriteOp,
 };
 use aptos_vm_logging::{flush_speculative_logs, init_speculative_logs};
@@ -391,7 +392,7 @@ impl BlockAptosVM {
     pub fn execute_block_on_thread_pool<
         S: StateView + Sync,
         L: TransactionCommitHook<Output = AptosTransactionOutput>,
-        TP: TxnProvider<SignatureVerifiedTransaction> + Sync,
+        TP: TxnProvider<SignatureVerifiedTransaction> + Sync + ?Sized,
     >(
         executor_thread_pool: Arc<rayon::ThreadPool>,
         signature_verified_block: &TP,
@@ -466,7 +467,7 @@ impl BlockAptosVM {
     pub fn execute_block<
         S: StateView + Sync,
         L: TransactionCommitHook<Output = AptosTransactionOutput>,
-        TP: TxnProvider<SignatureVerifiedTransaction> + Sync,
+        TP: TxnProvider<SignatureVerifiedTransaction> + Sync + ?Sized,
     >(
         signature_verified_block: &TP,
         state_view: &S,

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

@@ -126,7 +126,6 @@ pub mod verifier;
 
 pub use crate::aptos_vm::{AptosSimulationVM, AptosVM};
 use crate::sharded_block_executor::{executor_client::ExecutorClient, ShardedBlockExecutor};
-use aptos_block_executor::txn_provider::default::DefaultTxnProvider;
 use aptos_types::{
     block_executor::{
         config::BlockExecutorConfigFromOnchain, execution_state::TransactionSliceMetadata,
@@ -137,6 +136,7 @@ use aptos_types::{
         signature_verified_transaction::SignatureVerifiedTransaction, BlockOutput,
         SignedTransaction, TransactionOutput, VMValidatorResult,
     },
+    txn_provider::TxnProvider,
     vm_status::VMStatus,
 };
 use aptos_vm_types::module_and_script_storage::code_storage::AptosCodeStorage;
@@ -169,7 +169,7 @@ pub trait VMBlockExecutor: Send + Sync {
     /// Executes a block of transactions and returns output for each one of them.
     fn execute_block(
         &self,
-        txn_provider: &DefaultTxnProvider<SignatureVerifiedTransaction>,
+        txn_provider: &dyn TxnProvider<SignatureVerifiedTransaction>,
         state_view: &(impl StateView + Sync),
         onchain_config: BlockExecutorConfigFromOnchain,
         transaction_slice_metadata: TransactionSliceMetadata,
@@ -179,7 +179,7 @@ pub trait VMBlockExecutor: Send + Sync {
     /// any block limit.
     fn execute_block_no_limit(
         &self,
-        txn_provider: &DefaultTxnProvider<SignatureVerifiedTransaction>,
+        txn_provider: &dyn TxnProvider<SignatureVerifiedTransaction>,
         state_view: &(impl StateView + Sync),
     ) -> Result<Vec<TransactionOutput>, VMStatus> {
         self.execute_block(

aptos-move/aptos-vm/src/sharded_block_executor/sharded_executor_service.rs

@@ -16,9 +16,7 @@ use crate::{
         ExecutorShardCommand,
     },
 };
-use aptos_block_executor::{
-    code_cache_global_manager::AptosModuleCacheManager, txn_provider::default::DefaultTxnProvider,
-};
+use aptos_block_executor::code_cache_global_manager::AptosModuleCacheManager;
 use aptos_logger::{info, trace};
 use aptos_types::{
     block_executor::{
@@ -32,6 +30,7 @@ use aptos_types::{
         signature_verified_transaction::SignatureVerifiedTransaction, BlockOutput,
         TransactionOutput,
     },
+    txn_provider::default::DefaultTxnProvider,
 };
 use aptos_vm_logging::disable_speculative_logging;
 use futures::{channel::oneshot, executor::block_on};

aptos-move/aptos-vm/tests/sharded_block_executor.rs

@@ -186,7 +186,6 @@ fn test_partitioner_v2_connected_component_sharded_block_executor_with_random_tr
 }
 
 mod test_utils {
-    use aptos_block_executor::txn_provider::default::DefaultTxnProvider;
     use aptos_block_partitioner::BlockPartitioner;
     use aptos_language_e2e_tests::{
         account::AccountData, common_transactions::peer_to_peer_txn, data_store::FakeDataStore,
@@ -201,6 +200,7 @@ mod test_utils {
             signature_verified_transaction::SignatureVerifiedTransaction, Transaction,
             TransactionOutput,
         },
+        txn_provider::default::DefaultTxnProvider,
     };
     use aptos_vm::{
         aptos_vm::AptosVMBlockExecutor,

aptos-move/block-executor/src/counters.rs

@@ -165,6 +165,17 @@ pub static TASK_EXECUTE_SECONDS: Lazy<Histogram> = Lazy::new(|| {
     .unwrap()
 });
 
+pub static TXN_GET_SECONDS: Lazy<Histogram> = Lazy::new(|| {
+    register_histogram!(
+        // metric name
+        "aptos_execution_txn_get_seconds",
+        // metric description
+        "The time spent in txn get (for blocking txn provider) in Block STM",
+        time_buckets(),
+    )
+    .unwrap()
+});
+
 pub static DEPENDENCY_WAIT_SECONDS: Lazy<Histogram> = Lazy::new(|| {
     register_histogram!(
         "aptos_execution_dependency_wait",

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

@@ -18,7 +18,6 @@ use crate::{
     task::{ExecutionStatus, ExecutorTask, TransactionOutput},
     txn_commit_hook::TransactionCommitHook,
     txn_last_input_output::{KeyKind, TxnLastInputOutput},
-    txn_provider::TxnProvider,
     types::ReadWriteSummary,
     view::{LatestView, ParallelState, SequentialState, ViewState},
 };
@@ -43,6 +42,7 @@ use aptos_types::{
     transaction::{
         block_epilogue::BlockEndInfo, BlockExecutableTransaction as Transaction, BlockOutput,
     },
+    txn_provider::TxnProvider,
     vm::modules::AptosModuleExtension,
     write_set::{TransactionWrite, WriteOp},
 };
@@ -72,7 +72,7 @@ use std::{
     },
 };
 
-pub struct BlockExecutor<T, E, S, L, X, TP> {
+pub struct BlockExecutor<T, E, S, L, X, TP: ?Sized> {
     // Number of active concurrent tasks, corresponding to the maximum number of rayon
     // threads that may be concurrently participating in parallel execution.
     config: BlockExecutorConfig,
@@ -88,7 +88,7 @@ where
     S: TStateView<Key = T::Key> + Sync,
     L: TransactionCommitHook<Output = E::Output>,
     X: Executable + 'static,
-    TP: TxnProvider<T> + Sync,
+    TP: TxnProvider<T> + Sync + ?Sized,
 {
     /// The caller needs to ensure that concurrency_level > 1 (0 is illegal and 1 should
     /// be handled by sequential execution) and that concurrency_level <= num_cpus.