diff --git a/crates/blockifier/src/blockifier/transaction_executor.rs b/crates/blockifier/src/blockifier/transaction_executor.rs index 7e2941fc79..4536217c0f 100644 --- a/crates/blockifier/src/blockifier/transaction_executor.rs +++ b/crates/blockifier/src/blockifier/transaction_executor.rs @@ -1,6 +1,8 @@ #[cfg(feature = "concurrency")] use std::collections::{HashMap, HashSet}; #[cfg(feature = "concurrency")] +use std::panic::{self, catch_unwind, AssertUnwindSafe}; +#[cfg(feature = "concurrency")] use std::sync::Arc; #[cfg(feature = "concurrency")] use std::sync::Mutex; @@ -218,6 +220,8 @@ impl TransactionExecutor { &mut self, chunk: &[Transaction], ) -> Vec> { + use crate::concurrency::utils::AbortGuard; + let block_state = self.block_state.take().expect("The block state should be `Some`."); let worker_executor = Arc::new(WorkerExecutor::initialize( @@ -236,7 +240,25 @@ impl TransactionExecutor { for _ in 0..self.config.concurrency_config.n_workers { let worker_executor = Arc::clone(&worker_executor); s.spawn(move || { - worker_executor.run(); + // to make sure that if one of the threads panicked all threads would stop, + // and the main thread would panic. + let abort_guard = AbortGuard; + let result = catch_unwind(AssertUnwindSafe(|| { + worker_executor.run(); + })); + if let Err(err) = result { + // this make sure that the program will abort if a panic accured + // while halting the scheduler. + eprintln!( + "Worker executor thread panicked. {:?}", + err.downcast_ref::<&str>().unwrap() + ); + worker_executor.scheduler.halt(); + abort_guard.release(); + panic::resume_unwind(err); + } + abort_guard.release(); + }); } }); diff --git a/crates/blockifier/src/concurrency/scheduler.rs b/crates/blockifier/src/concurrency/scheduler.rs index 172918b365..b8301d2e59 100644 --- a/crates/blockifier/src/concurrency/scheduler.rs +++ b/crates/blockifier/src/concurrency/scheduler.rs @@ -52,7 +52,7 @@ impl<'a> TransactionCommitter<'a> { assert!(*self.commit_index_guard > 0, "Commit index underflow."); *self.commit_index_guard -= 1; - self.scheduler.done_marker.store(true, Ordering::Release); + self.scheduler.halt(); } } @@ -161,6 +161,10 @@ impl Scheduler { *self.commit_index.lock().unwrap() } + pub fn halt(&self) { + self.done_marker.store(true, Ordering::Release); + } + fn lock_tx_status(&self, tx_index: TxIndex) -> MutexGuard<'_, TransactionStatus> { lock_mutex_in_array(&self.tx_statuses, tx_index) } diff --git a/crates/blockifier/src/concurrency/utils.rs b/crates/blockifier/src/concurrency/utils.rs index 4ca2b2eb17..8b832e7c15 100644 --- a/crates/blockifier/src/concurrency/utils.rs +++ b/crates/blockifier/src/concurrency/utils.rs @@ -3,6 +3,23 @@ use std::sync::{Mutex, MutexGuard}; use crate::concurrency::TxIndex; +// This stract is used to abort the program if a +// panic ocurred in a place where it cannot be handled. +pub struct AbortGuard; + +impl Drop for AbortGuard { + fn drop(&mut self) { + eprintln!("detected unexpected panic; aborting"); + ::std::process::abort(); + } +} + +impl AbortGuard { + pub fn release(self) { + std::mem::forget(self); + } +} + pub fn lock_mutex_in_array(array: &[Mutex], tx_index: TxIndex) -> MutexGuard<'_, T> { array[tx_index].lock().unwrap_or_else(|error| { panic!("Cell of transaction index {} is poisoned. Data: {:?}.", tx_index, *error.get_ref()) diff --git a/crates/blockifier/src/concurrency/worker_logic.rs b/crates/blockifier/src/concurrency/worker_logic.rs index 60d5b19e71..de6753287e 100644 --- a/crates/blockifier/src/concurrency/worker_logic.rs +++ b/crates/blockifier/src/concurrency/worker_logic.rs @@ -118,7 +118,7 @@ impl<'a, S: StateReader> WorkerExecutor<'a, S> { fn execute(&self, tx_index: TxIndex) { self.execute_tx(tx_index); - self.scheduler.finish_execution(tx_index) + self.scheduler.finish_execution(tx_index); } fn execute_tx(&self, tx_index: TxIndex) {