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

flexaudio_core/
diagnostics.rs
