Skip to content

Fix deadlock in rayon pool #422

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
seanyoung wants to merge 1 commit into
base: move-rebase-e33
Choose a base branch
from
Draft

Fix deadlock in rayon pool #422

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 2 additions & 39 deletions aptos-move/aptos-vm/src/block_executor/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Arc<rayon::ThreadPool>> = 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)]
Expand Down Expand Up @@ -513,12 +502,11 @@ impl<
>,
> AptosBlockExecutorWrapper<E>
{
pub fn execute_block_on_thread_pool<
pub fn execute_block<
S: StateView + Sync,
L: TransactionCommitHook,
TP: TxnProvider<SignatureVerifiedTransaction, AuxiliaryInfo> + Sync,
>(
executor_thread_pool: Arc<rayon::ThreadPool>,
signature_verified_block: &TP,
state_view: &S,
module_cache_manager: &AptosModuleCacheManager,
Expand Down Expand Up @@ -546,7 +534,6 @@ impl<
let executor =
BlockExecutor::<SignatureVerifiedTransaction, E, S, L, TP, AuxiliaryInfo>::new(
config,
executor_thread_pool,
transaction_commit_listener,
);

Expand Down Expand Up @@ -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<SignatureVerifiedTransaction, AuxiliaryInfo> + Sync,
>(
signature_verified_block: &TP,
state_view: &S,
module_cache_manager: &AptosModuleCacheManager,
config: BlockExecutorConfig,
transaction_slice_metadata: TransactionSliceMetadata,
transaction_commit_listener: Option<L>,
) -> Result<BlockOutput<SignatureVerifiedTransaction, TransactionOutput>, VMStatus> {
Self::execute_block_on_thread_pool::<S, L, TP>(
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
Expand Down
3 changes: 1 addition & 2 deletions aptos-move/aptos-vm/src/sharded_block_executor/sharded_executor_service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,8 +142,7 @@ impl<S: StateView + Sync + Send + 'static> ShardedExecutorService<S> {
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
Expand Down
11 changes: 2 additions & 9 deletions aptos-move/block-executor/src/combinatorial_tests/bencher.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<K, V, E> {
transaction_size: usize,
Expand Down Expand Up @@ -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();

Expand All @@ -144,7 +137,7 @@ where
NoOpTransactionCommitHook<usize>,
DefaultTxnProvider<MockTransaction<KeyType<K>, E>, AuxiliaryInfo>,
AuxiliaryInfo,
>::new(config, executor_thread_pool, None)
>::new(config, None)
.execute_transactions_parallel(
&self.txns_provider,
&state_view,
Expand Down
5 changes: 1 addition & 4 deletions aptos-move/block-executor/src/combinatorial_tests/delayed_field_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
6 changes: 1 addition & 5 deletions aptos-move/block-executor/src/combinatorial_tests/delta_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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},
},
Expand All @@ -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);
Expand Down Expand Up @@ -67,7 +64,6 @@ fn run_transactions_deltas(
AuxiliaryInfo,
>,
>(
executor_thread_pool.clone(),
maybe_block_gas_limit,
&txn_provider,
&data_view,
Expand Down
10 changes: 1 addition & 9 deletions aptos-move/block-executor/src/combinatorial_tests/group_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
},
Expand All @@ -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
Expand All @@ -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<rayon::ThreadPool>,
gas_limits: Vec<Option<u64>>,
transactions: Vec<MockTransaction<KeyType<[u8; 32]>, MockEvent>>,
data_view: &NonEmptyGroupDataView<KeyType<[u8; 32]>>,
Expand All @@ -72,7 +68,6 @@ pub(crate) fn run_tests_with_groups(
AuxiliaryInfo,
>,
>(
executor_thread_pool.clone(),
*maybe_block_gas_limit,
&txn_provider,
data_view,
Expand All @@ -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(
Expand Down Expand Up @@ -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,
Expand Down
6 changes: 1 addition & 5 deletions aptos-move/block-executor/src/combinatorial_tests/module_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
},
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -125,7 +122,6 @@ fn execute_module_tests(
AuxiliaryInfo,
>,
>(
executor_thread_pool.clone(),
*maybe_block_gas_limit,
&txn_provider,
&state_view,
Expand Down
14 changes: 1 addition & 13 deletions aptos-move/block-executor/src/combinatorial_tests/resource_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,15 +50,6 @@ pub(crate) fn get_gas_limit_variants(
}
}

pub(crate) fn create_executor_thread_pool() -> Arc<rayon::ThreadPool> {
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.
///
Expand Down Expand Up @@ -99,7 +90,6 @@ pub(crate) fn populate_guard_with_modules(
}

pub(crate) fn execute_block_parallel<TxnType, ViewType, Provider>(
executor_thread_pool: Arc<rayon::ThreadPool>,
block_gas_limit: Option<u64>,
txn_provider: &Provider,
data_view: &ViewType,
Expand Down Expand Up @@ -127,7 +117,7 @@ where
NoOpTransactionCommitHook<usize>,
Provider,
AuxiliaryInfo,
>::new(config, executor_thread_pool, None);
>::new(config, None);

if block_stm_v2 {
block_executor.execute_transactions_parallel_v2(
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -247,7 +236,6 @@ pub(crate) fn run_transactions_resources(
AuxiliaryInfo,
>,
>(
executor_thread_pool.clone(),
*maybe_block_gas_limit,
&txn_provider,
&state_view,
Expand Down
Loading
Loading