improve parallel cycle tests

They now use signals to guarantee we are testing the code paths
we want to be testing.

tests/cycles.rs

@@ -36,7 +36,8 @@ use test_env_log::test;
 // | Intra  | Fallback | Both     | Tracked   | direct   | cycle_revalidate |
 // | Intra  | Fallback | New      | Tracked   | direct   | cycle_appears |
 // | Intra  | Fallback | Old      | Tracked   | direct   | cycle_disappears |
-// | Cross  | Fallback | N/A      | Tracked   | both     | parallel_cycle |
+// | Cross  | Fallback | N/A      | Tracked   | both     | parallel/cycles.rs: recover_parallel_cycle |
+// | Cross  | Panic    | N/A      | Tracked   | both     | parallel/cycles.rs: panic_parallel_cycle |
 
 #[derive(PartialEq, Eq, Hash, Clone, Debug)]
 struct Error {
@@ -198,42 +199,3 @@ fn cycle_disappears() {
     db.set_should_create_cycle(false);
     assert!(db.cycle_a().is_ok());
 }
-
-#[test]
-fn parallel_cycle() {
-    let _ = env_logger::try_init();
-
-    let db = DatabaseImpl::default();
-    let thread1 = std::thread::spawn({
-        let db = db.snapshot();
-        move || {
-            let result = db.cycle_a();
-            assert!(result.is_err(), "Expected cycle error");
-            let cycle = result.unwrap_err().cycle;
-            assert!(
-                cycle
-                    .iter()
-                    .all(|l| ["cycle_b", "cycle_a"].iter().any(|r| l.contains(r))),
-                "{:#?}",
-                cycle
-            );
-        }
-    });
-
-    let thread2 = std::thread::spawn(move || {
-        let result = db.cycle_c();
-        assert!(result.is_err(), "Expected cycle error");
-        let cycle = result.unwrap_err().cycle;
-        assert!(
-            cycle
-                .iter()
-                .all(|l| ["cycle_b", "cycle_a"].iter().any(|r| l.contains(r))),
-            "{:#?}",
-            cycle
-        );
-    });
-
-    thread1.join().unwrap();
-    thread2.join().unwrap();
-    eprintln!("OK");
-}

tests/parallel/cycles.rs

@@ -0,0 +1,112 @@
+//! Tests for cycles that occur across threads. See the
+//! `../cycles.rs` for a complete listing of cycle tests,
+//! both intra and cross thread.
+
+use crate::setup::{Knobs, ParDatabase, ParDatabaseImpl};
+use salsa::{Cancelled, ParallelDatabase};
+use test_env_log::test;
+
+pub(crate) fn recover_cycle(_db: &dyn ParDatabase, _cycle: &[String], key: &i32) -> i32 {
+    key * 10
+}
+
+pub(crate) fn recover_cycle_a(db: &dyn ParDatabase, key: i32) -> i32 {
+    // Wait to create the cycle until both threads have entered
+    db.signal(0);
+    db.wait_for(1);
+
+    db.recover_cycle_b(key)
+}
+
+pub(crate) fn recover_cycle_b(db: &dyn ParDatabase, key: i32) -> i32 {
+    // Wait to create the cycle until both threads have entered
+    db.wait_for(0);
+    db.signal(1);
+
+    if key < 0 && db.should_cycle() {
+        db.recover_cycle_a(key)
+    } else {
+        key
+    }
+}
+
+pub(crate) fn panic_cycle_a(db: &dyn ParDatabase, key: i32) -> i32 {
+    // Wait to create the cycle until both threads have entered
+    db.signal(0);
+    db.wait_for(1);
+
+    db.panic_cycle_b(key)
+}
+
+pub(crate) fn panic_cycle_b(db: &dyn ParDatabase, key: i32) -> i32 {
+    // Wait to create the cycle until both threads have entered
+    db.wait_for(0);
+    db.signal(1);
+
+    // Wait for thread A to block on this thread
+    db.wait_for(2);
+
+    // Now try to execute A
+    if key < 0 && db.should_cycle() {
+        db.panic_cycle_a(key)
+    } else {
+        key
+    }
+}
+
+#[test]
+fn recover_parallel_cycle() {
+    let mut db = ParDatabaseImpl::default();
+    db.set_should_cycle(true);
+
+    let thread_a = std::thread::spawn({
+        let db = db.snapshot();
+        move || db.recover_cycle_a(-1)
+    });
+
+    let thread_b = std::thread::spawn({
+        let db = db.snapshot();
+        move || db.recover_cycle_b(-1)
+    });
+
+    assert_eq!(thread_a.join().unwrap(), -10);
+    assert_eq!(thread_b.join().unwrap(), -10);
+}
+
+#[test]
+fn panic_parallel_cycle() {
+    let mut db = ParDatabaseImpl::default();
+    db.set_should_cycle(true);
+    db.knobs().signal_on_will_block.set(2);
+
+    let thread_a = std::thread::spawn({
+        let db = db.snapshot();
+        move || db.panic_cycle_a(-1)
+    });
+
+    let thread_b = std::thread::spawn({
+        let db = db.snapshot();
+        move || db.panic_cycle_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(str_b) = err_b.downcast_ref::<String>() {
+        assert!(
+            str_b.contains("cycle detected"),
+            "unexpeced string: {:?}",
+            str_b
+        );
+    } else {
+        panic!("b failed in an unexpected way");
+    }
+
+    // We expect A to propagate a panic, which causes us to use the sentinel
+    // type `Canceled`.
+    assert!(thread_a
+        .join()
+        .unwrap_err()
+        .downcast_ref::<Cancelled>()
+        .is_some());
+}

tests/parallel/main.rs

@@ -1,6 +1,7 @@
 mod setup;
 
 mod cancellation;
+mod cycles;
 mod frozen;
 mod independent;
 mod race;

tests/parallel/setup.rs

@@ -26,6 +26,29 @@ pub(crate) trait ParDatabase: Knobs {
 
     /// Invokes `sum2_drop_sum`
     fn sum3_drop_sum(&self, key: &'static str) -> usize;
+
+    // Cycle tests:
+    //
+    // A ───────► B
+    //
+    // ▲          │
+    // └──────────┘
+    //  if key > 0
+    //  && should_cycle
+
+    #[salsa::input]
+    fn should_cycle(&self) -> bool;
+    #[salsa::cycle(crate::cycles::recover_cycle)]
+    #[salsa::invoke(crate::cycles::recover_cycle_a)]
+    fn recover_cycle_a(&self, key: i32) -> i32;
+    #[salsa::cycle(crate::cycles::recover_cycle)]
+    #[salsa::invoke(crate::cycles::recover_cycle_b)]
+    fn recover_cycle_b(&self, key: i32) -> i32;
+
+    #[salsa::invoke(crate::cycles::panic_cycle_a)]
+    fn panic_cycle_a(&self, key: i32) -> i32;
+    #[salsa::invoke(crate::cycles::panic_cycle_b)]
+    fn panic_cycle_b(&self, key: i32) -> i32;
 }
 
 /// Various "knobs" and utilities used by tests to force