subms_spsc_ring_buffer/features/
bulk.rs

1//! Bulk transfer extensions on the base SPSC ring.
2//!
3//! `try_enqueue_bulk(&[T])` and `try_dequeue_bulk(&mut [T])` amortise the
4//! cost of the per-item atomic load + cache check by issuing exactly one
5//! acquire-load (only when the cached opposite-index is exhausted) and
6//! exactly one release-store per call. Items are still committed in the
7//! same order; consumers observing the new tail see all items that
8//! preceded it.
9//!
10//! Returns the number of items actually transferred. A return < `slice.len()`
11//! means the ring couldn't take the rest right now (full / empty); the
12//! caller decides whether to back off.
13
14use std::sync::atomic::Ordering;
15
16use crate::{Consumer, Producer};
17
18impl<T: Copy> Producer<T> {
19    /// Copy as many items from `values` into the ring as will fit right now.
20    /// Returns the count transferred. Single Release on the tail at the end.
21    pub fn try_enqueue_bulk(&mut self, values: &[T]) -> usize {
22        if values.is_empty() {
23            return 0;
24        }
25        let tail = self.inner.tail.0.load(Ordering::Relaxed);
26        let cap = self.inner.capacity;
27
28        let mut free = cap - tail.wrapping_sub(self.cached_head);
29        if free < values.len() {
30            // Cache says we can't take them all; re-read the real head once
31            // and recompute. One Acquire, not one per item.
32            self.cached_head = self.inner.head.0.load(Ordering::Acquire);
33            free = cap - tail.wrapping_sub(self.cached_head);
34        }
35        let n = free.min(values.len());
36        if n == 0 {
37            return 0;
38        }
39        for (i, v) in values.iter().take(n).enumerate() {
40            unsafe {
41                (*self.inner.buf[(tail.wrapping_add(i)) & self.inner.mask].get()).write(*v);
42            }
43        }
44        // Single Release publishes all `n` slots at once.
45        self.inner
46            .tail
47            .0
48            .store(tail.wrapping_add(n), Ordering::Release);
49        n
50    }
51}
52
53impl<T: Copy> Consumer<T> {
54    /// Drain up to `out.len()` items into `out`. Returns the count drained.
55    /// Single Release on the head at the end.
56    pub fn try_dequeue_bulk(&mut self, out: &mut [T]) -> usize {
57        if out.is_empty() {
58            return 0;
59        }
60        let head = self.inner.head.0.load(Ordering::Relaxed);
61
62        let mut avail = self.cached_tail.wrapping_sub(head);
63        if avail < out.len() {
64            self.cached_tail = self.inner.tail.0.load(Ordering::Acquire);
65            avail = self.cached_tail.wrapping_sub(head);
66        }
67        let n = avail.min(out.len());
68        if n == 0 {
69            return 0;
70        }
71        for (i, slot) in out.iter_mut().take(n).enumerate() {
72            *slot = unsafe {
73                (*self.inner.buf[(head.wrapping_add(i)) & self.inner.mask].get()).assume_init_read()
74            };
75        }
76        self.inner
77            .head
78            .0
79            .store(head.wrapping_add(n), Ordering::Release);
80        n
81    }
82}
83
84#[cfg(test)]
85#[path = "bulk_tests.rs"]
86mod tests;
subms_spsc_ring_buffer/features/bulk.rs

subms_spsc_ring_buffer/features/
bulk.rs
