diff options
Diffstat (limited to 'src/tools/rust-analyzer/crates/salsa/tests/parallel')
13 files changed, 0 insertions, 1178 deletions
diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/cancellation.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/cancellation.rs deleted file mode 100644 index 9a92e5cc1ff..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/cancellation.rs +++ /dev/null @@ -1,132 +0,0 @@ -use crate::setup::{CancellationFlag, Knobs, ParDatabase, ParDatabaseImpl, WithValue}; -use salsa::{Cancelled, ParallelDatabase}; - -macro_rules! assert_cancelled { - ($thread:expr) => { - match $thread.join() { - Ok(value) => panic!("expected cancellation, got {:?}", value), - Err(payload) => match payload.downcast::<Cancelled>() { - Ok(_) => {} - Err(payload) => ::std::panic::resume_unwind(payload), - }, - } - }; -} - -/// Add test where a call to `sum` is cancelled by a simultaneous -/// write. Check that we recompute the result in next revision, even -/// though none of the inputs have changed. -#[test] -fn in_par_get_set_cancellation_immediate() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 100); - db.set_input('b', 10); - db.set_input('c', 1); - db.set_input('d', 0); - - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - move || { - // This will not return until it sees cancellation is - // signaled. - db.knobs().sum_signal_on_entry.with_value(1, || { - db.knobs() - .sum_wait_for_cancellation - .with_value(CancellationFlag::Panic, || db.sum("abc")) - }) - } - }); - - // Wait until we have entered `sum` in the other thread. - db.wait_for(1); - - // Try to set the input. This will signal cancellation. - db.set_input('d', 1000); - - // This should re-compute the value (even though no input has changed). - let thread2 = std::thread::spawn({ - let db = db.snapshot(); - move || db.sum("abc") - }); - - assert_eq!(db.sum("d"), 1000); - assert_cancelled!(thread1); - assert_eq!(thread2.join().unwrap(), 111); -} - -/// Here, we check that `sum`'s cancellation is propagated -/// to `sum2` properly. -#[test] -fn in_par_get_set_cancellation_transitive() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 100); - db.set_input('b', 10); - db.set_input('c', 1); - db.set_input('d', 0); - - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - move || { - // This will not return until it sees cancellation is - // signaled. - db.knobs().sum_signal_on_entry.with_value(1, || { - db.knobs() - .sum_wait_for_cancellation - .with_value(CancellationFlag::Panic, || db.sum2("abc")) - }) - } - }); - - // Wait until we have entered `sum` in the other thread. - db.wait_for(1); - - // Try to set the input. This will signal cancellation. - db.set_input('d', 1000); - - // This should re-compute the value (even though no input has changed). - let thread2 = std::thread::spawn({ - let db = db.snapshot(); - move || db.sum2("abc") - }); - - assert_eq!(db.sum2("d"), 1000); - assert_cancelled!(thread1); - assert_eq!(thread2.join().unwrap(), 111); -} - -/// https://github.com/salsa-rs/salsa/issues/66 -#[test] -fn no_back_dating_in_cancellation() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 1); - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - move || { - // Here we compute a long-chain of queries, - // but the last one gets cancelled. - db.knobs().sum_signal_on_entry.with_value(1, || { - db.knobs() - .sum_wait_for_cancellation - .with_value(CancellationFlag::Panic, || db.sum3("a")) - }) - } - }); - - db.wait_for(1); - - // Set unrelated input to bump revision - db.set_input('b', 2); - - // Here we should recompuet the whole chain again, clearing the cancellation - // state. If we get `usize::max()` here, it is a bug! - assert_eq!(db.sum3("a"), 1); - - assert_cancelled!(thread1); - - db.set_input('a', 3); - db.set_input('a', 4); - assert_eq!(db.sum3("ab"), 6); -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/frozen.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/frozen.rs deleted file mode 100644 index 5359a8820e2..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/frozen.rs +++ /dev/null @@ -1,57 +0,0 @@ -use crate::setup::{ParDatabase, ParDatabaseImpl}; -use crate::signal::Signal; -use salsa::{Database, ParallelDatabase}; -use std::{ - panic::{catch_unwind, AssertUnwindSafe}, - sync::Arc, -}; - -/// Add test where a call to `sum` is cancelled by a simultaneous -/// write. Check that we recompute the result in next revision, even -/// though none of the inputs have changed. -#[test] -fn in_par_get_set_cancellation() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 1); - - let signal = Arc::new(Signal::default()); - - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - let signal = signal.clone(); - move || { - // Check that cancellation flag is not yet set, because - // `set` cannot have been called yet. - catch_unwind(AssertUnwindSafe(|| db.unwind_if_cancelled())).unwrap(); - - // Signal other thread to proceed. - signal.signal(1); - - // Wait for other thread to signal cancellation - catch_unwind(AssertUnwindSafe(|| loop { - db.unwind_if_cancelled(); - std::thread::yield_now(); - })) - .unwrap_err(); - } - }); - - let thread2 = std::thread::spawn({ - move || { - // Wait until thread 1 has asserted that they are not cancelled - // before we invoke `set.` - signal.wait_for(1); - - // This will block until thread1 drops the revision lock. - db.set_input('a', 2); - - db.input('a') - } - }); - - thread1.join().unwrap(); - - let c = thread2.join().unwrap(); - assert_eq!(c, 2); -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/independent.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/independent.rs deleted file mode 100644 index bd6ba3bf931..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/independent.rs +++ /dev/null @@ -1,29 +0,0 @@ -use crate::setup::{ParDatabase, ParDatabaseImpl}; -use salsa::ParallelDatabase; - -/// Test two `sum` queries (on distinct keys) executing in different -/// threads. Really just a test that `snapshot` etc compiles. -#[test] -fn in_par_two_independent_queries() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 100); - db.set_input('b', 10); - db.set_input('c', 1); - db.set_input('d', 200); - db.set_input('e', 20); - db.set_input('f', 2); - - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - move || db.sum("abc") - }); - - let thread2 = std::thread::spawn({ - let db = db.snapshot(); - move || db.sum("def") - }); - - assert_eq!(thread1.join().unwrap(), 111); - assert_eq!(thread2.join().unwrap(), 222); -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/main.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/main.rs deleted file mode 100644 index 31c0da18375..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/main.rs +++ /dev/null @@ -1,13 +0,0 @@ -mod setup; - -mod cancellation; -mod frozen; -mod independent; -mod parallel_cycle_all_recover; -mod parallel_cycle_mid_recover; -mod parallel_cycle_none_recover; -mod parallel_cycle_one_recovers; -mod race; -mod signal; -mod stress; -mod true_parallel; diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_all_recover.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_all_recover.rs deleted file mode 100644 index a13ae3418f2..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_all_recover.rs +++ /dev/null @@ -1,109 +0,0 @@ -//! Test for cycle recover spread across two threads. -//! See `../cycles.rs` for a complete listing of cycle tests, -//! both intra and cross thread. - -use crate::setup::{Knobs, ParDatabaseImpl}; -use salsa::ParallelDatabase; - -// Recover cycle test: -// -// The pattern is as follows. -// -// Thread A Thread B -// -------- -------- -// a1 b1 -// | wait for stage 1 (blocks) -// signal stage 1 | -// wait for stage 2 (blocks) (unblocked) -// | signal stage 2 -// (unblocked) wait for stage 3 (blocks) -// a2 | -// b1 (blocks -> stage 3) | -// | (unblocked) -// | b2 -// | a1 (cycle detected, recovers) -// | b2 completes, recovers -// | b1 completes, recovers -// a2 sees cycle, recovers -// a1 completes, recovers - -#[test] -fn parallel_cycle_all_recover() { - let db = ParDatabaseImpl::default(); - db.knobs().signal_on_will_block.set(3); - - let thread_a = std::thread::spawn({ - let db = db.snapshot(); - move || db.a1(1) - }); - - let thread_b = std::thread::spawn({ - let db = db.snapshot(); - move || db.b1(1) - }); - - assert_eq!(thread_a.join().unwrap(), 11); - assert_eq!(thread_b.join().unwrap(), 21); -} - -#[salsa::query_group(ParallelCycleAllRecover)] -pub(crate) trait TestDatabase: Knobs { - #[salsa::cycle(recover_a1)] - fn a1(&self, key: i32) -> i32; - - #[salsa::cycle(recover_a2)] - fn a2(&self, key: i32) -> i32; - - #[salsa::cycle(recover_b1)] - fn b1(&self, key: i32) -> i32; - - #[salsa::cycle(recover_b2)] - fn b2(&self, key: i32) -> i32; -} - -fn recover_a1(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 { - tracing::debug!("recover_a1"); - key * 10 + 1 -} - -fn recover_a2(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 { - tracing::debug!("recover_a2"); - key * 10 + 2 -} - -fn recover_b1(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 { - tracing::debug!("recover_b1"); - key * 20 + 1 -} - -fn recover_b2(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 { - tracing::debug!("recover_b2"); - key * 20 + 2 -} - -fn a1(db: &dyn TestDatabase, key: i32) -> i32 { - // Wait to create the cycle until both threads have entered - db.signal(1); - db.wait_for(2); - - db.a2(key) -} - -fn a2(db: &dyn TestDatabase, key: i32) -> i32 { - db.b1(key) -} - -fn b1(db: &dyn TestDatabase, key: i32) -> i32 { - // Wait to create the cycle until both threads have entered - db.wait_for(1); - db.signal(2); - - // Wait for thread A to block on this thread - db.wait_for(3); - - db.b2(key) -} - -fn b2(db: &dyn TestDatabase, key: i32) -> i32 { - db.a1(key) -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_mid_recover.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_mid_recover.rs deleted file mode 100644 index 971fe7ab120..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_mid_recover.rs +++ /dev/null @@ -1,109 +0,0 @@ -//! Test for cycle recover spread across two threads. -//! See `../cycles.rs` for a complete listing of cycle tests, -//! both intra and cross thread. - -use crate::setup::{Knobs, ParDatabaseImpl}; -use salsa::ParallelDatabase; - -// Recover cycle test: -// -// The pattern is as follows. -// -// Thread A Thread B -// -------- -------- -// a1 b1 -// | wait for stage 1 (blocks) -// signal stage 1 | -// wait for stage 2 (blocks) (unblocked) -// | | -// | b2 -// | b3 -// | a1 (blocks -> stage 2) -// (unblocked) | -// a2 (cycle detected) | -// b3 recovers -// b2 resumes -// b1 panics because bug - -#[test] -fn parallel_cycle_mid_recovers() { - let db = ParDatabaseImpl::default(); - db.knobs().signal_on_will_block.set(2); - - let thread_a = std::thread::spawn({ - let db = db.snapshot(); - move || db.a1(1) - }); - - let thread_b = std::thread::spawn({ - let db = db.snapshot(); - move || db.b1(1) - }); - - // We expect that the recovery function yields - // `1 * 20 + 2`, which is returned (and forwarded) - // to b1, and from there to a2 and a1. - assert_eq!(thread_a.join().unwrap(), 22); - assert_eq!(thread_b.join().unwrap(), 22); -} - -#[salsa::query_group(ParallelCycleMidRecovers)] -pub(crate) trait TestDatabase: Knobs { - fn a1(&self, key: i32) -> i32; - - fn a2(&self, key: i32) -> i32; - - #[salsa::cycle(recover_b1)] - fn b1(&self, key: i32) -> i32; - - fn b2(&self, key: i32) -> i32; - - #[salsa::cycle(recover_b3)] - fn b3(&self, key: i32) -> i32; -} - -fn recover_b1(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 { - tracing::debug!("recover_b1"); - key * 20 + 2 -} - -fn recover_b3(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 { - tracing::debug!("recover_b1"); - key * 200 + 2 -} - -fn a1(db: &dyn TestDatabase, key: i32) -> i32 { - // tell thread b we have started - db.signal(1); - - // wait for thread b to block on a1 - db.wait_for(2); - - db.a2(key) -} - -fn a2(db: &dyn TestDatabase, key: i32) -> i32 { - // create the cycle - db.b1(key) -} - -fn b1(db: &dyn TestDatabase, key: i32) -> i32 { - // wait for thread a to have started - db.wait_for(1); - - db.b2(key); - - 0 -} - -fn b2(db: &dyn TestDatabase, key: i32) -> i32 { - // will encounter a cycle but recover - db.b3(key); - db.b1(key); // hasn't recovered yet - 0 -} - -fn b3(db: &dyn TestDatabase, key: i32) -> i32 { - // will block on thread a, signaling stage 2 - db.a1(key) -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_none_recover.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_none_recover.rs deleted file mode 100644 index 3c73852eafb..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_none_recover.rs +++ /dev/null @@ -1,68 +0,0 @@ -//! Test a cycle where no queries recover that occurs across threads. -//! See the `../cycles.rs` for a complete listing of cycle tests, -//! both intra and cross thread. - -use crate::setup::{Knobs, ParDatabaseImpl}; -use expect_test::expect; -use salsa::ParallelDatabase; - -#[test] -fn parallel_cycle_none_recover() { - let db = ParDatabaseImpl::default(); - db.knobs().signal_on_will_block.set(3); - - let thread_a = std::thread::spawn({ - let db = db.snapshot(); - move || db.a(-1) - }); - - let thread_b = std::thread::spawn({ - let db = db.snapshot(); - move || db.b(-1) - }); - - // We expect B to panic because it detects a cycle (it is the one that calls A, ultimately). - // Right now, it panics with a string. - let err_b = thread_b.join().unwrap_err(); - if let Some(c) = err_b.downcast_ref::<salsa::Cycle>() { - expect![[r#" - [ - "parallel::parallel_cycle_none_recover::AQuery::a(-1)", - "parallel::parallel_cycle_none_recover::BQuery::b(-1)", - ] - "#]] - .assert_debug_eq(&c.unexpected_participants(&db)); - } else { - panic!("b failed in an unexpected way: {err_b:?}"); - } - - // We expect A to propagate a panic, which causes us to use the sentinel - // type `Canceled`. - assert!(thread_a.join().unwrap_err().downcast_ref::<salsa::Cycle>().is_some()); -} - -#[salsa::query_group(ParallelCycleNoneRecover)] -pub(crate) trait TestDatabase: Knobs { - fn a(&self, key: i32) -> i32; - fn b(&self, key: i32) -> i32; -} - -fn a(db: &dyn TestDatabase, key: i32) -> i32 { - // Wait to create the cycle until both threads have entered - db.signal(1); - db.wait_for(2); - - db.b(key) -} - -fn b(db: &dyn TestDatabase, key: i32) -> i32 { - // Wait to create the cycle until both threads have entered - db.wait_for(1); - db.signal(2); - - // Wait for thread A to block on this thread - db.wait_for(3); - - // Now try to execute A - db.a(key) -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_one_recovers.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_one_recovers.rs deleted file mode 100644 index 025fbf37477..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/parallel_cycle_one_recovers.rs +++ /dev/null @@ -1,94 +0,0 @@ -//! Test for cycle recover spread across two threads. -//! See `../cycles.rs` for a complete listing of cycle tests, -//! both intra and cross thread. - -use crate::setup::{Knobs, ParDatabaseImpl}; -use salsa::ParallelDatabase; - -// Recover cycle test: -// -// The pattern is as follows. -// -// Thread A Thread B -// -------- -------- -// a1 b1 -// | wait for stage 1 (blocks) -// signal stage 1 | -// wait for stage 2 (blocks) (unblocked) -// | signal stage 2 -// (unblocked) wait for stage 3 (blocks) -// a2 | -// b1 (blocks -> stage 3) | -// | (unblocked) -// | b2 -// | a1 (cycle detected) -// a2 recovery fn executes | -// a1 completes normally | -// b2 completes, recovers -// b1 completes, recovers - -#[test] -fn parallel_cycle_one_recovers() { - let db = ParDatabaseImpl::default(); - db.knobs().signal_on_will_block.set(3); - - let thread_a = std::thread::spawn({ - let db = db.snapshot(); - move || db.a1(1) - }); - - let thread_b = std::thread::spawn({ - let db = db.snapshot(); - move || db.b1(1) - }); - - // We expect that the recovery function yields - // `1 * 20 + 2`, which is returned (and forwarded) - // to b1, and from there to a2 and a1. - assert_eq!(thread_a.join().unwrap(), 22); - assert_eq!(thread_b.join().unwrap(), 22); -} - -#[salsa::query_group(ParallelCycleOneRecovers)] -pub(crate) trait TestDatabase: Knobs { - fn a1(&self, key: i32) -> i32; - - #[salsa::cycle(recover)] - fn a2(&self, key: i32) -> i32; - - fn b1(&self, key: i32) -> i32; - - fn b2(&self, key: i32) -> i32; -} - -fn recover(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 { - tracing::debug!("recover"); - key * 20 + 2 -} - -fn a1(db: &dyn TestDatabase, key: i32) -> i32 { - // Wait to create the cycle until both threads have entered - db.signal(1); - db.wait_for(2); - - db.a2(key) -} - -fn a2(db: &dyn TestDatabase, key: i32) -> i32 { - db.b1(key) -} - -fn b1(db: &dyn TestDatabase, key: i32) -> i32 { - // Wait to create the cycle until both threads have entered - db.wait_for(1); - db.signal(2); - - // Wait for thread A to block on this thread - db.wait_for(3); - - db.b2(key) -} - -fn b2(db: &dyn TestDatabase, key: i32) -> i32 { - db.a1(key) -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/race.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/race.rs deleted file mode 100644 index c53d4b464ea..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/race.rs +++ /dev/null @@ -1,37 +0,0 @@ -use std::panic::AssertUnwindSafe; - -use crate::setup::{ParDatabase, ParDatabaseImpl}; -use salsa::{Cancelled, ParallelDatabase}; - -/// Test where a read and a set are racing with one another. -/// Should be atomic. -#[test] -fn in_par_get_set_race() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 100); - db.set_input('b', 10); - db.set_input('c', 1); - - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - move || Cancelled::catch(AssertUnwindSafe(|| db.sum("abc"))) - }); - - let thread2 = std::thread::spawn(move || { - db.set_input('a', 1000); - db.sum("a") - }); - - // If the 1st thread runs first, you get 111, otherwise you get - // 1011; if they run concurrently and the 1st thread observes the - // cancellation, it'll unwind. - let result1 = thread1.join().unwrap(); - if let Ok(value1) = result1 { - assert!(value1 == 111 || value1 == 1011, "illegal result {value1}"); - } - - // thread2 can not observe a cancellation because it performs a - // database write before running any other queries. - assert_eq!(thread2.join().unwrap(), 1000); -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/setup.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/setup.rs deleted file mode 100644 index 0a35902b435..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/setup.rs +++ /dev/null @@ -1,197 +0,0 @@ -use crate::signal::Signal; -use salsa::Database; -use salsa::ParallelDatabase; -use salsa::Snapshot; -use std::sync::Arc; -use std::{ - cell::Cell, - panic::{catch_unwind, resume_unwind, AssertUnwindSafe}, -}; - -#[salsa::query_group(Par)] -pub(crate) trait ParDatabase: Knobs { - #[salsa::input] - fn input(&self, key: char) -> usize; - - fn sum(&self, key: &'static str) -> usize; - - /// Invokes `sum` - fn sum2(&self, key: &'static str) -> usize; - - /// Invokes `sum` but doesn't really care about the result. - fn sum2_drop_sum(&self, key: &'static str) -> usize; - - /// Invokes `sum2` - fn sum3(&self, key: &'static str) -> usize; - - /// Invokes `sum2_drop_sum` - fn sum3_drop_sum(&self, key: &'static str) -> usize; -} - -/// Various "knobs" and utilities used by tests to force -/// a certain behavior. -pub(crate) trait Knobs { - fn knobs(&self) -> &KnobsStruct; - - fn signal(&self, stage: usize); - - fn wait_for(&self, stage: usize); -} - -pub(crate) trait WithValue<T> { - fn with_value<R>(&self, value: T, closure: impl FnOnce() -> R) -> R; -} - -impl<T> WithValue<T> for Cell<T> { - fn with_value<R>(&self, value: T, closure: impl FnOnce() -> R) -> R { - let old_value = self.replace(value); - - let result = catch_unwind(AssertUnwindSafe(closure)); - - self.set(old_value); - - match result { - Ok(r) => r, - Err(payload) => resume_unwind(payload), - } - } -} - -#[derive(Default, Clone, Copy, PartialEq, Eq)] -pub(crate) enum CancellationFlag { - #[default] - Down, - Panic, -} - -/// Various "knobs" that can be used to customize how the queries -/// behave on one specific thread. Note that this state is -/// intentionally thread-local (apart from `signal`). -#[derive(Clone, Default)] -pub(crate) struct KnobsStruct { - /// A kind of flexible barrier used to coordinate execution across - /// threads to ensure we reach various weird states. - pub(crate) signal: Arc<Signal>, - - /// When this database is about to block, send a signal. - pub(crate) signal_on_will_block: Cell<usize>, - - /// Invocations of `sum` will signal this stage on entry. - pub(crate) sum_signal_on_entry: Cell<usize>, - - /// Invocations of `sum` will wait for this stage on entry. - pub(crate) sum_wait_for_on_entry: Cell<usize>, - - /// If true, invocations of `sum` will panic before they exit. - pub(crate) sum_should_panic: Cell<bool>, - - /// If true, invocations of `sum` will wait for cancellation before - /// they exit. - pub(crate) sum_wait_for_cancellation: Cell<CancellationFlag>, - - /// Invocations of `sum` will wait for this stage prior to exiting. - pub(crate) sum_wait_for_on_exit: Cell<usize>, - - /// Invocations of `sum` will signal this stage prior to exiting. - pub(crate) sum_signal_on_exit: Cell<usize>, - - /// Invocations of `sum3_drop_sum` will panic unconditionally - pub(crate) sum3_drop_sum_should_panic: Cell<bool>, -} - -fn sum(db: &dyn ParDatabase, key: &'static str) -> usize { - let mut sum = 0; - - db.signal(db.knobs().sum_signal_on_entry.get()); - - db.wait_for(db.knobs().sum_wait_for_on_entry.get()); - - if db.knobs().sum_should_panic.get() { - panic!("query set to panic before exit") - } - - for ch in key.chars() { - sum += db.input(ch); - } - - match db.knobs().sum_wait_for_cancellation.get() { - CancellationFlag::Down => (), - CancellationFlag::Panic => { - tracing::debug!("waiting for cancellation"); - loop { - db.unwind_if_cancelled(); - std::thread::yield_now(); - } - } - } - - db.wait_for(db.knobs().sum_wait_for_on_exit.get()); - - db.signal(db.knobs().sum_signal_on_exit.get()); - - sum -} - -fn sum2(db: &dyn ParDatabase, key: &'static str) -> usize { - db.sum(key) -} - -fn sum2_drop_sum(db: &dyn ParDatabase, key: &'static str) -> usize { - let _ = db.sum(key); - 22 -} - -fn sum3(db: &dyn ParDatabase, key: &'static str) -> usize { - db.sum2(key) -} - -fn sum3_drop_sum(db: &dyn ParDatabase, key: &'static str) -> usize { - if db.knobs().sum3_drop_sum_should_panic.get() { - panic!("sum3_drop_sum executed") - } - db.sum2_drop_sum(key) -} - -#[salsa::database( - Par, - crate::parallel_cycle_all_recover::ParallelCycleAllRecover, - crate::parallel_cycle_none_recover::ParallelCycleNoneRecover, - crate::parallel_cycle_mid_recover::ParallelCycleMidRecovers, - crate::parallel_cycle_one_recovers::ParallelCycleOneRecovers -)] -#[derive(Default)] -pub(crate) struct ParDatabaseImpl { - storage: salsa::Storage<Self>, - knobs: KnobsStruct, -} - -impl Database for ParDatabaseImpl { - fn salsa_event(&self, event: salsa::Event) { - if let salsa::EventKind::WillBlockOn { .. } = event.kind { - self.signal(self.knobs().signal_on_will_block.get()); - } - } -} - -impl ParallelDatabase for ParDatabaseImpl { - fn snapshot(&self) -> Snapshot<Self> { - Snapshot::new(ParDatabaseImpl { - storage: self.storage.snapshot(), - knobs: self.knobs.clone(), - }) - } -} - -impl Knobs for ParDatabaseImpl { - fn knobs(&self) -> &KnobsStruct { - &self.knobs - } - - fn signal(&self, stage: usize) { - self.knobs.signal.signal(stage); - } - - fn wait_for(&self, stage: usize) { - self.knobs.signal.wait_for(stage); - } -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/signal.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/signal.rs deleted file mode 100644 index 0af7b66e482..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/signal.rs +++ /dev/null @@ -1,40 +0,0 @@ -use parking_lot::{Condvar, Mutex}; - -#[derive(Default)] -pub(crate) struct Signal { - value: Mutex<usize>, - cond_var: Condvar, -} - -impl Signal { - pub(crate) fn signal(&self, stage: usize) { - tracing::debug!("signal({})", stage); - - // This check avoids acquiring the lock for things that will - // clearly be a no-op. Not *necessary* but helps to ensure we - // are more likely to encounter weird race conditions; - // otherwise calls to `sum` will tend to be unnecessarily - // synchronous. - if stage > 0 { - let mut v = self.value.lock(); - if stage > *v { - *v = stage; - self.cond_var.notify_all(); - } - } - } - - /// Waits until the given condition is true; the fn is invoked - /// with the current stage. - pub(crate) fn wait_for(&self, stage: usize) { - tracing::debug!("wait_for({})", stage); - - // As above, avoid lock if clearly a no-op. - if stage > 0 { - let mut v = self.value.lock(); - while *v < stage { - self.cond_var.wait(&mut v); - } - } - } -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/stress.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/stress.rs deleted file mode 100644 index 2fa317b2b90..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/stress.rs +++ /dev/null @@ -1,168 +0,0 @@ -use rand::seq::SliceRandom; -use rand::Rng; - -use salsa::ParallelDatabase; -use salsa::Snapshot; -use salsa::{Cancelled, Database}; - -// Number of operations a reader performs -const N_MUTATOR_OPS: usize = 100; -const N_READER_OPS: usize = 100; - -#[salsa::query_group(Stress)] -trait StressDatabase: salsa::Database { - #[salsa::input] - fn a(&self, key: usize) -> usize; - - fn b(&self, key: usize) -> usize; - - fn c(&self, key: usize) -> usize; -} - -fn b(db: &dyn StressDatabase, key: usize) -> usize { - db.unwind_if_cancelled(); - db.a(key) -} - -fn c(db: &dyn StressDatabase, key: usize) -> usize { - db.b(key) -} - -#[salsa::database(Stress)] -#[derive(Default)] -struct StressDatabaseImpl { - storage: salsa::Storage<Self>, -} - -impl salsa::Database for StressDatabaseImpl {} - -impl salsa::ParallelDatabase for StressDatabaseImpl { - fn snapshot(&self) -> Snapshot<StressDatabaseImpl> { - Snapshot::new(StressDatabaseImpl { storage: self.storage.snapshot() }) - } -} - -#[derive(Clone, Copy, Debug)] -enum Query { - A, - B, - C, -} - -enum MutatorOp { - WriteOp(WriteOp), - LaunchReader { ops: Vec<ReadOp>, check_cancellation: bool }, -} - -#[derive(Debug)] -enum WriteOp { - SetA(usize, usize), -} - -#[derive(Debug)] -enum ReadOp { - Get(Query, usize), -} - -impl rand::distributions::Distribution<Query> for rand::distributions::Standard { - fn sample<R: rand::Rng + ?Sized>(&self, rng: &mut R) -> Query { - *[Query::A, Query::B, Query::C].choose(rng).unwrap() - } -} - -impl rand::distributions::Distribution<MutatorOp> for rand::distributions::Standard { - fn sample<R: rand::Rng + ?Sized>(&self, rng: &mut R) -> MutatorOp { - if rng.gen_bool(0.5) { - MutatorOp::WriteOp(rng.gen()) - } else { - MutatorOp::LaunchReader { - ops: (0..N_READER_OPS).map(|_| rng.gen()).collect(), - check_cancellation: rng.gen(), - } - } - } -} - -impl rand::distributions::Distribution<WriteOp> for rand::distributions::Standard { - fn sample<R: rand::Rng + ?Sized>(&self, rng: &mut R) -> WriteOp { - let key = rng.gen::<usize>() % 10; - let value = rng.gen::<usize>() % 10; - WriteOp::SetA(key, value) - } -} - -impl rand::distributions::Distribution<ReadOp> for rand::distributions::Standard { - fn sample<R: rand::Rng + ?Sized>(&self, rng: &mut R) -> ReadOp { - let query = rng.gen::<Query>(); - let key = rng.gen::<usize>() % 10; - ReadOp::Get(query, key) - } -} - -fn db_reader_thread(db: &StressDatabaseImpl, ops: Vec<ReadOp>, check_cancellation: bool) { - for op in ops { - if check_cancellation { - db.unwind_if_cancelled(); - } - op.execute(db); - } -} - -impl WriteOp { - fn execute(self, db: &mut StressDatabaseImpl) { - match self { - WriteOp::SetA(key, value) => { - db.set_a(key, value); - } - } - } -} - -impl ReadOp { - fn execute(self, db: &StressDatabaseImpl) { - match self { - ReadOp::Get(query, key) => match query { - Query::A => { - db.a(key); - } - Query::B => { - let _ = db.b(key); - } - Query::C => { - let _ = db.c(key); - } - }, - } - } -} - -#[test] -fn stress_test() { - let mut db = StressDatabaseImpl::default(); - for i in 0..10 { - db.set_a(i, i); - } - - let mut rng = rand::thread_rng(); - - // generate the ops that the mutator thread will perform - let write_ops: Vec<MutatorOp> = (0..N_MUTATOR_OPS).map(|_| rng.gen()).collect(); - - // execute the "main thread", which sometimes snapshots off other threads - let mut all_threads = vec![]; - for op in write_ops { - match op { - MutatorOp::WriteOp(w) => w.execute(&mut db), - MutatorOp::LaunchReader { ops, check_cancellation } => { - all_threads.push(std::thread::spawn({ - let db = db.snapshot(); - move || Cancelled::catch(|| db_reader_thread(&db, ops, check_cancellation)) - })) - } - } - } - - for thread in all_threads { - thread.join().unwrap().ok(); - } -} diff --git a/src/tools/rust-analyzer/crates/salsa/tests/parallel/true_parallel.rs b/src/tools/rust-analyzer/crates/salsa/tests/parallel/true_parallel.rs deleted file mode 100644 index d0e58efd1ac..00000000000 --- a/src/tools/rust-analyzer/crates/salsa/tests/parallel/true_parallel.rs +++ /dev/null @@ -1,125 +0,0 @@ -use crate::setup::{Knobs, ParDatabase, ParDatabaseImpl, WithValue}; -use salsa::ParallelDatabase; -use std::panic::{self, AssertUnwindSafe}; - -/// Test where two threads are executing sum. We show that they can -/// both be executing sum in parallel by having thread1 wait for -/// thread2 to send a signal before it leaves (similarly, thread2 -/// waits for thread1 to send a signal before it enters). -#[test] -fn true_parallel_different_keys() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 100); - db.set_input('b', 10); - db.set_input('c', 1); - - // Thread 1 will signal stage 1 when it enters and wait for stage 2. - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - move || { - let v = db - .knobs() - .sum_signal_on_entry - .with_value(1, || db.knobs().sum_wait_for_on_exit.with_value(2, || db.sum("a"))); - v - } - }); - - // Thread 2 will wait_for stage 1 when it enters and signal stage 2 - // when it leaves. - let thread2 = std::thread::spawn({ - let db = db.snapshot(); - move || { - let v = db - .knobs() - .sum_wait_for_on_entry - .with_value(1, || db.knobs().sum_signal_on_exit.with_value(2, || db.sum("b"))); - v - } - }); - - assert_eq!(thread1.join().unwrap(), 100); - assert_eq!(thread2.join().unwrap(), 10); -} - -/// Add a test that tries to trigger a conflict, where we fetch -/// `sum("abc")` from two threads simultaneously, and of them -/// therefore has to block. -#[test] -fn true_parallel_same_keys() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 100); - db.set_input('b', 10); - db.set_input('c', 1); - - // Thread 1 will wait_for a barrier in the start of `sum` - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - move || { - let v = db - .knobs() - .sum_signal_on_entry - .with_value(1, || db.knobs().sum_wait_for_on_entry.with_value(2, || db.sum("abc"))); - v - } - }); - - // Thread 2 will wait until Thread 1 has entered sum and then -- - // once it has set itself to block -- signal Thread 1 to - // continue. This way, we test out the mechanism of one thread - // blocking on another. - let thread2 = std::thread::spawn({ - let db = db.snapshot(); - move || { - db.knobs().signal.wait_for(1); - db.knobs().signal_on_will_block.set(2); - db.sum("abc") - } - }); - - assert_eq!(thread1.join().unwrap(), 111); - assert_eq!(thread2.join().unwrap(), 111); -} - -/// Add a test that tries to trigger a conflict, where we fetch `sum("a")` -/// from two threads simultaneously. After `thread2` begins blocking, -/// we force `thread1` to panic and should see that propagate to `thread2`. -#[test] -fn true_parallel_propagate_panic() { - let mut db = ParDatabaseImpl::default(); - - db.set_input('a', 1); - - // `thread1` will wait_for a barrier in the start of `sum`. Once it can - // continue, it will panic. - let thread1 = std::thread::spawn({ - let db = db.snapshot(); - move || { - let v = db.knobs().sum_signal_on_entry.with_value(1, || { - db.knobs() - .sum_wait_for_on_entry - .with_value(2, || db.knobs().sum_should_panic.with_value(true, || db.sum("a"))) - }); - v - } - }); - - // `thread2` will wait until `thread1` has entered sum and then -- once it - // has set itself to block -- signal `thread1` to continue. - let thread2 = std::thread::spawn({ - let db = db.snapshot(); - move || { - db.knobs().signal.wait_for(1); - db.knobs().signal_on_will_block.set(2); - db.sum("a") - } - }); - - let result1 = panic::catch_unwind(AssertUnwindSafe(|| thread1.join().unwrap())); - let result2 = panic::catch_unwind(AssertUnwindSafe(|| thread2.join().unwrap())); - - assert!(result1.is_err()); - assert!(result2.is_err()); -} |