Put IngredientIndex into SyncTable

Author: Veykril

src/function.rs

@@ -157,7 +157,7 @@ where
             lru: lru::Lru::new(lru),
             deleted_entries: Default::default(),
             view_caster,
-            sync_table: Default::default(),
+            sync_table: SyncTable::new(index),
         }
     }
 
@@ -258,10 +258,7 @@ where
     /// Attempts to claim `key_index`, returning `false` if a cycle occurs.
     fn wait_for(&self, db: &dyn Database, key_index: Id) -> bool {
         let zalsa = db.zalsa();
-        match self
-            .sync_table
-            .try_claim(db, zalsa, self.database_key_index(key_index))
-        {
+        match self.sync_table.try_claim(db, zalsa, key_index) {
             ClaimResult::Retry | ClaimResult::Claimed(_) => true,
             ClaimResult::Cycle => false,
         }

src/function/fetch.rs

@@ -92,12 +92,11 @@ where
         id: Id,
         memo_ingredient_index: MemoIngredientIndex,
     ) -> Option<&'db Memo<C::Output<'db>>> {
-        let database_key_index = self.database_key_index(id);
-
         // Try to claim this query: if someone else has claimed it already, go back and start again.
-        let _claim_guard = match self.sync_table.try_claim(db, zalsa, database_key_index) {
+        let _claim_guard = match self.sync_table.try_claim(db, zalsa, id) {
             ClaimResult::Retry => return None,
             ClaimResult::Cycle => {
+                let database_key_index = self.database_key_index(id);
                 // check if there's a provisional value for this query
                 // Note we don't `validate_may_be_provisional` the memo here as we want to reuse an
                 // existing provisional memo if it exists
@@ -123,7 +122,7 @@ where
                 }
                 // no provisional value; create/insert/return initial provisional value
                 return self
-                    .initial_value(db, database_key_index.key_index())
+                    .initial_value(db, id)
                     .map(|initial_value| {
                         tracing::debug!(
                             "hit cycle at {database_key_index:#?}, \
@@ -157,7 +156,7 @@ where
             ClaimResult::Claimed(guard) => guard,
         };
 
-        let mut active_query = LazyActiveQueryGuard::new(database_key_index);
+        let mut active_query = LazyActiveQueryGuard::new(self.database_key_index(id));
 
         // Now that we've claimed the item, check again to see if there's a "hot" value.
         let opt_old_memo = self.get_memo_from_table_for(zalsa, id, memo_ingredient_index);

src/function/maybe_changed_after.rs

@@ -103,7 +103,7 @@ where
     ) -> Option<VerifyResult> {
         let database_key_index = self.database_key_index(key_index);
 
-        let _claim_guard = match self.sync_table.try_claim(db, zalsa, database_key_index) {
+        let _claim_guard = match self.sync_table.try_claim(db, zalsa, key_index) {
             ClaimResult::Retry => return None,
             ClaimResult::Cycle => match C::CYCLE_STRATEGY {
                 CycleRecoveryStrategy::Panic => db.zalsa_local().with_query_stack(|stack| {

src/function/sync.rs

@@ -7,14 +7,14 @@ use crate::{
     key::DatabaseKeyIndex,
     runtime::{BlockResult, WaitResult},
     zalsa::Zalsa,
-    Database, Id,
+    Database, Id, IngredientIndex,
 };
 
 /// Tracks the keys that are currently being processed; used to coordinate between
 /// worker threads.
-#[derive(Default)]
 pub(crate) struct SyncTable {
     syncs: Mutex<FxHashMap<Id, SyncState>>,
+    ingredient: IngredientIndex,
 }
 
 pub(crate) enum ClaimResult<'a> {
@@ -32,14 +32,21 @@ struct SyncState {
 }
 
 impl SyncTable {
+    pub(crate) fn new(ingredient: IngredientIndex) -> Self {
+        Self {
+            syncs: Default::default(),
+            ingredient,
+        }
+    }
+
     pub(crate) fn try_claim<'me>(
         &'me self,
         db: &'me (impl ?Sized + Database),
         zalsa: &'me Zalsa,
-        database_key_index: DatabaseKeyIndex,
+        key_index: Id,
     ) -> ClaimResult<'me> {
         let mut write = self.syncs.lock();
-        match write.entry(database_key_index.key_index()) {
+        match write.entry(key_index) {
             std::collections::hash_map::Entry::Occupied(occupied_entry) => {
                 let &mut SyncState {
                     id,
@@ -52,7 +59,12 @@ impl SyncTable {
                 // boolean is to decide *whether* to acquire the lock,
                 // not to gate future atomic reads.
                 *anyone_waiting = true;
-                match zalsa.runtime().block_on(db, database_key_index, id, write) {
+                match zalsa.runtime().block_on(
+                    db,
+                    DatabaseKeyIndex::new(self.ingredient, key_index),
+                    id,
+                    write,
+                ) {
                     BlockResult::Completed => ClaimResult::Retry,
                     BlockResult::Cycle => ClaimResult::Cycle,
                 }
@@ -63,7 +75,7 @@ impl SyncTable {
                     anyone_waiting: false,
                 });
                 ClaimResult::Claimed(ClaimGuard {
-                    database_key_index,
+                    key_index,
                     zalsa,
                     sync_table: self,
                     _padding: false,
@@ -77,7 +89,7 @@ impl SyncTable {
 /// released when this value is dropped.
 #[must_use]
 pub(crate) struct ClaimGuard<'me> {
-    database_key_index: DatabaseKeyIndex,
+    key_index: Id,
     zalsa: &'me Zalsa,
     sync_table: &'me SyncTable,
     // Reduce the size of ClaimResult by making more niches available in ClaimGuard; this fits into
@@ -89,14 +101,13 @@ impl ClaimGuard<'_> {
     fn remove_from_map_and_unblock_queries(&self) {
         let mut syncs = self.sync_table.syncs.lock();
 
-        let SyncState { anyone_waiting, .. } =
-            syncs.remove(&self.database_key_index.key_index()).unwrap();
+        let SyncState { anyone_waiting, .. } = syncs.remove(&self.key_index).unwrap();
 
         drop(syncs);
 
         if anyone_waiting {
             self.zalsa.runtime().unblock_queries_blocked_on(
-                self.database_key_index,
+                DatabaseKeyIndex::new(self.sync_table.ingredient, self.key_index),
                 if std::thread::panicking() {
                     WaitResult::Panicked
                 } else {

src/runtime.rs

@@ -170,7 +170,7 @@ impl Runtime {
         other_id: ThreadId,
         query_mutex_guard: QueryMutexGuard,
     ) -> BlockResult {
-        let mut dg = self.dependency_graph.lock();
+        let dg = self.dependency_graph.lock();
         let thread_id = std::thread::current().id();
 
         if dg.depends_on(other_id, thread_id) {

src/runtime/dependency_graph.rs

@@ -31,7 +31,7 @@ impl DependencyGraph {
     /// True if `from_id` depends on `to_id`.
     ///
     /// (i.e., there is a path from `from_id` to `to_id` in the graph.)
-    pub(super) fn depends_on(&mut self, from_id: ThreadId, to_id: ThreadId) -> bool {
+    pub(super) fn depends_on(&self, from_id: ThreadId, to_id: ThreadId) -> bool {
         let mut p = from_id;
         while let Some(q) = self.edges.get(&p).map(|edge| edge.blocked_on_id) {
             if q == to_id {