use super::{ observer::{DelegateObserver, Observer}, utils::{add_mod, modulus}, }; #[cfg(feature = "std")] use crate::utils::slice_assume_init_mut; use crate::utils::write_slice; use core::mem::MaybeUninit; #[cfg(feature = "std")] use std::{ cmp, io::{self, Read}, }; /// Producer part of ring buffer. pub trait Producer: Observer { /// Set read index. /// /// # Safety /// /// Index must go only forward, never backward. It is recommended to use [`Self::advance_write_index`] instead. /// /// All slots with index less than `value` must be initialized until write index, all slots with index equal or greater - must be uninitialized. unsafe fn set_write_index(&self, value: usize); /// Moves `write` pointer by `count` places forward. /// /// # Safety /// /// First `count` items in free space must be initialized. /// /// Must not be called concurrently. unsafe fn advance_write_index(&self, count: usize) { unsafe { self.set_write_index(add_mod(self.write_index(), count, modulus(self))) }; } /// Provides a direct access to the ring buffer vacant memory. /// /// Returns a pair of slices of uninitialized memory, the second one may be empty. fn vacant_slices(&self) -> (&[MaybeUninit], &[MaybeUninit]) { unsafe { self.unsafe_slices(self.write_index(), add_mod(self.read_index(), self.capacity().get(), modulus(self))) } } /// Mutable version of [`Self::vacant_slices`]. /// /// Vacant memory is uninitialized. Initialized items must be put starting from the beginning of first slice. /// When first slice is fully filled then items must be put to the beginning of the second slice. /// /// *This method must be followed by [`Self::advance_write_index`] call with the number of items being put previously as argument.* /// *No other mutating calls allowed before that.* /// /// *Vacant slices must not be used to store any data because their contents aren't synchronized properly.* fn vacant_slices_mut(&mut self) -> (&mut [MaybeUninit], &mut [MaybeUninit]) { unsafe { self.unsafe_slices_mut(self.write_index(), add_mod(self.read_index(), self.capacity().get(), modulus(self))) } } /// Appends an item to the ring buffer. /// /// If buffer is full returns an `Err` containing the item that hasn't been appended. fn try_push(&mut self, elem: Self::Item) -> Result<(), Self::Item> { if !self.is_full() { unsafe { self.vacant_slices_mut().0.get_unchecked_mut(0).write(elem); self.advance_write_index(1) }; Ok(()) } else { Err(elem) } } /// Appends items from an iterator to the ring buffer. /// Elements that haven't been added to the ring buffer remain in the iterator. /// /// Returns count of items been appended to the ring buffer. /// /// *Inserted items are committed to the ring buffer all at once in the end,* /// *e.g. when buffer is full or iterator has ended.* fn push_iter>(&mut self, mut iter: I) -> usize { let (left, right) = self.vacant_slices_mut(); let mut count = 0; for place in left.iter_mut().chain(right.iter_mut()) { match iter.next() { Some(elem) => unsafe { place.as_mut_ptr().write(elem) }, None => break, } count += 1; } unsafe { self.advance_write_index(count) }; count } /// Appends items from slice to the ring buffer. /// /// Returns count of items been appended to the ring buffer. fn push_slice(&mut self, elems: &[Self::Item]) -> usize where Self::Item: Copy, { let (left, right) = self.vacant_slices_mut(); let count = if elems.len() < left.len() { write_slice(&mut left[..elems.len()], elems); elems.len() } else { let (left_elems, elems) = elems.split_at(left.len()); write_slice(left, left_elems); left.len() + if elems.len() < right.len() { write_slice(&mut right[..elems.len()], elems); elems.len() } else { write_slice(right, &elems[..right.len()]); right.len() } }; unsafe { self.advance_write_index(count) }; count } #[cfg(feature = "std")] /// Reads at most `count` bytes from `Read` instance and appends them to the ring buffer. /// If `count` is `None` then as much as possible bytes will be read. /// /// Returns: /// + `None`: ring buffer is full or `count` is `0`. In this case `read` isn't called at all. /// + `Some(Ok(n))`: `read` succeeded. `n` is number of bytes been read. `n == 0` means that `read` also returned `0`. /// + `Some(Err(e))` `read` is failed and `e` is original error. In this case it is guaranteed that no items was read from the reader. /// To achieve this we read only one contiguous slice at once. So this call may read less than `vacant_len` items in the buffer even if the reader is ready to provide more. fn read_from(&mut self, reader: &mut S, count: Option) -> Option> where Self: Producer, { let (left, _) = self.vacant_slices_mut(); let count = cmp::min(count.unwrap_or(left.len()), left.len()); if count == 0 { return None; } let buf = &mut left[..count]; // Initialize memory before read. It's an overhead but there's no way to read to uninit buffer in stable Rust yet. // TODO: Use `reader.read_buf` when it stabilized (see https://github.com/rust-lang/rust/issues/78485). buf.fill(MaybeUninit::new(0)); let left_init = unsafe { slice_assume_init_mut(buf) }; let read_count = match reader.read(left_init) { Ok(n) => n, Err(e) => return Some(Err(e)), }; assert!(read_count <= count); unsafe { self.advance_write_index(read_count) }; Some(Ok(read_count)) } } /// Trait used for delegating consumer methods. pub trait DelegateProducer: DelegateObserver where Self::Base: Producer, { } impl Producer for D where D::Base: Producer, { #[inline] unsafe fn set_write_index(&self, value: usize) { unsafe { self.base().set_write_index(value) } } #[inline] unsafe fn advance_write_index(&self, count: usize) { unsafe { self.base().advance_write_index(count) } } #[inline] fn vacant_slices(&self) -> (&[core::mem::MaybeUninit], &[core::mem::MaybeUninit]) { self.base().vacant_slices() } #[inline] fn vacant_slices_mut(&mut self) -> (&mut [core::mem::MaybeUninit], &mut [core::mem::MaybeUninit]) { self.base_mut().vacant_slices_mut() } #[inline] fn try_push(&mut self, elem: Self::Item) -> Result<(), Self::Item> { self.base_mut().try_push(elem) } #[inline] fn push_iter>(&mut self, iter: I) -> usize { self.base_mut().push_iter(iter) } #[inline] fn push_slice(&mut self, elems: &[Self::Item]) -> usize where Self::Item: Copy, { self.base_mut().push_slice(elems) } } macro_rules! impl_producer_traits { ($type:ident $(< $( $param:tt $( : $first_bound:tt $(+ $next_bound:tt )* )? ),+ >)?) => { #[cfg(feature = "std")] impl $(< $( $param $( : $first_bound $(+ $next_bound )* )? ),+ >)? std::io::Write for $type $(< $( $param ),+ >)? where Self: $crate::traits::Producer, { fn write(&mut self, buf: &[u8]) -> std::io::Result { let n = self.push_slice(buf); if n == 0 { Err(std::io::ErrorKind::WouldBlock.into()) } else { Ok(n) } } fn flush(&mut self) -> std::io::Result<()> { Ok(()) } } impl $(< $( $param $( : $first_bound $(+ $next_bound )* )? ),+ >)? core::fmt::Write for $type $(< $( $param ),+ >)? where Self: $crate::traits::Producer, { fn write_str(&mut self, s: &str) -> core::fmt::Result { let n = self.push_slice(s.as_bytes()); if n != s.len() { Err(core::fmt::Error::default()) } else { Ok(()) } } } }; } pub(crate) use impl_producer_traits;