use crate::error::StreamError;
use std::cell::UnsafeCell;
use std::mem::MaybeUninit;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
pub struct SpscRing<T, const N: usize> {
buf: Box<[UnsafeCell<MaybeUninit<T>>; N]>,
head: AtomicUsize,
tail: AtomicUsize,
}
unsafe impl<T: Send, const N: usize> Send for SpscRing<T, N> {}
impl<T, const N: usize> SpscRing<T, N> {
const _ASSERT_N_GE_2: () = assert!(N >= 2, "SpscRing: N must be >= 2 (capacity = N-1)");
const _ASSERT_N_POW2: () = assert!(
N.is_power_of_two(),
"SpscRing: N must be a power of two (e.g. 4, 8, 16, 32, 64, 128, 256, 512, 1024)"
);
pub fn new() -> Self {
let _ = Self::_ASSERT_N_GE_2;
let _ = Self::_ASSERT_N_POW2;
let buf: Vec<UnsafeCell<MaybeUninit<T>>> =
(0..N).map(|_| UnsafeCell::new(MaybeUninit::uninit())).collect();
let buf: Box<[UnsafeCell<MaybeUninit<T>>; N]> = buf
.try_into()
.unwrap_or_else(|_| unreachable!("length is exactly N"));
Self {
buf,
head: AtomicUsize::new(0),
tail: AtomicUsize::new(0),
}
}
#[inline]
pub fn is_empty(&self) -> bool {
self.head.load(Ordering::Acquire) == self.tail.load(Ordering::Acquire)
}
#[inline]
pub fn is_full(&self) -> bool {
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
tail.wrapping_sub(head) >= N - 1
}
#[inline]
pub fn len(&self) -> usize {
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
tail.wrapping_sub(head)
}
#[inline]
pub fn capacity(&self) -> usize {
N - 1
}
#[inline]
#[must_use = "dropping a push result silently discards the item when full"]
pub fn push(&self, item: T) -> Result<(), StreamError> {
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Relaxed);
if tail.wrapping_sub(head) >= N - 1 {
return Err(StreamError::RingBufferFull { capacity: N - 1 });
}
let slot = tail & (N - 1);
unsafe {
(*self.buf[slot].get()).write(item);
}
self.tail.store(tail.wrapping_add(1), Ordering::Release);
Ok(())
}
#[inline]
#[must_use = "dropping a pop result discards the dequeued item"]
pub fn pop(&self) -> Result<T, StreamError> {
let tail = self.tail.load(Ordering::Acquire);
let head = self.head.load(Ordering::Relaxed);
if head == tail {
return Err(StreamError::RingBufferEmpty);
}
let slot = head & (N - 1);
let item = unsafe { (*self.buf[slot].get()).assume_init_read() };
self.head.store(head.wrapping_add(1), Ordering::Release);
Ok(item)
}
#[inline]
pub fn try_push_or_drop(&self, item: T) -> bool {
self.push(item).is_ok()
}
pub fn peek_clone(&self) -> Option<T>
where
T: Clone,
{
let tail = self.tail.load(Ordering::Acquire);
let head = self.head.load(Ordering::Relaxed);
if head == tail {
return None;
}
let slot = head & (N - 1);
Some(unsafe { (*self.buf[slot].get()).assume_init_ref() }.clone())
}
pub fn peek_all(&self) -> Vec<T>
where
T: Clone,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let count = tail.wrapping_sub(head);
let mut out = Vec::with_capacity(count);
for i in 0..count {
let slot = head.wrapping_add(i) & (N - 1);
out.push(unsafe { (*self.buf[slot].get()).assume_init_ref() }.clone());
}
out
}
pub fn peek_newest(&self) -> Option<T>
where
T: Copy,
{
self.peek_back().copied()
}
pub fn peek_oldest(&self) -> Option<T>
where
T: Copy,
{
self.first()
}
pub fn fill_ratio(&self) -> f64 {
let cap = self.capacity();
if cap == 0 {
return 0.0;
}
self.len() as f64 / cap as f64
}
pub fn has_capacity(&self, n: usize) -> bool {
n <= self.remaining_capacity()
}
pub fn utilization_pct(&self) -> f64 {
self.fill_ratio() * 100.0
}
#[inline]
pub fn remaining_capacity(&self) -> usize {
self.capacity().saturating_sub(self.len())
}
pub fn is_nearly_full(&self, threshold: f64) -> bool {
self.fill_ratio() >= threshold
}
pub fn first(&self) -> Option<T>
where
T: Copy,
{
self.peek_front().copied()
}
pub fn peek_front(&self) -> Option<&T> {
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
if head == tail {
return None;
}
Some(unsafe { (*self.buf[head & (N - 1)].get()).assume_init_ref() })
}
pub fn peek_back(&self) -> Option<&T> {
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
if head == tail {
return None;
}
let back = tail.wrapping_sub(1);
Some(unsafe { (*self.buf[back & (N - 1)].get()).assume_init_ref() })
}
pub fn drain(&self) -> Vec<T> {
let mut out = Vec::with_capacity(self.len());
while let Ok(item) = self.pop() {
out.push(item);
}
out
}
pub fn drain_into(&self, buf: &mut Vec<T>) {
while let Ok(item) = self.pop() {
buf.push(item);
}
}
pub fn to_vec_cloned(&self) -> Vec<T>
where
T: Clone,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
let mut out = Vec::with_capacity(len);
for i in 0..len {
let slot = head.wrapping_add(i) & (N - 1);
let item = unsafe { (*self.buf[slot].get()).assume_init_ref() };
out.push(item.clone());
}
out
}
pub fn to_vec_sorted(&self) -> Vec<T>
where
T: Clone + Ord,
{
let mut v = self.to_vec_cloned();
v.sort();
v
}
pub fn min_cloned(&self) -> Option<T>
where
T: Clone + Ord,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
if len == 0 {
return None;
}
let mut min_val = unsafe { (*self.buf[head & (N - 1)].get()).assume_init_ref() }.clone();
for i in 1..len {
let slot = head.wrapping_add(i) & (N - 1);
let item = unsafe { (*self.buf[slot].get()).assume_init_ref() };
if item < &min_val {
min_val = item.clone();
}
}
Some(min_val)
}
pub fn max_cloned(&self) -> Option<T>
where
T: Clone + Ord,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
if len == 0 {
return None;
}
let mut max_val = unsafe { (*self.buf[head & (N - 1)].get()).assume_init_ref() }.clone();
for i in 1..len {
let slot = head.wrapping_add(i) & (N - 1);
let item = unsafe { (*self.buf[slot].get()).assume_init_ref() };
if item > &max_val {
max_val = item.clone();
}
}
Some(max_val)
}
pub fn count_if<F>(&self, predicate: F) -> usize
where
F: Fn(&T) -> bool,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
let mut count = 0;
for i in 0..len {
let slot = head.wrapping_add(i) & (N - 1);
let item = unsafe { (*self.buf[slot].get()).assume_init_ref() };
if predicate(item) {
count += 1;
}
}
count
}
pub fn peek_nth(&self, n: usize) -> Option<T>
where
T: Clone,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
if n >= len {
return None;
}
let slot = head.wrapping_add(n) & (N - 1);
Some(unsafe { (*self.buf[slot].get()).assume_init_ref() }.clone())
}
pub fn min_cloned_by<F, K>(&self, key: F) -> Option<T>
where
T: Clone,
F: Fn(&T) -> K,
K: Ord,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
if len == 0 {
return None;
}
let first = unsafe { (*self.buf[head & (N - 1)].get()).assume_init_ref() }.clone();
let mut best_key = key(&first);
let mut best = first;
for i in 1..len {
let slot = head.wrapping_add(i) & (N - 1);
let item = unsafe { (*self.buf[slot].get()).assume_init_ref() }.clone();
let k = key(&item);
if k < best_key {
best_key = k;
best = item;
}
}
Some(best)
}
pub fn max_cloned_by<F, K>(&self, key: F) -> Option<T>
where
T: Clone,
F: Fn(&T) -> K,
K: Ord,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
if len == 0 {
return None;
}
let first = unsafe { (*self.buf[head & (N - 1)].get()).assume_init_ref() }.clone();
let mut best_key = key(&first);
let mut best = first;
for i in 1..len {
let slot = head.wrapping_add(i) & (N - 1);
let item = unsafe { (*self.buf[slot].get()).assume_init_ref() }.clone();
let k = key(&item);
if k > best_key {
best_key = k;
best = item;
}
}
Some(best)
}
pub fn contains_cloned(&self, value: &T) -> bool
where
T: Clone + PartialEq,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
for i in 0..len {
let slot = head.wrapping_add(i) & (N - 1);
let item = unsafe { (*self.buf[slot].get()).assume_init_ref() };
if item == value {
return true;
}
}
false
}
pub fn average_cloned(&self) -> Option<f64>
where
T: Clone + Into<f64>,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
if len == 0 {
return None;
}
let sum: f64 = (0..len)
.map(|i| {
let slot = head.wrapping_add(i) & (N - 1);
unsafe { (*self.buf[slot].get()).assume_init_ref() }.clone().into()
})
.sum();
Some(sum / len as f64)
}
pub fn sum_cloned(&self) -> T
where
T: Clone + std::iter::Sum + Default,
{
let head = self.head.load(Ordering::Acquire);
let tail = self.tail.load(Ordering::Acquire);
let len = tail.wrapping_sub(head);
if len == 0 {
return T::default();
}
(0..len)
.map(|i| {
let slot = head.wrapping_add(i) & (N - 1);
unsafe { (*self.buf[slot].get()).assume_init_ref() }.clone()
})
.sum()
}
pub fn split(self) -> (SpscProducer<T, N>, SpscConsumer<T, N>) {
let shared = Arc::new(self);
(
SpscProducer {
inner: Arc::clone(&shared),
},
SpscConsumer { inner: shared },
)
}
}
impl<T, const N: usize> Drop for SpscRing<T, N> {
fn drop(&mut self) {
let head = self.head.load(Ordering::Relaxed);
let tail = self.tail.load(Ordering::Relaxed);
let mut idx = head;
while idx != tail {
let slot = idx & (N - 1);
unsafe {
(*self.buf[slot].get()).assume_init_drop();
}
idx = idx.wrapping_add(1);
}
}
}
impl<T, const N: usize> Default for SpscRing<T, N> {
fn default() -> Self {
Self::new()
}
}
pub struct SpscProducer<T, const N: usize> {
inner: Arc<SpscRing<T, N>>,
}
unsafe impl<T: Send, const N: usize> Send for SpscProducer<T, N> {}
impl<T, const N: usize> SpscProducer<T, N> {
#[inline]
pub fn push(&self, item: T) -> Result<(), StreamError> {
self.inner.push(item)
}
#[inline]
pub fn try_push_or_drop(&self, item: T) -> bool {
self.inner.try_push_or_drop(item)
}
#[inline]
pub fn is_full(&self) -> bool {
self.inner.is_full()
}
#[inline]
pub fn is_empty(&self) -> bool {
self.inner.is_empty()
}
#[inline]
pub fn len(&self) -> usize {
self.inner.len()
}
#[inline]
pub fn available(&self) -> usize {
self.inner.remaining_capacity()
}
#[inline]
pub fn capacity(&self) -> usize {
self.inner.capacity()
}
#[inline]
pub fn fill_ratio(&self) -> f64 {
self.inner.fill_ratio()
}
}
pub struct SpscConsumer<T, const N: usize> {
inner: Arc<SpscRing<T, N>>,
}
unsafe impl<T: Send, const N: usize> Send for SpscConsumer<T, N> {}
impl<T, const N: usize> SpscConsumer<T, N> {
#[inline]
pub fn pop(&self) -> Result<T, StreamError> {
self.inner.pop()
}
pub fn drain(&self) -> Vec<T> {
let mut out = Vec::with_capacity(self.inner.len());
while let Ok(item) = self.inner.pop() {
out.push(item);
}
out
}
#[inline]
pub fn is_empty(&self) -> bool {
self.inner.is_empty()
}
#[inline]
pub fn len(&self) -> usize {
self.inner.len()
}
#[inline]
pub fn capacity(&self) -> usize {
self.inner.capacity()
}
#[inline]
pub fn fill_ratio(&self) -> f64 {
self.inner.fill_ratio()
}
pub fn peek_clone(&self) -> Option<T>
where
T: Clone,
{
self.inner.peek_clone()
}
pub fn try_pop_n(&self, max: usize) -> Vec<T> {
let mut out = Vec::with_capacity(max.min(self.inner.len()));
while out.len() < max {
match self.inner.pop() {
Ok(item) => out.push(item),
Err(_) => break,
}
}
out
}
pub fn into_iter_drain(self) -> SpscDrainIter<T, N> {
SpscDrainIter { consumer: self }
}
}
pub struct SpscDrainIter<T, const N: usize> {
consumer: SpscConsumer<T, N>,
}
impl<T, const N: usize> Iterator for SpscDrainIter<T, N> {
type Item = T;
fn next(&mut self) -> Option<Self::Item> {
self.consumer.pop().ok()
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::thread;
#[test]
fn test_new_ring_is_empty() {
let r: SpscRing<u32, 8> = SpscRing::new();
assert!(r.is_empty());
assert_eq!(r.len(), 0);
}
#[test]
fn test_push_pop_single_item() {
let r: SpscRing<u32, 8> = SpscRing::new();
r.push(42).unwrap();
assert_eq!(r.pop().unwrap(), 42);
}
#[test]
fn test_pop_empty_returns_ring_buffer_empty() {
let r: SpscRing<u32, 8> = SpscRing::new();
let err = r.pop().unwrap_err();
assert!(matches!(err, StreamError::RingBufferEmpty));
}
#[test]
fn test_capacity_is_n_minus_1() {
let r: SpscRing<u32, 8> = SpscRing::new();
assert_eq!(r.capacity(), 7);
}
#[test]
fn test_fill_to_exact_capacity_then_overflow() {
let r: SpscRing<u32, 8> = SpscRing::new(); for i in 0..7u32 {
r.push(i).unwrap();
}
assert!(r.is_full());
let err = r.push(99).unwrap_err();
assert!(matches!(err, StreamError::RingBufferFull { capacity: 7 }));
}
#[test]
fn test_push_n_minus_1_pop_one_push_one() {
let r: SpscRing<u32, 8> = SpscRing::new();
for i in 0..7u32 {
r.push(i).unwrap();
}
assert_eq!(r.pop().unwrap(), 0);
r.push(100).unwrap();
assert_eq!(r.len(), 7);
}
#[test]
fn test_push_n_plus_1_returns_full_error() {
let r: SpscRing<u32, 4> = SpscRing::new(); r.push(1).unwrap();
r.push(2).unwrap();
r.push(3).unwrap();
assert!(r.is_full());
let e1 = r.push(4).unwrap_err();
let e2 = r.push(5).unwrap_err();
assert!(matches!(e1, StreamError::RingBufferFull { .. }));
assert!(matches!(e2, StreamError::RingBufferFull { .. }));
}
#[test]
fn test_fifo_ordering() {
let r: SpscRing<u32, 16> = SpscRing::new();
for i in 0..10u32 {
r.push(i).unwrap();
}
for i in 0..10u32 {
assert_eq!(r.pop().unwrap(), i);
}
}
#[test]
fn test_wraparound_correctness() {
let r: SpscRing<u32, 4> = SpscRing::new(); r.push(1).unwrap();
r.push(2).unwrap();
r.push(3).unwrap();
assert_eq!(r.pop().unwrap(), 1);
assert_eq!(r.pop().unwrap(), 2);
assert_eq!(r.pop().unwrap(), 3);
r.push(10).unwrap();
r.push(20).unwrap();
r.push(30).unwrap();
assert_eq!(r.pop().unwrap(), 10);
assert_eq!(r.pop().unwrap(), 20);
assert_eq!(r.pop().unwrap(), 30);
}
#[test]
fn test_wraparound_many_cycles() {
let r: SpscRing<u64, 8> = SpscRing::new(); for cycle in 0u64..20 {
for i in 0..5 {
r.push(cycle * 100 + i).unwrap();
}
for i in 0..5 {
let v = r.pop().unwrap();
assert_eq!(v, cycle * 100 + i);
}
}
}
#[test]
fn test_is_full_false_when_one_slot_free() {
let r: SpscRing<u32, 4> = SpscRing::new(); r.push(1).unwrap();
r.push(2).unwrap();
assert!(!r.is_full());
r.push(3).unwrap();
assert!(r.is_full());
}
#[test]
fn test_is_empty_after_drain() {
let r: SpscRing<u32, 4> = SpscRing::new();
r.push(1).unwrap();
r.push(2).unwrap();
r.pop().unwrap();
r.pop().unwrap();
assert!(r.is_empty());
}
#[test]
fn test_drop_drains_remaining_items() {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
let drop_count = Arc::new(AtomicUsize::new(0));
struct Counted(Arc<AtomicUsize>);
impl Drop for Counted {
fn drop(&mut self) {
self.0.fetch_add(1, Ordering::Relaxed);
}
}
let ring: SpscRing<Counted, 8> = SpscRing::new();
ring.push(Counted(Arc::clone(&drop_count))).unwrap();
ring.push(Counted(Arc::clone(&drop_count))).unwrap();
ring.push(Counted(Arc::clone(&drop_count))).unwrap();
drop(ring);
assert_eq!(drop_count.load(Ordering::Relaxed), 3);
}
#[test]
fn test_concurrent_producer_consumer() {
const ITEMS: u64 = 10_000;
let ring: SpscRing<u64, 256> = SpscRing::new();
let (prod, cons) = ring.split();
let producer = thread::spawn(move || {
let mut sent = 0u64;
while sent < ITEMS {
if prod.push(sent).is_ok() {
sent += 1;
}
}
});
let consumer = thread::spawn(move || {
let mut received = Vec::with_capacity(ITEMS as usize);
while received.len() < ITEMS as usize {
if let Ok(v) = cons.pop() {
received.push(v);
}
}
received
});
producer.join().unwrap();
let received = consumer.join().unwrap();
assert_eq!(received.len(), ITEMS as usize);
for (i, &v) in received.iter().enumerate() {
assert_eq!(v, i as u64, "FIFO ordering violated at index {i}");
}
}
#[test]
fn test_throughput_100k_round_trips() {
const ITEMS: usize = 100_000;
let ring: SpscRing<u64, 1024> = SpscRing::new();
let (prod, cons) = ring.split();
let producer = thread::spawn(move || {
let mut sent = 0usize;
while sent < ITEMS {
if prod.push(sent as u64).is_ok() {
sent += 1;
}
}
});
let consumer = thread::spawn(move || {
let mut count = 0usize;
while count < ITEMS {
if cons.pop().is_ok() {
count += 1;
}
}
count
});
producer.join().unwrap();
let count = consumer.join().unwrap();
assert_eq!(count, ITEMS);
}
#[test]
fn test_split_producer_push_consumer_pop() {
let ring: SpscRing<u32, 16> = SpscRing::new();
let (prod, cons) = ring.split();
prod.push(7).unwrap();
assert_eq!(cons.pop().unwrap(), 7);
}
#[test]
fn test_producer_is_full_matches_ring() {
let ring: SpscRing<u32, 4> = SpscRing::new();
let (prod, cons) = ring.split();
prod.push(1).unwrap();
prod.push(2).unwrap();
prod.push(3).unwrap();
assert!(prod.is_full());
cons.pop().unwrap();
assert!(!prod.is_full());
}
#[test]
fn test_consumer_len_and_is_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (prod, cons) = ring.split();
assert!(cons.is_empty());
prod.push(1).unwrap();
prod.push(2).unwrap();
assert_eq!(cons.len(), 2);
assert!(!cons.is_empty());
}
#[test]
fn test_producer_is_empty_initially_true() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (prod, _cons) = ring.split();
assert!(prod.is_empty());
}
#[test]
fn test_producer_is_empty_false_after_push() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (prod, _cons) = ring.split();
prod.push(1).unwrap();
assert!(!prod.is_empty());
}
#[test]
fn test_producer_len_matches_consumer_len() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (prod, cons) = ring.split();
assert_eq!(prod.len(), 0);
prod.push(10).unwrap();
prod.push(20).unwrap();
assert_eq!(prod.len(), 2);
assert_eq!(cons.len(), 2);
}
#[test]
fn test_minimum_power_of_two_size() {
let ring: SpscRing<u32, 2> = SpscRing::new(); assert_eq!(ring.capacity(), 1);
ring.push(99).unwrap();
assert!(ring.is_full());
assert_eq!(ring.pop().unwrap(), 99);
assert!(ring.is_empty());
}
#[test]
fn test_large_power_of_two_size() {
let ring: SpscRing<u64, 1024> = SpscRing::new();
assert_eq!(ring.capacity(), 1023);
}
#[test]
fn test_drain_iter_yields_fifo_order() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (prod, cons) = ring.split();
prod.push(1).unwrap();
prod.push(2).unwrap();
prod.push(3).unwrap();
let items: Vec<u32> = cons.into_iter_drain().collect();
assert_eq!(items, vec![1, 2, 3]);
}
#[test]
fn test_drain_iter_empty_ring_yields_nothing() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (_, cons) = ring.split();
let items: Vec<u32> = cons.into_iter_drain().collect();
assert!(items.is_empty());
}
#[test]
fn test_peek_clone_empty_returns_none() {
let r: SpscRing<u32, 8> = SpscRing::new();
assert!(r.peek_clone().is_none());
}
#[test]
fn test_peek_clone_does_not_consume() {
let r: SpscRing<u32, 8> = SpscRing::new();
r.push(42).unwrap();
assert_eq!(r.peek_clone(), Some(42));
assert_eq!(r.peek_clone(), Some(42)); assert_eq!(r.pop().unwrap(), 42);
assert!(r.is_empty());
}
#[test]
fn test_peek_clone_via_consumer() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (prod, cons) = ring.split();
prod.push(7).unwrap();
prod.push(8).unwrap();
assert_eq!(cons.peek_clone(), Some(7)); assert_eq!(cons.pop().unwrap(), 7); assert_eq!(cons.peek_clone(), Some(8)); }
proptest::proptest! {
#[test]
fn prop_fifo_ordering_with_wraparound(
batches in proptest::collection::vec(
proptest::collection::vec(0u32..=u32::MAX, 1..=7),
1..=20,
)
) {
let ring: SpscRing<u32, 8> = SpscRing::new();
let mut oracle: std::collections::VecDeque<u32> = std::collections::VecDeque::new();
for batch in &batches {
for &item in batch {
if ring.push(item).is_ok() {
oracle.push_back(item);
}
}
while let Ok(popped) = ring.pop() {
let expected = oracle.pop_front().expect("oracle must have matching item");
proptest::prop_assert_eq!(popped, expected);
}
}
}
}
#[test]
fn test_try_pop_n_empty_returns_empty_vec() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (_, consumer) = ring.split();
assert!(consumer.try_pop_n(5).is_empty());
}
#[test]
fn test_try_pop_n_bounded_by_max() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (producer, consumer) = ring.split();
for i in 0..5 {
producer.push(i).unwrap();
}
let batch = consumer.try_pop_n(3);
assert_eq!(batch.len(), 3);
assert_eq!(batch, vec![0, 1, 2]);
assert_eq!(consumer.len(), 2);
}
#[test]
fn test_try_pop_n_larger_than_available_returns_all() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (producer, consumer) = ring.split();
for i in 0..3 {
producer.push(i).unwrap();
}
let batch = consumer.try_pop_n(100);
assert_eq!(batch.len(), 3);
assert!(consumer.is_empty());
}
#[test]
fn test_producer_capacity_equals_ring_capacity() {
let ring: SpscRing<u32, 8> = SpscRing::new(); let (producer, consumer) = ring.split();
assert_eq!(producer.capacity(), 7);
assert_eq!(consumer.capacity(), 7);
}
#[test]
fn test_capacity_consistent_with_max_items() {
let ring: SpscRing<u32, 4> = SpscRing::new(); let (producer, consumer) = ring.split();
for i in 0..3 {
producer.push(i).unwrap();
}
assert_eq!(consumer.capacity(), 3);
assert_eq!(consumer.len(), 3);
}
#[test]
fn test_fill_ratio_empty_is_zero() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (_, consumer) = ring.split();
assert!((consumer.fill_ratio() - 0.0).abs() < 1e-9);
}
#[test]
fn test_fill_ratio_full_is_one() {
let ring: SpscRing<u32, 4> = SpscRing::new(); let (producer, consumer) = ring.split();
for i in 0..3 {
producer.push(i).unwrap();
}
assert!((consumer.fill_ratio() - 1.0).abs() < 1e-9);
}
#[test]
fn test_fill_ratio_partial() {
let ring: SpscRing<u32, 8> = SpscRing::new(); let (producer, consumer) = ring.split();
for i in 0..7 {
producer.push(i).unwrap();
}
consumer.pop().unwrap();
consumer.pop().unwrap();
consumer.pop().unwrap();
consumer.pop().unwrap();
let ratio = consumer.fill_ratio();
assert!((ratio - 3.0 / 7.0).abs() < 1e-9, "got {ratio}");
}
#[test]
fn test_producer_fill_ratio_empty_is_zero() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (producer, _) = ring.split();
assert!((producer.fill_ratio() - 0.0).abs() < 1e-9);
}
#[test]
fn test_producer_fill_ratio_full_is_one() {
let ring: SpscRing<u32, 4> = SpscRing::new(); let (producer, _) = ring.split();
for i in 0..3 {
producer.push(i).unwrap();
}
assert!((producer.fill_ratio() - 1.0).abs() < 1e-9);
}
#[test]
fn test_producer_and_consumer_fill_ratio_agree() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let (producer, consumer) = ring.split();
producer.push(1).unwrap();
producer.push(2).unwrap();
assert!((producer.fill_ratio() - consumer.fill_ratio()).abs() < 1e-9);
}
#[test]
fn test_peek_all_empty_returns_empty_vec() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.peek_all().is_empty());
}
#[test]
fn test_peek_all_does_not_consume() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(1).unwrap();
ring.push(2).unwrap();
ring.push(3).unwrap();
let snapshot = ring.peek_all();
assert_eq!(snapshot, vec![1, 2, 3]);
assert_eq!(ring.len(), 3);
}
#[test]
fn test_peek_all_fifo_order_after_pop() {
let ring: SpscRing<u32, 16> = SpscRing::new();
for i in 0..5u32 {
ring.push(i).unwrap();
}
ring.pop().unwrap(); let snapshot = ring.peek_all();
assert_eq!(snapshot, vec![1, 2, 3, 4]);
}
#[test]
fn test_drain_into_appends_to_buf() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10).unwrap();
ring.push(20).unwrap();
let mut buf = vec![1u32, 2];
ring.drain_into(&mut buf);
assert_eq!(buf, vec![1, 2, 10, 20]);
assert!(ring.is_empty());
}
#[test]
fn test_drain_into_empty_ring_leaves_buf_unchanged() {
let ring: SpscRing<u32, 8> = SpscRing::new();
let mut buf = vec![42u32];
ring.drain_into(&mut buf);
assert_eq!(buf, vec![42]);
}
#[test]
fn test_peek_newest_none_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.peek_newest().is_none());
}
#[test]
fn test_peek_newest_returns_last_pushed() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10).unwrap();
ring.push(20).unwrap();
ring.push(30).unwrap();
assert_eq!(ring.peek_newest(), Some(30));
}
#[test]
fn test_peek_newest_does_not_consume() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(42).unwrap();
let _ = ring.peek_newest();
assert_eq!(ring.len(), 1);
}
#[test]
fn test_fill_ratio_zero_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert_eq!(ring.fill_ratio(), 0.0);
}
#[test]
fn test_fill_ratio_one_when_full() {
let ring: SpscRing<u32, 8> = SpscRing::new(); for i in 0..7u32 {
ring.push(i).unwrap();
}
assert!((ring.fill_ratio() - 1.0).abs() < 1e-10);
}
#[test]
fn test_fill_ratio_half_when_half_full() {
let ring: SpscRing<u32, 8> = SpscRing::new(); ring.push(1).unwrap();
ring.push(2).unwrap();
ring.push(3).unwrap();
let ratio = ring.fill_ratio();
assert!((ratio - 3.0 / 7.0).abs() < 1e-10);
}
#[test]
fn test_utilization_pct_zero_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert_eq!(ring.utilization_pct(), 0.0);
}
#[test]
fn test_utilization_pct_100_when_full() {
let ring: SpscRing<u32, 8> = SpscRing::new(); for i in 0..7u32 {
ring.push(i).unwrap();
}
assert!((ring.utilization_pct() - 100.0).abs() < 1e-10);
}
#[test]
fn test_utilization_pct_equals_fill_ratio_times_100() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(1u32).unwrap();
ring.push(2u32).unwrap();
let ratio = ring.fill_ratio();
assert!((ring.utilization_pct() - ratio * 100.0).abs() < 1e-10);
}
#[test]
fn test_remaining_capacity_full_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new(); assert_eq!(ring.remaining_capacity(), 7);
}
#[test]
fn test_remaining_capacity_decreases_on_push() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(1u32).unwrap();
ring.push(2u32).unwrap();
assert_eq!(ring.remaining_capacity(), 5);
}
#[test]
fn test_remaining_capacity_zero_when_full() {
let ring: SpscRing<u32, 8> = SpscRing::new();
for i in 0..7u32 {
ring.push(i).unwrap();
}
assert_eq!(ring.remaining_capacity(), 0);
}
#[test]
fn test_is_nearly_full_false_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(!ring.is_nearly_full(0.5));
}
#[test]
fn test_is_nearly_full_true_when_at_threshold() {
let ring: SpscRing<u32, 8> = SpscRing::new(); for i in 0..7u32 {
ring.push(i).unwrap();
}
assert!(ring.is_nearly_full(0.9));
}
#[test]
fn test_is_nearly_full_false_when_below_threshold() {
let ring: SpscRing<u32, 8> = SpscRing::new(); ring.push(1u32).unwrap(); assert!(!ring.is_nearly_full(0.9));
}
#[test]
fn test_first_none_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.first().is_none());
}
#[test]
fn test_first_returns_oldest_copy() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(42u32).unwrap();
ring.push(99u32).unwrap();
assert_eq!(ring.first(), Some(42u32));
}
#[test]
fn test_first_does_not_remove() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(7u32).unwrap();
let _ = ring.first();
assert_eq!(ring.len(), 1);
}
#[test]
fn test_peek_front_none_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.peek_front().is_none());
}
#[test]
fn test_peek_front_returns_oldest_item() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
ring.push(20u32).unwrap();
assert_eq!(ring.peek_front(), Some(&10u32));
}
#[test]
fn test_peek_front_does_not_remove_item() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(42u32).unwrap();
let _ = ring.peek_front();
assert_eq!(ring.len(), 1);
}
#[test]
fn test_peek_back_none_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.peek_back().is_none());
}
#[test]
fn test_peek_back_returns_newest_item() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
ring.push(20u32).unwrap();
assert_eq!(ring.peek_back(), Some(&20u32));
}
#[test]
fn test_to_vec_cloned_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert_eq!(ring.to_vec_cloned(), Vec::<u32>::new());
}
#[test]
fn test_to_vec_cloned_preserves_fifo_order() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(1u32).unwrap();
ring.push(2u32).unwrap();
ring.push(3u32).unwrap();
assert_eq!(ring.to_vec_cloned(), vec![1u32, 2, 3]);
}
#[test]
fn test_to_vec_cloned_does_not_drain() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(42u32).unwrap();
let _ = ring.to_vec_cloned();
assert_eq!(ring.len(), 1);
}
#[test]
fn test_min_cloned_none_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.min_cloned().is_none());
}
#[test]
fn test_min_cloned_returns_minimum() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(3u32).unwrap();
ring.push(1u32).unwrap();
ring.push(4u32).unwrap();
ring.push(2u32).unwrap();
assert_eq!(ring.min_cloned(), Some(1u32));
}
#[test]
fn test_min_cloned_does_not_drain() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
let _ = ring.min_cloned();
assert_eq!(ring.len(), 1);
}
#[test]
fn test_max_cloned_none_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.max_cloned().is_none());
}
#[test]
fn test_max_cloned_returns_maximum() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(3u32).unwrap();
ring.push(1u32).unwrap();
ring.push(4u32).unwrap();
ring.push(2u32).unwrap();
assert_eq!(ring.max_cloned(), Some(4u32));
}
#[test]
fn test_max_cloned_does_not_drain() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
let _ = ring.max_cloned();
assert_eq!(ring.len(), 1);
}
#[test]
fn test_count_if_zero_when_empty() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert_eq!(ring.count_if(|_| true), 0);
}
#[test]
fn test_count_if_counts_matching_items() {
let ring: SpscRing<u32, 8> = SpscRing::new();
for i in 1u32..=6 {
ring.push(i).unwrap();
}
assert_eq!(ring.count_if(|x| x % 2 == 0), 3);
}
#[test]
fn test_count_if_all_match() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
ring.push(20u32).unwrap();
assert_eq!(ring.count_if(|_| true), 2);
}
#[test]
fn test_has_capacity_true_on_empty_ring() {
let ring: SpscRing<u32, 8> = SpscRing::new(); assert!(ring.has_capacity(7));
}
#[test]
fn test_has_capacity_false_when_full() {
let ring: SpscRing<u32, 8> = SpscRing::new();
for i in 0..7u32 {
ring.push(i).unwrap();
}
assert!(!ring.has_capacity(1));
}
#[test]
fn test_has_capacity_false_for_zero_capacity_needed() {
let ring: SpscRing<u32, 4> = SpscRing::new();
assert!(ring.has_capacity(0));
}
#[test]
fn test_has_capacity_partial_fill() {
let ring: SpscRing<u32, 8> = SpscRing::new(); ring.push(1u32).unwrap();
ring.push(2u32).unwrap();
assert!(ring.has_capacity(5));
assert!(!ring.has_capacity(6));
}
#[test]
fn test_is_empty_true_for_new_ring() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.is_empty());
}
#[test]
fn test_is_empty_false_after_push() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(42u32).unwrap();
assert!(!ring.is_empty());
}
#[test]
fn test_is_empty_true_after_push_and_pop() {
let ring: SpscRing<u32, 4> = SpscRing::new();
ring.push(1u32).unwrap();
let _ = ring.pop();
assert!(ring.is_empty());
}
#[test]
fn test_peek_oldest_none_on_empty_ring() {
let ring: SpscRing<u32, 8> = SpscRing::new();
assert!(ring.peek_oldest().is_none());
}
#[test]
fn test_peek_oldest_returns_first_pushed_item() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
ring.push(20u32).unwrap();
ring.push(30u32).unwrap();
assert_eq!(ring.peek_oldest(), Some(10));
}
#[test]
fn test_peek_oldest_does_not_remove_item() {
let ring: SpscRing<u32, 4> = SpscRing::new();
ring.push(5u32).unwrap();
let _ = ring.peek_oldest();
assert_eq!(ring.pop().unwrap(), 5);
}
#[test]
fn test_peek_oldest_different_from_peek_newest_when_multiple_items() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(1u32).unwrap();
ring.push(2u32).unwrap();
ring.push(3u32).unwrap();
assert_eq!(ring.peek_oldest(), Some(1));
assert_eq!(ring.peek_newest(), Some(3));
}
#[test]
fn test_sum_cloned_empty_returns_default() {
let ring: SpscRing<u32, 4> = SpscRing::new();
assert_eq!(ring.sum_cloned(), 0u32);
}
#[test]
fn test_sum_cloned_single_element() {
let ring: SpscRing<u32, 4> = SpscRing::new();
ring.push(42u32).unwrap();
assert_eq!(ring.sum_cloned(), 42u32);
}
#[test]
fn test_sum_cloned_multiple_elements() {
let ring: SpscRing<u32, 8> = SpscRing::new();
for v in [1u32, 2, 3, 4, 5] { ring.push(v).unwrap(); }
assert_eq!(ring.sum_cloned(), 15u32);
}
#[test]
fn test_sum_cloned_after_pop_reflects_remaining() {
let ring: SpscRing<u32, 4> = SpscRing::new();
ring.push(10u32).unwrap();
ring.push(20u32).unwrap();
ring.pop().unwrap(); assert_eq!(ring.sum_cloned(), 20u32);
}
#[test]
fn test_average_cloned_none_when_empty() {
let ring: SpscRing<f64, 4> = SpscRing::new();
assert!(ring.average_cloned().is_none());
}
#[test]
fn test_average_cloned_single_element() {
let ring: SpscRing<f64, 4> = SpscRing::new();
ring.push(6.0f64).unwrap();
assert_eq!(ring.average_cloned(), Some(6.0));
}
#[test]
fn test_average_cloned_multiple_elements() {
let ring: SpscRing<f64, 8> = SpscRing::new();
for v in [2.0f64, 4.0, 6.0, 8.0] { ring.push(v).unwrap(); }
assert_eq!(ring.average_cloned(), Some(5.0));
}
#[test]
fn test_peek_nth_returns_oldest_at_index_zero() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
ring.push(20u32).unwrap();
ring.push(30u32).unwrap();
assert_eq!(ring.peek_nth(0), Some(10));
}
#[test]
fn test_peek_nth_returns_correct_element() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
ring.push(20u32).unwrap();
ring.push(30u32).unwrap();
assert_eq!(ring.peek_nth(1), Some(20));
assert_eq!(ring.peek_nth(2), Some(30));
}
#[test]
fn test_peek_nth_returns_none_when_out_of_bounds() {
let ring: SpscRing<u32, 4> = SpscRing::new();
ring.push(5u32).unwrap();
assert!(ring.peek_nth(1).is_none());
}
#[test]
fn test_contains_cloned_false_when_empty() {
let ring: SpscRing<u32, 4> = SpscRing::new();
assert!(!ring.contains_cloned(&42u32));
}
#[test]
fn test_contains_cloned_true_when_value_present() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(10u32).unwrap();
ring.push(20u32).unwrap();
assert!(ring.contains_cloned(&10u32));
assert!(ring.contains_cloned(&20u32));
}
#[test]
fn test_contains_cloned_false_when_value_absent() {
let ring: SpscRing<u32, 4> = SpscRing::new();
ring.push(5u32).unwrap();
assert!(!ring.contains_cloned(&99u32));
}
#[test]
fn test_max_cloned_by_none_when_empty() {
let ring: SpscRing<u32, 4> = SpscRing::new();
assert!(ring.max_cloned_by(|&x| x).is_none());
}
#[test]
fn test_max_cloned_by_returns_max_element() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(3u32).unwrap();
ring.push(1u32).unwrap();
ring.push(7u32).unwrap();
ring.push(2u32).unwrap();
assert_eq!(ring.max_cloned_by(|&x| x), Some(7));
}
#[test]
fn test_max_cloned_by_custom_key() {
let ring: SpscRing<i32, 8> = SpscRing::new();
ring.push(-5i32).unwrap();
ring.push(3i32).unwrap();
ring.push(-10i32).unwrap();
assert_eq!(ring.max_cloned_by(|&x| x.abs()), Some(-10));
}
#[test]
fn test_min_cloned_by_none_when_empty() {
let ring: SpscRing<u32, 4> = SpscRing::new();
assert!(ring.min_cloned_by(|&x| x).is_none());
}
#[test]
fn test_min_cloned_by_returns_min_element() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(3u32).unwrap();
ring.push(1u32).unwrap();
ring.push(7u32).unwrap();
assert_eq!(ring.min_cloned_by(|&x| x), Some(1));
}
#[test]
fn test_min_cloned_by_custom_key() {
let ring: SpscRing<i32, 8> = SpscRing::new();
ring.push(-5i32).unwrap();
ring.push(3i32).unwrap();
ring.push(-1i32).unwrap();
assert_eq!(ring.min_cloned_by(|&x| x.abs()), Some(-1));
}
#[test]
fn test_to_vec_sorted_empty() {
let ring: SpscRing<u32, 4> = SpscRing::new();
assert_eq!(ring.to_vec_sorted(), Vec::<u32>::new());
}
#[test]
fn test_to_vec_sorted_returns_sorted_elements() {
let ring: SpscRing<u32, 8> = SpscRing::new();
ring.push(5u32).unwrap();
ring.push(1u32).unwrap();
ring.push(3u32).unwrap();
assert_eq!(ring.to_vec_sorted(), vec![1u32, 3, 5]);
}
}