1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290
use alloc::sync::Arc;
use core::{
mem::{self, MaybeUninit},
ptr::copy_nonoverlapping,
slice,
sync::atomic::Ordering,
};
#[cfg(feature = "std")]
use std::io::{self, Read, Write};
use crate::{consumer::Consumer, ring_buffer::*};
/// Producer part of ring buffer.
pub struct Producer<T> {
pub(crate) rb: Arc<RingBuffer<T>>,
}
impl<T: Sized> Producer<T> {
/// Returns capacity of the ring buffer.
///
/// The capacity of the buffer is constant.
pub fn capacity(&self) -> usize {
self.rb.capacity()
}
/// Checks if the ring buffer is empty.
///
/// The result is relevant until you push items to the producer.
pub fn is_empty(&self) -> bool {
self.rb.is_empty()
}
/// Checks if the ring buffer is full.
///
/// *The result may become irrelevant at any time because of concurring activity of the consumer.*
pub fn is_full(&self) -> bool {
self.rb.is_full()
}
/// The length of the data stored in the buffer.
///
/// Actual length may be equal to or less than the returned value.
pub fn len(&self) -> usize {
self.rb.len()
}
/// The remaining space in the buffer.
///
/// Actual remaining space may be equal to or greater than the returning value.
pub fn remaining(&self) -> usize {
self.rb.remaining()
}
/// Allows to write into ring buffer memory directly.
///
/// *This function is unsafe because it gives access to possibly uninitialized memory*
///
/// The method takes a function `f` as argument.
/// `f` takes two slices of ring buffer content (the second one or both of them may be empty).
/// First slice contains older elements.
///
/// `f` should return number of elements been written.
/// *There is no checks for returned number - it remains on the developer's conscience.*
///
/// The method **always** calls `f` even if ring buffer is full.
///
/// The method returns number returned from `f`.
///
/// # Safety
///
/// The method gives access to ring buffer underlying memory which may be uninitialized.
///
pub unsafe fn push_access<F>(&mut self, f: F) -> usize
where
F: FnOnce(&mut [MaybeUninit<T>], &mut [MaybeUninit<T>]) -> usize,
{
let head = self.rb.head.load(Ordering::Acquire);
let tail = self.rb.tail.load(Ordering::Acquire);
let len = self.rb.data.len();
let ranges = if tail >= head {
if head > 0 {
(tail..len, 0..(head - 1))
} else if tail < len - 1 {
(tail..(len - 1), 0..0)
} else {
(0..0, 0..0)
}
} else if tail < head - 1 {
(tail..(head - 1), 0..0)
} else {
(0..0, 0..0)
};
let ptr = self.rb.data.get_mut().as_mut_ptr();
let slices = (
slice::from_raw_parts_mut(ptr.add(ranges.0.start), ranges.0.len()),
slice::from_raw_parts_mut(ptr.add(ranges.1.start), ranges.1.len()),
);
let n = f(slices.0, slices.1);
if n > 0 {
let new_tail = (tail + n) % len;
self.rb.tail.store(new_tail, Ordering::Release);
}
n
}
/// Copies data from the slice to the ring buffer in byte-to-byte manner.
///
/// The `elems` slice should contain **initialized** data before the method call.
/// After the call the copied part of data in `elems` should be interpreted as **un-initialized**.
///
/// Returns the number of items been copied.
///
/// # Safety
///
/// The method copies raw data into the ring buffer.
///
/// *You should properly fill the slice and manage remaining elements after copy.*
///
pub unsafe fn push_copy(&mut self, elems: &[MaybeUninit<T>]) -> usize {
self.push_access(|left, right| -> usize {
if elems.len() < left.len() {
copy_nonoverlapping(elems.as_ptr(), left.as_mut_ptr(), elems.len());
elems.len()
} else {
copy_nonoverlapping(elems.as_ptr(), left.as_mut_ptr(), left.len());
if elems.len() < left.len() + right.len() {
copy_nonoverlapping(
elems.as_ptr().add(left.len()),
right.as_mut_ptr(),
elems.len() - left.len(),
);
elems.len()
} else {
copy_nonoverlapping(
elems.as_ptr().add(left.len()),
right.as_mut_ptr(),
right.len(),
);
left.len() + right.len()
}
}
})
}
/// Appends an element to the ring buffer.
/// On failure returns an error containing the element that hasn't been appended.
pub fn push(&mut self, elem: T) -> Result<(), T> {
let mut elem_mu = MaybeUninit::new(elem);
let n = unsafe {
self.push_access(|slice, _| {
if !slice.is_empty() {
mem::swap(slice.get_unchecked_mut(0), &mut elem_mu);
1
} else {
0
}
})
};
match n {
0 => Err(unsafe { elem_mu.assume_init() }),
1 => Ok(()),
_ => unreachable!(),
}
}
/// Repeatedly calls the closure `f` and pushes elements returned from it to the ring buffer.
///
/// The closure is called until it returns `None` or the ring buffer is full.
///
/// The method returns number of elements been put into the buffer.
pub fn push_each<F: FnMut() -> Option<T>>(&mut self, mut f: F) -> usize {
unsafe {
self.push_access(|left, right| {
for (i, dst) in left.iter_mut().enumerate() {
match f() {
Some(e) => dst.as_mut_ptr().write(e),
None => return i,
};
}
for (i, dst) in right.iter_mut().enumerate() {
match f() {
Some(e) => dst.as_mut_ptr().write(e),
None => return i + left.len(),
};
}
left.len() + right.len()
})
}
}
/// Appends elements from an iterator to the ring buffer.
/// Elements that haven't been added to the ring buffer remain in the iterator.
///
/// Returns count of elements been appended to the ring buffer.
pub fn push_iter<I: Iterator<Item = T>>(&mut self, elems: &mut I) -> usize {
self.push_each(|| elems.next())
}
/// Removes at most `count` elements from the consumer and appends them to the producer.
/// If `count` is `None` then as much as possible elements will be moved.
/// The producer and consumer parts may be of different buffers as well as of the same one.
///
/// On success returns number of elements been moved.
pub fn move_from(&mut self, other: &mut Consumer<T>, count: Option<usize>) -> usize {
move_items(other, self, count)
}
}
impl<T: Sized + Copy> Producer<T> {
/// Appends elements from slice to the ring buffer.
/// Elements should be [`Copy`](https://doc.rust-lang.org/std/marker/trait.Copy.html).
///
/// Returns count of elements been appended to the ring buffer.
pub fn push_slice(&mut self, elems: &[T]) -> usize {
unsafe { self.push_copy(&*(elems as *const [T] as *const [MaybeUninit<T>])) }
}
}
#[cfg(feature = "std")]
impl Producer<u8> {
/// Reads at most `count` bytes
/// from [`Read`](https://doc.rust-lang.org/std/io/trait.Read.html) instance
/// and appends them to the ring buffer.
/// If `count` is `None` then as much as possible bytes will be read.
///
/// Returns `Ok(n)` if `read` succeeded. `n` is number of bytes been read.
/// `n == 0` means that either `read` returned zero or ring buffer is full.
///
/// If `read` is failed or returned an invalid number then error is returned.
pub fn read_from(&mut self, reader: &mut dyn Read, count: Option<usize>) -> io::Result<usize> {
let mut err = None;
let n = unsafe {
self.push_access(|left, _| -> usize {
let left = match count {
Some(c) => {
if c < left.len() {
&mut left[0..c]
} else {
left
}
}
None => left,
};
match reader
.read(&mut *(left as *mut [MaybeUninit<u8>] as *mut [u8]))
.and_then(|n| {
if n <= left.len() {
Ok(n)
} else {
Err(io::Error::new(
io::ErrorKind::InvalidInput,
"Read operation returned an invalid number",
))
}
}) {
Ok(n) => n,
Err(e) => {
err = Some(e);
0
}
}
})
};
match err {
Some(e) => Err(e),
None => Ok(n),
}
}
}
#[cfg(feature = "std")]
impl Write for Producer<u8> {
fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
let n = self.push_slice(buffer);
if n == 0 && !buffer.is_empty() {
Err(io::ErrorKind::WouldBlock.into())
} else {
Ok(n)
}
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}