use super::{ macros::rb_impl_init, utils::{assert_capacity, ranges}, }; #[cfg(feature = "alloc")] use crate::traits::Split; use crate::{ storage::Storage, traits::{ Observer, RingBuffer, SplitRef, consumer::{Consumer, impl_consumer_traits}, producer::{Producer, impl_producer_traits}, }, wrap::{CachingCons, CachingProd}, }; use core::{ mem::{ManuallyDrop, MaybeUninit}, num::NonZeroUsize, ptr, }; use crossbeam_utils::CachePadded; #[cfg(not(feature = "portable-atomic"))] use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; #[cfg(feature = "portable-atomic")] use portable_atomic::{AtomicBool, AtomicUsize, Ordering}; #[cfg(feature = "alloc")] use {crate::alias::Arc, alloc::boxed::Box}; /// Ring buffer that can be shared between threads. /// /// Note that there is no explicit requirement of `T: Send`. Instead ring buffer will work just fine even with `T: !Send` /// until you try to send its producer or consumer to another thread. #[cfg_attr( feature = "std", doc = r##" ``` use std::thread; use ringbuf::{SharedRb, storage::Heap, traits::*}; let rb = SharedRb::>::new(256); let (mut prod, mut cons) = rb.split(); thread::spawn(move || { prod.try_push(123).unwrap(); }) .join(); thread::spawn(move || { assert_eq!(cons.try_pop().unwrap(), 123); }) .join(); ``` "## )] pub struct SharedRb { read_index: CachePadded, write_index: CachePadded, read_held: AtomicBool, write_held: AtomicBool, storage: S, } impl SharedRb { /// Constructs ring buffer from storage and indices. /// /// Panics if the storage is empty or its length exceeds `usize::MAX / 2`. /// /// # Safety /// /// The items in storage inside `read..write` range must be initialized, items outside this range must be uninitialized. /// `read` and `write` positions must be valid (see implementation details). pub unsafe fn from_raw_parts(storage: S, read: usize, write: usize) -> Self { assert_capacity(storage.len()); Self { storage, read_index: CachePadded::new(AtomicUsize::new(read)), write_index: CachePadded::new(AtomicUsize::new(write)), read_held: AtomicBool::new(false), write_held: AtomicBool::new(false), } } /// Destructures ring buffer into underlying storage and `read` and `write` indices. /// /// # Safety /// /// Initialized contents of the storage must be properly dropped. pub unsafe fn into_raw_parts(self) -> (S, usize, usize) { let this = ManuallyDrop::new(self); (unsafe { ptr::read(&this.storage) }, this.read_index(), this.write_index()) } } impl Observer for SharedRb { type Item = S::Item; #[inline] fn capacity(&self) -> NonZeroUsize { unsafe { NonZeroUsize::new_unchecked(self.storage.len()) } } #[inline] fn read_index(&self) -> usize { self.read_index.load(Ordering::Acquire) } #[inline] fn write_index(&self) -> usize { self.write_index.load(Ordering::Acquire) } unsafe fn unsafe_slices(&self, start: usize, end: usize) -> (&[MaybeUninit], &[MaybeUninit]) { let (first, second) = ranges(self.capacity(), start, end); unsafe { (self.storage.slice(first), self.storage.slice(second)) } } unsafe fn unsafe_slices_mut(&self, start: usize, end: usize) -> (&mut [MaybeUninit], &mut [MaybeUninit]) { let (first, second) = ranges(self.capacity(), start, end); unsafe { (self.storage.slice_mut(first), self.storage.slice_mut(second)) } } #[inline] fn read_is_held(&self) -> bool { self.read_held.load(Ordering::Acquire) } #[inline] fn write_is_held(&self) -> bool { self.write_held.load(Ordering::Acquire) } } impl Producer for SharedRb { #[inline] unsafe fn set_write_index(&self, value: usize) { self.write_index.store(value, Ordering::Release); } } impl Consumer for SharedRb { #[inline] unsafe fn set_read_index(&self, value: usize) { self.read_index.store(value, Ordering::Release); } } impl RingBuffer for SharedRb { #[inline] unsafe fn hold_read(&self, flag: bool) -> bool { self.read_held.swap(flag, Ordering::AcqRel) } #[inline] unsafe fn hold_write(&self, flag: bool) -> bool { self.write_held.swap(flag, Ordering::AcqRel) } } impl Drop for SharedRb { fn drop(&mut self) { self.clear(); } } #[cfg(feature = "alloc")] impl Split for SharedRb { type Prod = CachingProd>; type Cons = CachingCons>; fn split(self) -> (Self::Prod, Self::Cons) { Arc::new(self).split() } } #[cfg(feature = "alloc")] impl Split for Arc> { type Prod = CachingProd; type Cons = CachingCons; fn split(self) -> (Self::Prod, Self::Cons) { (CachingProd::new(self.clone()), CachingCons::new(self)) } } #[cfg(feature = "alloc")] impl Split for Box> { type Prod = CachingProd>>; type Cons = CachingCons>>; fn split(self) -> (Self::Prod, Self::Cons) { Arc::>::from(self).split() } } impl SplitRef for SharedRb { type RefProd<'a> = CachingProd<&'a Self> where Self: 'a; type RefCons<'a> = CachingCons<&'a Self> where Self: 'a; fn split_ref(&mut self) -> (Self::RefProd<'_>, Self::RefCons<'_>) { (CachingProd::new(self), CachingCons::new(self)) } } rb_impl_init!(SharedRb); impl_producer_traits!(SharedRb); impl_consumer_traits!(SharedRb); impl AsRef for SharedRb { fn as_ref(&self) -> &Self { self } } impl AsMut for SharedRb { fn as_mut(&mut self) -> &mut Self { self } }