diff options
Diffstat (limited to 'vendor/rust-analyzer-salsa/tests/parallel')
13 files changed, 1196 insertions, 0 deletions
diff --git a/vendor/rust-analyzer-salsa/tests/parallel/cancellation.rs b/vendor/rust-analyzer-salsa/tests/parallel/cancellation.rs new file mode 100644 index 000000000..72c0ed290 --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/cancellation.rs @@ -0,0 +1,132 @@ +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/vendor/rust-analyzer-salsa/tests/parallel/frozen.rs b/vendor/rust-analyzer-salsa/tests/parallel/frozen.rs new file mode 100644 index 000000000..40b736f5f --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/frozen.rs @@ -0,0 +1,57 @@ +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/vendor/rust-analyzer-salsa/tests/parallel/independent.rs b/vendor/rust-analyzer-salsa/tests/parallel/independent.rs new file mode 100644 index 000000000..29bcc2df4 --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/independent.rs @@ -0,0 +1,29 @@ +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/vendor/rust-analyzer-salsa/tests/parallel/main.rs b/vendor/rust-analyzer-salsa/tests/parallel/main.rs new file mode 100644 index 000000000..c1ad5620a --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/main.rs @@ -0,0 +1,13 @@ +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/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_all_recover.rs b/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_all_recover.rs new file mode 100644 index 000000000..e9cfbaffc --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_all_recover.rs @@ -0,0 +1,110 @@ +//! 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;
+use test_env_log::test;
+
+// 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 {
+ log::debug!("recover_a1");
+ key * 10 + 1
+}
+
+fn recover_a2(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 {
+ log::debug!("recover_a2");
+ key * 10 + 2
+}
+
+fn recover_b1(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 {
+ log::debug!("recover_b1");
+ key * 20 + 1
+}
+
+fn recover_b2(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 {
+ log::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/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_mid_recover.rs b/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_mid_recover.rs new file mode 100644 index 000000000..d702f89c2 --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_mid_recover.rs @@ -0,0 +1,110 @@ +//! 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;
+use test_env_log::test;
+
+// 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 {
+ log::debug!("recover_b1");
+ key * 20 + 2
+}
+
+fn recover_b3(_db: &dyn TestDatabase, _cycle: &salsa::Cycle, key: &i32) -> i32 {
+ log::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/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_none_recover.rs b/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_none_recover.rs new file mode 100644 index 000000000..fabfb39c6 --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_none_recover.rs @@ -0,0 +1,71 @@ +//! 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 salsa::ParallelDatabase;
+use test_env_log::test;
+
+#[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>() {
+ insta::assert_debug_snapshot!(c.unexpected_participants(&db), @r###"
+ [
+ "a(-1)",
+ "b(-1)",
+ ]
+ "###);
+ } 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/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_one_recovers.rs b/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_one_recovers.rs new file mode 100644 index 000000000..d12aa569d --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/parallel_cycle_one_recovers.rs @@ -0,0 +1,95 @@ +//! 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;
+use test_env_log::test;
+
+// 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 {
+ log::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/vendor/rust-analyzer-salsa/tests/parallel/race.rs b/vendor/rust-analyzer-salsa/tests/parallel/race.rs new file mode 100644 index 000000000..d1b3b5ccc --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/race.rs @@ -0,0 +1,37 @@ +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/vendor/rust-analyzer-salsa/tests/parallel/setup.rs b/vendor/rust-analyzer-salsa/tests/parallel/setup.rs new file mode 100644 index 000000000..6a7f10322 --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/setup.rs @@ -0,0 +1,202 @@ +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(Clone, Copy, PartialEq, Eq)]
+pub(crate) enum CancellationFlag {
+ Down,
+ Panic,
+}
+
+impl Default for CancellationFlag {
+ fn default() -> CancellationFlag {
+ CancellationFlag::Down
+ }
+}
+
+/// 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 => {
+ log::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/vendor/rust-analyzer-salsa/tests/parallel/signal.rs b/vendor/rust-analyzer-salsa/tests/parallel/signal.rs new file mode 100644 index 000000000..ad89e771e --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/signal.rs @@ -0,0 +1,40 @@ +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) {
+ log::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) {
+ log::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/vendor/rust-analyzer-salsa/tests/parallel/stress.rs b/vendor/rust-analyzer-salsa/tests/parallel/stress.rs new file mode 100644 index 000000000..1e140d21a --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/stress.rs @@ -0,0 +1,174 @@ +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/vendor/rust-analyzer-salsa/tests/parallel/true_parallel.rs b/vendor/rust-analyzer-salsa/tests/parallel/true_parallel.rs new file mode 100644 index 000000000..dd080f01e --- /dev/null +++ b/vendor/rust-analyzer-salsa/tests/parallel/true_parallel.rs @@ -0,0 +1,126 @@ +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());
+}
|