use super::{ observer::{DelegateObserver, Observer}, utils::{add_mod, modulus}, }; use crate::utils::{move_uninit_slice, slice_as_uninit_mut, slice_assume_init_mut, slice_assume_init_ref}; use core::{ iter::Chain, mem::{self, MaybeUninit}, ptr, slice, }; #[cfg(feature = "std")] use std::io::{self, Write}; /// Consumer part of ring buffer. pub trait Consumer: Observer { /// Set read index. /// /// # Safety /// /// Index must go only forward, never backward. It is recommended to use [`Self::advance_read_index`] instead. /// /// All slots with index less than `value` must be uninitialized until write index, all slots with index equal or greater - must be initialized. unsafe fn set_read_index(&self, value: usize); /// Moves `read` pointer by `count` places forward. /// /// # Safety /// /// First `count` items in occupied memory must be moved out or dropped. /// /// Must not be called concurrently. unsafe fn advance_read_index(&self, count: usize) { unsafe { self.set_read_index(add_mod(self.read_index(), count, modulus(self))) }; } /// Provides a direct access to the ring buffer occupied memory. /// The difference from [`Self::as_slices`] is that this method provides slices of [`MaybeUninit`], so items may be moved out of slices. /// /// Returns a pair of slices of stored items, the second one may be empty. /// Elements with lower indices in slice are older. First slice contains older items that second one. /// /// # Safety /// /// All items are initialized. Elements must be removed starting from the beginning of first slice. /// When all items are removed from the first slice then items must be removed from the beginning of the second slice. /// /// *This method must be followed by [`Self::advance_read_index`] call with the number of items being removed previously as argument.* /// *No other mutating calls allowed before that.* fn occupied_slices(&self) -> (&[MaybeUninit], &[MaybeUninit]) { unsafe { self.unsafe_slices(self.read_index(), self.write_index()) } } /// Provides a direct mutable access to the ring buffer occupied memory. /// /// Same as [`Self::occupied_slices`]. /// /// # Safety /// /// When some item is replaced with uninitialized value then it must not be read anymore. unsafe fn occupied_slices_mut(&mut self) -> (&mut [MaybeUninit], &mut [MaybeUninit]) { unsafe { self.unsafe_slices_mut(self.read_index(), self.write_index()) } } /// Returns a pair of slices which contain, in order, the contents of the ring buffer. #[inline] fn as_slices(&self) -> (&[Self::Item], &[Self::Item]) { unsafe { let (left, right) = self.occupied_slices(); (slice_assume_init_ref(left), slice_assume_init_ref(right)) } } /// Returns a pair of mutable slices which contain, in order, the contents of the ring buffer. #[inline] fn as_mut_slices(&mut self) -> (&mut [Self::Item], &mut [Self::Item]) { unsafe { let (left, right) = self.occupied_slices_mut(); (slice_assume_init_mut(left), slice_assume_init_mut(right)) } } /// Returns a reference to the eldest item in the ring buffer, if exists. #[inline] fn first(&self) -> Option<&Self::Item> { self.as_slices().0.first() } /// Returns a mutable reference to the eldest item in the ring buffer, if exists. #[inline] fn first_mut(&mut self) -> Option<&mut Self::Item> { self.as_mut_slices().0.first_mut() } /// Returns a reference to the most recent item in the ring buffer, if exists. /// /// *Returned item may not be actually the most recent if there is a concurrent producer activity.* fn last(&self) -> Option<&Self::Item> { let (first, second) = self.as_slices(); if second.is_empty() { first.last() } else { second.last() } } /// Returns a mutable reference to the most recent item in the ring buffer, if exists. /// /// *Returned item may not be actually the most recent if there is a concurrent producer activity.* fn last_mut(&mut self) -> Option<&mut Self::Item> { let (first, second) = self.as_mut_slices(); if second.is_empty() { first.last_mut() } else { second.last_mut() } } /// Removes the eldest item from the ring buffer and returns it. /// /// Returns `None` if the ring buffer is empty. fn try_pop(&mut self) -> Option { if !self.is_empty() { let elem = unsafe { self.occupied_slices().0.get_unchecked(0).assume_init_read() }; unsafe { self.advance_read_index(1) }; Some(elem) } else { None } } /// Returns the reference to the eldest item without removing it from the buffer. /// /// Returns `None` if the ring buffer is empty. fn try_peek(&self) -> Option<&Self::Item> { if !self.is_empty() { Some(unsafe { self.occupied_slices().0.get_unchecked(0).assume_init_ref() }) } else { None } } /// Copies items from the ring buffer to an uninit slice without removing them from the ring buffer. /// /// Returns a number of items being copied. fn peek_slice_uninit(&self, elems: &mut [MaybeUninit]) -> usize { let (left, right) = self.occupied_slices(); if elems.len() < left.len() { move_uninit_slice(elems, unsafe { left.get_unchecked(..elems.len()) }); elems.len() } else { let (left_elems, elems) = elems.split_at_mut(left.len()); move_uninit_slice(left_elems, left); left.len() + if elems.len() < right.len() { move_uninit_slice(elems, unsafe { right.get_unchecked(..elems.len()) }); elems.len() } else { move_uninit_slice(unsafe { elems.get_unchecked_mut(..right.len()) }, right); right.len() } } } /// Copies items from the ring buffer to a slice without removing them from the ring buffer. /// /// Returns a number of items being copied. fn peek_slice(&self, elems: &mut [Self::Item]) -> usize where Self::Item: Copy, { self.peek_slice_uninit(unsafe { slice_as_uninit_mut(elems) }) } /// Removes items from the ring buffer and writes them into an uninit slice. /// /// Returns count of items been removed. fn pop_slice_uninit(&mut self, elems: &mut [MaybeUninit]) -> usize { let count = self.peek_slice_uninit(elems); unsafe { self.advance_read_index(count) }; count } /// Removes items from the ring buffer and writes them into a slice. /// /// Returns count of items been removed. fn pop_slice(&mut self, elems: &mut [Self::Item]) -> usize where Self::Item: Copy, { self.pop_slice_uninit(unsafe { slice_as_uninit_mut(elems) }) } /// Returns an iterator that removes items one by one from the ring buffer. fn pop_iter(&mut self) -> PopIter<'_, Self> { PopIter::new(self) } /// Returns a front-to-back iterator containing references to items in the ring buffer. /// /// This iterator does not remove items out of the ring buffer. fn iter(&self) -> Iter<'_, Self> { let (left, right) = self.as_slices(); left.iter().chain(right.iter()) } /// Returns a front-to-back iterator that returns mutable references to items in the ring buffer. /// /// This iterator does not remove items out of the ring buffer. fn iter_mut(&mut self) -> IterMut<'_, Self> { let (left, right) = self.as_mut_slices(); left.iter_mut().chain(right.iter_mut()) } /// Removes and drops at most `count` items, returning the number removed. /// /// Without concurrent producer activity, removes exactly `min(count, self.occupied_len())` items, /// where the length is measured before removal. Items inserted after the occupied range is read are left in the buffer. /// /// Takes constant time for item types for which [`core::mem::needs_drop`] returns false. /// Otherwise, removed items are dropped in order and their slots remain occupied until destruction finishes. /// If a destructor panics, that item and earlier items are removed; later items remain in the buffer. /// /// ``` /// # extern crate ringbuf; /// # use ringbuf::{LocalRb, storage::Array, traits::*}; /// # fn main() { /// let mut rb = LocalRb::>::default(); /// /// assert_eq!(rb.push_iter(0..8), 8); /// /// assert_eq!(rb.skip(4), 4); /// assert_eq!(rb.skip(8), 4); /// assert_eq!(rb.skip(4), 0); /// # } /// ``` fn skip(&mut self, count: usize) -> usize { if !mem::needs_drop::() { // No destructor can run or unwind, so release all slots at once // instead of visiting each item. let count = usize::min(count, self.occupied_len()); unsafe { self.advance_read_index(count) }; return count; } struct Guard<'a, C: Consumer + ?Sized> { consumer: &'a mut C, count: usize, } impl Drop for Guard<'_, C> { fn drop(&mut self) { // The slice iterator and any panicking item destructor have // finished before this guard publishes the freed slots. unsafe { self.consumer.advance_read_index(self.count) }; } } let mut guard = Guard { consumer: self, count: 0 }; let (left, right) = unsafe { guard.consumer.occupied_slices_mut() }; for elem in left.iter_mut().chain(right.iter_mut()).take(count) { // Include this item if its destructor unwinds, but keep its slot // occupied until destruction ends so a producer cannot reuse it. guard.count += 1; unsafe { ptr::drop_in_place(elem.as_mut_ptr()) }; } guard.count } /// Removes and drops all items present when the occupied range is read. /// /// Returns the number of deleted items. See [`Self::skip`] for complexity and behavior when a destructor panics. fn clear(&mut self) -> usize { self.skip(usize::MAX) } #[cfg(feature = "std")] /// Removes at most first `count` bytes from the ring buffer and writes them into a [`Write`] instance. /// If `count` is `None` then as much as possible bytes will be written. /// /// Returns: /// /// + `None`: ring buffer is empty or `count` is `0`. In this case `write` isn't called at all. /// + `Some(Ok(n))`: `write` succeeded. `n` is number of bytes been written. `n == 0` means that `write` also returned `0`. /// + `Some(Err(e))`: `write` is failed and `e` is original error. In this case it is guaranteed that no items was written to the writer. /// To achieve this we write only one contiguous slice at once. So this call may write less than `occupied_len` items even if the writer is ready to get more. fn write_into(&mut self, writer: &mut S, count: Option) -> Option> where Self: Consumer, { let (left, _) = self.occupied_slices(); let count = usize::min(count.unwrap_or(left.len()), left.len()); if count == 0 { return None; } let left_init = unsafe { slice_assume_init_ref(&left[..count]) }; let write_count = match writer.write(left_init) { Ok(n) => n, Err(e) => return Some(Err(e)), }; assert!(write_count <= count); unsafe { self.advance_read_index(write_count) }; Some(Ok(write_count)) } } /// Owning ring buffer iterator. pub struct IntoIter { inner: C, } impl IntoIter { pub fn new(inner: C) -> Self { Self { inner } } pub fn into_inner(self) -> C { self.inner } } impl Iterator for IntoIter { type Item = C::Item; #[inline] fn next(&mut self) -> Option { self.inner.try_pop() } #[inline] fn size_hint(&self) -> (usize, Option) { (self.inner.occupied_len(), None) } } /// An iterator that removes items from the ring buffer. /// /// Each yielded item is removed immediately, making its slot available to the producer. /// Only items present when the iterator is created are yielded. pub struct PopIter<'a, C: Consumer + ?Sized> { inner: &'a C, iter: Chain>, slice::Iter<'a, MaybeUninit>>, read: usize, modulus: usize, remaining: usize, } impl<'a, C: Consumer + ?Sized> PopIter<'a, C> { /// Create an iterator. pub fn new(inner: &'a mut C) -> Self { let (left, right) = inner.occupied_slices(); Self { inner, iter: left.iter().chain(right), read: inner.read_index(), modulus: modulus(inner).get(), remaining: left.len() + right.len(), } } /// Does nothing: each yielded item has already been removed from the ring buffer. #[deprecated(note = "each yielded item is removed immediately; remove this call")] pub fn commit(&mut self) {} } impl Iterator for PopIter<'_, C> { type Item = C::Item; #[inline] fn next(&mut self) -> Option { let item = unsafe { self.iter.next()?.assume_init_read() }; self.read += 1; if self.read == self.modulus { self.read = 0; } self.remaining -= 1; // The slice iterator no longer accesses this item. Publish its removal before // handing it to the caller, who may forget the iterator or drop the item. unsafe { self.inner.set_read_index(self.read) }; Some(item) } #[inline] fn size_hint(&self) -> (usize, Option) { (self.remaining, Some(self.remaining)) } } impl ExactSizeIterator for PopIter<'_, C> {} /// Iterator over ring buffer contents. /// /// *Please do not rely on actual type, it may change in future.* #[allow(type_alias_bounds)] pub type Iter<'a, C: Consumer> = Chain, slice::Iter<'a, C::Item>>; /// Mutable iterator over ring buffer contents. /// /// *Please do not rely on actual type, it may change in future.* #[allow(type_alias_bounds)] pub type IterMut<'a, C: Consumer> = Chain, slice::IterMut<'a, C::Item>>; /// Trait used for delegating producer methods. pub trait DelegateConsumer: DelegateObserver where Self::Base: Consumer, { } impl Consumer for D where D::Base: Consumer, { #[inline] unsafe fn set_read_index(&self, value: usize) { unsafe { self.base().set_read_index(value) } } #[inline] unsafe fn advance_read_index(&self, count: usize) { unsafe { self.base().advance_read_index(count) } } #[inline] fn occupied_slices(&self) -> (&[core::mem::MaybeUninit], &[core::mem::MaybeUninit]) { self.base().occupied_slices() } #[inline] unsafe fn occupied_slices_mut(&mut self) -> (&mut [core::mem::MaybeUninit], &mut [core::mem::MaybeUninit]) { unsafe { self.base_mut().occupied_slices_mut() } } #[inline] fn as_slices(&self) -> (&[Self::Item], &[Self::Item]) { self.base().as_slices() } #[inline] fn as_mut_slices(&mut self) -> (&mut [Self::Item], &mut [Self::Item]) { self.base_mut().as_mut_slices() } #[inline] fn try_pop(&mut self) -> Option { self.base_mut().try_pop() } #[inline] fn pop_slice(&mut self, elems: &mut [Self::Item]) -> usize where Self::Item: Copy, { self.base_mut().pop_slice(elems) } #[inline] fn iter(&self) -> Iter<'_, Self> { self.base().iter() } #[inline] fn iter_mut(&mut self) -> IterMut<'_, Self> { self.base_mut().iter_mut() } #[inline] fn skip(&mut self, count: usize) -> usize { self.base_mut().skip(count) } #[inline] fn clear(&mut self) -> usize { self.base_mut().clear() } } macro_rules! impl_consumer_traits { ($type:ident $(< $( $param:tt $( : $first_bound:tt $(+ $next_bound:tt )* )? ),+ >)?) => { impl $(< $( $param $( : $first_bound $(+ $next_bound )* )? ),+ >)? core::iter::IntoIterator for $type $(< $( $param ),+ >)? where Self: Sized { type Item = ::Item; type IntoIter = $crate::traits::consumer::IntoIter; fn into_iter(self) -> Self::IntoIter { $crate::traits::consumer::IntoIter::new(self) } } #[cfg(feature = "std")] impl $(< $( $param $( : $first_bound $(+ $next_bound )* )? ),+ >)? std::io::Read for $type $(< $( $param ),+ >)? where Self: $crate::traits::Consumer, { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { let n = self.pop_slice(buf); if n == 0 { Err(std::io::ErrorKind::WouldBlock.into()) } else { Ok(n) } } } }; } pub(crate) use impl_consumer_traits;