//! Caching implementation. //! //! Fetches changes from the ring buffer only when there is no more slots to perform requested operation. //! Changes to this endpoint's index are always published before an operation returns. #[allow(deprecated)] use super::frozen::Frozen; use super::{ direct::{Direct, Obs}, traits::Wrap, }; use crate::{ rb::RbRef, traits::{ Observer, consumer::{Consumer, impl_consumer_traits}, producer::{Producer, impl_producer_traits}, }, }; use core::{cell::Cell, mem::MaybeUninit, num::NonZeroUsize}; /// Caching wrapper of a ring buffer. pub struct Caching { base: Direct, read: Cell, write: Cell, } /// Caching producer implementation. pub type CachingProd = Caching; /// Caching consumer implementation. pub type CachingCons = Caching; impl Caching { /// Create a new ring buffer cached wrapper. /// /// Panics if wrapper with matching rights already exists. pub fn new(rb: R) -> Self { Self::from_direct(Direct::new(rb)) } pub(crate) fn from_direct(base: Direct) -> Self { Self { read: Cell::new(base.read_index()), write: Cell::new(base.write_index()), base, } } /// Get ring buffer observer. pub fn observe(&self) -> Obs { self.base.observe() } /// Convert to the deprecated compatibility wrapper. Changes remain immediately visible. #[deprecated(note = "use the caching endpoint directly; freezing no longer delays publication")] #[allow(deprecated)] pub fn freeze(self) -> Frozen { Frozen::from_caching(self) } pub(crate) fn fetch(&self) { if P { self.read.set(self.base.read_index()); } if C { self.write.set(self.base.write_index()); } } } impl Wrap for Caching { type RbRef = R; fn rb_ref(&self) -> &R { self.base.rb_ref() } fn into_rb_ref(self) -> R { self.base.into_rb_ref() } } impl AsRef for Caching { fn as_ref(&self) -> &Self { self } } impl AsMut for Caching { fn as_mut(&mut self) -> &mut Self { self } } impl Observer for Caching { type Item = ::Item; #[inline] fn capacity(&self) -> NonZeroUsize { self.base.capacity() } #[inline] fn read_index(&self) -> usize { if P { self.fetch(); } self.read.get() } #[inline] fn write_index(&self) -> usize { if C { self.fetch(); } self.write.get() } unsafe fn unsafe_slices(&self, start: usize, end: usize) -> (&[MaybeUninit], &[MaybeUninit]) { unsafe { self.base.unsafe_slices(start, end) } } unsafe fn unsafe_slices_mut(&self, start: usize, end: usize) -> (&mut [MaybeUninit], &mut [MaybeUninit]) { unsafe { self.base.unsafe_slices_mut(start, end) } } #[inline] fn read_is_held(&self) -> bool { self.base.read_is_held() } #[inline] fn write_is_held(&self) -> bool { self.base.write_is_held() } } impl Producer for CachingProd { #[inline] unsafe fn set_write_index(&self, value: usize) { self.write.set(value); unsafe { self.base.set_write_index(value) }; } fn try_push(&mut self, elem: Self::Item) -> Result<(), Self::Item> { let capacity = self.capacity().get(); if self.write.get().abs_diff(self.read.get()) == capacity { self.fetch(); } if self.write.get().abs_diff(self.read.get()) == capacity { return Err(elem); } let write = self.write.get(); // The cached read index only underestimates the available space. unsafe { self.unsafe_slices_mut(write, write + 1).0.get_unchecked_mut(0).write(elem); self.advance_write_index(1); } Ok(()) } } impl Consumer for CachingCons { #[inline] unsafe fn set_read_index(&self, value: usize) { self.read.set(value); unsafe { self.base.set_read_index(value) }; } fn try_pop(&mut self) -> Option<::Item> { if self.read.get() == self.write.get() { self.fetch(); } if self.read.get() == self.write.get() { return None; } let read = self.read.get(); // Move the item out before publishing the slot for reuse. let item = unsafe { self.unsafe_slices(read, read + 1).0.get_unchecked(0).assume_init_read() }; unsafe { self.advance_read_index(1) }; Some(item) } } impl_producer_traits!(CachingProd); impl_consumer_traits!(CachingCons);