From ea68ae8dc6068a27ac45f9ccb38df42e3dce61f9 Mon Sep 17 00:00:00 2001 From: Sean Young Date: Wed, 2 Sep 2026 15:42:03 +0100 Subject: [PATCH] Do not use rayon pool --- aptos-move/aptos-vm/src/block_executor/mod.rs | 41 +------ .../sharded_executor_service.rs | 3 +- .../src/combinatorial_tests/bencher.rs | 11 +- .../delayed_field_tests.rs | 5 +- .../src/combinatorial_tests/delta_tests.rs | 6 +- .../src/combinatorial_tests/group_tests.rs | 10 +- .../src/combinatorial_tests/module_tests.rs | 6 +- .../src/combinatorial_tests/resource_tests.rs | 14 +-- .../src/combinatorial_tests/tests.rs | 49 -------- aptos-move/block-executor/src/executor.rs | 114 +++++++++++++++--- .../block-executor/src/unit_tests/mod.rs | 59 --------- aptos-move/e2e-tests/src/executor.rs | 46 ++----- .../src/native/native_vm.rs | 7 +- .../move/aptosvm_authenticators.rs | 13 +- .../fuzz/fuzz_targets/move/aptosvm_publish.rs | 15 +-- .../move/aptosvm_publish_and_run.rs | 15 +-- .../aptosvm_publish_and_run_transactional.rs | 18 +-- 17 files changed, 133 insertions(+), 299 deletions(-) diff --git a/aptos-move/aptos-vm/src/block_executor/mod.rs b/aptos-move/aptos-vm/src/block_executor/mod.rs index ffdec39a1d2..256652246de 100644 --- a/aptos-move/aptos-vm/src/block_executor/mod.rs +++ b/aptos-move/aptos-vm/src/block_executor/mod.rs @@ -46,25 +46,14 @@ use move_core_types::{ }; use move_vm_runtime::execution_tracing::Trace; use move_vm_types::delayed_values::delayed_field_id::DelayedFieldID; -use once_cell::sync::{Lazy, OnceCell}; +use once_cell::sync::OnceCell; use std::{ collections::{BTreeMap, HashMap, HashSet}, marker::PhantomData, - sync::Arc, }; use triomphe::Arc as TriompheArc; use vm_wrapper::AptosExecutorTask; -static RAYON_EXEC_POOL: Lazy> = Lazy::new(|| { - Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .thread_name(|index| format!("par_exec-{}", index)) - .build() - .unwrap(), - ) -}); - /// Output type wrapper used by block executor. VM output is stored first, then /// transformed into TransactionOutput type that is returned. #[derive(Debug)] @@ -513,12 +502,11 @@ impl< >, > AptosBlockExecutorWrapper { - pub fn execute_block_on_thread_pool< + pub fn execute_block< S: StateView + Sync, L: TransactionCommitHook, TP: TxnProvider + Sync, >( - executor_thread_pool: Arc, signature_verified_block: &TP, state_view: &S, module_cache_manager: &AptosModuleCacheManager, @@ -546,7 +534,6 @@ impl< let executor = BlockExecutor::::new( config, - executor_thread_pool, transaction_commit_listener, ); @@ -585,30 +572,6 @@ impl< Err(BlockExecutionError::FatalVMError(err)) => Err(err), } } - - /// Uses shared thread pool to execute blocks. - pub(crate) fn execute_block< - S: StateView + Sync, - L: TransactionCommitHook, - TP: TxnProvider + Sync, - >( - signature_verified_block: &TP, - state_view: &S, - module_cache_manager: &AptosModuleCacheManager, - config: BlockExecutorConfig, - transaction_slice_metadata: TransactionSliceMetadata, - transaction_commit_listener: Option, - ) -> Result, VMStatus> { - Self::execute_block_on_thread_pool::( - Arc::clone(&RAYON_EXEC_POOL), - signature_verified_block, - state_view, - module_cache_manager, - config, - transaction_slice_metadata, - transaction_commit_listener, - ) - } } // Same as AptosBlockExecutorWrapper with AptosExecutorTask diff --git a/aptos-move/aptos-vm/src/sharded_block_executor/sharded_executor_service.rs b/aptos-move/aptos-vm/src/sharded_block_executor/sharded_executor_service.rs index aa94cde8922..e0d848a96eb 100644 --- a/aptos-move/aptos-vm/src/sharded_block_executor/sharded_executor_service.rs +++ b/aptos-move/aptos-vm/src/sharded_block_executor/sharded_executor_service.rs @@ -142,8 +142,7 @@ impl ShardedExecutorService { s.spawn(move |_| { let txn_provider = DefaultTxnProvider::new_without_info(signature_verified_transactions); - let ret = AptosVMBlockExecutorWrapper::execute_block_on_thread_pool( - executor_thread_pool, + let ret = AptosVMBlockExecutorWrapper::execute_block( &txn_provider, aggr_overridden_state_view.as_ref(), // Since we execute blocks in parallel, we cannot share module caches, so each diff --git a/aptos-move/block-executor/src/combinatorial_tests/bencher.rs b/aptos-move/block-executor/src/combinatorial_tests/bencher.rs index cd7b9259c67..36f48c7526c 100644 --- a/aptos-move/block-executor/src/combinatorial_tests/bencher.rs +++ b/aptos-move/block-executor/src/combinatorial_tests/bencher.rs @@ -32,7 +32,7 @@ use proptest::{ strategy::{Strategy, ValueTree}, test_runner::TestRunner, }; -use std::{fmt::Debug, hash::Hash, marker::PhantomData, sync::Arc}; +use std::{fmt::Debug, hash::Hash, marker::PhantomData}; pub struct Bencher { transaction_size: usize, @@ -127,13 +127,6 @@ where pub(crate) fn run(self) { let state_view = MockStateView::empty(); - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); - let config = BlockExecutorConfig::new_no_block_limit(num_cpus::get()); let mut guard = AptosModuleCacheManagerGuard::none(); @@ -144,7 +137,7 @@ where NoOpTransactionCommitHook, DefaultTxnProvider, E>, AuxiliaryInfo>, AuxiliaryInfo, - >::new(config, executor_thread_pool, None) + >::new(config, None) .execute_transactions_parallel( &self.txns_provider, &state_view, diff --git a/aptos-move/block-executor/src/combinatorial_tests/delayed_field_tests.rs b/aptos-move/block-executor/src/combinatorial_tests/delayed_field_tests.rs index 6f87fc9288e..ff3735a037f 100644 --- a/aptos-move/block-executor/src/combinatorial_tests/delayed_field_tests.rs +++ b/aptos-move/block-executor/src/combinatorial_tests/delayed_field_tests.rs @@ -6,7 +6,7 @@ use crate::{ combinatorial_tests::{ group_tests::{create_non_empty_group_data_view, run_tests_with_groups}, mock_executor::{MockEvent, MockTask}, - resource_tests::{create_executor_thread_pool, get_gas_limit_variants}, + resource_tests::get_gas_limit_variants, types::{KeyType, MockTransaction, TransactionGen, TransactionGenParams}, }, task::ExecutorTask, @@ -66,12 +66,9 @@ fn delayed_field_transaction_tests( let data_view = create_non_empty_group_data_view(&key_universe, universe_size, true); - let executor_thread_pool = create_executor_thread_pool(); - let gas_limits = get_gas_limit_variants(use_gas_limit, transaction_count); run_tests_with_groups( - executor_thread_pool, gas_limits, transactions, &data_view, diff --git a/aptos-move/block-executor/src/combinatorial_tests/delta_tests.rs b/aptos-move/block-executor/src/combinatorial_tests/delta_tests.rs index eae348c360b..ea75df388f6 100644 --- a/aptos-move/block-executor/src/combinatorial_tests/delta_tests.rs +++ b/aptos-move/block-executor/src/combinatorial_tests/delta_tests.rs @@ -7,8 +7,7 @@ use crate::{ baseline::BaselineOutput, mock_executor::{MockEvent, MockTask}, resource_tests::{ - create_executor_thread_pool, execute_block_parallel, - generate_universe_and_transactions, get_gas_limit_variants, + execute_block_parallel, generate_universe_and_transactions, get_gas_limit_variants, }, types::{DeltaDataView, KeyType, MockTransaction}, }, @@ -28,8 +27,6 @@ fn run_transactions_deltas( num_executions: usize, num_random_generations: usize, ) { - let executor_thread_pool = create_executor_thread_pool(); - // The delta threshold controls how many keys / paths are guaranteed r/w resources even // in the presence of deltas. let delta_threshold = std::cmp::min(15, universe_size / 2); @@ -67,7 +64,6 @@ fn run_transactions_deltas( AuxiliaryInfo, >, >( - executor_thread_pool.clone(), maybe_block_gas_limit, &txn_provider, &data_view, diff --git a/aptos-move/block-executor/src/combinatorial_tests/group_tests.rs b/aptos-move/block-executor/src/combinatorial_tests/group_tests.rs index e5c799f8afc..f8a015cf908 100644 --- a/aptos-move/block-executor/src/combinatorial_tests/group_tests.rs +++ b/aptos-move/block-executor/src/combinatorial_tests/group_tests.rs @@ -7,9 +7,7 @@ use crate::{ combinatorial_tests::{ baseline::BaselineOutput, mock_executor::{MockEvent, MockTask}, - resource_tests::{ - create_executor_thread_pool, execute_block_parallel, get_gas_limit_variants, - }, + resource_tests::{execute_block_parallel, get_gas_limit_variants}, types::{ KeyType, MockTransaction, NonEmptyGroupDataView, TransactionGen, TransactionGenParams, }, @@ -27,7 +25,6 @@ use aptos_types::{ transaction::AuxiliaryInfo, }; use proptest::{collection::vec, prelude::*, strategy::ValueTree, test_runner::TestRunner}; -use std::sync::Arc; use test_case::test_case; /// Create a data view for testing with non-empty groups @@ -47,7 +44,6 @@ pub(crate) fn create_non_empty_group_data_view( /// Run both parallel and sequential execution tests for a transaction provider pub(crate) fn run_tests_with_groups( - executor_thread_pool: Arc, gas_limits: Vec>, transactions: Vec, MockEvent>>, data_view: &NonEmptyGroupDataView>, @@ -72,7 +68,6 @@ pub(crate) fn run_tests_with_groups( AuxiliaryInfo, >, >( - executor_thread_pool.clone(), *maybe_block_gas_limit, &txn_provider, data_view, @@ -99,7 +94,6 @@ pub(crate) fn run_tests_with_groups( AuxiliaryInfo, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool.clone(), None, ) .execute_transactions_sequential( @@ -170,11 +164,9 @@ fn non_empty_group_transaction_tests( .collect(); let data_view = create_non_empty_group_data_view(&key_universe, universe_size, false); - let executor_thread_pool = create_executor_thread_pool(); let gas_limits = get_gas_limit_variants(use_gas_limit, transaction_count); run_tests_with_groups( - executor_thread_pool, gas_limits, transactions, &data_view, diff --git a/aptos-move/block-executor/src/combinatorial_tests/module_tests.rs b/aptos-move/block-executor/src/combinatorial_tests/module_tests.rs index 4d4017b04c2..2e9a54aceca 100644 --- a/aptos-move/block-executor/src/combinatorial_tests/module_tests.rs +++ b/aptos-move/block-executor/src/combinatorial_tests/module_tests.rs @@ -6,9 +6,7 @@ use crate::{ combinatorial_tests::{ baseline::BaselineOutput, mock_executor::{MockEvent, MockTask}, - resource_tests::{ - create_executor_thread_pool, execute_block_parallel, get_gas_limit_variants, - }, + resource_tests::{execute_block_parallel, get_gas_limit_variants}, types::{ key_to_mock_module_id, KeyType, MockTransaction, TransactionGen, TransactionGenParams, }, @@ -49,7 +47,6 @@ fn execute_module_tests( assert!(fail::has_failpoints()); fail::cfg("module_test", "return").unwrap(); - let executor_thread_pool = create_executor_thread_pool(); let mut runner = TestRunner::default(); let module_id_pool = InternedModuleIdPool::new(); @@ -125,7 +122,6 @@ fn execute_module_tests( AuxiliaryInfo, >, >( - executor_thread_pool.clone(), *maybe_block_gas_limit, &txn_provider, &state_view, diff --git a/aptos-move/block-executor/src/combinatorial_tests/resource_tests.rs b/aptos-move/block-executor/src/combinatorial_tests/resource_tests.rs index 474117ea904..6c264e75c22 100644 --- a/aptos-move/block-executor/src/combinatorial_tests/resource_tests.rs +++ b/aptos-move/block-executor/src/combinatorial_tests/resource_tests.rs @@ -50,15 +50,6 @@ pub(crate) fn get_gas_limit_variants( } } -pub(crate) fn create_executor_thread_pool() -> Arc { - Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ) -} - /// Populates a module cache manager guard with empty modules for testing. /// This function creates empty modules for each ModuleId in the provided list and adds them to the guard's module cache. /// @@ -99,7 +90,6 @@ pub(crate) fn populate_guard_with_modules( } pub(crate) fn execute_block_parallel( - executor_thread_pool: Arc, block_gas_limit: Option, txn_provider: &Provider, data_view: &ViewType, @@ -127,7 +117,7 @@ where NoOpTransactionCommitHook, Provider, AuxiliaryInfo, - >::new(config, executor_thread_pool, None); + >::new(config, None); if block_stm_v2 { block_executor.execute_transactions_parallel_v2( @@ -184,7 +174,6 @@ pub(crate) fn run_transactions_resources( num_executions: usize, num_random_generations: usize, ) { - let executor_thread_pool = create_executor_thread_pool(); let mut runner = TestRunner::default(); let gas_limits = get_gas_limit_variants(use_gas_limit, transaction_count); @@ -247,7 +236,6 @@ pub(crate) fn run_transactions_resources( AuxiliaryInfo, >, >( - executor_thread_pool.clone(), *maybe_block_gas_limit, &txn_provider, &state_view, diff --git a/aptos-move/block-executor/src/combinatorial_tests/tests.rs b/aptos-move/block-executor/src/combinatorial_tests/tests.rs index 082b05a79af..5d45429a5d0 100644 --- a/aptos-move/block-executor/src/combinatorial_tests/tests.rs +++ b/aptos-move/block-executor/src/combinatorial_tests/tests.rs @@ -65,13 +65,6 @@ fn run_transactions( let state_view = MockStateView::empty(); - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); - let txn_provider = DefaultTxnProvider::new_without_info(transactions); for _ in 0..num_repeat { let mut guard = AptosModuleCacheManagerGuard::none(); @@ -84,7 +77,6 @@ fn run_transactions( DefaultTxnProvider, E>>, >::new( BlockExecutorConfig::new_maybe_block_limit(num_cpus::get(), maybe_block_gas_limit), - executor_thread_pool.clone(), None, ) .execute_transactions_parallel( @@ -209,13 +201,6 @@ fn deltas_writes_mixed_with_block_gas_limit(num_txns: usize, maybe_block_gas_lim phantom: PhantomData, }; - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); - for _ in 0..20 { let mut guard = AptosModuleCacheManagerGuard::none(); @@ -227,7 +212,6 @@ fn deltas_writes_mixed_with_block_gas_limit(num_txns: usize, maybe_block_gas_lim DefaultTxnProvider, MockEvent>>, >::new( BlockExecutorConfig::new_maybe_block_limit(num_cpus::get(), maybe_block_gas_limit), - executor_thread_pool.clone(), None, ) .execute_transactions_parallel( @@ -268,13 +252,6 @@ fn deltas_resolver_with_block_gas_limit(num_txns: usize, maybe_block_gas_limit: .collect(); let txn_provider = DefaultTxnProvider::new_without_info(transactions); - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); - for _ in 0..20 { let mut guard = AptosModuleCacheManagerGuard::none(); @@ -286,7 +263,6 @@ fn deltas_resolver_with_block_gas_limit(num_txns: usize, maybe_block_gas_limit: DefaultTxnProvider, MockEvent>>, >::new( BlockExecutorConfig::new_maybe_block_limit(num_cpus::get(), maybe_block_gas_limit), - executor_thread_pool.clone(), None, ) .execute_transactions_parallel( @@ -384,13 +360,6 @@ fn publishing_fixed_params_with_block_gas_limit( phantom: PhantomData, }; - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); - let txn_provider = DefaultTxnProvider::new_without_info(transactions.clone()); // Confirm still no intersection let mut guard = AptosModuleCacheManagerGuard::none(); @@ -402,7 +371,6 @@ fn publishing_fixed_params_with_block_gas_limit( DefaultTxnProvider, MockEvent>>, >::new( BlockExecutorConfig::new_maybe_block_limit(num_cpus::get(), maybe_block_gas_limit), - executor_thread_pool, None, ) .execute_transactions_parallel( @@ -433,13 +401,6 @@ fn publishing_fixed_params_with_block_gas_limit( }, }; - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); - let txn_provider = DefaultTxnProvider::new_without_info(transactions); for _ in 0..200 { let mut guard = AptosModuleCacheManagerGuard::none(); @@ -455,7 +416,6 @@ fn publishing_fixed_params_with_block_gas_limit( num_cpus::get(), Some(max(w_index, r_index) as u64 * MAX_GAS_PER_TXN + 1), ), - executor_thread_pool.clone(), None, ) // Ensure enough gas limit to commit the module txns (4 is maximum gas per txn) .execute_transactions_parallel( @@ -524,13 +484,6 @@ fn non_empty_group( .collect(), }; - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); - for _ in 0..num_repeat_parallel { let mut guard = AptosModuleCacheManagerGuard::none(); @@ -542,7 +495,6 @@ fn non_empty_group( DefaultTxnProvider, MockEvent>>, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool.clone(), None, ) .execute_transactions_parallel( @@ -566,7 +518,6 @@ fn non_empty_group( DefaultTxnProvider, MockEvent>>, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool.clone(), None, ) .execute_transactions_sequential( diff --git a/aptos-move/block-executor/src/executor.rs b/aptos-move/block-executor/src/executor.rs index f3365307b4c..25080cfbfb1 100644 --- a/aptos-move/block-executor/src/executor.rs +++ b/aptos-move/block-executor/src/executor.rs @@ -68,15 +68,12 @@ use move_core_types::{language_storage::ModuleId, value::MoveTypeLayout, vm_stat use move_vm_runtime::{Module, RuntimeEnvironment, TypeChecker, WithRuntimeEnvironment}; use move_vm_types::delayed_values::delayed_field_id::DelayedFieldID; use num_cpus; -use rayon::ThreadPool; use std::{ cell::RefCell, collections::{BTreeMap, BTreeSet, HashMap, HashSet}, marker::{PhantomData, Sync}, - sync::{ - atomic::{AtomicBool, AtomicU32, Ordering}, - Arc, - }, + sync::atomic::{AtomicBool, AtomicU32, Ordering}, + thread, }; use triomphe::Arc as TriompheArc; @@ -99,11 +96,40 @@ where maybe_block_epilogue_txn_idx: &'a ExplicitSyncWrapper>, } +/// Spawns one Block-STM worker as a plain OS thread bound to `scope`. +/// +/// Block-STM workers deliberately do **not** run on a rayon pool. A rayon worker +/// that blocks is not idle: rayon's `wait_until` loop keeps it busy by stealing +/// other jobs from its own pool onto that thread. So a worker that blocks inside a +/// Move native's nested `par_iter()` can pick up a sibling worker task, and that +/// stolen task can then block on state the original task still holds -- a +/// writer-preferring `RwLock` over per-txn status, or a dependency condvar for the +/// very transaction this thread had been executing. Neither side can make progress +/// and the scheduler still believes the transaction is being executed. +/// +/// Plain `std` threads are not registered with rayon, so a nested `par_iter()` +/// runs on rayon's global pool while this thread parks on OS primitives, owing +/// rayon nothing. That removes the cycle structurally, for every native, rather +/// than requiring each rayon-using native to remember to isolate itself. +fn spawn_block_stm_worker<'scope, F>( + scope: &'scope thread::Scope<'scope, '_>, + worker_id: u32, + worker: F, +) where + F: FnOnce() + Send + 'scope, +{ + thread::Builder::new() + // Keep this short: Linux truncates thread names to 15 bytes, and the name is + // the fastest signal in a thread dump that these are not rayon workers. + .name(format!("blockstm-{}", worker_id)) + .spawn_scoped(scope, worker) + .expect("failed to spawn Block-STM worker thread"); +} + pub struct BlockExecutor { - // Number of active concurrent tasks, corresponding to the maximum number of rayon - // threads that may be concurrently participating in parallel execution. + // Number of active concurrent tasks, corresponding to the maximum number of + // worker threads that may be concurrently participating in parallel execution. config: BlockExecutorConfig, - executor_thread_pool: Arc, transaction_commit_hook: Option, phantom: PhantomData (T, E, S, L, TP, A)>, } @@ -119,11 +145,7 @@ where { /// 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. - pub fn new( - config: BlockExecutorConfig, - executor_thread_pool: Arc, - transaction_commit_hook: Option, - ) -> Self { + pub fn new(config: BlockExecutorConfig, transaction_commit_hook: Option) -> Self { let num_cpus = num_cpus::get(); assert!( config.local.concurrency_level > 0 && config.local.concurrency_level <= num_cpus, @@ -133,7 +155,6 @@ where ); Self { config, - executor_thread_pool, transaction_commit_hook, phantom: PhantomData, } @@ -1763,9 +1784,9 @@ where let timer = RAYON_EXECUTION_SECONDS.start_timer(); let worker_ids: Vec = (0..num_workers).collect(); let maybe_executor = ExplicitSyncWrapper::new(None); - self.executor_thread_pool.scope(|s| { + thread::scope(|s| { for worker_id in &worker_ids { - s.spawn(|_| { + spawn_block_stm_worker(s, *worker_id, || { let environment = module_cache_manager_guard.environment(); let executor = { let _init_timer = VM_INIT_SECONDS.start_timer(); @@ -1920,9 +1941,9 @@ where worker_ids.len() as u32, ); - self.executor_thread_pool.scope(|s| { + thread::scope(|s| { for worker_id in &worker_ids { - s.spawn(|_| { + spawn_block_stm_worker(s, *worker_id, || { let environment = module_cache_manager_guard.environment(); let executor = { let _init_timer = VM_INIT_SECONDS.start_timer(); @@ -2713,3 +2734,60 @@ fn should_perform_async_runtime_checks_for_block( ) -> bool { environment.async_runtime_checks_enabled() && num_txns > 3 } + +#[cfg(test)] +mod worker_thread_tests { + use super::*; + use std::sync::Mutex; + + /// Pins the invariant the Block-STM worker threading exists to guarantee: workers + /// must not be rayon workers, even when block execution is driven from inside a + /// rayon pool (as the sharded executor does). If a worker were rayon-registered, + /// blocking inside a Move native's nested `par_iter()` would let rayon steal + /// sibling worker tasks onto it, which is the deadlock this design rules out. + #[test] + fn workers_are_not_rayon_registered() { + let driver = rayon::ThreadPoolBuilder::new() + .num_threads(2) + .build() + .unwrap(); + let observed: Mutex, Option)>> = Mutex::new(Vec::new()); + + // Drive from a rayon worker: the worst case, and the one that used to deadlock. + driver.install(|| { + // Guards against the assertion below going vacuous: the driving thread really + // is rayon-registered, so `None` in a worker is a meaningful difference. + assert!( + rayon::current_thread_index().is_some(), + "test setup: driver should be a rayon worker" + ); + thread::scope(|s| { + for worker_id in 0..4u32 { + spawn_block_stm_worker(s, worker_id, || { + observed.lock().unwrap().push(( + rayon::current_thread_index(), + thread::current().name().map(str::to_string), + )); + }); + } + }); + }); + + let observed = observed.into_inner().unwrap(); + assert_eq!(observed.len(), 4, "every worker should have run"); + for (rayon_index, name) in &observed { + assert_eq!( + *rayon_index, None, + "Block-STM worker is registered with rayon as index {:?}; nested par_iter \ + in a native could steal sibling worker tasks onto it", + rayon_index + ); + let name = name.as_deref().unwrap_or(""); + assert!( + name.starts_with("blockstm-"), + "worker thread should be identifiable in a thread dump, got {:?}", + name + ); + } + } +} diff --git a/aptos-move/block-executor/src/unit_tests/mod.rs b/aptos-move/block-executor/src/unit_tests/mod.rs index de59e88b973..94d5ff09618 100644 --- a/aptos-move/block-executor/src/unit_tests/mod.rs +++ b/aptos-move/block-executor/src/unit_tests/mod.rs @@ -47,7 +47,6 @@ use std::{ fmt::Debug, hash::Hash, marker::PhantomData, - sync::Arc, }; #[test] @@ -57,12 +56,6 @@ fn test_block_epilogue_happy_path() { let t_1 = MockTransaction::from_behavior(behaivor); let transactions = vec![t_0, t_1]; - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); let block_executor = BlockExecutor::< MockTransaction, MockEvent>, MockTask, MockEvent>, @@ -72,7 +65,6 @@ fn test_block_epilogue_happy_path() { AuxiliaryInfo, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool, None, ); let data_view = MockStateView::empty(); @@ -128,12 +120,6 @@ fn test_block_epilogue_block_gas_limit_reached() { let t_1 = MockTransaction::from_behavior(behaivor); let transactions = vec![t_0, t_1]; - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); let block_executor = BlockExecutor::< MockTransaction, MockEvent>, MockTask, MockEvent>, @@ -143,7 +129,6 @@ fn test_block_epilogue_block_gas_limit_reached() { AuxiliaryInfo, >::new( BlockExecutorConfig::new_maybe_block_limit(num_cpus::get(), Some(1)), - executor_thread_pool, None, ); let data_view = MockStateView::empty(); @@ -223,12 +208,6 @@ fn test_resource_group_deletion() { group_keys: HashSet::new(), delayed_field_testing: false, }; - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); let block_executor = BlockExecutor::< MockTransaction, MockEvent>, MockTask, MockEvent>, @@ -238,7 +217,6 @@ fn test_resource_group_deletion() { AuxiliaryInfo, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool, None, ); @@ -302,12 +280,6 @@ fn resource_group_bcs_fallback() { group_keys: HashSet::new(), delayed_field_testing: false, }; - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); let block_executor = BlockExecutor::< MockTransaction, MockEvent>, MockTask, MockEvent>, @@ -317,7 +289,6 @@ fn resource_group_bcs_fallback() { AuxiliaryInfo, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool, None, ); @@ -418,12 +389,6 @@ fn interrupt_requested() { let mut guard = AptosModuleCacheManagerGuard::none(); let data_view = MockStateView::empty(); - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); let block_executor = BlockExecutor::< MockTransaction, MockEvent>, MockTask, MockEvent>, @@ -433,7 +398,6 @@ fn interrupt_requested() { AuxiliaryInfo, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool, None, ); @@ -465,12 +429,6 @@ fn block_output_err_precedence() { let txn_provider = DefaultTxnProvider::new_without_info(transactions); let data_view = MockStateView::empty(); - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); let block_executor = BlockExecutor::< MockTransaction, MockEvent>, MockTask, MockEvent>, @@ -480,7 +438,6 @@ fn block_output_err_precedence() { AuxiliaryInfo, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool, None, ); @@ -509,12 +466,6 @@ fn skip_rest_gas_limit() { let txn_provider = DefaultTxnProvider::new_without_info(transactions); let data_view = MockStateView::empty(); - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); let block_executor = BlockExecutor::< MockTransaction, MockEvent>, MockTask, MockEvent>, @@ -524,7 +475,6 @@ fn skip_rest_gas_limit() { AuxiliaryInfo, >::new( BlockExecutorConfig::new_maybe_block_limit(num_cpus::get(), Some(5)), - executor_thread_pool, None, ); @@ -544,13 +494,6 @@ where K: PartialOrd + Ord + Send + Sync + Clone + Hash + Eq + ModulePath + Debug + 'static, E: Send + Sync + Debug + Clone + TransactionEvent + 'static, { - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); - let mut guard = AptosModuleCacheManagerGuard::none(); let txn_provider = DefaultTxnProvider::new_without_info(transactions); @@ -568,7 +511,6 @@ where AuxiliaryInfo, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool, None, ) .execute_transactions_parallel( @@ -588,7 +530,6 @@ where AuxiliaryInfo, >::new( BlockExecutorConfig::new_no_block_limit(num_cpus::get()), - executor_thread_pool, None, ) .execute_transactions_parallel( diff --git a/aptos-move/e2e-tests/src/executor.rs b/aptos-move/e2e-tests/src/executor.rs index 20cfcfe8b05..8e575f5e7f0 100644 --- a/aptos-move/e2e-tests/src/executor.rs +++ b/aptos-move/e2e-tests/src/executor.rs @@ -159,7 +159,7 @@ struct SharedCacheState { pub struct FakeExecutorImpl { state_store: FakeExecutorStateStore, event_store: Vec, - executor_thread_pool: Arc, + concurrency_level: usize, block_time: u64, executed_output: Option, trace_dir: Option, @@ -229,12 +229,7 @@ pub enum ExecFuncTimerDynamicArgs { impl FakeExecutorImpl { /// Creates an executor from a genesis [`WriteSet`]. pub fn from_genesis(write_set: &WriteSet, chain_id: ChainId) -> Self { - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); + let concurrency_level = num_cpus::get(); let state_store = empty_in_memory_state_store(); state_store.set_chain_id(chain_id).unwrap(); @@ -242,7 +237,7 @@ impl FakeExecutorImpl { let mut executor = Self { state_store, event_store: Vec::new(), - executor_thread_pool, + concurrency_level, block_time: 0, executed_output: None, trace_dir: None, @@ -256,10 +251,10 @@ impl FakeExecutorImpl { } #[cfg(any(test, feature = "fuzzing"))] - pub fn from_genesis_with_existing_thread_pool( + pub fn from_genesis_with_concurrency_level( write_set: &WriteSet, chain_id: ChainId, - executor_thread_pool: Arc, + concurrency_level: usize, module_cache_manager: Option, ) -> Self { let state_store = empty_in_memory_state_store(); @@ -268,7 +263,7 @@ impl FakeExecutorImpl { let mut executor = Self { state_store, event_store: Vec::new(), - executor_thread_pool, + concurrency_level, block_time: 0, executed_output: None, trace_dir: None, @@ -310,17 +305,12 @@ impl FakeExecutorImpl { .get_on_chain_config::() .expect("failed to get block time from remote"); - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); + let concurrency_level = num_cpus::get(); Self { state_store, event_store: Vec::new(), - executor_thread_pool, + concurrency_level, block_time: timestamp.microseconds, executed_output: None, trace_dir: None, @@ -461,16 +451,11 @@ impl FakeExecutorImpl { /// Creates an executor in which no genesis state has been applied yet. pub fn no_genesis() -> Self { - let executor_thread_pool = Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(num_cpus::get()) - .build() - .unwrap(), - ); + let concurrency_level = num_cpus::get(); Self { state_store: empty_in_memory_state_store(), event_store: Vec::new(), - executor_thread_pool, + concurrency_level, block_time: 0, executed_output: None, trace_dir: None, @@ -844,12 +829,7 @@ impl FakeExecutorImpl { let txn_provider = DefaultTxnProvider::new(txn_block, auxiliary_info); let metadata = self.get_txn_slice_metadata(); let result = { - AptosVMBlockExecutorWrapper::execute_block_on_thread_pool::< - _, - NoOpTransactionCommitHook, - _, - >( - self.executor_thread_pool.clone(), + AptosVMBlockExecutorWrapper::execute_block::<_, NoOpTransactionCommitHook, _>( &txn_provider, &state_view, self.module_cache_manager_opt() @@ -1019,8 +999,8 @@ impl FakeExecutorImpl { } let parallel_output = if mode != ExecutorMode::SequentialOnly { - // use the number of threads specified in the executor thread pool as specified at construction time - config.local.concurrency_level = self.executor_thread_pool.current_num_threads(); + // use the concurrency level specified at construction time + config.local.concurrency_level = self.concurrency_level; Some(self.execute_transaction_block_impl_with_state_view( sig_verified_block, state_view, diff --git a/execution/executor-benchmark/src/native/native_vm.rs b/execution/executor-benchmark/src/native/native_vm.rs index db01a954dc0..994219368e5 100644 --- a/execution/executor-benchmark/src/native/native_vm.rs +++ b/execution/executor-benchmark/src/native/native_vm.rs @@ -5,7 +5,7 @@ use crate::{ db_access::DbAccessUtil, native::{ - native_config::{NativeConfig, NATIVE_EXECUTOR_POOL}, + native_config::NativeConfig, native_transaction::{compute_deltas_for_batch, NativeTransaction}, }, }; @@ -69,7 +69,7 @@ use move_core_types::{ }; use move_vm_types::delayed_values::delayed_field_id::DelayedFieldID; use serde::{de::DeserializeOwned, Serialize}; -use std::{collections::BTreeMap, fmt::Debug, sync::Arc}; +use std::{collections::BTreeMap, fmt::Debug}; pub struct NativeVMBlockExecutor; @@ -91,12 +91,11 @@ impl VMBlockExecutor for NativeVMBlockExecutor { onchain_config: BlockExecutorConfigFromOnchain, transaction_slice_metadata: TransactionSliceMetadata, ) -> Result, VMStatus> { - AptosBlockExecutorWrapper::::execute_block_on_thread_pool::< + AptosBlockExecutorWrapper::::execute_block::< _, NoOpTransactionCommitHook, _, >( - Arc::clone(&NATIVE_EXECUTOR_POOL), txn_provider, state_view, &AptosModuleCacheManager::new(), diff --git a/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_authenticators.rs b/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_authenticators.rs index b32fefbbbc6..8f646905c76 100644 --- a/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_authenticators.rs +++ b/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_authenticators.rs @@ -52,23 +52,14 @@ use utils::{ static VM: Lazy = Lazy::new(|| GENESIS_CHANGE_SET_HEAD.write_set().clone()); const FUZZER_CONCURRENCY_LEVEL: usize = 1; -static TP: Lazy> = Lazy::new(|| { - Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(FUZZER_CONCURRENCY_LEVEL) - .build() - .unwrap(), - ) -}); - fn run_case(input: TransactionState) -> Result<(), Corpus> { tdbg!(&input); AptosVM::set_concurrency_level_once(FUZZER_CONCURRENCY_LEVEL); - let mut vm = FakeExecutor::from_genesis_with_existing_thread_pool( + let mut vm = FakeExecutor::from_genesis_with_concurrency_level( &VM, ChainId::mainnet(), - Arc::clone(&TP), + FUZZER_CONCURRENCY_LEVEL, None, ) .set_not_parallel(); diff --git a/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish.rs b/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish.rs index cd37385d623..1a10a831c41 100644 --- a/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish.rs +++ b/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish.rs @@ -16,7 +16,7 @@ use move_binary_format::{ file_format::{CompiledModule, CompiledScript}, }; use once_cell::sync::Lazy; -use std::{collections::HashSet, sync::Arc}; +use std::collections::HashSet; use utils::vm::{group_modules_by_address_topo, publish_group}; // genesis write set generated once for each fuzzing session @@ -24,15 +24,6 @@ static VM: Lazy = Lazy::new(|| GENESIS_CHANGE_SET_HEAD.write_set().clo const TEST_UPGRADE: bool = true; const FUZZER_CONCURRENCY_LEVEL: usize = 1; -static TP: Lazy> = Lazy::new(|| { - Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(FUZZER_CONCURRENCY_LEVEL) - .build() - .unwrap(), - ) -}); - fn run_case(mut input: RunnableState) -> Result<(), Corpus> { tdbg!(&input); @@ -72,10 +63,10 @@ fn run_case(mut input: RunnableState) -> Result<(), Corpus> { let packages = group_modules_by_address_topo(input.dep_modules.clone())?; AptosVM::set_concurrency_level_once(FUZZER_CONCURRENCY_LEVEL); - let mut vm = FakeExecutor::from_genesis_with_existing_thread_pool( + let mut vm = FakeExecutor::from_genesis_with_concurrency_level( &VM, ChainId::mainnet(), - Arc::clone(&TP), + FUZZER_CONCURRENCY_LEVEL, None, ) .set_not_parallel(); diff --git a/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish_and_run.rs b/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish_and_run.rs index a1dea2a53d0..0486184e7bf 100644 --- a/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish_and_run.rs +++ b/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish_and_run.rs @@ -22,7 +22,7 @@ use move_binary_format::{ }; use move_core_types::vm_status::{StatusCode, StatusType}; use once_cell::sync::Lazy; -use std::{collections::HashSet, sync::Arc, time::Instant}; +use std::{collections::HashSet, time::Instant}; mod utils; use fuzzer::{Authenticator, ExecVariant, RunnableState}; use move_vm_runtime::RuntimeEnvironment; @@ -35,15 +35,6 @@ use utils::vm::{ static VM_WRITE_SET: Lazy = Lazy::new(|| GENESIS_CHANGE_SET_HEAD.write_set().clone()); const FUZZER_CONCURRENCY_LEVEL: usize = 1; -static TP: Lazy> = Lazy::new(|| { - Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(FUZZER_CONCURRENCY_LEVEL) - .build() - .unwrap(), - ) -}); - const MAX_TYPE_PARAMETER_VALUE: u16 = 64 / 4 * 16; // third_party/move/move-bytecode-verifier/src/signature_v2.rs#L1306-L1312 const EXECUTION_TIME_GAS_RATIO: u8 = 100; @@ -109,10 +100,10 @@ fn run_case(mut input: RunnableState) -> Result<(), Corpus> { AptosVM::set_concurrency_level_once(FUZZER_CONCURRENCY_LEVEL); // Enable runtime reference-safety checks for the Move VM // prod_configs::set_paranoid_ref_checks(true); - let mut vm = FakeExecutor::from_genesis_with_existing_thread_pool( + let mut vm = FakeExecutor::from_genesis_with_concurrency_level( &VM_WRITE_SET, ChainId::mainnet(), - Arc::clone(&TP), + FUZZER_CONCURRENCY_LEVEL, None, ) .set_not_parallel(); diff --git a/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish_and_run_transactional.rs b/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish_and_run_transactional.rs index db62a0ea671..54818152ccd 100644 --- a/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish_and_run_transactional.rs +++ b/testsuite/fuzzer/fuzz/fuzz_targets/move/aptosvm_publish_and_run_transactional.rs @@ -29,10 +29,7 @@ use move_core_types::{ }; use move_transactional_test_runner::transactional_ops::TransactionalOperation; use once_cell::sync::Lazy; -use std::{ - collections::{HashMap, HashSet}, - sync::Arc, -}; +use std::collections::{HashMap, HashSet}; mod utils; use fuzzer::RunnableStateWithOperations; use utils::vm::{ @@ -44,15 +41,6 @@ use utils::vm::{ static VM_WRITE_SET: Lazy = Lazy::new(|| GENESIS_CHANGE_SET_HEAD.write_set().clone()); const FUZZER_CONCURRENCY_LEVEL: usize = 4; -static TP: Lazy> = Lazy::new(|| { - Arc::new( - rayon::ThreadPoolBuilder::new() - .num_threads(FUZZER_CONCURRENCY_LEVEL) - .build() - .unwrap(), - ) -}); - const MAX_TYPE_PARAMETER_VALUE: u16 = 64 / 4 * 16; // third_party/move/move-bytecode-verifier/src/signature_v2.rs#L1306-L1312 // filter modules @@ -136,10 +124,10 @@ fn run_case(input: RunnableStateWithOperations) -> Result<(), Corpus> { let module_cache_manager = AptosModuleCacheManager::new(); AptosVM::set_concurrency_level_once(FUZZER_CONCURRENCY_LEVEL); - let mut vm = FakeExecutor::from_genesis_with_existing_thread_pool( + let mut vm = FakeExecutor::from_genesis_with_concurrency_level( &VM_WRITE_SET, ChainId::mainnet(), - Arc::clone(&TP), + FUZZER_CONCURRENCY_LEVEL, Some(module_cache_manager), ) .set_parallel();