vector_calls/
ring.rs

1//! A single-producer, single-consumer sample ring the audio callbacks can touch:
2//! no lock, no allocation, a dropped sample on overflow rather than a stall.
3
4use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
5
6pub struct SpscRing {
7    buf: Box<[AtomicU32]>,
8    mask: usize,
9    /// Next slot to write; only the producer advances it.
10    tail: AtomicUsize,
11    /// Next slot to read; only the consumer advances it.
12    head: AtomicUsize,
13}
14
15impl SpscRing {
16    /// Capacity rounds up to a power of two.
17    pub fn new(min_capacity: usize) -> Self {
18        let cap = min_capacity.next_power_of_two().max(2);
19        let buf = (0..cap).map(|_| AtomicU32::new(0)).collect::<Vec<_>>().into_boxed_slice();
20        Self { buf, mask: cap - 1, tail: AtomicUsize::new(0), head: AtomicUsize::new(0) }
21    }
22
23    pub fn capacity(&self) -> usize {
24        self.buf.len()
25    }
26
27    /// Samples waiting to be read.
28    pub fn len(&self) -> usize {
29        self.tail.load(Ordering::Acquire).wrapping_sub(self.head.load(Ordering::Acquire))
30    }
31
32    /// Writes what fits and returns how many samples were taken.
33    pub fn push(&self, samples: &[f32]) -> usize {
34        let tail = self.tail.load(Ordering::Relaxed);
35        let head = self.head.load(Ordering::Acquire);
36        let free = self.capacity() - tail.wrapping_sub(head);
37        let n = samples.len().min(free);
38        for (i, s) in samples[..n].iter().enumerate() {
39            self.buf[(tail.wrapping_add(i)) & self.mask].store(s.to_bits(), Ordering::Relaxed);
40        }
41        self.tail.store(tail.wrapping_add(n), Ordering::Release);
42        n
43    }
44
45    /// Reads up to `out.len()` samples and returns how many were filled.
46    pub fn pop(&self, out: &mut [f32]) -> usize {
47        let head = self.head.load(Ordering::Relaxed);
48        let tail = self.tail.load(Ordering::Acquire);
49        let n = out.len().min(tail.wrapping_sub(head));
50        for (i, o) in out[..n].iter_mut().enumerate() {
51            *o = f32::from_bits(self.buf[(head.wrapping_add(i)) & self.mask].load(Ordering::Relaxed));
52        }
53        self.head.store(head.wrapping_add(n), Ordering::Release);
54        n
55    }
56
57    /// Consumer side: throws away the oldest `n` samples.
58    pub fn skip(&self, n: usize) {
59        let head = self.head.load(Ordering::Relaxed);
60        let tail = self.tail.load(Ordering::Acquire);
61        let n = n.min(tail.wrapping_sub(head));
62        self.head.store(head.wrapping_add(n), Ordering::Release);
63    }
64}
65
66#[cfg(test)]
67mod tests {
68    use super::*;
69
70    #[test]
71    fn samples_come_out_in_order_and_overflow_drops_the_newest() {
72        let ring = SpscRing::new(8);
73        assert_eq!(ring.push(&[1.0, 2.0, 3.0]), 3);
74        let mut out = [0.0; 2];
75        assert_eq!(ring.pop(&mut out), 2);
76        assert_eq!(out, [1.0, 2.0]);
77        let ten: Vec<f32> = (0..10).map(|i| i as f32).collect();
78        assert_eq!(ring.push(&ten), 7);
79        assert_eq!(ring.len(), 8);
80        ring.skip(1);
81        let mut rest = [0.0; 8];
82        assert_eq!(ring.pop(&mut rest), 7);
83        assert_eq!(&rest[..7], &[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0]);
84    }
85}
vector_calls/ring.rs

vector_calls/
ring.rs
