remove locks from AtomTable
there are technically still two locks - the Mutex to serialize consurrent updates to the AtomTable - the RwLock as part of GLOBAL_ATOM_TABLE the former is the age lock of RESIZE ATOM TABLE in <https://arxiv.org/pdf/1608.00989.pdf>, though we use it for the whole update as we don't match the whole structure and can't reserve an atom slot as is, so we need to also lock concurrent updates without resizing the later should be uncontendet as we only write AtomTable::new iff the current value is a dangling Weak
This commit is contained in:
Bennet Bleßmann
213
src/rcu.rs
Normal file
213
src/rcu.rs
Normal file
@@ -0,0 +1,213 @@
|
||||
use std::{
|
||||
cell::OnceCell,
|
||||
fmt::Debug,
|
||||
mem::ManuallyDrop,
|
||||
ops::Deref,
|
||||
ptr::NonNull,
|
||||
sync::{
|
||||
atomic::{AtomicPtr, AtomicU8},
|
||||
Arc, Weak,
|
||||
},
|
||||
};
|
||||
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
// the epoch counters of all threads that have ever accessed an Rcu
|
||||
// threads that have finished will have a dangling Weak reference and can be cleand up
|
||||
// having this be shared between all Rcu's is a tradeof,
|
||||
// writes will be slower as more epoch counters need to be waited for
|
||||
// reads should be faster as a thread only needs to register itself once on the first read
|
||||
//
|
||||
static EPOCH_COUNTERS: RwLock<Vec<Weak<AtomicU8>>> = RwLock::const_new(Vec::new());
|
||||
|
||||
thread_local! {
|
||||
// odd value means the current thread is about to access the active_epoch of an Rcu
|
||||
// a thread has a single epoch counter for all Rcu it accesses,
|
||||
// as a thread can only access one Rcu at a time
|
||||
static THREAD_EPOCH_COUNTER: OnceCell<Arc<AtomicU8>> = OnceCell::new();
|
||||
}
|
||||
|
||||
pub struct Rcu<T> {
|
||||
active_value: AtomicPtr<T>,
|
||||
}
|
||||
|
||||
impl<T: std::fmt::Debug> std::fmt::Debug for Rcu<T> {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
let active_epoch = self.active_epoch();
|
||||
f.debug_struct("Rcu")
|
||||
.field("active_value", &active_epoch)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Rcu<T> {
|
||||
pub fn new(initial_value: T) -> Self {
|
||||
Rcu {
|
||||
active_value: AtomicPtr::new(Arc::into_raw(Arc::new(initial_value)).cast_mut()),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn active_epoch(&self) -> RcuRef<T, T> {
|
||||
THREAD_EPOCH_COUNTER.with(|epoch_counter| {
|
||||
let epoch_counter = epoch_counter.get_or_init(|| {
|
||||
let epoch_counter = Arc::new(AtomicU8::new(0));
|
||||
// register the current threads epoch counter on init
|
||||
EPOCH_COUNTERS
|
||||
.blocking_write()
|
||||
.push(Arc::downgrade(&epoch_counter));
|
||||
epoch_counter
|
||||
});
|
||||
|
||||
let old = epoch_counter.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
||||
assert!(old % 2 == 0, "Old Epoch counter value should be even!");
|
||||
});
|
||||
|
||||
let arc_ptr = self.active_value.load(std::sync::atomic::Ordering::Acquire);
|
||||
|
||||
let arc = unsafe {
|
||||
// Safety:
|
||||
// - the ptr was created in Rcu::new or Rcu::replace with Arc::into_raw
|
||||
// - the Rcu is responsible for of the arc's strong refrences
|
||||
// - the Rcu is alive as this function takes a reference to the Rcu
|
||||
// - replace will wait with decrementing the old values strong count until our epoich counter is even again
|
||||
Arc::increment_strong_count(arc_ptr);
|
||||
// Safety:
|
||||
// - the ptr was created in Rcu::new or Rcu::replace with Arc::into_raw
|
||||
// - we have just ensured an additional strong count by incrementing the count
|
||||
Arc::from_raw(arc_ptr)
|
||||
};
|
||||
|
||||
THREAD_EPOCH_COUNTER.with(|epoch_counter| {
|
||||
let old = epoch_counter
|
||||
.get().expect("we initialized the OnceCell when we incremented the epoch counter the fist time")
|
||||
.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
|
||||
assert!(old % 2 != 0, "Old Epoch counter value should be odd!");
|
||||
});
|
||||
|
||||
RcuRef {
|
||||
data: arc.deref().into(),
|
||||
arc,
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* replace the Rcu'S content with a new value
|
||||
*
|
||||
* This does not syncronize write and last to update the active_value pointer wins,
|
||||
* all writes that do not win will be lost, though not leaked.
|
||||
* This will block untill the old value can be reclaimed,
|
||||
* i.e. all threads whitnest to be in the read critical sections
|
||||
* have been witnest to have left the critical section at least once
|
||||
*/
|
||||
pub fn replace(&self, new_value: T) {
|
||||
let arc_ptr = self.active_value.swap(
|
||||
Arc::into_raw(Arc::new(new_value)).cast_mut(),
|
||||
std::sync::atomic::Ordering::AcqRel,
|
||||
);
|
||||
|
||||
// maually drop as we need to ensure not to drop the arc while
|
||||
// we have not witnest all threads to be or have been outside the read critical section
|
||||
// i.e. even epoch counter or different odd epoch counter
|
||||
// Safety:
|
||||
// - the ptr was created in Rcu::new or Rcu::replace with Arc::into_raw
|
||||
// - the Rcu itself holds one strong count
|
||||
let arc = unsafe { ManuallyDrop::new(Arc::from_raw(arc_ptr)) };
|
||||
|
||||
let epochs = EPOCH_COUNTERS.blocking_read().clone();
|
||||
let mut epochs = epochs
|
||||
.into_iter()
|
||||
.flat_map(|elem| {
|
||||
let arc = elem.upgrade()?;
|
||||
let init_val = arc.load(std::sync::atomic::Ordering::Acquire);
|
||||
if init_val % 2 == 0 {
|
||||
// already even can be ignored
|
||||
return None;
|
||||
}
|
||||
// odd initial value thread is in read critical section
|
||||
// need to wait for the value to change before we can drop the arc
|
||||
Some((init_val, elem))
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
while !epochs.is_empty() {
|
||||
epochs.retain(|elem| {
|
||||
let Some(arc) = elem.1.upgrade() else {
|
||||
// as the thread is dead it can't have a ref to old arc
|
||||
return false;
|
||||
};
|
||||
// the epoch counter has not changed so the thread is still in the same instance of the critical section
|
||||
// any different value is ok as
|
||||
// - even values indicate the thread is outside the critical section
|
||||
// - a diffrent odd value indicates the thread has left the critical section and can subsequently only read the new active_value
|
||||
arc.load(std::sync::atomic::Ordering::Acquire) == elem.0
|
||||
})
|
||||
}
|
||||
|
||||
// Safety:
|
||||
// - we have not dropped the arc another way
|
||||
// - we witnessed all threads either with an even epoch count or with a new odd count
|
||||
// as such they must have left the critical section at some point
|
||||
ManuallyDrop::into_inner(arc);
|
||||
}
|
||||
}
|
||||
|
||||
pub struct RcuRef<T, M>
|
||||
where
|
||||
T: ?Sized,
|
||||
M: ?Sized,
|
||||
{
|
||||
arc: Arc<T>,
|
||||
data: NonNull<M>,
|
||||
}
|
||||
|
||||
impl<T: ?Sized, M: ?Sized + Debug> Debug for RcuRef<T, M> {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("RcuRef")
|
||||
.field("data", &self.deref())
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
// use assoiated functions rather than methods so that we don't overlap
|
||||
// with functions of the Deref Target type
|
||||
impl<T: ?Sized, M: ?Sized> RcuRef<T, M> {
|
||||
pub fn map<N: ?Sized, F: for<'a> FnOnce(&'a M) -> &'a N>(referece: Self, f: F) -> RcuRef<T, N> {
|
||||
RcuRef {
|
||||
arc: referece.arc,
|
||||
data: f(unsafe { referece.data.as_ref() }).into(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn try_map<N: ?Sized, F: for<'a> FnOnce(&'a M) -> Option<&'a N>>(
|
||||
referece: Self,
|
||||
f: F,
|
||||
) -> Option<RcuRef<T, N>> {
|
||||
let val = f(unsafe { referece.data.as_ref() })?;
|
||||
Some(RcuRef {
|
||||
arc: Arc::clone(&referece.arc),
|
||||
data: val.into(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn same_epoch(this: &Self, other: &Self) -> bool {
|
||||
Arc::ptr_eq(&this.arc, &other.arc)
|
||||
}
|
||||
|
||||
pub fn clone(this: &Self) -> Self {
|
||||
Self {
|
||||
arc: Arc::clone(&this.arc),
|
||||
data: this.data,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: ?Sized, M: ?Sized> Deref for RcuRef<T, M> {
|
||||
type Target = M;
|
||||
|
||||
fn deref(&self) -> &Self::Target {
|
||||
// Safety: The pointer points into the arc we are holding
|
||||
// while we are alive so is the target
|
||||
// as the content is in an Rcu no mutable acess is given out
|
||||
unsafe { self.data.as_ref() }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user