Run nested cycles in a single fixpoint iteration

Fix serde attribute

Author: MichaReiser

src/active_query.rs

@@ -225,6 +225,7 @@ impl ActiveQuery {
             active_tracked_structs,
             mem::take(cycle_heads),
             iteration_count,
+            false,
         );
 
         let revisions = QueryRevisions {
@@ -498,7 +499,7 @@ impl fmt::Display for Backtrace {
             if full {
                 write!(fmt, " -> ({changed_at:?}, {durability:#?}")?;
                 if !cycle_heads.is_empty() || !iteration_count.is_initial() {
-                    write!(fmt, ", iteration = {iteration_count:?}")?;
+                    write!(fmt, ", iteration = {iteration_count}")?;
                 }
                 write!(fmt, ")")?;
             }
@@ -517,7 +518,7 @@ impl fmt::Display for Backtrace {
                         }
                         write!(
                             fmt,
-                            "{:?} -> {:?}",
+                            "{:?} -> {}",
                             head.database_key_index, head.iteration_count
                         )?;
                     }

src/cycle.rs

@@ -52,6 +52,7 @@
 use thin_vec::{thin_vec, ThinVec};
 
 use crate::key::DatabaseKeyIndex;
+use crate::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use crate::sync::OnceLock;
 use crate::Revision;
 
@@ -96,14 +97,26 @@ pub enum CycleRecoveryStrategy {
 /// would be the cycle head. It returns an "initial value" when the cycle is encountered (if
 /// fixpoint iteration is enabled for that query), and then is responsible for re-iterating the
 /// cycle until it converges.
-#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+#[derive(Debug)]
 #[cfg_attr(feature = "persistence", derive(serde::Serialize, serde::Deserialize))]
 pub struct CycleHead {
     pub(crate) database_key_index: DatabaseKeyIndex,
-    pub(crate) iteration_count: IterationCount,
+    pub(crate) iteration_count: AtomicIterationCount,
+    #[cfg_attr(feature = "persistence", serde(skip))]
+    removed: AtomicBool,
+}
+
+impl Clone for CycleHead {
+    fn clone(&self) -> Self {
+        Self {
+            database_key_index: self.database_key_index,
+            iteration_count: self.iteration_count.load().into(),
+            removed: self.removed.load(Ordering::Relaxed).into(),
+        }
+    }
 }
 
-#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Default)]
+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Default, PartialOrd, Ord)]
 #[cfg_attr(feature = "persistence", derive(serde::Serialize, serde::Deserialize))]
 #[cfg_attr(feature = "persistence", serde(transparent))]
 pub struct IterationCount(u8);
@@ -131,11 +144,65 @@ impl IterationCount {
     }
 }
 
+impl std::fmt::Display for IterationCount {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "iteration={}", self.0)
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct AtomicIterationCount(AtomicU8);
+
+impl AtomicIterationCount {
+    pub(crate) fn load(&self) -> IterationCount {
+        IterationCount(self.0.load(Ordering::Relaxed))
+    }
+
+    pub(crate) fn store(&self, value: IterationCount) {
+        self.0.store(value.0, Ordering::Release);
+    }
+
+    pub(crate) fn store_mut(&mut self, value: IterationCount) {
+        *self.0.get_mut() = value.0;
+    }
+}
+
+impl std::fmt::Display for AtomicIterationCount {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.load().fmt(f)
+    }
+}
+
+impl From<IterationCount> for AtomicIterationCount {
+    fn from(iteration_count: IterationCount) -> Self {
+        AtomicIterationCount(iteration_count.0.into())
+    }
+}
+
+#[cfg(feature = "persistence")]
+impl serde::Serialize for AtomicIterationCount {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        self.load().serialize(serializer)
+    }
+}
+
+#[cfg(feature = "persistence")]
+impl<'de> serde::Deserialize<'de> for AtomicIterationCount {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        IterationCount::deserialize(deserializer).map(Into::into)
+    }
+}
+
 /// Any provisional value generated by any query in a cycle will track the cycle head(s) (can be
 /// plural in case of nested cycles) representing the cycles it is part of, and the current
 /// iteration count for each cycle head. This struct tracks these cycle heads.
 #[derive(Clone, Debug, Default)]
