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

flexaudio_core/
diagnostics.rs
