feat: add xrun reporting on AAudio, macOS, PipeWire and WASAPI

CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **AAudio**: Xruns are now reported as `ErrorKind::Xrun`.
+- **CoreAudio**: Xruns are now reported as `ErrorKind::Xrun`.
+- **PipeWire**: Xruns are now reported as `ErrorKind::Xrun`.
+- **WASAPI**: Capture xruns are now reported as `ErrorKind::Xrun`.
+
 ### Changed
 
 - **CoreAudio**: Bump `objc2-core-foundation` dependency lower bound to 0.3.1.

src/host/aaudio/mod.rs

@@ -25,7 +25,6 @@ use crate::{
 
 extern crate ndk;
 use self::ndk::audio::AudioStream;
-#[cfg(feature = "realtime")]
 use crate::host::try_emit_error;
 
 mod convert;
@@ -322,6 +321,8 @@ where
     let error_callback: ErrorCallbackArc = Arc::new(Mutex::new(error_callback));
     let error_callback_for_stream = error_callback.clone();
     let error_callback_for_data = error_callback.clone();
+    let error_callback_for_xrun = error_callback.clone();
+    let last_xrun_count = Arc::new(AtomicI32::new(0));
 
     #[cfg(feature = "realtime")]
     let mut rt_checked = false;
@@ -346,6 +347,11 @@ where
                 }
             }
 
+            let xrun_count = stream.x_run_count();
+            if last_xrun_count.swap(xrun_count, Ordering::Relaxed) < xrun_count {
+                let _ = try_emit_error(&error_callback_for_xrun, ErrorKind::Xrun.into());
+            }
+
             let Some(n_samples) = u64::try_from(num_frames)
                 .ok()
                 .and_then(|f| f.checked_mul(channel_count as u64))
@@ -412,6 +418,8 @@ where
     let error_callback: ErrorCallbackArc = Arc::new(Mutex::new(error_callback));
     let error_callback_for_stream = error_callback.clone();
     let error_callback_for_data = error_callback.clone();
+    let error_callback_for_xrun = error_callback.clone();
+    let last_xrun_count = Arc::new(AtomicI32::new(0));
 
     #[cfg(feature = "realtime")]
     let mut rt_checked = false;
@@ -436,6 +444,11 @@ where
                 }
             }
 
+            let xrun_count = stream.x_run_count();
+            if last_xrun_count.swap(xrun_count, Ordering::Relaxed) < xrun_count {
+                let _ = try_emit_error(&error_callback_for_xrun, ErrorKind::Xrun.into());
+            }
+
             let Some(n_samples) = u64::try_from(num_frames)
                 .ok()
                 .and_then(|f| f.checked_mul(channel_count as u64))

src/host/coreaudio/macos/mod.rs

