Nits

Author: MichaReiser

src/cycle.rs

@@ -130,6 +130,10 @@ impl IterationCount {
         self.0 == 0
     }
 
+    pub(crate) const fn panicked() -> Self {
+        Self(u8::MAX)
+    }
+
     pub(crate) const fn increment(self) -> Option<Self> {
         let next = Self(self.0 + 1);
         if next.0 <= MAX_ITERATIONS.0 {
@@ -226,7 +230,7 @@ impl CycleHeads {
 
     pub(crate) fn contains(&self, value: &DatabaseKeyIndex) -> bool {
         self.into_iter()
-            .any(|head| head.database_key_index == *value)
+            .any(|head| head.database_key_index == *value && !head.removed.load(Ordering::Relaxed))
     }
 
     pub(crate) fn clear_except(&self, except: DatabaseKeyIndex) {

src/function/execute.rs

@@ -187,12 +187,27 @@ where
 
             let mut cycle_heads = std::mem::take(cycle_heads);
 
+            // Recursively resolve all cycle heads that this head depends on.
+            // This isn't required in a single-threaded execution but it's not guaranteed that all nested cycles are listed
+            // in cycle heads in a multi-threaded execution:
+            //
+            // t1: a -> b
+            // t2: c -> b (blocks on t1)
+            // t1: a -> b -> c (cycle, returns fixpoint initial with c(0) in heads)
+            // t1: a -> b (completes b, b has c(0) in its cycle heads, releases `b`, which resumes `t2`, and `retry_provisional` blocks on `c` (t2))
+            // t2: c -> a (cycle, returns fixpoint initial for a with a(0) in heads)
+            // t2: completes c, `provisional_retry` blocks on `a` (t2)
+            // t1: a (complets `b` with `c` in heads)
+            //
+            // Note how `a` only depends on `c` but not `a`. This is because `a` only saw the initial value of `c` and wasn't updated when `c` completed.
+            // That's why we need to resolve the cycle heads recursively to `cycle_heads` contains all cycle heads at the moment this query completed.
             let mut queue: SmallVec<[DatabaseKeyIndex; 4]> = cycle_heads
                 .iter()
                 .map(|head| head.database_key_index)
                 .filter(|head| *head != database_key_index)
                 .collect();
 
+            // TODO: Can we also resolve whether the cycles have converged here?
             while let Some(head) = queue.pop() {
                 let ingredient = zalsa.lookup_ingredient(head.ingredient_index());
                 let nested_heads = ingredient.cycle_heads(zalsa, head.key_index());
@@ -230,10 +245,9 @@ where
                 memo.value.as_ref()
             };
 
-
             let last_provisional_value = last_provisional_value.expect(
-                    "`fetch_cold_cycle` should have inserted a provisional memo with Cycle::initial",
-                );
+                "`fetch_cold_cycle` should have inserted a provisional memo with Cycle::initial",
+            );
             crate::tracing::debug!(
                 "{database_key_index:?}: execute: \
                         I am a cycle head, comparing last provisional value with new value"
@@ -487,6 +501,10 @@ impl<C: Configuration> Drop for ClearCycleHeadIfPanicking<'_, C> {
         if std::thread::panicking() {
             let revisions =
                 QueryRevisions::fixpoint_initial(self.ingredient.database_key_index(self.id));
+            revisions.update_iteration_count_mut(
+                self.ingredient.database_key_index(self.id),
+                IterationCount::panicked(),
+            );
 
             let memo = Memo::new(None, self.zalsa.current_revision(), revisions);
             self.ingredient

src/function/fetch.rs

@@ -232,7 +232,7 @@ where
             }
         }
 
-        let memo = self.execute(db,zalsa, zalsa_local, database_key_index, opt_old_memo);
+        let memo = self.execute(db, zalsa, zalsa_local, database_key_index, opt_old_memo);
 
         Some(memo)
     }

src/function/maybe_changed_after.rs

@@ -446,6 +446,7 @@ where
     /// If this is a provisional memo, validate that it was cached in the same iteration of the
     /// same cycle(s) that we are still executing. If so, it is valid for reuse. This avoids
     /// runaway re-execution of the same queries within a fixpoint iteration.
+    #[inline]
     fn validate_same_iteration(
         &self,
         zalsa: &Zalsa,

src/zalsa_local.rs

@@ -568,7 +568,10 @@ struct QueryRevisionsExtraInner {
 
     cycle_converged: bool,
 
-    #[cfg_attr(feature = "persistence", serde(with = "crate::zalsa_local::persistence::atomic_bool"))]
+    #[cfg_attr(
+        feature = "persistence",
+        serde(with = "crate::zalsa_local::persistence::atomic_bool")
+    )]
     nested_cycle: AtomicBool,
 }
 
@@ -1265,7 +1268,7 @@ pub(crate) mod persistence {
             }
         }
     }
-    
+
     // A workaround the fact that `shuttle` atomic types do not implement `serde::{Serialize, Deserialize}`.
     pub(super) mod verified_final {
         use crate::sync::atomic::{AtomicBool, Ordering};
@@ -1302,5 +1305,4 @@ pub(crate) mod persistence {
             serde::Deserialize::deserialize(deserializer).map(AtomicBool::new)
         }
     }
-
 }

tests/parallel/cycle_nested_deep_conditional_changed.rs

@@ -108,34 +108,31 @@ fn the_test() {
         }
 
         let t1 = thread::spawn(move || {
-            let _span = tracing::debug_span!("t1", thread_id = ?thread::current().id()).entered();
+            let _span = tracing::info_span!("t1", thread_id = ?thread::current().id()).entered();
             let (db, input) = get_db(|db, input| {
                 query_a(db, input);
             });
 
-            let _span = tracing::debug_span!("t1", thread_id = ?thread::current().id()).entered();
-
             query_a(&db, input)
         });
         let t2 = thread::spawn(move || {
-            let _span = tracing::debug_span!("t2", thread_id = ?thread::current().id()).entered();
+            let _span = tracing::info_span!("t2", thread_id = ?thread::current().id()).entered();
             let (db, input) = get_db(|db, input| {
                 query_b(db, input);
             });
 
-            let _span = tracing::debug_span!("t4", thread_id = ?thread::current().id()).entered();
             query_b(&db, input)
         });
         let t3 = thread::spawn(move || {
-            let _span = tracing::debug_span!("t3", thread_id = ?thread::current().id()).entered();
+            let _span = tracing::info_span!("t3", thread_id = ?thread::current().id()).entered();
             let (db, input) = get_db(|db, input| {
                 query_d(db, input);
             });
 
             query_d(&db, input)
         });
         let t4 = thread::spawn(move || {
-            let _span = tracing::debug_span!("t4", thread_id = ?thread::current().id()).entered();
+            let _span = tracing::info_span!("t4", thread_id = ?thread::current().id()).entered();
 
             let (db, input) = get_db(|db, input| {
                 query_e(db, input);