diff --git a/src/index/arctic.rs b/src/index/arctic.rs index 3f124d1..108dc59 100644 --- a/src/index/arctic.rs +++ b/src/index/arctic.rs @@ -6,6 +6,7 @@ use std::ops::{Bound, RangeBounds}; use std::sync::atomic::{AtomicUsize, Ordering}; use arctic::{ConcurrentMap, Key, Order}; +use parking_lot::RwLock; use super::UniqueIndex; @@ -71,6 +72,10 @@ impl_arctic_key!(u16, u32, u64, u128); /// satisfy WorkTable's double-ended query interface. pub struct ArcticIndex { inner: ConcurrentMap>, + // arctic-wt 0.1.4 can trip its raw-cursor structural assertion when + // mutations overlap and can expose a transient structural rewrite to a + // point read. Reads share access; structural mutations take it exclusively. + access: RwLock<()>, len: AtomicUsize, } @@ -89,6 +94,7 @@ where fn default() -> Self { Self { inner: ConcurrentMap::default(), + access: RwLock::new(()), len: AtomicUsize::new(0), } } @@ -115,6 +121,7 @@ where let len = inner.all().entries(Order::Ascend).count(); Ok(Self { inner, + access: RwLock::new(()), len: AtomicUsize::new(len), }) } @@ -127,23 +134,27 @@ where { #[inline] fn get_value(&self, key: &K) -> Option { - self.with_value(key, Clone::clone) + let _access = self.access.read(); + let key = key.to_arctic(); + self.inner.get(key.borrow()).map(|value| value.clone()) } #[inline] fn with_value(&self, key: &K, read: impl FnOnce(&V) -> R) -> Option { - let key = key.to_arctic(); - self.inner.get(key.borrow()).map(|value| read(&value)) + let value = self.get_value(key)?; + Some(read(&value)) } #[inline] fn contains_key(&self, key: &K) -> bool { + let _access = self.access.read(); let key = key.to_arctic(); self.inner.get(key.borrow()).is_some() } #[inline] fn insert_value(&self, key: K, value: V) -> Option { + let _access = self.access.write(); let key = key.to_arctic(); let updated = self.inner.upsert(key.as_insert(), Box::new(value)); let old = updated.old().cloned(); @@ -155,6 +166,7 @@ where #[inline] fn insert_value_checked(&self, key: K, value: V) -> Option<()> { + let _access = self.access.write(); let key = key.to_arctic(); match self.inner.insert(key.as_insert(), Box::new(value)) { Ok(_) => { @@ -170,6 +182,7 @@ where #[inline] fn remove_value(&self, key: &K) -> Option<(K, V)> { + let _access = self.access.write(); let raw_key = key.to_arctic(); let old = self.inner.remove(raw_key.borrow())?; self.len.fetch_sub(1, Ordering::Relaxed); @@ -182,6 +195,7 @@ where } fn iter_values(&self) -> impl DoubleEndedIterator + '_ { + let _access = self.access.read(); let shard = self.inner.all(); shard .entries(Order::Ascend) @@ -198,6 +212,7 @@ where where R: RangeBounds + 'a, { + let _access = self.access.read(); let lower = match range.start_bound() { Bound::Included(key) => Some(key.to_arctic()), Bound::Excluded(key) => key.to_arctic().next(), diff --git a/src/index/congee.rs b/src/index/congee.rs index 06e8a4d..ef94ac4 100644 --- a/src/index/congee.rs +++ b/src/index/congee.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use congee::{CongeeRaw, DefaultAllocator}; -use parking_lot::Mutex; +use parking_lot::RwLock; use super::UniqueIndex; @@ -48,10 +48,10 @@ impl_congee_key!(u64); /// reclamation pattern used by Congee's own `CongeeArc` implementation. pub struct CongeeIndex { inner: CongeeRaw, - // congee-wt 0.4.1 can lose disjoint insert/remove mutations when their - // structural updates overlap. Keep point reads native and concurrent, but - // serialize mutations until the backend offers the required visibility. - mutation: Mutex<()>, + // congee-wt 0.4.1 can lose disjoint structural mutations and let point + // reads miss while another mutation rewrites the tree. Reads remain + // concurrent with reads; structural mutations take exclusive access. + access: RwLock<()>, len: AtomicUsize, marker: std::marker::PhantomData<(K, V)>, } @@ -77,7 +77,7 @@ where }; Self { inner: CongeeRaw::new_with_drainer(DefaultAllocator {}, drainer), - mutation: Mutex::new(()), + access: RwLock::new(()), len: AtomicUsize::new(0), marker: std::marker::PhantomData, } @@ -116,6 +116,7 @@ where where R: RangeBounds, { + let _access = self.access.read(); let start = match range.start_bound() { Bound::Included(key) => key.into_congee(), Bound::Excluded(key) => match key.into_congee().checked_add(1) { @@ -196,7 +197,7 @@ where let len = inner.keys().len(); Ok(Self { inner, - mutation: Mutex::new(()), + access: RwLock::new(()), len: AtomicUsize::new(len), marker: std::marker::PhantomData, }) @@ -210,28 +211,30 @@ where { #[inline] fn get_value(&self, key: &K) -> Option { - self.with_value(key, Clone::clone) + let _access = self.access.read(); + let guard = self.inner.pin(); + let pointer = self.inner.get(&key.into_congee(), &guard)?; + // SAFETY: the read lock prevents removal and the epoch guard retains + // the tree-owned allocation while its value is cloned. + Some(unsafe { &*std::ptr::with_exposed_provenance::(pointer) }.clone()) } #[inline] fn with_value(&self, key: &K, read: impl FnOnce(&V) -> R) -> Option { - let guard = self.inner.pin(); - let pointer = self.inner.get(&key.into_congee(), &guard)?; - // SAFETY: the epoch guard keeps the tree-owned `Arc` alive for the - // duration of `read`, and the pointer originated from `Arc::into_raw`. - let value = unsafe { &*std::ptr::with_exposed_provenance::(pointer) }; - Some(read(value)) + let value = self.get_value(key)?; + Some(read(&value)) } #[inline] fn contains_key(&self, key: &K) -> bool { + let _access = self.access.read(); let guard = self.inner.pin(); self.inner.get(&key.into_congee(), &guard).is_some() } #[inline] fn insert_value(&self, key: K, value: V) -> Option { - let _mutation = self.mutation.lock(); + let _access = self.access.write(); let guard = self.inner.pin(); let pointer = Arc::into_raw(Arc::new(value)).expose_provenance(); match self.inner.insert(key.into_congee(), pointer, &guard) { @@ -250,7 +253,7 @@ where #[inline] fn insert_value_checked(&self, key: K, value: V) -> Option<()> { - let _mutation = self.mutation.lock(); + let _access = self.access.write(); let guard = self.inner.pin(); let pointer = Arc::into_raw(Arc::new(value)).expose_provenance(); let result = self @@ -278,7 +281,7 @@ where #[inline] fn remove_value(&self, key: &K) -> Option<(K, V)> { - let _mutation = self.mutation.lock(); + let _access = self.access.write(); let guard = self.inner.pin(); let pointer = self.inner.remove(&key.into_congee(), &guard)?; self.len.fetch_sub(1, Ordering::Relaxed); diff --git a/src/index/unique.rs b/src/index/unique.rs index 55e559f..d89716f 100644 --- a/src/index/unique.rs +++ b/src/index/unique.rs @@ -206,7 +206,7 @@ pub type UpstreamIndexPair = VanillaPair; #[cfg(test)] mod tests { - use std::sync::Arc; + use std::sync::{Arc, Barrier}; use super::{UniqueIndex, UpstreamIndexMap}; use crate::{ArcticIndex, CongeeIndex, IndexMap}; @@ -317,6 +317,32 @@ mod tests { assert!(index.is_empty()); } + fn assert_separate_instances_do_not_interfere() + where + I: UniqueIndex + Send + Sync + 'static, + { + let barrier = Arc::new(Barrier::new(9)); + let mut threads = Vec::new(); + for worker in 0..8_u64 { + let barrier = Arc::clone(&barrier); + threads.push(std::thread::spawn(move || { + let index = I::default(); + barrier.wait(); + for sequence in 0..10_000_u64 { + let key = worker * 10_000 + sequence; + assert_eq!(index.insert_value_checked(key, key + 1), Some(())); + assert_eq!(index.get_value(&key), Some(key + 1)); + assert_eq!(index.remove_value(&key), Some((key, key + 1))); + } + assert!(index.is_empty()); + })); + } + barrier.wait(); + for thread in threads { + thread.join().unwrap(); + } + } + #[test] fn worktables_index_implements_contract() { assert_unique_index_contract::>(); @@ -335,9 +361,29 @@ mod tests { assert_disjoint_concurrent_insert_then_remove::>(); } + #[test] + fn worktables_index_preserves_disjoint_concurrent_mutations() { + assert_disjoint_concurrent_insert_then_remove::>(); + } + + #[test] + fn upstream_indexset_preserves_disjoint_concurrent_mutations() { + assert_disjoint_concurrent_insert_then_remove::>(); + } + #[test] fn art_backends_make_disjoint_mutations_immediately_visible() { assert_immediate_disjoint_crud::>(); assert_immediate_disjoint_crud::>(); } + + #[test] + fn congee_instances_do_not_share_mutation_state() { + assert_separate_instances_do_not_interfere::>(); + } + + #[test] + fn arctic_instances_do_not_share_mutation_state() { + assert_separate_instances_do_not_interfere::>(); + } }