Panic if provisional_retry runs too many times

Author: MichaReiser

src/function/fetch.rs

@@ -58,11 +58,19 @@ where
         id: Id,
     ) -> &'db Memo<'db, C> {
         let memo_ingredient_index = self.memo_ingredient_index(zalsa, id);
+        let mut retry_count = 0;
         loop {
             if let Some(memo) = self
                 .fetch_hot(zalsa, id, memo_ingredient_index)
                 .or_else(|| {
-                    self.fetch_cold_with_retry(zalsa, zalsa_local, db, id, memo_ingredient_index)
+                    self.fetch_cold_with_retry(
+                        zalsa,
+                        zalsa_local,
+                        db,
+                        id,
+                        memo_ingredient_index,
+                        &mut retry_count,
+                    )
                 })
             {
                 return memo;
@@ -103,6 +111,7 @@ where
         db: &'db C::DbView,
         id: Id,
         memo_ingredient_index: MemoIngredientIndex,
+        retry_count: &mut u32,
     ) -> Option<&'db Memo<'db, C>> {
         let memo = self.fetch_cold(zalsa, zalsa_local, db, id, memo_ingredient_index)?;
 
@@ -114,7 +123,7 @@ where
         // That is only correct for fixpoint cycles, though: `FallbackImmediate` cycles
         // never have provisional entries.
         if C::CYCLE_STRATEGY == CycleRecoveryStrategy::FallbackImmediate
-            || !memo.provisional_retry(zalsa, zalsa_local, self.database_key_index(id))
+            || !memo.provisional_retry(zalsa, zalsa_local, self.database_key_index(id), retry_count)
         {
             Some(memo)
         } else {

src/function/memo.rs

@@ -144,13 +144,23 @@ impl<'db, C: Configuration> Memo<'db, C> {
         zalsa: &Zalsa,
         zalsa_local: &ZalsaLocal,
         database_key_index: DatabaseKeyIndex,
+        retry_count: &mut u32,
     ) -> bool {
         if self.block_on_heads(zalsa, zalsa_local) {
             // If we get here, we are a provisional value of
             // the cycle head (either initial value, or from a later iteration) and should be
             // returned to caller to allow fixpoint iteration to proceed.
             false
         } else {
+            assert!(
+                *retry_count <= 20000,
+                "Provisional memo retry limit exceeded for {database_key_index:?}; \
+                     this usually indicates a bug in salsa's cycle caching/locking. \
+                     (retried {retry_count} times)",
+            );
+
+            *retry_count += 1;
+
             // all our cycle heads are complete; re-fetch
             // and we should get a non-provisional memo.
             crate::tracing::debug!(

src/function/sync.rs

@@ -219,7 +219,7 @@ pub enum SyncOwnerId {
 /// Marks an active 'claim' in the synchronization map. The claim is
 /// released when this value is dropped.
 #[must_use]
-pub struct ClaimGuard<'me> {
+pub(crate) struct ClaimGuard<'me> {
     key_index: Id,
     zalsa: &'me Zalsa,
     sync_table: &'me SyncTable,

tests/parallel/cycle_nested_deep_panic.rs

@@ -5,6 +5,7 @@
 
 use crate::sync::thread;
 use crate::{Knobs, KnobsDatabase};
+use std::fmt;
 use std::panic::catch_unwind;
 
 use salsa::CycleRecoveryAction;
@@ -17,7 +18,6 @@ const MAX: CycleValue = CycleValue(3);
 
 #[salsa::tracked(cycle_fn=cycle_fn, cycle_initial=initial)]
 fn query_a(db: &dyn KnobsDatabase) -> CycleValue {
-    db.signal(1);
     query_b(db)
 }
 
@@ -27,16 +27,14 @@ fn query_b(db: &dyn KnobsDatabase) -> CycleValue {
     CycleValue(c_value.0 + 1).min(MAX)
 }
 
-#[salsa::tracked(cycle_fn=cycle_fn, cycle_initial=initial)]
+#[salsa::tracked]
 fn query_c(db: &dyn KnobsDatabase) -> CycleValue {
     let d_value = query_d(db);
 
     if d_value > CycleValue(0) {
-        let _e_value = query_e(db);
-        let _b = query_b(db);
-        db.wait_for(2);
-        db.signal(3);
-        panic!("Dragons are real");
+        let e_value = query_e(db);
+        let b_value = query_b(db);
+        CycleValue(d_value.0.max(e_value.0).max(b_value.0))
     } else {
         let a_value = query_a(db);
         CycleValue(d_value.0.max(a_value.0))
@@ -45,7 +43,7 @@ fn query_c(db: &dyn KnobsDatabase) -> CycleValue {
 
 #[salsa::tracked(cycle_fn=cycle_fn, cycle_initial=initial)]
 fn query_d(db: &dyn KnobsDatabase) -> CycleValue {
-    query_c(db)
+    query_b(db)
 }
 
 #[salsa::tracked(cycle_fn=cycle_fn, cycle_initial=initial)]
@@ -75,29 +73,62 @@ fn the_test() {
 
     let t1 = thread::spawn(move || {
         let _span = tracing::debug_span!("t1", thread_id = ?thread::current().id()).entered();
-
-        query_a(&db_t1)
+        catch_unwind(|| {
+            db_t1.wait_for(1);
+            query_a(&db_t1)
+        })
     });
     let t2 = thread::spawn(move || {
-        let _span = tracing::debug_span!("t4", thread_id = ?thread::current().id()).entered();
-        db_t4.wait_for(1);
-        db_t4.signal(2);
-        query_b(&db_t4)
+        let _span = tracing::debug_span!("t2", thread_id = ?thread::current().id()).entered();
+        catch_unwind(|| {
+            db_t2.wait_for(1);
+
+            query_b(&db_t2)
+        })
     });
     let t3 = thread::spawn(move || {
-        let _span = tracing::debug_span!("t2", thread_id = ?thread::current().id()).entered();
-        db_t2.wait_for(1);
-        query_d(&db_t2)
+        let _span = tracing::debug_span!("t3", thread_id = ?thread::current().id()).entered();
+        catch_unwind(|| {
+            db_t3.signal(2);
+            query_d(&db_t3)
+        })
     });
 
-    let r_t1 = t1.join();
-    let r_t2 = t2.join();
-    let r_t3 = t3.join();
+    let r_t1 = t1.join().unwrap();
+    let r_t2 = t2.join().unwrap();
+    let r_t3 = t3.join().unwrap();
 
-    assert!(r_t1.is_err());
-    assert!(r_t2.is_err());
-    assert!(r_t3.is_err());
+    assert_is_set_cycle_error(r_t1);
+    assert_is_set_cycle_error(r_t2);
+    assert_is_set_cycle_error(r_t3);
 
     // Pulling the cycle again at a later point should still result in a panic.
-    assert!(catch_unwind(|| query_d(&db_t3)).is_err());
+    assert_is_set_cycle_error(catch_unwind(|| query_d(&db_t4)));
+}
+
+#[track_caller]
+fn assert_is_set_cycle_error<T>(result: Result<T, Box<dyn std::any::Any + Send>>)
+where
+    T: fmt::Debug,
+{
+    let err = result.expect_err("expected an error");
+
+    if let Some(message) = err.downcast_ref::<&str>() {
+        assert!(
+            message.contains("set cycle_fn/cycle_initial to fixpoint iterate"),
+            "Expected error message to contain 'set cycle_fn/cycle_initial to fixpoint iterate', but got: {}",
+            message
+        );
+    } else if let Some(message) = err.downcast_ref::<String>() {
+        assert!(
+            message.contains("set cycle_fn/cycle_initial to fixpoint iterate"),
+            "Expected error message to contain 'set cycle_fn/cycle_initial to fixpoint iterate', but got: {}",
+            message
+        );
+    } else if let Some(_) = err.downcast_ref::<salsa::Cancelled>() {
+        // This is okay, because Salsa throws a Cancelled::PropagatedPanic when a panic occurs in a query
+        // that it blocks on.
+    } else {
+        std::panic::resume_unwind(err);
+    }
 }