flexaudio_core/
diagnostics.rs

1//! Allocation-free realtime rejection summaries, materialized off callback.
2use crate::loss::CaptureRejection;
3use crate::{AudioLoss, Result};
4use std::num::NonZeroU64;
5use std::sync::{
6    atomic::{AtomicBool, AtomicU64, Ordering},
7    Arc, Mutex,
8};
9
10#[cfg(not(target_has_atomic = "64"))]
11compile_error!("capture diagnostics require lock-free 64-bit atomics");
12
13/// Clone before entering the callback; recording never allocates or locks.
14#[derive(Clone)]
15pub struct CaptureDiagnostics {
16    inner: Arc<Pending>,
17}
18struct Pending {
19    discontinuity: AtomicBool,
20    startup_drop: Mutex<Option<(u64, u32)>>,
21    summaries: [AtomicU64; 3],
22    rate: u32,
23    channels: u16,
24}
25impl CaptureDiagnostics {
26    /// Configure a handle for one negotiated generation (off callback).
27    pub fn new(rate: u32, channels: u16) -> Self {
28        Self {
29            inner: Arc::new(Pending {
30                discontinuity: AtomicBool::new(false),
31                startup_drop: Mutex::new(None),
32                summaries: std::array::from_fn(|_| AtomicU64::new(0)),
33                rate,
34                channels,
35            }),
36        }
37    }
38    /// Publish a microphone startup gap before the first composite Mix samples.
39    /// Called only by the non-realtime mixer, once per generation.
40    pub fn record_startup_drop(&self, frames: u64, native_rate: u32) {
41        *self
42            .inner
43            .startup_drop
44            .lock()
45            .unwrap_or_else(|error| error.into_inner()) = Some((frames, native_rate));
46    }
47    /// Drain the composite microphone gap once, off callback and before PCM.
48    pub fn take_startup_drop(&self) -> Option<(u64, u32)> {
49        self.inner
50            .startup_drop
51            .lock()
52            .unwrap_or_else(|error| error.into_inner())
53            .take()
54    }
55    /// Mark routing changes without allocating or locking in the native callback.
56    pub fn mark_discontinuity(&self) {
57        self.inner.discontinuity.store(true, Ordering::Release);
58    }
59    /// Consume a routing change once, outside realtime callbacks.
60    pub fn take_discontinuity(&self) -> bool {
61        self.inner.discontinuity.swap(false, Ordering::AcqRel)
62    }
63    fn record(&self, index: usize, samples: Option<NonZeroU64>) {
64        let summary = &self.inner.summaries[index];
65        let Some(samples) = samples else {
66            summary.store(u64::MAX, Ordering::Release);
67            return;
68        };
69        let mut previous = summary.load(Ordering::Acquire);
70        for _ in 0..8 {
71            if previous == u64::MAX {
72                return;
73            }
74            let next = previous.saturating_add(samples.get());
75            match summary.compare_exchange_weak(previous, next, Ordering::AcqRel, Ordering::Acquire)
76            {
77                Ok(_) => return,
78                Err(current) => previous = current,
79            }
80        }
81        // Saturate rather than silently lose a diagnostic under contention.
82        summary.store(u64::MAX, Ordering::Release);
83    }
84    /// Reject a corrupt buffer without trusting invalid metadata for a count.
85    pub fn record_corrupt_buffer(&self, samples: Option<NonZeroU64>) {
86        self.record(0, samples);
87    }
88    /// Reject malformed native buffer layout.
89    pub fn record_malformed_buffer(&self, samples: Option<NonZeroU64>) {
90        self.record(1, samples);
91    }
92    /// Reject a callback whose exclusive state cannot be accessed.
93    pub fn record_callback_rejected(&self, samples: Option<NonZeroU64>) {
94        self.record(2, samples);
95    }
96    /// Drain once after a control-thread poll or producer shutdown; no duplicate reports.
97    pub fn drain(&self) -> Result<Vec<AudioLoss>> {
98        let mut losses = Vec::new();
99        for (summary, reason) in self.inner.summaries.iter().zip([
100            CaptureRejection::CorruptBuffer,
101            CaptureRejection::MalformedBuffer,
102            CaptureRejection::CallbackRejected,
103        ]) {
104            let pending = summary.swap(0, Ordering::AcqRel);
105            if pending != 0 {
106                losses.push(AudioLoss::rejected(
107                    reason,
108                    if pending == u64::MAX {
109                        None
110                    } else {
111                        NonZeroU64::new(pending)
112                    },
113                    self.inner.rate,
114                    self.inner.channels,
115                )?);
116            }
117        }
118        Ok(losses)
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125    #[test]
126    fn known_unknown_saturated_and_final_drain_are_observable_once() {
127        let diagnostics = CaptureDiagnostics::new(48_000, 2);
128        diagnostics.record_corrupt_buffer(NonZeroU64::new(100));
129        diagnostics.record_corrupt_buffer(NonZeroU64::new(100));
130        diagnostics.record_malformed_buffer(None);
131        diagnostics.record_malformed_buffer(NonZeroU64::new(7));
132        diagnostics.record_callback_rejected(NonZeroU64::new(u64::MAX - 1));
133        diagnostics.record_callback_rejected(NonZeroU64::new(2));
134        let losses = diagnostics.drain().unwrap();
135        assert_eq!(losses.len(), 3);
136        assert_eq!(losses[0].samples().unwrap().get(), 200);
137        assert_eq!(losses[1].samples(), None);
138        assert_eq!(losses[2].samples(), None);
139        for loss in &losses {
140            assert_eq!((loss.sample_rate(), loss.channels()), (48_000, 2));
141        }
142        assert!(diagnostics.drain().unwrap().is_empty());
143    }
144    #[test]
145    fn concurrent_recording_preserves_every_observation() {
146        let diagnostics = CaptureDiagnostics::new(48_000, 2);
147        let handles: Vec<_> = (0..4)
148            .map(|_| {
149                let diagnostics = diagnostics.clone();
150                std::thread::spawn(move || {
151                    for _ in 0..1_000 {
152                        diagnostics.record_corrupt_buffer(NonZeroU64::new(1));
153                    }
154                })
155            })
156            .collect();
157        for handle in handles {
158            handle.join().unwrap();
159        }
160        let loss = diagnostics.drain().unwrap()[0];
161        assert!(loss.samples().is_none_or(|samples| samples.get() == 4_000));
162        assert!(diagnostics.drain().unwrap().is_empty());
163    }
164}
165
166#[cfg(test)]
167mod startup_tests {
168    use super::*;
169    #[test]
170    fn composite_startup_gap_is_drained_once_with_native_rate() {
171        let diagnostics = CaptureDiagnostics::new(48_000, 2);
172        assert_eq!(diagnostics.take_startup_drop(), None);
173        diagnostics.record_startup_drop(441, 44_100);
174        assert_eq!(diagnostics.clone().take_startup_drop(), Some((441, 44_100)));
175        assert_eq!(diagnostics.take_startup_drop(), None);
176    }
177}
178
179#[cfg(test)]
180mod routing_tests {
181    use super::*;
182    #[test]
183    fn routing_discontinuity_is_sticky_and_consumed_once() {
184        let diagnostics = CaptureDiagnostics::new(48_000, 2);
185        assert!(!diagnostics.take_discontinuity());
186        diagnostics.mark_discontinuity();
187        diagnostics.mark_discontinuity();
188        assert!(diagnostics.clone().take_discontinuity());
189        assert!(!diagnostics.take_discontinuity());
190    }
191}
flexaudio_core/diagnostics.rs

flexaudio_core/
diagnostics.rs