@@ -5,17 +5,17 @@ use std::sync::{
 
 use coreaudio::audio_unit::{AudioUnit, Scope};
 use objc2_core_audio::{
-    kAudioDevicePropertyDeviceIsAlive, kAudioDevicePropertyNominalSampleRate,
-    kAudioHardwarePropertyDefaultOutputDevice, kAudioObjectPropertyElementMain,
-    kAudioObjectPropertyScopeGlobal, kAudioObjectSystemObject, AudioDeviceID, AudioObjectID,
-    AudioObjectPropertyAddress,
+    kAudioDeviceProcessorOverload, kAudioDevicePropertyDeviceIsAlive,
+    kAudioDevicePropertyNominalSampleRate, kAudioHardwarePropertyDefaultOutputDevice,
+    kAudioObjectPropertyElementMain, kAudioObjectPropertyScopeGlobal, kAudioObjectSystemObject,
+    AudioDeviceID, AudioObjectID, AudioObjectPropertyAddress,
 };
 use property_listener::AudioObjectPropertyListener;
 
 pub use self::enumerate::{default_input_device, default_output_device, Devices};
 use super::{asbd_from_config, check_os_status, host_time_to_stream_instant, OSStatus};
 use crate::{
-    host::{coreaudio::macos::loopback::LoopbackDevice, emit_error, latch::Latch},
+    host::{coreaudio::macos::loopback::LoopbackDevice, emit_error, latch::Latch, try_emit_error},
     traits::{HostTrait, StreamTrait},
     Error, ErrorKind, FrameCount, ResultExt, StreamInstant,
 };
@@ -127,10 +127,11 @@ impl DisconnectManager {
         let (disconnect_tx, disconnect_rx) = mpsc::channel::<Error>();
         let (ready_tx, ready_rx) = mpsc::channel();
 
-        // Spawn a dedicated thread to own both listeners. CoreAudio requires that
+        // Spawn a dedicated thread to own all listeners. CoreAudio requires that
         // AudioObjectPropertyListeners are added and removed on the same thread.
         let disconnect_tx_alive = disconnect_tx.clone();
         let disconnect_tx_rate = disconnect_tx;
+        let error_callback_overload = error_callback.clone();
         std::thread::spawn(move || {
             let alive_address = AudioObjectPropertyAddress {
                 mSelector: kAudioDevicePropertyDeviceIsAlive,
@@ -158,13 +159,24 @@ impl DisconnectManager {
                     ));
                 });
 
-            match (alive_listener, rate_listener) {
-                (Ok(_alive), Ok(_rate)) => {
+            // Overload notifications fire on the RT thread.
+            let overload_address = AudioObjectPropertyAddress {
+                mSelector: kAudioDeviceProcessorOverload,
+                mScope: kAudioObjectPropertyScopeGlobal,
+                mElement: kAudioObjectPropertyElementMain,
+            };
+            let overload_listener =
+                AudioObjectPropertyListener::new(device_id, overload_address, move || {
+                    let _ = try_emit_error(&error_callback_overload, ErrorKind::Xrun.into());
+                });
+
+            match (alive_listener, rate_listener, overload_listener) {
+                (Ok(_alive), Ok(_rate), Ok(_overload)) => {
                     let _ = ready_tx.send(Ok(()));
                     // Block until the stream is dropped; listeners are removed on drop.
                     let _ = shutdown_rx.recv();
                 }
-                (Err(e), _) | (_, Err(e)) => {
+                (Err(e), _, _) | (_, Err(e), _) | (_, _, Err(e)) => {
                     let _ = ready_tx.send(Err(e));
                 }
             }

src/host/mod.rs

@@ -196,32 +196,7 @@ pub(crate) mod error_emit;
         )
     ),
 ))]
-pub(crate) use error_emit::emit_error;
-#[cfg(any(
-    target_vendor = "apple",
-    all(target_os = "android", feature = "realtime"),
-    all(
-        feature = "jack",
-        any(
-            target_os = "linux",
-            target_os = "dragonfly",
-            target_os = "freebsd",
-            target_os = "netbsd",
-            target_os = "macos",
-            target_os = "windows",
-        )
-    ),
-    all(
-        feature = "pipewire",
-        any(
-            target_os = "linux",
-            target_os = "dragonfly",
-            target_os = "freebsd",
-            target_os = "netbsd",
-        )
-    ),
-))]
-pub(crate) use error_emit::try_emit_error;
+pub(crate) use error_emit::{emit_error, try_emit_error};
 
 /// Convert a frame count at a given sample rate to a [`std::time::Duration`].
 #[cfg(any(

src/host/pipewire/stream.rs

@@ -24,6 +24,7 @@ use pipewire::{
             format_utils, ParamType,
         },
         pod::{serialize::PodSerializer, Object, Pod, Value},
+        sys::{spa_io_clock, SPA_IO_Clock},
         utils::{Direction, SpaTypes},
     },
     stream::{StreamFlags, StreamListener, StreamRc, StreamState, Time},
@@ -229,6 +230,8 @@ pub struct UserData<D> {
     has_connected: bool,
     invalidated: Arc<AtomicBool>,
     pending_device_changed: Arc<AtomicBool>,
+    spa_io_clock: *const spa_io_clock,
+    xrun_recovering: bool,
     #[cfg(feature = "realtime")]
     rt_promoted: bool,
 }
@@ -273,6 +276,21 @@ impl<D> UserData<D> {
             StreamState::Paused | StreamState::Connecting => {}
         }
     }
+
+    fn check_xrun(&mut self) {
+        if self.spa_io_clock.is_null() {
+            return;
+        }
+        // spa/node/io.h; not present in bindgen output on all libspa versions.
+        const SPA_IO_CLOCK_FLAG_XRUN_RECOVER: u32 = 1 << 1;
+        // io_changed and process run on the same thread.
+        let flags = unsafe { (*self.spa_io_clock).flags };
+        let recovering = flags & SPA_IO_CLOCK_FLAG_XRUN_RECOVER != 0;
+        if recovering && !self.xrun_recovering {
+            let _ = try_emit_error(&self.error_callback, ErrorKind::Xrun.into());
+        }
+        self.xrun_recovering = recovering;
+    }
 }
 
 /// Returns a hardware timestamp for the current graph cycle, or `None` if
