Source code
Revision control
Copy as Markdown
Other Tools
//! 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<R: RbRef, const P: bool, const C: bool> {
base: Direct<R, P, C>,
read: Cell<usize>,
write: Cell<usize>,
}
/// Caching producer implementation.
pub type CachingProd<R> = Caching<R, true, false>;
/// Caching consumer implementation.
pub type CachingCons<R> = Caching<R, false, true>;
impl<R: RbRef, const P: bool, const C: bool> Caching<R, P, C> {
/// 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<R, P, C>) -> Self {
Self {
read: Cell::new(base.read_index()),
write: Cell::new(base.write_index()),
base,
}
}
/// Get ring buffer observer.
pub fn observe(&self) -> Obs<R> {
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<R, P, C> {
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<R: RbRef, const P: bool, const C: bool> Wrap for Caching<R, P, C> {
type RbRef = R;
fn rb_ref(&self) -> &R {
self.base.rb_ref()
}
fn into_rb_ref(self) -> R {
self.base.into_rb_ref()
}
}
impl<R: RbRef, const P: bool, const C: bool> AsRef<Self> for Caching<R, P, C> {
fn as_ref(&self) -> &Self {
self
}
}
impl<R: RbRef, const P: bool, const C: bool> AsMut<Self> for Caching<R, P, C> {
fn as_mut(&mut self) -> &mut Self {
self
}
}
impl<R: RbRef, const P: bool, const C: bool> Observer for Caching<R, P, C> {
type Item = <R::Rb as Observer>::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<Self::Item>], &[MaybeUninit<Self::Item>]) {
unsafe { self.base.unsafe_slices(start, end) }
}
unsafe fn unsafe_slices_mut(&self, start: usize, end: usize) -> (&mut [MaybeUninit<Self::Item>], &mut [MaybeUninit<Self::Item>]) {
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<R: RbRef> Producer for CachingProd<R> {
#[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<R: RbRef> Consumer for CachingCons<R> {
#[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<<Self as Observer>::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<R: RbRef>);
impl_consumer_traits!(CachingCons<R: RbRef>);