audio_graph_bsd/
ring.rs

1//! Lock-free ring-buffer bridges between worker threads and the RT graph.
2//!
3//! [`RingSource`] feeds audio **into** the graph from a worker producer, and
4//! [`RingSink`] taps audio **out of** the graph for a worker consumer. Both
5//! wrap an [`rtrb`] ring, whose `pop`/`push` operations are wait-free and
6//! therefore safe to touch from the real-time thread — subject to the caveat
7//! documented on [`RingSink`] (the `push` path clones, so it is deferred to a
8//! non-RT [`RingSink::flush`]).
9
10use audio_core_bsd::{
11    AudioFrame, AudioNode, PortDescriptor, PortDirection, ProcessContext, SampleFormat,
12};
13
14/// A source node that drains audio from a worker thread via a lock-free ring.
15///
16/// `RingSource` holds the `Consumer` end of an [`rtrb`] ring. On every cycle it
17/// pops at most one [`AudioFrame`] (wait-free) and writes it into its single
18/// output port. If the ring is empty it repeats the last frame received, so the
19/// downstream signal holds rather than clicking to silence.
20///
21/// # Real-time safety
22///
23/// `process` performs only a wait-free `pop` and bounded slice copies — no
24/// allocation, lock, panic, or syscall. The `last` cache is pre-sized in
25/// [`RingSource::new`] so mirroring an incoming frame never grows it; if an
26/// incoming frame is shorter/longer than `last` the copy is bounded to the
27/// smaller length (no realloc).
28pub struct RingSource {
29    /// Consumer end of the lock-free ring (wait-free `pop`).
30    consumer: rtrb::Consumer<AudioFrame>,
31    /// The single output port descriptor.
32    out_port: [PortDescriptor; 1],
33    /// Last frame received; reused when the ring is empty to avoid clicks.
34    last: AudioFrame,
35}
36
37impl RingSource {
38    /// Creates a new source draining `consumer`.
39    ///
40    /// `last` is pre-allocated to silence of the given size so that cycles
41    /// before the first frame arrives emit silence rather than uninitialised
42    /// data.
43    #[must_use]
44    pub fn new(
45        consumer: rtrb::Consumer<AudioFrame>,
46        channels: u16,
47        sample_rate: u32,
48        num_frames: usize,
49    ) -> Self {
50        Self {
51            consumer,
52            out_port: [PortDescriptor::new(
53                PortDirection::Output,
54                channels,
55                SampleFormat::F32,
56            )],
57            last: AudioFrame::silence(channels, num_frames, sample_rate),
58        }
59    }
60
61    /// Returns a shared reference to the underlying consumer (for diagnostics).
62    #[must_use]
63    pub fn consumer(&self) -> &rtrb::Consumer<AudioFrame> {
64        &self.consumer
65    }
66
67    /// Returns a shared reference to the last received frame.
68    #[must_use]
69    pub fn last(&self) -> &AudioFrame {
70        &self.last
71    }
72}
73
74impl AudioNode for RingSource {
75    fn inputs(&self) -> &[PortDescriptor] {
76        &[]
77    }
78    fn outputs(&self) -> &[PortDescriptor] {
79        &self.out_port
80    }
81    fn process(
82        &mut self,
83        _ctx: &mut ProcessContext,
84        _in_frames: &[AudioFrame],
85        out_frames: &mut [AudioFrame],
86    ) {
87        let Some(out) = out_frames.get_mut(0) else {
88            return;
89        };
90        // Wait-free pop. On success, copy into both the output and `last`;
91        // on failure, reuse `last`.
92        if let Ok(frame) = self.consumer.pop() {
93            out.channels = frame.channels;
94            out.sample_rate = frame.sample_rate;
95            let n = frame.samples.len().min(out.samples.len());
96            out.samples[..n].copy_from_slice(&frame.samples[..n]);
97
98            // Mirror into the pre-sized `last` cache (bounded, alloc-free).
99            self.last.channels = frame.channels;
100            self.last.sample_rate = frame.sample_rate;
101            let cn = frame.samples.len().min(self.last.samples.len());
102            self.last.samples[..cn].copy_from_slice(&frame.samples[..cn]);
103        } else {
104            // Repeat the last frame: bounded copy, no alloc.
105            out.channels = self.last.channels;
106            out.sample_rate = self.last.sample_rate;
107            let n = self.last.samples.len().min(out.samples.len());
108            out.samples[..n].copy_from_slice(&self.last.samples[..n]);
109        }
110    }
111}
112
113/// A sink node that stashes audio from the graph for a worker thread to drain.
114///
115/// `RingSink` holds the `Producer` end of an [`rtrb`] ring plus a stash frame.
116/// On every cycle it copies its single input into the stash (bounded, RT-safe).
117/// The stash is **not** pushed to the ring inside `process`, because
118/// [`rtrb::Producer::push`] takes ownership and would force a heap clone of the
119/// frame — that clone must happen off the real-time thread.
120///
121/// Call [`RingSink::flush`](Self::flush) from a worker thread to ship the
122/// stashed frame across the ring to its consumer.
123///
124/// # Real-time safety
125///
126/// `process` performs only a bounded copy into the pre-sized `stash` — no
127/// allocation, lock, panic, or syscall. The cloning `push` is deferred to the
128/// non-RT [`RingSink::flush`].
129pub struct RingSink {
130    /// Producer end of the lock-free ring (pushed from a worker thread).
131    producer: rtrb::Producer<AudioFrame>,
132    /// The single input port descriptor.
133    in_port: [PortDescriptor; 1],
134    /// Stashed copy of the most recent input; drained by `flush`.
135    stash: AudioFrame,
136}
137
138impl RingSink {
139    /// Creates a new sink feeding `producer`.
140    ///
141    /// `stash` is pre-allocated to silence of the given size so the RT copy
142    /// never grows it.
143    #[must_use]
144    pub fn new(
145        producer: rtrb::Producer<AudioFrame>,
146        channels: u16,
147        sample_rate: u32,
148        num_frames: usize,
149    ) -> Self {
150        Self {
151            producer,
152            in_port: [PortDescriptor::new(
153                PortDirection::Input,
154                channels,
155                SampleFormat::F32,
156            )],
157            stash: AudioFrame::silence(channels, num_frames, sample_rate),
158        }
159    }
160
161    /// Pushes the stashed frame into the ring.
162    ///
163    /// This **clones** the stash (allocating) and is therefore intended for the
164    /// worker thread, **not** the real-time thread. The stash itself is left
165    /// in place so the next cycle can overwrite it.
166    ///
167    /// # Errors
168    ///
169    /// Returns [`rtrb::PushError`] containing the unpushed frame when the ring
170    /// has no free slot (treat as an xrun / dropped frame).
171    pub fn flush(&mut self) -> Result<(), rtrb::PushError<AudioFrame>> {
172        self.producer.push(self.stash.clone())
173    }
174
175    /// Returns a shared reference to the underlying producer (for diagnostics).
176    #[must_use]
177    pub fn producer(&self) -> &rtrb::Producer<AudioFrame> {
178        &self.producer
179    }
180
181    /// Returns a shared reference to the stashed frame.
182    #[must_use]
183    pub fn stash(&self) -> &AudioFrame {
184        &self.stash
185    }
186}
187
188impl AudioNode for RingSink {
189    fn inputs(&self) -> &[PortDescriptor] {
190        &self.in_port
191    }
192    fn outputs(&self) -> &[PortDescriptor] {
193        &[]
194    }
195    fn process(
196        &mut self,
197        _ctx: &mut ProcessContext,
198        in_frames: &[AudioFrame],
199        _out_frames: &mut [AudioFrame],
200    ) {
201        let Some(inp) = in_frames.first() else {
202            return;
203        };
204        // Bounded copy into the pre-sized stash — no allocation.
205        self.stash.channels = inp.channels;
206        self.stash.sample_rate = inp.sample_rate;
207        let n = inp.samples.len().min(self.stash.samples.len());
208        self.stash.samples[..n].copy_from_slice(&inp.samples[..n]);
209    }
210}
211
212#[cfg(test)]
213mod tests {
214    use super::*;
215    use audio_core_bsd::{AudioNode, PortDirection, ProcessContext};
216
217    /// Approximate float equality (uses `<`, never `==`).
218    fn approx_eq(a: f32, b: f32) -> bool {
219        (a - b).abs() < 1e-6
220    }
221
222    #[test]
223    fn ring_source_outputs_one_port_and_no_inputs() {
224        let (_producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
225        let src = RingSource::new(consumer, 1, 48_000, 8);
226        assert_eq!(src.inputs().len(), 0);
227        assert_eq!(src.outputs().len(), 1);
228        assert_eq!(src.outputs()[0].direction, PortDirection::Output);
229        assert_eq!(src.outputs()[0].channels, 1);
230        assert_eq!(src.outputs()[0].sample_format, SampleFormat::F32);
231    }
232
233    #[test]
234    fn ring_sink_inputs_one_port_and_no_outputs() {
235        let (producer, _consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
236        let sink = RingSink::new(producer, 2, 48_000, 8);
237        assert_eq!(sink.inputs().len(), 1);
238        assert_eq!(sink.outputs().len(), 0);
239        assert_eq!(sink.inputs()[0].channels, 2);
240        assert_eq!(sink.inputs()[0].direction, PortDirection::Input);
241    }
242
243    #[test]
244    fn ring_source_pops_and_outputs_frame() {
245        let (mut producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
246        producer
247            .push(AudioFrame::from_planar(1, 48_000, vec![0.5; 8]))
248            .unwrap();
249
250        let mut src = RingSource::new(consumer, 1, 48_000, 8);
251        let mut ctx = ProcessContext::new(8, 0, 48_000);
252        let mut out = [AudioFrame::silence(1, 8, 48_000)];
253        src.process(&mut ctx, &[], &mut out);
254        assert!(out[0].samples.iter().all(|&s| approx_eq(s, 0.5)));
255        // last should also mirror the popped frame.
256        assert!(src.last().samples.iter().all(|&s| approx_eq(s, 0.5)));
257    }
258
259    #[test]
260    fn ring_source_repeats_last_when_empty() {
261        let (_producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
262        let mut src = RingSource::new(consumer, 1, 48_000, 8);
263        let mut ctx = ProcessContext::new(8, 0, 48_000);
264        let mut out = [AudioFrame::silence(1, 8, 48_000)];
265        // Fill out with a sentinel to detect overwrite.
266        for s in &mut out[0].samples {
267            *s = 9.0;
268        }
269        // No frame available -> repeats last (silence) -> out becomes 0.0.
270        src.process(&mut ctx, &[], &mut out);
271        assert!(out[0].samples.iter().all(|&s| approx_eq(s, 0.0)));
272    }
273
274    #[test]
275    fn ring_source_holds_after_underrun() {
276        // After receiving one non-silence frame, an underrun must hold it.
277        let (mut producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
278        producer
279            .push(AudioFrame::from_planar(1, 48_000, vec![0.7; 4]))
280            .unwrap();
281        let mut src = RingSource::new(consumer, 1, 48_000, 4);
282        let mut ctx = ProcessContext::new(4, 0, 48_000);
283        let mut out = [AudioFrame::silence(1, 4, 48_000)];
284        src.process(&mut ctx, &[], &mut out); // pops 0.7
285        assert!(out[0].samples.iter().all(|&s| approx_eq(s, 0.7)));
286        // Now underrun: must hold 0.7, not collapse to silence.
287        src.process(&mut ctx, &[], &mut out);
288        assert!(out[0].samples.iter().all(|&s| approx_eq(s, 0.7)));
289    }
290
291    #[test]
292    fn ring_sink_stashes_input() {
293        let (producer, _consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
294        let mut sink = RingSink::new(producer, 1, 48_000, 8);
295        let inp = [AudioFrame::from_planar(1, 48_000, vec![0.25; 8])];
296        let mut ctx = ProcessContext::new(8, 0, 48_000);
297        let mut out_dummy: [AudioFrame; 0] = [];
298        sink.process(&mut ctx, &inp, &mut out_dummy);
299        assert!(sink.stash().samples.iter().all(|&s| approx_eq(s, 0.25)));
300    }
301
302    #[test]
303    fn ring_sink_flush_pushes_to_consumer() {
304        let (producer, mut consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
305        let mut sink = RingSink::new(producer, 1, 48_000, 4);
306        let inp = [AudioFrame::from_planar(1, 48_000, vec![1.0; 4])];
307        let mut ctx = ProcessContext::new(4, 0, 48_000);
308        let mut out_dummy: [AudioFrame; 0] = [];
309        sink.process(&mut ctx, &inp, &mut out_dummy);
310        sink.flush().unwrap();
311        let popped = consumer.pop().unwrap();
312        assert!(popped.samples.iter().all(|&s| approx_eq(s, 1.0)));
313    }
314
315    #[test]
316    fn ring_sink_flush_full_ring_errors() {
317        let (producer, _consumer) = rtrb::RingBuffer::<AudioFrame>::new(1);
318        let mut sink = RingSink::new(producer, 1, 48_000, 2);
319        // Fill the ring (capacity 1).
320        sink.flush().unwrap();
321        // Second flush must fail with PushError::Full.
322        assert!(sink.flush().is_err());
323    }
324
325    #[test]
326    fn ring_source_last_starts_silent() {
327        let (_producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(2);
328        let src = RingSource::new(consumer, 1, 48_000, 4);
329        assert_eq!(src.last().channels, 1);
330        assert_eq!(src.last().sample_rate, 48_000);
331        assert!(src.last().samples.iter().all(|&s| approx_eq(s, 0.0)));
332    }
333
334    #[test]
335    fn ring_sink_stash_starts_silent() {
336        let (producer, _consumer) = rtrb::RingBuffer::<AudioFrame>::new(2);
337        let sink = RingSink::new(producer, 2, 44_100, 4);
338        assert_eq!(sink.stash().channels, 2);
339        assert_eq!(sink.stash().sample_rate, 44_100);
340        assert_eq!(sink.stash().samples.len(), 8);
341        assert!(sink.stash().samples.iter().all(|&s| approx_eq(s, 0.0)));
342    }
343}
audio_graph_bsd/ring.rs

audio_graph_bsd/
ring.rs
