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 crate::Flushable for RingSink {
189    /// Pushes the stashed frame into the ring (OFF-RT ONLY — clones/allocates).
190    ///
191    /// This is the [`Flushable`](crate::Flushable) contract the graph engine
192    /// calls in the between-cycle window via
193    /// [`Graph::flush_sinks`](crate::Graph::flush_sinks). It is equivalent to
194    /// the inherent [`RingSink::flush`] but returns the graph-level
195    /// [`FlushError`](crate::FlushError).
196    fn flush(&mut self) -> Result<(), crate::FlushError> {
197        self.producer
198            .push(self.stash.clone())
199            .map_err(|_| crate::FlushError::RingFull(1))
200    }
201}
202
203impl AudioNode for RingSink {
204    fn inputs(&self) -> &[PortDescriptor] {
205        &self.in_port
206    }
207    fn outputs(&self) -> &[PortDescriptor] {
208        &[]
209    }
210    fn process(
211        &mut self,
212        _ctx: &mut ProcessContext,
213        in_frames: &[AudioFrame],
214        _out_frames: &mut [AudioFrame],
215    ) {
216        let Some(inp) = in_frames.first() else {
217            return;
218        };
219        // Bounded copy into the pre-sized stash — no allocation.
220        self.stash.channels = inp.channels;
221        self.stash.sample_rate = inp.sample_rate;
222        let n = inp.samples.len().min(self.stash.samples.len());
223        self.stash.samples[..n].copy_from_slice(&inp.samples[..n]);
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use super::*;
230    use audio_core_bsd::{AudioNode, PortDirection, ProcessContext};
231
232    /// Approximate float equality (uses `<`, never `==`).
233    fn approx_eq(a: f32, b: f32) -> bool {
234        (a - b).abs() < 1e-6
235    }
236
237    #[test]
238    fn ring_source_outputs_one_port_and_no_inputs() {
239        let (_producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
240        let src = RingSource::new(consumer, 1, 48_000, 8);
241        assert_eq!(src.inputs().len(), 0);
242        assert_eq!(src.outputs().len(), 1);
243        assert_eq!(src.outputs()[0].direction, PortDirection::Output);
244        assert_eq!(src.outputs()[0].channels, 1);
245        assert_eq!(src.outputs()[0].sample_format, SampleFormat::F32);
246    }
247
248    #[test]
249    fn ring_sink_inputs_one_port_and_no_outputs() {
250        let (producer, _consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
251        let sink = RingSink::new(producer, 2, 48_000, 8);
252        assert_eq!(sink.inputs().len(), 1);
253        assert_eq!(sink.outputs().len(), 0);
254        assert_eq!(sink.inputs()[0].channels, 2);
255        assert_eq!(sink.inputs()[0].direction, PortDirection::Input);
256    }
257
258    #[test]
259    fn ring_source_pops_and_outputs_frame() {
260        let (mut producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
261        producer
262            .push(AudioFrame::from_planar(1, 48_000, vec![0.5; 8]))
263            .unwrap();
264
265        let mut src = RingSource::new(consumer, 1, 48_000, 8);
266        let mut ctx = ProcessContext::new(8, 0, 48_000);
267        let mut out = [AudioFrame::silence(1, 8, 48_000)];
268        src.process(&mut ctx, &[], &mut out);
269        assert!(out[0].samples.iter().all(|&s| approx_eq(s, 0.5)));
270        // last should also mirror the popped frame.
271        assert!(src.last().samples.iter().all(|&s| approx_eq(s, 0.5)));
272    }
273
274    #[test]
275    fn ring_source_repeats_last_when_empty() {
276        let (_producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
277        let mut src = RingSource::new(consumer, 1, 48_000, 8);
278        let mut ctx = ProcessContext::new(8, 0, 48_000);
279        let mut out = [AudioFrame::silence(1, 8, 48_000)];
280        // Fill out with a sentinel to detect overwrite.
281        for s in &mut out[0].samples {
282            *s = 9.0;
283        }
284        // No frame available -> repeats last (silence) -> out becomes 0.0.
285        src.process(&mut ctx, &[], &mut out);
286        assert!(out[0].samples.iter().all(|&s| approx_eq(s, 0.0)));
287    }
288
289    #[test]
290    fn ring_source_holds_after_underrun() {
291        // After receiving one non-silence frame, an underrun must hold it.
292        let (mut producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
293        producer
294            .push(AudioFrame::from_planar(1, 48_000, vec![0.7; 4]))
295            .unwrap();
296        let mut src = RingSource::new(consumer, 1, 48_000, 4);
297        let mut ctx = ProcessContext::new(4, 0, 48_000);
298        let mut out = [AudioFrame::silence(1, 4, 48_000)];
299        src.process(&mut ctx, &[], &mut out); // pops 0.7
300        assert!(out[0].samples.iter().all(|&s| approx_eq(s, 0.7)));
301        // Now underrun: must hold 0.7, not collapse to silence.
302        src.process(&mut ctx, &[], &mut out);
303        assert!(out[0].samples.iter().all(|&s| approx_eq(s, 0.7)));
304    }
305
306    #[test]
307    fn ring_sink_stashes_input() {
308        let (producer, _consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
309        let mut sink = RingSink::new(producer, 1, 48_000, 8);
310        let inp = [AudioFrame::from_planar(1, 48_000, vec![0.25; 8])];
311        let mut ctx = ProcessContext::new(8, 0, 48_000);
312        let mut out_dummy: [AudioFrame; 0] = [];
313        sink.process(&mut ctx, &inp, &mut out_dummy);
314        assert!(sink.stash().samples.iter().all(|&s| approx_eq(s, 0.25)));
315    }
316
317    #[test]
318    fn ring_sink_flush_pushes_to_consumer() {
319        let (producer, mut consumer) = rtrb::RingBuffer::<AudioFrame>::new(4);
320        let mut sink = RingSink::new(producer, 1, 48_000, 4);
321        let inp = [AudioFrame::from_planar(1, 48_000, vec![1.0; 4])];
322        let mut ctx = ProcessContext::new(4, 0, 48_000);
323        let mut out_dummy: [AudioFrame; 0] = [];
324        sink.process(&mut ctx, &inp, &mut out_dummy);
325        sink.flush().unwrap();
326        let popped = consumer.pop().unwrap();
327        assert!(popped.samples.iter().all(|&s| approx_eq(s, 1.0)));
328    }
329
330    #[test]
331    fn ring_sink_flush_full_ring_errors() {
332        let (producer, _consumer) = rtrb::RingBuffer::<AudioFrame>::new(1);
333        let mut sink = RingSink::new(producer, 1, 48_000, 2);
334        // Fill the ring (capacity 1).
335        sink.flush().unwrap();
336        // Second flush must fail with PushError::Full.
337        assert!(sink.flush().is_err());
338    }
339
340    #[test]
341    fn ring_source_last_starts_silent() {
342        let (_producer, consumer) = rtrb::RingBuffer::<AudioFrame>::new(2);
343        let src = RingSource::new(consumer, 1, 48_000, 4);
344        assert_eq!(src.last().channels, 1);
345        assert_eq!(src.last().sample_rate, 48_000);
346        assert!(src.last().samples.iter().all(|&s| approx_eq(s, 0.0)));
347    }
348
349    #[test]
350    fn ring_sink_stash_starts_silent() {
351        let (producer, _consumer) = rtrb::RingBuffer::<AudioFrame>::new(2);
352        let sink = RingSink::new(producer, 2, 44_100, 4);
353        assert_eq!(sink.stash().channels, 2);
354        assert_eq!(sink.stash().sample_rate, 44_100);
355        assert_eq!(sink.stash().samples.len(), 8);
356        assert!(sink.stash().samples.iter().all(|&s| approx_eq(s, 0.0)));
357    }
358}
audio_graph_bsd/ring.rs

audio_graph_bsd/
ring.rs
