Per ingredient sync table

Author: Veykril

src/function.rs

@@ -3,6 +3,7 @@ use std::{any::Any, fmt, sync::Arc};
 use crate::{
     accumulator::accumulated_map::AccumulatedMap,
     cycle::CycleRecoveryStrategy,
+    function::sync::SyncTable,
     ingredient::fmt_index,
     key::DatabaseKeyIndex,
     plumbing::JarAux,
@@ -27,6 +28,7 @@ mod lru;
 mod maybe_changed_after;
 mod memo;
 mod specify;
+mod sync;
 
 pub trait Configuration: Any {
     const DEBUG_NAME: &'static str;
@@ -100,6 +102,8 @@ pub struct IngredientImpl<C: Configuration> {
     /// Used to find memos to throw out when we have too many memoized values.
     lru: lru::Lru,
 
+    sync_table: SyncTable,
+
     /// When `fetch` and friends executes, they return a reference to the
     /// value stored in the memo that is extended to live as long as the `&self`
     /// reference we start with. This means that whenever we remove something
@@ -132,6 +136,7 @@ where
             memo_ingredient_index: aux.next_memo_ingredient_index(struct_index, index),
             lru: Default::default(),
             deleted_entries: Default::default(),
+            sync_table: Default::default(),
         }
     }
 

src/function/fetch.rs

@@ -65,18 +65,18 @@ where
         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 = zalsa.sync_table_for(id).claim(
+        let _claim_guard = self.sync_table.try_claim(
             db.as_dyn_database(),
             zalsa_local,
+            zalsa,
             database_key_index,
-            self.memo_ingredient_index,
+            id,
         )?;
 
         // Push the query on the stack.
         let active_query = zalsa_local.push_query(database_key_index);
 
         // Now that we've claimed the item, check again to see if there's a "hot" value.
-        let zalsa = db.zalsa();
         let opt_old_memo = self.get_memo_from_table_for(zalsa, id);
         if let Some(old_memo) = &opt_old_memo {
             if old_memo.value.is_some() && self.deep_verify_memo(db, old_memo, &active_query) {

src/function/maybe_changed_after.rs

@@ -53,11 +53,12 @@ where
         let (zalsa, zalsa_local) = db.zalsas();
         let database_key_index = self.database_key_index(key_index);
 
-        let _claim_guard = zalsa.sync_table_for(key_index).claim(
+        let _claim_guard = self.sync_table.try_claim(
             db.as_dyn_database(),
             zalsa_local,
+            zalsa,
             database_key_index,
-            self.memo_ingredient_index,
+            key_index,
         )?;
         let active_query = zalsa_local.push_query(database_key_index);
 

src/table/sync.rs renamed to src/function/sync.rs

@@ -3,23 +3,18 @@ use std::{
     thread::ThreadId,
 };
 
-use parking_lot::RwLock;
+use parking_lot::Mutex;
+use rustc_hash::FxHashMap;
 
 use crate::{
-    key::DatabaseKeyIndex,
-    runtime::WaitResult,
-    zalsa::{MemoIngredientIndex, Zalsa},
-    zalsa_local::ZalsaLocal,
-    Database,
+    key::DatabaseKeyIndex, runtime::WaitResult, zalsa::Zalsa, zalsa_local::ZalsaLocal, Database, Id,
 };
 
-use super::util;
-
 /// Tracks the keys that are currently being processed; used to coordinate between
 /// worker threads.
 #[derive(Default)]
 pub(crate) struct SyncTable {
-    syncs: RwLock<Vec<Option<SyncState>>>,
+    syncs: Mutex<FxHashMap<Id, SyncState>>,
 }
 
 struct SyncState {
@@ -31,46 +26,43 @@ struct SyncState {
 }
 
 impl SyncTable {
-    pub(crate) fn claim<'me>(
+    pub(crate) fn try_claim<'me>(
         &'me self,
         db: &'me dyn Database,
         zalsa_local: &ZalsaLocal,
+        zalsa: &'me Zalsa,
         database_key_index: DatabaseKeyIndex,
-        memo_ingredient_index: MemoIngredientIndex,
+        id: Id,
     ) -> Option<ClaimGuard<'me>> {
-        let mut syncs = self.syncs.write();
-        let zalsa = db.zalsa();
-        let thread_id = std::thread::current().id();
-
-        util::ensure_vec_len(&mut syncs, memo_ingredient_index.as_usize() + 1);
-
-        match &syncs[memo_ingredient_index.as_usize()] {
-            None => {
-                syncs[memo_ingredient_index.as_usize()] = Some(SyncState {
-                    id: thread_id,
-                    anyone_waiting: AtomicBool::new(false),
-                });
-                Some(ClaimGuard {
-                    database_key_index,
-                    memo_ingredient_index,
-                    zalsa,
-                    sync_table: self,
-                })
-            }
-            Some(SyncState {
-                id: other_id,
-                anyone_waiting,
-            }) => {
+        let mut write = self.syncs.lock();
+        match write.entry(id) {
+            std::collections::hash_map::Entry::Occupied(occupied_entry) => {
+                let &mut SyncState {
+                    id,
+                    ref anyone_waiting,
+                } = occupied_entry.into_mut();
                 // NB: `Ordering::Relaxed` is sufficient here,
                 // as there are no loads that are "gated" on this
                 // value. Everything that is written is also protected
                 // by a lock that must be acquired. The role of this
                 // boolean is to decide *whether* to acquire the lock,
                 // not to gate future atomic reads.
                 anyone_waiting.store(true, Ordering::Relaxed);
-                zalsa.block_on_or_unwind(db, zalsa_local, database_key_index, *other_id, syncs);
+                zalsa.block_on_or_unwind(db, zalsa_local, database_key_index, id, write);
                 None
             }
+            std::collections::hash_map::Entry::Vacant(vacant_entry) => {
+                vacant_entry.insert(SyncState {
+                    id: std::thread::current().id(),
+                    anyone_waiting: AtomicBool::new(false),
+                });
+                Some(ClaimGuard {
+                    database_key_index,
+                    id,
+                    zalsa,
+                    sync_table: self,
+                })
+            }
         }
     }
 }
@@ -80,17 +72,18 @@ impl SyncTable {
 #[must_use]
 pub(crate) struct ClaimGuard<'me> {
     database_key_index: DatabaseKeyIndex,
-    memo_ingredient_index: MemoIngredientIndex,
+    id: Id,
     zalsa: &'me Zalsa,
     sync_table: &'me SyncTable,
 }
 
 impl ClaimGuard<'_> {
     fn remove_from_map_and_unblock_queries(&self, wait_result: WaitResult) {
-        let mut syncs = self.sync_table.syncs.write();
+        let mut syncs = self.sync_table.syncs.lock();
+
+        let SyncState { anyone_waiting, .. } = syncs.remove(&self.id).unwrap();
 
-        let SyncState { anyone_waiting, .. } =
-            syncs[self.memo_ingredient_index.as_usize()].take().unwrap();
+        drop(syncs);
 
         // NB: `Ordering::Relaxed` is sufficient here,
         // see `store` above for explanation.

src/input.rs

@@ -18,7 +18,7 @@ use crate::{
     ingredient::{fmt_index, Ingredient},
     key::{DatabaseKeyIndex, InputDependencyIndex},
     plumbing::{Jar, JarAux, Stamp},
-    table::{memo::MemoTable, sync::SyncTable, Slot, Table},
+    table::{memo::MemoTable, Slot, Table},
     zalsa::{IngredientIndex, Zalsa},
     zalsa_local::QueryOrigin,
     Database, Durability, Id, Revision, Runtime,
@@ -121,7 +121,6 @@ impl<C: Configuration> IngredientImpl<C> {
             fields,
             stamps,
             memos: Default::default(),
-            syncs: Default::default(),
         });
 
         if C::IS_SINGLETON {
@@ -299,9 +298,6 @@ where
 
     /// Memos
     memos: MemoTable,
-
-    /// Syncs
-    syncs: SyncTable,
 }
 
 pub trait HasBuilder {
@@ -315,8 +311,4 @@ where
     unsafe fn memos(&self, _current_revision: Revision) -> &crate::table::memo::MemoTable {
         &self.memos
     }
-
-    unsafe fn syncs(&self, _current_revision: Revision) -> &SyncTable {
-        &self.syncs
-    }
 }

src/interned.rs

@@ -5,7 +5,6 @@ use crate::ingredient::fmt_index;
 use crate::key::InputDependencyIndex;
 use crate::plumbing::{Jar, JarAux};
 use crate::table::memo::MemoTable;
-use crate::table::sync::SyncTable;
 use crate::table::Slot;
 use crate::zalsa::IngredientIndex;
 use crate::zalsa_local::QueryOrigin;
@@ -75,7 +74,6 @@ where
 {
     data: C::Data<'static>,
     memos: MemoTable,
-    syncs: SyncTable,
 }
 
 impl<C: Configuration> Default for JarImpl<C> {
@@ -183,7 +181,6 @@ where
                 let next_id = zalsa_local.allocate(table, self.ingredient_index, || Value::<C> {
                     data: internal_data,
                     memos: Default::default(),
-                    syncs: Default::default(),
                 });
                 entry.insert(next_id);
                 C::struct_from_id(next_id)
@@ -302,10 +299,6 @@ where
     unsafe fn memos(&self, _current_revision: Revision) -> &MemoTable {
         &self.memos
     }
-
-    unsafe fn syncs(&self, _current_revision: Revision) -> &crate::table::sync::SyncTable {
-        &self.syncs
-    }
 }
 
 /// The `Lookup` trait is a more flexible variant on [`std::borrow::Borrow`]

src/table.rs

@@ -10,13 +10,10 @@ use std::{
 use append_only_vec::AppendOnlyVec;
 use memo::MemoTable;
 use parking_lot::Mutex;
-use sync::SyncTable;
 
 use crate::{zalsa::transmute_data_ptr, Id, IngredientIndex, Revision};
 
 pub(crate) mod memo;
-pub(crate) mod sync;
-mod util;
 
 const PAGE_LEN_BITS: usize = 10;
 const PAGE_LEN_MASK: usize = PAGE_LEN - 1;
@@ -36,13 +33,6 @@ pub(crate) trait TablePage: Any + Send + Sync {
     ///
     /// The `current_revision` MUST be the current revision of the database owning this table page.
     unsafe fn memos(&self, slot: SlotIndex, current_revision: Revision) -> &MemoTable;
-
-    /// Access the syncs attached to `slot`.
-    ///
-    /// # Safety condition
-    ///
-    /// The `current_revision` MUST be the current revision of the database owning this table page.
-    unsafe fn syncs(&self, slot: SlotIndex, current_revision: Revision) -> &SyncTable;
 }
 
 pub(crate) struct Page<T: Slot> {
@@ -74,13 +64,6 @@ pub(crate) trait Slot: Any + Send + Sync {
     ///
     /// The current revision MUST be the current revision of the database containing this slot.
     unsafe fn memos(&self, current_revision: Revision) -> &MemoTable;
-
-    /// Access the [`SyncTable`][] for this slot.
-    ///
-    /// # Safety condition
-    ///
-    /// The current revision MUST be the current revision of the database containing this slot.
-    unsafe fn syncs(&self, current_revision: Revision) -> &SyncTable;
 }
 
 unsafe impl<T: Slot> Send for Page<T> {}
@@ -169,17 +152,6 @@ impl Table {
         let (page, slot) = split_id(id);
         self.pages[page.0].memos(slot, current_revision)
     }
-
-    /// Get the sync table associated with `id`
-    ///
-    /// # Safety condition
-    ///
-    /// The parameter `current_revision` MUST be the current revision
-    /// of the owner of database owning this table.
-    pub unsafe fn syncs(&self, id: Id, current_revision: Revision) -> &SyncTable {
-        let (page, slot) = split_id(id);
-        self.pages[page.0].syncs(slot, current_revision)
-    }
 }
 
 impl<T: Slot> Page<T> {
@@ -256,10 +228,6 @@ impl<T: Slot> TablePage for Page<T> {
     unsafe fn memos(&self, slot: SlotIndex, current_revision: Revision) -> &MemoTable {
         self.get(slot).memos(current_revision)
     }
-
-    unsafe fn syncs(&self, slot: SlotIndex, current_revision: Revision) -> &SyncTable {
-        self.get(slot).syncs(current_revision)
-    }
 }
 
 impl<T: Slot> Drop for Page<T> {

src/table/util.rs

@@ -11,7 +11,7 @@ use crate::{
     plumbing::ZalsaLocal,
     runtime::StampedValue,
     salsa_struct::SalsaStructInDb,
-    table::{memo::MemoTable, sync::SyncTable, Slot, Table},
+    table::{memo::MemoTable, Slot, Table},
     zalsa::{IngredientIndex, Zalsa},
     zalsa_local::QueryOrigin,
     Database, Durability, Event, EventKind, Id, Revision,
@@ -225,9 +225,6 @@ where
 
     /// Memo table storing the results of query functions etc.
     memos: MemoTable,
-
-    /// Sync table storing the results of query functions etc.
-    syncs: SyncTable,
 }
 // ANCHOR_END: ValueStruct
 
@@ -326,7 +323,6 @@ where
             fields: unsafe { self.to_static(fields) },
             revisions: C::new_revisions(current_deps.changed_at),
             memos: Default::default(),
-            syncs: Default::default(),
         };
 
         if let Some(id) = self.free_list.pop() {
@@ -695,12 +691,4 @@ where
         self.read_lock(current_revision);
         &self.memos
     }
-
-    unsafe fn syncs(&self, current_revision: Revision) -> &crate::table::sync::SyncTable {
-        // Acquiring the read lock here with the current revision
-        // ensures that there is no danger of a race
-        // when deleting a tracked struct.
-        self.read_lock(current_revision);
-        &self.syncs
-    }
 }