-#[cfg_attr(feature = "persistence", derive(serde::Serialize, serde::Deserialize))]
 pub struct CycleHeads(ThinVec<CycleHead>);
 
 impl CycleHeads {
@@ -146,30 +213,34 @@ impl CycleHeads {
     pub(crate) fn initial(database_key_index: DatabaseKeyIndex) -> Self {
         Self(thin_vec![CycleHead {
             database_key_index,
-            iteration_count: IterationCount::initial(),
+            iteration_count: IterationCount::initial().into(),
+            removed: false.into()
         }])
     }
 
-    pub(crate) fn iter(&self) -> std::slice::Iter<'_, CycleHead> {
-        self.0.iter()
+    pub(crate) fn iter(&self) -> CycleHeadsIterator<'_> {
+        CycleHeadsIterator {
+            inner: self.0.iter(),
+        }
     }
 
     pub(crate) fn contains(&self, value: &DatabaseKeyIndex) -> bool {
         self.into_iter()
             .any(|head| head.database_key_index == *value)
     }
 
-    pub(crate) fn remove(&mut self, value: &DatabaseKeyIndex) -> bool {
-        let found = self
-            .0
-            .iter()
-            .position(|&head| head.database_key_index == *value);
-        let Some(found) = found else { return false };
-        self.0.swap_remove(found);
-        true
+    pub(crate) fn clear_except(&self, except: DatabaseKeyIndex) {
+        for head in self.0.iter() {
+            if head.database_key_index == except {
+                continue;
+            }
+
+            // TODO: verify ordering
+            head.removed.store(true, Ordering::Release);
+        }
     }
 
-    pub(crate) fn update_iteration_count(
+    pub(crate) fn update_iteration_count_mut(
         &mut self,
         cycle_head_index: DatabaseKeyIndex,
         new_iteration_count: IterationCount,
@@ -179,7 +250,21 @@ impl CycleHeads {
             .iter_mut()
             .find(|cycle_head| cycle_head.database_key_index == cycle_head_index)
         {
-            cycle_head.iteration_count = new_iteration_count;
+            cycle_head.iteration_count.store_mut(new_iteration_count);
+        }
+    }
+
+    pub(crate) fn update_iteration_count(
+        &self,
+        cycle_head_index: DatabaseKeyIndex,
+        new_iteration_count: IterationCount,
+    ) {
+        if let Some(cycle_head) = self
+            .0
+            .iter()
+            .find(|cycle_head| cycle_head.database_key_index == cycle_head_index)
+        {
+            cycle_head.iteration_count.store(new_iteration_count);
         }
     }
 
@@ -188,15 +273,41 @@ impl CycleHeads {
         self.0.reserve(other.0.len());
 
         for head in other {
-            if let Some(existing) = self
-                .0
-                .iter()
-                .find(|candidate| candidate.database_key_index == head.database_key_index)
-            {
-                assert_eq!(existing.iteration_count, head.iteration_count);
+            self.insert(head);
+        }
+    }
+
+    pub(crate) fn insert(&mut self, head: &CycleHead) -> bool {
+        if let Some(existing) = self
+            .0
+            .iter_mut()
+            .find(|candidate| candidate.database_key_index == head.database_key_index)
+        {
+            let removed = existing.removed.get_mut();
+
+            if *removed {
+                *removed = false;
+
+                true
             } else {
-                self.0.push(*head);
+                let existing_count = existing.iteration_count.load();
+                let head_count = head.iteration_count.load();
+
+                // It's now possible that a query can depend on different iteration counts of the same query
+                // This because some queries (inner) read the provisional value of the last iteration
+                // while outer queries read the value from the last iteration (which is i+1 if the head didn't converge).
+                assert_eq!(
+                    existing_count, head_count,
+                    "Can't merge cycle heads {:?} with different iteration counts ({existing_count:?}, {head_count:?})",
+                    existing.database_key_index
+                );
+
+                false
             }
+        } else {
+            debug_assert!(!head.removed.load(Ordering::Relaxed));
+            self.0.push(head.clone());
+            true
         }
     }
 
@@ -206,6 +317,37 @@ impl CycleHeads {
     }
 }
 
+#[cfg(feature = "persistence")]
+impl serde::Serialize for CycleHeads {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        use serde::ser::SerializeSeq;
+
+        let mut seq = serializer.serialize_seq(None)?;
+        for e in self {
+            if e.removed.load(Ordering::Relaxed) {
+                continue;
+            }
+
+            seq.serialize_element(e)?;
+        }
+        seq.end()
+    }
+}
+
+#[cfg(feature = "persistence")]
+impl<'de> serde::Deserialize<'de> for CycleHeads {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let vec: ThinVec<CycleHead> = serde::Deserialize::deserialize(deserializer)?;
+        Ok(CycleHeads(vec))
+    }
+}
+
 impl IntoIterator for CycleHeads {
     type Item = CycleHead;
     type IntoIter = <ThinVec<Self::Item> as IntoIterator>::IntoIter;
@@ -215,9 +357,29 @@ impl IntoIterator for CycleHeads {
     }
 }
 
+pub struct CycleHeadsIterator<'a> {
+    inner: std::slice::Iter<'a, CycleHead>,
+}
+
+impl<'a> Iterator for CycleHeadsIterator<'a> {
+    type Item = &'a CycleHead;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        loop {
+            let next = self.inner.next()?;
+
+            if next.removed.load(Ordering::Relaxed) {
+                continue;
+            }
+
+            return Some(next);
+        }
+    }
+}
+
 impl<'a> std::iter::IntoIterator for &'a CycleHeads {
     type Item = &'a CycleHead;
-    type IntoIter = std::slice::Iter<'a, CycleHead>;
+    type IntoIter = CycleHeadsIterator<'a>;
 
     fn into_iter(self) -> Self::IntoIter {
         self.iter()
@@ -241,28 +403,22 @@ pub enum ProvisionalStatus {
     Provisional {
         iteration: IterationCount,
         verified_at: Revision,
+        nested: bool,
     },
     Final {
         iteration: IterationCount,
         verified_at: Revision,
+        nested: bool,
     },
     FallbackImmediate,
 }
 
 impl ProvisionalStatus {
-    pub(crate) const fn iteration(&self) -> Option<IterationCount> {
-        match self {
-            ProvisionalStatus::Provisional { iteration, .. } => Some(*iteration),
-            ProvisionalStatus::Final { iteration, .. } => Some(*iteration),
-            ProvisionalStatus::FallbackImmediate => None,
-        }
-    }
-
-    pub(crate) const fn verified_at(&self) -> Option<Revision> {
+    pub(crate) fn nested(&self) -> bool {
         match self {
-            ProvisionalStatus::Provisional { verified_at, .. } => Some(*verified_at),
-            ProvisionalStatus::Final { verified_at, .. } => Some(*verified_at),
-            ProvisionalStatus::FallbackImmediate => None,
+            ProvisionalStatus::Provisional { nested, .. } => *nested,
+            ProvisionalStatus::Final { nested, .. } => *nested,
+            ProvisionalStatus::FallbackImmediate => false,
         }
     }
 }