Something went wrong. Try again.
Realtime safe, waitfree, concurrency library
Something went wrong. Try again.
12 kB · 357 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358// SPDX-FileCopyrightText: 2025 Lucas Baumann// SPDX-FileCopyrightText: Lucas Baumann//// SPDX-License-Identifier: Apache-2.0// SPDX-License-Identifier: MIT
//! Simpler version of the left-right from Jon Gjengset library.//!//! Uses two copies of the value to allow doing small changes, while still allowing non-blocking reading.//! Writing can block, while reading doesn't.#![no_std]
extern crate alloc;
use core::{marker::PhantomData, ops::Deref, ptr::NonNull};
use alloc::collections::vec_deque::VecDeque;
mod shared;
use shared::{Ptr, Shared};
/// Should be implemented on structs that want to be shared with this librarypub trait Absorb<O> { /// has to be deterministic. Operations will be applied in the same order to both buffers fn absorb(&mut self, operation: O);}
/// Dropping the Reader isn't realtime safe, because if dropped after the Writer, it deallocates./// Should only get dropped, when closing the real-time thread////// Reader will be able to read data even if Writer has been dropped. Obviously that data won't change anymore/// When there is no Reader the Writer is able to create a new one. The other way around doesn't work.////// Isn't Sync as there is no methos that takes &self, so it is useless anyways.#[derive(Debug)]pub struct Reader<T> { shared: NonNull<Shared<T>>, locked: bool, /// for drop check _own: PhantomData<Shared<T>>,}
impl<T> Reader<T> { fn shared_ref(&self) -> &Shared<T> { // SAFETY: Reader always has a valid Shared<T>, a mut ref to a shared is never created, // only to the UnsafeCell<T>s inside of it unsafe { self.shared.as_ref() } }
/// this function never blocks. (`fetch_update` loop doesn't count) pub fn lock(&mut self) -> ReadGuard<'_, T> { if self.locked { self.locked = false; panic!("ReadGuard was forgotten"); } self.locked = true; // SAFETY: value just locked let value = unsafe { &*self.shared_ref().lock_read().get() }; ReadGuard { value, reader: self, } }}
/// SAFETY: Owns a Tunsafe impl<T: Send> Send for Reader<T> {}
impl<T> Drop for Reader<T> { fn drop(&mut self) { // SAFETY: self.shared is valid and not used after this. unsafe { Shared::drop(self.shared) }; assert!(!self.locked, "ReadGuard was forgotten"); }}
/// Data won't change while holding the Guard. This also means the Writer can only issue one swap, while Guard is being held/// If T: !Sync this is guaranteed to be the only ref to this T////// Doesn't implement Clone as that would require refcounting to know when to unlock.#[derive(Debug)]pub struct ReadGuard<'a, T> { reader: &'a mut Reader<T>, value: &'a T,}
impl<T> Deref for ReadGuard<'_, T> { type Target = T;
fn deref(&self) -> &Self::Target { self.value }}
impl<T, E> AsRef<E> for ReadGuard<'_, T>where E: ?Sized, T: AsRef<E>,{ fn as_ref(&self) -> &E { self.deref().as_ref() }}
impl<T> Drop for ReadGuard<'_, T> { fn drop(&mut self) { // release the read lock self.reader.shared_ref().release_read_lock(); self.reader.locked = false; }}
/// Not realtime safe object which can change the internal T value.#[derive(Debug)]pub struct Writer<T, O> { shared: NonNull<Shared<T>>, // sets which buffer the next write is applied to // write_ptr doesn't need to be Atomics as it only changes, when the Writer itself swaps write_ptr: Ptr, // buffer is pushed at the back and popped at the front. op_buffer: VecDeque<O>, locked: bool, // needed for drop_check _own: PhantomData<Shared<T>>,}
impl<T, O> Writer<T, O> { fn shared_ref(&self) -> &Shared<T> { // SAFETY: Reader always has a valid Shared<T>, the only possibility to get a &mut Shared requires &mut self unsafe { self.shared.as_ref() } }
/// if no Reader exists this gives a mut ref to Shared. fn shared_mut(&mut self) -> Option<&mut Shared<T>> { self.shared_ref() .is_unique() // SAFETY: No `Reader` exists, as `is_unique` returns true .then(|| unsafe { &mut *self.shared.as_ptr() }) }
/// swaps the read and write values. If no changes were made since the last swap nothing happens. Never blocks /// not public as swapping without creating a before `WriteGuard` is pretty useless fn swap(&mut self) { if self.op_buffer.is_empty() { return; }
self.shared_ref().set_read_ptr(self.write_ptr);
self.write_ptr.switch(); }
/// get a Reader if none exists pub fn build_reader(&mut self) -> Option<Reader<T>> { let shared_ref = self.shared_ref(); // SAFETY: all is_unique_with_increase requirements are satisfied. unsafe { shared_ref.is_unique().then(|| { shared_ref.set_shared(); Reader { shared: self.shared, _own: PhantomData, locked: false, } }) } }}
impl<T: Absorb<O>, O> Writer<T, O> { /// doesn't block. Returns None if the Reader has a `ReadGuard` pointing to the old value. pub fn try_lock(&mut self) -> Option<WriteGuard<'_, T, O>> { if self.locked { self.locked = false; panic!("WriteGuard was forgotten"); } self.shared_ref() .lock_write(self.write_ptr) .ok() // locking was successful .map(|()| { self.locked = true; let mut guard = WriteGuard { writer: self }; while let Some(operation) = guard.writer.op_buffer.pop_front() { guard.get_data_mut().absorb(operation); } guard }) }}
impl<T: Clone, O> Writer<T, O> { /// Creates a new Writer by cloning the value once to get two values /// `T::clone()` shoulnd't give a different value, as that would make this library pretty useless pub fn new(value: T) -> Self { let (shared, write_ptr) = Shared::new(value, |value_1| value_1.clone()); Self { shared, write_ptr, op_buffer: VecDeque::new(), _own: PhantomData, locked: false, } }}
impl<T: Default, O> Default for Writer<T, O> { /// Creates a new Writer by calling `T::default()` twice to create the two values /// /// Default impl of T needs to give the same result every time. Not upholding this doens't lead to UB, but turns the library basically useless fn default() -> Self { let (shared, write_ptr) = Shared::new(T::default(), |_| T::default()); Self { shared, write_ptr, op_buffer: VecDeque::new(), _own: PhantomData, locked: false, } }}
impl<T: Sync, O> Writer<T, O> { /// The Value returned may be newer than the version the reader is currently seeing. /// This value will be written to next. /// If this is called after swapping the write_lock this will return an older value. /// To get the newest value lock the writer and call `Writeguard::read`. /// /// Needs T: Sync because maybe this is the value the reader is curently reading pub fn read(&self) -> &T { // SAFETY: Only the WriteGuard can write to the values / create mut refs to them. // The WriteGuard holds a mut ref to the writer so this function can't be called while a writeguard exists // This means that reading them / creating refs is safe to do unsafe { self.shared_ref().get_value_ref(self.write_ptr) } }}
/// SAFETY: owns T and Ounsafe impl<T: Send, O: Send> Send for Writer<T, O> {}/// SAFETY: &self fn can only create a &T and never gives shared access to Ounsafe impl<T: Sync, O> Sync for Writer<T, O> {}
impl<T, O> Drop for Writer<T, O> { fn drop(&mut self) { // SAFETY: self.shared is valid and not used after this. unsafe { Shared::drop(self.shared) }; assert!(!self.locked, "WriteGuard was forgotten"); }}
// Don't create a WriteGuard directly, as that wouldn't sync with old Operations/// Can be used to write to the Data structure.////// When this structure exists the Reader already switched to the other value////// Dropping this makes all changes available to the Reader.#[derive(Debug)]pub struct WriteGuard<'a, T, O> { // can't hold a mut ref to T, as then it wouldn't be possible to write to both at the same time, // which is an optimization i want to keep. writer: &'a mut Writer<T, O>,}
impl<T, O> WriteGuard<'_, T, O> { /// Makes the changes available to the reader. Equivalent to `std::mem::drop(self)` pub fn swap(self) {}
/// Gets the value currently being written to. pub fn read(&self) -> &T { // SAFETY: Only the WriteGuard can write to the values / create mut refs to them. // The WriteGuard holds a mut ref to the writer so this function can't be called while a writeguard exists // This means that reading them / creating refs is safe to do unsafe { self.writer .shared_ref() .get_value_ref(self.writer.write_ptr) } }
/// Isn't public as this could easily create disconnects between the two versions. /// While that wouldn't lead to UB it goes against the purpose of this library fn get_data_mut(&mut self) -> &mut T { // SAFETY: When creating the writeguad it is checked that the reader doesnt have access to the same data // This function requires &mut self so there also isn't any ref created by writeguard. unsafe { &mut *self .writer .shared_ref() .get_value(self.writer.write_ptr) .get() } }}
impl<T: Absorb<O>, O: Clone> WriteGuard<'_, T, O> { /// applies operation to the current write Value and stores it to apply to the other later. /// If there is no reader the operation is applied to both values immediately and not stored. pub fn apply_op(&mut self, operation: O) { if let Some(shared) = self.writer.shared_mut() { shared.value_1.get_mut().absorb(operation.clone()); shared.value_2.get_mut().absorb(operation); } else { self.writer.op_buffer.push_back(operation.clone()); self.get_data_mut().absorb(operation); } }}
impl<T, O> Drop for WriteGuard<'_, T, O> { fn drop(&mut self) { self.writer.swap(); self.writer.locked = false; }}
#[cfg(test)]mod internal_test { use core::cell::Cell;
use crate::{Absorb, Writer};
#[derive(Clone, Copy, Debug)] pub struct CounterAddOp(i32);
impl Absorb<CounterAddOp> for i32 { fn absorb(&mut self, operation: CounterAddOp) { *self += operation.0; } }
impl Absorb<CounterAddOp> for Cell<i32> { fn absorb(&mut self, operation: CounterAddOp) { self.set(self.get() + operation.0); } }
#[test] fn drop_reader() { let mut writer: Writer<i32, CounterAddOp> = Writer::default(); let reader = writer.build_reader().unwrap();
assert!(!writer.shared_ref().is_unique()); drop(reader); assert!(writer.shared_ref().is_unique()); }
#[test] fn drop_writer() { let mut writer: Writer<i32, CounterAddOp> = Writer::default(); let reader = writer.build_reader().unwrap();
assert!(!reader.shared_ref().is_unique()); drop(writer); assert!(reader.shared_ref().is_unique()); }}