@@ -585,6 +603,8 @@ where
         is_default_device,
         has_connected: false,
         pending_device_changed: pending_device_changed.clone(),
+        spa_io_clock: std::ptr::null(),
+        xrun_recovering: false,
         #[cfg(feature = "realtime")]
         rt_promoted: false,
     };
@@ -677,6 +697,11 @@ where
         .state_changed(|_stream, user_data, _old, new| {
             user_data.state_changed(new);
         })
+        .io_changed(|_stream, user_data, id, area, _size| {
+            if id == SPA_IO_Clock {
+                user_data.spa_io_clock = area as *const spa_io_clock;
+            }
+        })
         .process(|stream, user_data| {
             if user_data.pending_device_changed.load(Ordering::Relaxed)
                 && try_emit_error(
@@ -687,6 +712,7 @@ where
             {
                 user_data.pending_device_changed.store(false, Ordering::Relaxed);
             }
+            user_data.check_xrun();
 
             let n_channels = user_data.format.channels();
             if n_channels == 0 {
@@ -833,6 +859,8 @@ where
         is_default_device,
         has_connected: false,
         pending_device_changed: pending_device_changed.clone(),
+        spa_io_clock: std::ptr::null(),
+        xrun_recovering: false,
         #[cfg(feature = "realtime")]
         rt_promoted: false,
     };
@@ -926,6 +954,11 @@ where
         .state_changed(|_stream, user_data, _old, new| {
             user_data.state_changed(new);
         })
+        .io_changed(|_stream, user_data, id, area, _size| {
+            if id == SPA_IO_Clock {
+                user_data.spa_io_clock = area as *const spa_io_clock;
+            }
+        })
         .process(|stream, user_data| {
             if user_data.pending_device_changed.load(Ordering::Relaxed)
                 && try_emit_error(
@@ -936,6 +969,7 @@ where
             {
                 user_data.pending_device_changed.store(false, Ordering::Relaxed);
             }
+            user_data.check_xrun();
 
             let n_channels = user_data.format.channels();
             if n_channels == 0 {

src/host/wasapi/stream.rs

@@ -18,7 +18,9 @@ use windows::Win32::{
 };
 
 use crate::{
-    host::{emit_error, equilibrium::fill_equilibrium, latch::Latch, ErrorCallbackArc},
+    host::{
+        emit_error, equilibrium::fill_equilibrium, latch::Latch, try_emit_error, ErrorCallbackArc,
+    },
     traits::StreamTrait,
     Data, Error, ErrorKind, FrameCount, InputCallbackInfo, InputStreamTimestamp,
     OutputCallbackInfo, OutputStreamTimestamp, ResultExt, SampleFormat, SampleRate, StreamConfig,
@@ -629,7 +631,12 @@ fn run_input(
             AudioClientFlow::Capture { ref capture_client } => capture_client.clone(),
             _ => unreachable!(),
         };
-        if let Err(err) = process_input(&run_ctxt.stream, capture_client, data_callback) {
+        if let Err(err) = process_input(
+            &run_ctxt.stream,
+            capture_client,
+            data_callback,
+            error_callback,
+        ) {
             emit_error(error_callback, err);
             break;
         }
@@ -769,6 +776,7 @@ fn process_input(
     stream: &StreamInner,
     capture_client: Audio::IAudioCaptureClient,
     data_callback: &mut dyn FnMut(&Data, &InputCallbackInfo),
+    error_callback: &ErrorCallbackArc,
 ) -> Result<(), Error> {
     unsafe {
         // Get the available data in the shared buffer.
@@ -796,6 +804,11 @@ fn process_input(
                 Ok(_) => (),
             }
 
+            let flags = flags.assume_init();
+            if flags & Audio::AUDCLNT_BUFFERFLAGS_DATA_DISCONTINUITY.0 as u32 != 0 {
+                let _ = try_emit_error(error_callback, ErrorKind::Xrun.into());
+            }
+
             debug_assert!(!buffer.is_null());
 
             let data = buffer as *mut ();