diff --git a/CHANGELOG.md b/CHANGELOG.md index d4ff9d27e..217cefb25 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/host/aaudio/mod.rs b/src/host/aaudio/mod.rs index b6cdb4584..b6ba7367e 100644 --- a/src/host/aaudio/mod.rs +++ b/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)) diff --git a/src/host/coreaudio/macos/mod.rs b/src/host/coreaudio/macos/mod.rs index 3fe7370e5..81be095cb 100644 --- a/src/host/coreaudio/macos/mod.rs +++ b/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::(); 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)); } } diff --git a/src/host/mod.rs b/src/host/mod.rs index 5a266e072..c5aeff56c 100644 --- a/src/host/mod.rs +++ b/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( diff --git a/src/host/pipewire/stream.rs b/src/host/pipewire/stream.rs index 5a19255be..c1c24bd79 100644 --- a/src/host/pipewire/stream.rs +++ b/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 { has_connected: bool, invalidated: Arc, pending_device_changed: Arc, + spa_io_clock: *const spa_io_clock, + xrun_recovering: bool, #[cfg(feature = "realtime")] rt_promoted: bool, } @@ -273,6 +276,21 @@ impl UserData { 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 { diff --git a/src/host/wasapi/stream.rs b/src/host/wasapi/stream.rs index 301274793..28082e4f9 100644 --- a/src/host/wasapi/stream.rs +++ b/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 ();