ringbuf/wrap/
caching.rs

1//! Caching implementation.
2//!
3//! Fetches changes from the ring buffer only when there is no more slots to perform requested operation.
4//! Changes to this endpoint's index are always published before an operation returns.
5
6#[allow(deprecated)]
7use super::frozen::Frozen;
8use super::{
9    direct::{Direct, Obs},
10    traits::Wrap,
11};
12use crate::{
13    rb::RbRef,
14    traits::{
15        Observer,
16        consumer::{Consumer, impl_consumer_traits},
17        producer::{Producer, impl_producer_traits},
18    },
19};
20use core::{cell::Cell, mem::MaybeUninit, num::NonZeroUsize};
21
22/// Caching wrapper of a ring buffer.
23pub struct Caching<R: RbRef, const P: bool, const C: bool> {
24    base: Direct<R, P, C>,
25    read: Cell<usize>,
26    write: Cell<usize>,
27}
28
29/// Caching producer implementation.
30pub type CachingProd<R> = Caching<R, true, false>;
31/// Caching consumer implementation.
32pub type CachingCons<R> = Caching<R, false, true>;
33
34impl<R: RbRef, const P: bool, const C: bool> Caching<R, P, C> {
35    /// Create a new ring buffer cached wrapper.
36    ///
37    /// Panics if wrapper with matching rights already exists.
38    pub fn new(rb: R) -> Self {
39        Self::from_direct(Direct::new(rb))
40    }
41
42    pub(crate) fn from_direct(base: Direct<R, P, C>) -> Self {
43        Self {
44            read: Cell::new(base.read_index()),
45            write: Cell::new(base.write_index()),
46            base,
47        }
48    }
49
50    /// Get ring buffer observer.
51    pub fn observe(&self) -> Obs<R> {
52        self.base.observe()
53    }
54
55    /// Convert to the deprecated compatibility wrapper. Changes remain immediately visible.
56    #[deprecated(note = "use the caching endpoint directly; freezing no longer delays publication")]
57    #[allow(deprecated)]
58    pub fn freeze(self) -> Frozen<R, P, C> {
59        Frozen::from_caching(self)
60    }
61
62    pub(crate) fn fetch(&self) {
63        if P {
64            self.read.set(self.base.read_index());
65        }
66        if C {
67            self.write.set(self.base.write_index());
68        }
69    }
70}
71
72impl<R: RbRef, const P: bool, const C: bool> Wrap for Caching<R, P, C> {
73    type RbRef = R;
74
75    fn rb_ref(&self) -> &R {
76        self.base.rb_ref()
77    }
78    fn into_rb_ref(self) -> R {
79        self.base.into_rb_ref()
80    }
81}
82
83impl<R: RbRef, const P: bool, const C: bool> AsRef<Self> for Caching<R, P, C> {
84    fn as_ref(&self) -> &Self {
85        self
86    }
87}
88impl<R: RbRef, const P: bool, const C: bool> AsMut<Self> for Caching<R, P, C> {
89    fn as_mut(&mut self) -> &mut Self {
90        self
91    }
92}
93
94impl<R: RbRef, const P: bool, const C: bool> Observer for Caching<R, P, C> {
95    type Item = <R::Rb as Observer>::Item;
96
97    #[inline]
98    fn capacity(&self) -> NonZeroUsize {
99        self.base.capacity()
100    }
101
102    #[inline]
103    fn read_index(&self) -> usize {
104        if P {
105            self.fetch();
106        }
107        self.read.get()
108    }
109    #[inline]
110    fn write_index(&self) -> usize {
111        if C {
112            self.fetch();
113        }
114        self.write.get()
115    }
116
117    unsafe fn unsafe_slices(&self, start: usize, end: usize) -> (&[MaybeUninit<Self::Item>], &[MaybeUninit<Self::Item>]) {
118        unsafe { self.base.unsafe_slices(start, end) }
119    }
120    unsafe fn unsafe_slices_mut(&self, start: usize, end: usize) -> (&mut [MaybeUninit<Self::Item>], &mut [MaybeUninit<Self::Item>]) {
121        unsafe { self.base.unsafe_slices_mut(start, end) }
122    }
123
124    #[inline]
125    fn read_is_held(&self) -> bool {
126        self.base.read_is_held()
127    }
128    #[inline]
129    fn write_is_held(&self) -> bool {
130        self.base.write_is_held()
131    }
132}
133
134impl<R: RbRef> Producer for CachingProd<R> {
135    #[inline]
136    unsafe fn set_write_index(&self, value: usize) {
137        self.write.set(value);
138        unsafe { self.base.set_write_index(value) };
139    }
140
141    fn try_push(&mut self, elem: Self::Item) -> Result<(), Self::Item> {
142        let capacity = self.capacity().get();
143        if self.write.get().abs_diff(self.read.get()) == capacity {
144            self.fetch();
145        }
146        if self.write.get().abs_diff(self.read.get()) == capacity {
147            return Err(elem);
148        }
149        let write = self.write.get();
150        // The cached read index only underestimates the available space.
151        unsafe {
152            self.unsafe_slices_mut(write, write + 1).0.get_unchecked_mut(0).write(elem);
153            self.advance_write_index(1);
154        }
155        Ok(())
156    }
157}
158
159impl<R: RbRef> Consumer for CachingCons<R> {
160    #[inline]
161    unsafe fn set_read_index(&self, value: usize) {
162        self.read.set(value);
163        unsafe { self.base.set_read_index(value) };
164    }
165
166    fn try_pop(&mut self) -> Option<<Self as Observer>::Item> {
167        if self.read.get() == self.write.get() {
168            self.fetch();
169        }
170        if self.read.get() == self.write.get() {
171            return None;
172        }
173        let read = self.read.get();
174        // Move the item out before publishing the slot for reuse.
175        let item = unsafe { self.unsafe_slices(read, read + 1).0.get_unchecked(0).assume_init_read() };
176        unsafe { self.advance_read_index(1) };
177        Some(item)
178    }
179}
180
181impl_producer_traits!(CachingProd<R: RbRef>);
182impl_consumer_traits!(CachingCons<R: RbRef>);
ringbuf/wrap/caching.rs

ringbuf/wrap/
caching.rs
