diff --git a/CHANGELOG.md b/CHANGELOG.md index 54e6def03..d4c35c86a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -107,7 +107,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `realtime` feature for real-time audio thread scheduling without a D-Bus build dependency. - `StreamTrait::now()` to query the current instant on the stream's clock. - `StreamTrait::buffer_size()` to query the stream's current buffer size in frames per callback. -- `DeviceTrait::build_duplex_stream()`, `build_duplex_stream_raw()`, and `supports_duplex()` for synchronized capture and playback on a shared clock (no backend support yet). +- `DeviceTrait::build_duplex_stream()`, `build_duplex_stream_raw()`, and `supports_duplex()` for synchronized capture and playback on a shared clock (CoreAudio on macOS; other backends to follow). - `SAMPLE_RATE_CD` (44100 Hz) and `SAMPLE_RATE_48K` (48000 Hz) constants. - `SupportedStreamConfigRange::try_with_standard_sample_rate()` and `with_standard_sample_rate()` to select 48 kHz or 44.1 kHz from a range. diff --git a/Cargo.toml b/Cargo.toml index 5afca71b4..04ac045ec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -226,6 +226,9 @@ name = "enumerate" [[example]] name = "feedback" +[[example]] +name = "duplex_feedback" + [[example]] name = "record_wav" diff --git a/examples/duplex_feedback.rs b/examples/duplex_feedback.rs new file mode 100644 index 000000000..2ed1fb49d --- /dev/null +++ b/examples/duplex_feedback.rs @@ -0,0 +1,185 @@ +// Duplex feedback example. + +#[cfg(target_os = "macos")] +mod imp { + use clap::Parser; + use cpal::{ + BufferSize, ChannelCount, DuplexStreamConfig, FrameCount, Sample, SampleRate, + traits::{DeviceTrait, HostTrait, StreamTrait}, + }; + + #[derive(Parser, Debug)] + #[command(version, about = "CPAL duplex feedback example", long_about = None)] + struct Opt { + /// List devices that can build duplex streams, then exit. + #[arg(long)] + list: bool, + + /// Device ID to use for the duplex stream. If omitted, the default output device is + /// used. + #[arg(short, long, value_name = "ID")] + device: Option, + + /// Number of input channels to capture. + #[arg(long, default_value_t = 1)] + input_channels: ChannelCount, + + /// Number of output channels to render. + #[arg(long, default_value_t = 2)] + output_channels: ChannelCount, + + /// Sample rate. + #[arg(long, default_value_t = 48_000)] + sample_rate: SampleRate, + + /// Optional fixed buffer size, in frames. Omit to use the device default. + #[arg(long)] + buffer_size: Option, + } + + pub fn run() -> Result<(), cpal::Error> { + let opt = Opt::parse(); + let host = cpal::default_host(); + + if opt.list { + return list_duplex_devices(&host); + } + + let device = match opt.device.as_deref() { + Some(id_str) => { + let id = id_str.parse().map_err(|e| { + cpal::Error::with_message( + cpal::ErrorKind::InvalidInput, + format!("failed to parse device id {id_str:?}: {e}"), + ) + })?; + host.device_by_id(&id).ok_or_else(|| { + cpal::Error::with_message( + cpal::ErrorKind::DeviceNotAvailable, + format!("no device with id {id_str:?}"), + ) + })? + } + None => host.default_output_device().ok_or_else(|| { + cpal::Error::with_message( + cpal::ErrorKind::DeviceNotAvailable, + "no default output device", + ) + })?, + }; + + let device_name = device + .description() + .map(|d| d.name().to_string()) + .unwrap_or_else(|_| "".to_string()); + println!("using device: {device_name}"); + + if !device.supports_duplex() { + return Err(cpal::Error::with_message( + cpal::ErrorKind::UnsupportedOperation, + "this device does not support duplex streams \ + (run with --list to see candidates)", + )); + } + + let config = DuplexStreamConfig { + input_channels: opt.input_channels, + output_channels: opt.output_channels, + sample_rate: opt.sample_rate, + buffer_size: opt + .buffer_size + .map_or(BufferSize::Default, BufferSize::Fixed), + }; + + let input_channels = opt.input_channels; + let output_channels = opt.output_channels; + + let stream = device.build_duplex_stream::( + config, + move |input, output, _info| { + mix_to_output(input, output, input_channels, output_channels) + }, + |err| eprintln!("duplex stream error: {err}"), + None, + )?; + + stream.start()?; + + println!("playing duplex feedback. Ctrl-C to exit."); + std::thread::park(); + Ok(()) + } + + /// Mix interleaved input frames to mono and broadcast across all output channels, + /// padding any trailing output samples with silence. + fn mix_to_output( + input: &[f32], + output: &mut [f32], + input_channels: ChannelCount, + output_channels: ChannelCount, + ) { + let input_channels = input_channels as usize; + let output_channels = output_channels as usize; + let input_frames = input.len() / input_channels.max(1); + let output_frames = output.len() / output_channels.max(1); + let frames = input_frames.min(output_frames); + + for frame in 0..frames { + let mut acc = f32::EQUILIBRIUM; + for ch in 0..input_channels { + acc += input[frame * input_channels + ch]; + } + let mixed = acc / input_channels.max(1) as f32; + for ch in 0..output_channels { + output[frame * output_channels + ch] = mixed; + } + } + + output[frames * output_channels..].fill(f32::EQUILIBRIUM); + } + + /// Print the devices on the active host that report `supports_duplex() == true`, with their + /// IDs (for use with `--device`) and descriptions. + fn list_duplex_devices(host: &cpal::Host) -> Result<(), cpal::Error> { + let default_id = host.default_output_device().and_then(|d| d.id().ok()); + + let mut found = false; + println!("Devices supporting duplex on this host:"); + for device in host.devices()? { + if !device.supports_duplex() { + continue; + } + found = true; + let id = device.id().ok(); + let name = device + .description() + .map_or_else(|_| "".to_string(), |d| d.name().to_string()); + let default_marker = if id.is_some() && id == default_id { + " [default]" + } else { + "" + }; + let id_str = id + .as_ref() + .map_or_else(|| "".to_string(), ToString::to_string); + println!(" {id_str}{default_marker} — {name}"); + } + if !found { + println!(" (none)"); + } + Ok(()) + } +} + +fn main() { + #[cfg(target_os = "macos")] + if let Err(e) = imp::run() { + eprintln!("duplex_feedback: {e}"); + std::process::exit(1); + } + + #[cfg(not(target_os = "macos"))] + { + eprintln!("duplex streams are not supported on this platform"); + } +} diff --git a/src/host/coreaudio/macos/device.rs b/src/host/coreaudio/macos/device.rs index bde36af6b..2d9be7efa 100644 --- a/src/host/coreaudio/macos/device.rs +++ b/src/host/coreaudio/macos/device.rs @@ -1,6 +1,6 @@ use std::{ fmt, - mem::{self, size_of}, + mem::{self, ManuallyDrop, size_of}, ptr::{NonNull, null}, sync::{ Arc, Mutex, @@ -46,9 +46,9 @@ use super::{ }; use crate::{ BufferSize, CallbackInfo, ChannelCount, Data, DeviceDescription, DeviceDescriptionBuilder, - DeviceId, Error, ErrorKind, FrameCount, InterfaceType, ResultExt, SampleFormat, SampleRate, - StreamConfig, StreamInstant, StreamTimestamp, SupportedBufferSize, SupportedStreamConfig, - SupportedStreamConfigRange, + DeviceId, DuplexCallbackInfo, DuplexStreamConfig, Error, ErrorKind, FrameCount, InterfaceType, + ResultExt, SampleFormat, SampleRate, StreamConfig, StreamInstant, StreamTimestamp, + SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange, host::{ ErrorCallbackArc, coreaudio::macos::{StreamInner, loopback::LoopbackDevice}, @@ -91,7 +91,7 @@ fn set_physical_format( /// /// Unlike [`set_physical_format`], this only changes the device clock rate. The AudioUnit bridges /// any remaining format difference to the virtual stream format seen by the callback. -fn set_sample_rate( +pub(super) fn set_sample_rate( audio_device_id: AudioObjectID, target_sample_rate: SampleRate, timeout: Option, @@ -285,6 +285,51 @@ fn get_io_buffer_frame_size_range(device_id: AudioDeviceID) -> Result StreamInstant { + callback.checked_sub(delay).unwrap_or_else(|| { + let _ = try_emit_error( + error_callback, + Error::with_message( + ErrorKind::BackendError, + "timestamp underflow computing capture instant", + ), + ); + callback + }) +} + +/// Compute the playback-side timestamp from a callback instant and the output latency. +/// +/// Falls back to `callback` and reports a [`ErrorKind::BackendError`] via `error_callback` if +/// `callback + delay` would overflow. The representation supports ~585 billion years of stream +/// uptime, so overflow indicates a pathological latency value and should not happen in practice. +pub(super) fn estimate_playback_instant( + callback: StreamInstant, + delay: Duration, + error_callback: &ErrorCallbackArc, +) -> StreamInstant { + callback.checked_add(delay).unwrap_or_else(|| { + let _ = try_emit_error( + error_callback, + Error::with_message( + ErrorKind::BackendError, + "timestamp overflow computing playback instant", + ), + ); + callback + }) +} + impl DeviceTrait for Device { type SupportedInputConfigs = SupportedInputConfigs; type SupportedOutputConfigs = SupportedOutputConfigs; @@ -357,6 +402,43 @@ impl DeviceTrait for Device { timeout, ) } + + fn supports_duplex(&self) -> bool { + // Any `AudioDeviceID` that exposes both directions can be driven by a single HALOutput + // AudioUnit, which delivers input and output to one render callback. + // + // For non-aggregate devices the clock is shared by construction (one piece of hardware). + // For aggregate devices, CoreAudio drift-corrects across sub-device clocks — drift + // correction is configured per-aggregate in Audio MIDI Setup and is enabled by default. + // The resulting callback is sample-aligned (any drift between physical clocks is + // absorbed by the aggregate). We trust that user-configured aggregates are what the + // user wants and accept them here. + self.supports_input() && self.supports_output() + } + + fn build_duplex_stream_raw( + &self, + config: DuplexStreamConfig, + input_sample_format: SampleFormat, + output_sample_format: SampleFormat, + data_callback: D, + error_callback: E, + timeout: Option, + ) -> Result + where + D: FnMut(&Data, &mut Data, &DuplexCallbackInfo) + Send + 'static, + E: FnMut(Error) + Send + 'static, + { + Device::build_duplex_stream_raw( + self, + config, + input_sample_format, + output_sample_format, + data_callback, + error_callback, + timeout, + ) + } } #[derive(Clone)] @@ -809,9 +891,10 @@ impl Device { let inner_arc = Arc::new(Mutex::new(StreamInner { playing: false, - audio_unit, + audio_unit: ManuallyDrop::new(audio_unit), _device_id: self.audio_device_id, _loopback_device: loopback_aggregate, + duplex_callback_ptr: None, })); let weak_inner = Arc::downgrade(&inner_arc); let monitor: Box = Box::new(DisconnectManager::new( @@ -819,6 +902,7 @@ impl Device { weak_inner, error_callback_disconnect, pending_xrun_overload, + false, )?); let stream = Stream::new(inner_arc, monitor, draining, Duration::ZERO); stream.signal_ready(); @@ -956,9 +1040,10 @@ impl Device { let inner_arc = Arc::new(Mutex::new(StreamInner { playing: false, - audio_unit, + audio_unit: ManuallyDrop::new(audio_unit), _device_id: self.audio_device_id, _loopback_device: None, + duplex_callback_ptr: None, })); let weak_inner = Arc::downgrade(&inner_arc); let monitor: Box = if matches!(mode, AudioUnitMode::DefaultOutput) { @@ -975,6 +1060,7 @@ impl Device { weak_inner, error_callback, pending_xrun_overload, + false, )?) }; let stream = Stream::new(inner_arc, monitor, draining, drain_window); diff --git a/src/host/coreaudio/macos/duplex.rs b/src/host/coreaudio/macos/duplex.rs new file mode 100644 index 000000000..6ee1c2648 --- /dev/null +++ b/src/host/coreaudio/macos/duplex.rs @@ -0,0 +1,390 @@ +use std::{ + ffi::c_void, + mem::ManuallyDrop, + panic::{AssertUnwindSafe, catch_unwind}, + ptr::NonNull, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, +}; + +use coreaudio::audio_unit::{AudioUnit, Element, Scope}; +use objc2_audio_toolbox::{ + AURenderCallbackStruct, AudioUnitRender, AudioUnitRenderActionFlags, + kAudioOutputUnitProperty_CurrentDevice, kAudioOutputUnitProperty_EnableIO, + kAudioUnitProperty_SetRenderCallback, kAudioUnitProperty_StreamFormat, +}; +use objc2_core_audio::kAudioDevicePropertyBufferFrameSize; +use objc2_core_audio_types::{AudioBuffer, AudioBufferList, AudioTimeStamp, kAudio_ParamError}; + +use super::{ + DisconnectManager, DuplexCallbackPtr, Monitor, Stream, StreamInner, asbd_from_config, + device::{ + Device, estimate_capture_instant, estimate_playback_instant, get_device_buffer_frame_size, + set_sample_rate, + }, + host_time_to_stream_instant, +}; +use crate::{ + BufferSize, CallbackInfo, Data, DuplexCallbackInfo, DuplexStreamConfig, Error, ErrorKind, + FrameCount, SampleFormat, StreamConfig, StreamTimestamp, + host::{ErrorCallbackArc, equilibrium::fill_equilibrium, frames_to_duration, try_emit_error}, + traits::DeviceTrait, +}; + +type DuplexProcFn = dyn FnMut( + NonNull, + NonNull, + u32, // bus_number + u32, // num_frames + *mut AudioBufferList, +) -> i32; + +pub(crate) struct DuplexProcWrapper { + callback: Box, +} + +// SAFETY: DuplexProcWrapper is Send because: +// 1. The boxed closure captures only Send types (the DuplexCallback trait requires Send) +// 2. The raw pointer stored in StreamInner is accessed: +// - By CoreAudio's audio thread via `duplex_input_proc` (as the refcon) +// - During Drop, after stopping the audio unit (callback no longer running) +// These never overlap: Drop stops the audio unit before reclaiming the pointer. +// 3. CoreAudio guarantees single-threaded callback invocation +unsafe impl Send for DuplexProcWrapper {} + +// `extern "C-unwind"` matches `AURenderCallbackStruct::inputProc`. +// `catch_unwind` prevents panics from unwinding through CoreAudio's C frames. +extern "C-unwind" fn duplex_input_proc( + in_ref_con: NonNull, + io_action_flags: NonNull, + in_time_stamp: NonNull, + in_bus_number: u32, + in_number_frames: u32, + io_data: *mut AudioBufferList, +) -> i32 { + // SAFETY: `in_ref_con` originates from `Box::into_raw` in `build_duplex_stream_raw`. + // `StreamInner::drop` stops the audio unit before reclaiming the pointer, + // so it remains valid for the lifetime of the callback. + // Called from a single render thread per audio unit, so `as_mut()` has exclusive access. + let wrapper = unsafe { in_ref_con.cast::().as_mut() }; + match catch_unwind(AssertUnwindSafe(|| { + (wrapper.callback)( + io_action_flags, + in_time_stamp, + in_bus_number, + in_number_frames, + io_data, + ) + })) { + Ok(result) => result, + Err(_) => kAudio_ParamError, + } +} + +impl Device { + // See: https://developer.apple.com/library/archive/technotes/tn2091/_index.html + pub(crate) fn build_duplex_stream_raw( + &self, + config: DuplexStreamConfig, + input_sample_format: SampleFormat, + output_sample_format: SampleFormat, + data_callback: D, + error_callback: E, + timeout: Option, + ) -> Result + where + D: FnMut(&Data, &mut Data, &DuplexCallbackInfo) + Send + 'static, + E: FnMut(Error) + Send + 'static, + { + if !self.supports_duplex() { + return Err(Error::with_message( + ErrorKind::UnsupportedOperation, + "device does not support both input and output", + )); + } + + set_sample_rate(self.audio_device_id, config.sample_rate, timeout)?; + + let mut audio_unit = + AudioUnit::new_uninitialized(coreaudio::audio_unit::IOType::HalOutput)?; + + const ENABLED: u32 = 1; + audio_unit.set_property( + kAudioOutputUnitProperty_EnableIO, + Scope::Input, + Element::Input, + Some(&ENABLED), + )?; + + audio_unit.set_property( + kAudioOutputUnitProperty_EnableIO, + Scope::Output, + Element::Output, + Some(&ENABLED), + )?; + + audio_unit.set_property( + kAudioOutputUnitProperty_CurrentDevice, + Scope::Global, + Element::Output, + Some(&self.audio_device_id), + )?; + + let input_stream_config = StreamConfig { + channels: config.input_channels, + sample_rate: config.sample_rate, + buffer_size: config.buffer_size, + }; + + let output_stream_config = StreamConfig { + channels: config.output_channels, + sample_rate: config.sample_rate, + buffer_size: config.buffer_size, + }; + + // Client-side format: Scope::Output for input bus, Scope::Input for output bus. + let input_asbd = asbd_from_config(input_stream_config, input_sample_format); + audio_unit.set_property( + kAudioUnitProperty_StreamFormat, + Scope::Output, + Element::Input, + Some(&input_asbd), + )?; + + let output_asbd = asbd_from_config(output_stream_config, output_sample_format); + audio_unit.set_property( + kAudioUnitProperty_StreamFormat, + Scope::Input, + Element::Output, + Some(&output_asbd), + )?; + + if let BufferSize::Fixed(buffer_size) = &config.buffer_size { + audio_unit.set_property( + kAudioDevicePropertyBufferFrameSize, + Scope::Global, + Element::Output, + Some(buffer_size), + )?; + } + + audio_unit.initialize()?; + + let current_buffer_size = get_device_buffer_frame_size(&audio_unit).map_err(|e| { + Error::with_message( + ErrorKind::BackendError, + format!("failed to query device buffer size: {e}"), + ) + })?; + + let sample_rate = config.sample_rate; + let device_buffer_frames = current_buffer_size; + let raw_audio_unit = *audio_unit.as_ref(); + let input_channels = config.input_channels as usize; + let input_sample_bytes = input_sample_format.sample_size(); + let output_sample_bytes = output_sample_format.sample_size(); + + let input_buffer_bytes = current_buffer_size * input_channels * input_sample_bytes; + let mut input_buffer: Box<[u8]> = vec![0u8; input_buffer_bytes].into_boxed_slice(); + + let error_callback: ErrorCallbackArc = Arc::new(Mutex::new(error_callback)); + let error_callback_for_callback = error_callback.clone(); + + let pending_xrun = Arc::new(AtomicBool::new(false)); + let pending_xrun_overload = pending_xrun.clone(); + let draining = Arc::new(AtomicBool::new(false)); + let draining_render = draining.clone(); + let drain_window = frames_to_duration(current_buffer_size as FrameCount, sample_rate); + + let mut data_callback = data_callback; + let buffer_size_changed = AtomicBool::new(false); + + let duplex_proc: Box = Box::new( + move |io_action_flags: NonNull, + in_time_stamp: NonNull, + _in_bus_number: u32, + in_number_frames: u32, + io_data: *mut AudioBufferList| + -> i32 { + if buffer_size_changed.load(Ordering::Relaxed) { + return kAudio_ParamError; + } + + if io_data.is_null() { + return kAudio_ParamError; + } + // SAFETY: io_data validated as non-null above. + let buffer_list = unsafe { &mut *io_data }; + if buffer_list.mNumberBuffers == 0 { + return kAudio_ParamError; + } + + let num_frames = in_number_frames as usize; + let input_samples = num_frames * input_channels; + let input_bytes = input_samples * input_sample_bytes; + + if input_bytes != input_buffer.len() { + buffer_size_changed.store(true, Ordering::Relaxed); + return kAudio_ParamError; + } + + let buffer = &mut buffer_list.mBuffers[0]; + if buffer.mData.is_null() { + return kAudio_ParamError; + } + let output_byte_len = buffer.mDataByteSize as usize; + let output_samples = output_byte_len / output_sample_bytes; + + // Pre-fill the output with silence so any early return below (draining, + // timestamp failure) plays silence rather than stale buffer contents. + // SAFETY: buffer.mData validated as non-null above; CoreAudio guarantees the + // buffer holds mDataByteSize bytes. + unsafe { + let bytes = + std::slice::from_raw_parts_mut(buffer.mData as *mut u8, output_byte_len); + fill_equilibrium(bytes, output_sample_format); + } + + if draining_render.load(Ordering::Relaxed) { + return 0; + } + + // SAFETY: in_time_stamp is valid per CoreAudio callback contract. + let timestamp: &AudioTimeStamp = unsafe { in_time_stamp.as_ref() }; + + let callback_instant = match host_time_to_stream_instant(timestamp.mHostTime) { + Err(err) => { + let _ = try_emit_error(&error_callback_for_callback, err); + return 0; + } + Ok(cb) => cb, + }; + + // SAFETY: buffer.mData validated as non-null above. + let mut output_data = unsafe { + Data::from_parts( + buffer.mData as *mut (), + output_samples, + output_sample_format, + ) + }; + + let delay = frames_to_duration(device_buffer_frames as FrameCount, sample_rate); + + let capture = + estimate_capture_instant(callback_instant, delay, &error_callback_for_callback); + let playback = estimate_playback_instant( + callback_instant, + delay, + &error_callback_for_callback, + ); + + // A processor-overload notification does not say which direction glitched, so + // report it on both. + let xrun = pending_xrun.swap(false, Ordering::Relaxed); + let input_info = CallbackInfo { + timestamp: StreamTimestamp { + callback: callback_instant, + device: capture, + }, + xrun, + }; + let output_info = CallbackInfo { + timestamp: StreamTimestamp { + callback: callback_instant, + device: playback, + }, + xrun, + }; + + let mut input_buffer_list = AudioBufferList { + mNumberBuffers: 1, + mBuffers: [AudioBuffer { + mNumberChannels: input_channels as u32, + mDataByteSize: input_bytes as u32, + mData: input_buffer.as_mut_ptr() as *mut c_void, + }], + }; + + // SAFETY: raw_audio_unit is valid for the callback duration, + // input_buffer_list points to bounds-checked input_buffer. + let status = unsafe { + AudioUnitRender( + raw_audio_unit, + io_action_flags.as_ptr(), + in_time_stamp, + 1, // Element 1 = input + in_number_frames, + NonNull::new_unchecked(&mut input_buffer_list), + ) + }; + + if status != 0 { + let _ = try_emit_error( + &error_callback_for_callback, + Error::with_message( + ErrorKind::BackendError, + format!("AudioUnitRender failed for input: OSStatus {status}"), + ), + ); + input_buffer[..input_bytes].fill(0); + } + + // SAFETY: input_buffer is bounds-checked, filled by AudioUnitRender + // (or zeroed on error), and outlives this Data reference. + let input_data = unsafe { + Data::from_parts( + input_buffer.as_mut_ptr() as *mut (), + input_samples, + input_sample_format, + ) + }; + + let callback_info = DuplexCallbackInfo::new(input_info, output_info); + data_callback(&input_data, &mut output_data, &callback_info); + + 0 + }, + ); + + let wrapper = Box::new(DuplexProcWrapper { + callback: duplex_proc, + }); + let wrapper_ptr = Box::into_raw(wrapper); + + let render_callback = AURenderCallbackStruct { + inputProc: Some(duplex_input_proc), + inputProcRefCon: wrapper_ptr as *mut c_void, + }; + + audio_unit.set_property( + kAudioUnitProperty_SetRenderCallback, + Scope::Global, + Element::Output, + Some(&render_callback), + )?; + + let inner_arc = Arc::new(Mutex::new(StreamInner { + playing: false, + audio_unit: ManuallyDrop::new(audio_unit), + _device_id: self.audio_device_id, + _loopback_device: None, + duplex_callback_ptr: Some(DuplexCallbackPtr(wrapper_ptr)), + })); + let weak_inner = Arc::downgrade(&inner_arc); + let monitor: Box = Box::new(DisconnectManager::new( + self.audio_device_id, + weak_inner, + error_callback, + pending_xrun_overload, + true, + )?); + + let stream = Stream::new(inner_arc, monitor, draining, drain_window); + stream.signal_ready(); + Ok(stream) + } +} diff --git a/src/host/coreaudio/macos/mod.rs b/src/host/coreaudio/macos/mod.rs index 7b53a40a5..91c09c131 100644 --- a/src/host/coreaudio/macos/mod.rs +++ b/src/host/coreaudio/macos/mod.rs @@ -1,4 +1,5 @@ use std::{ + mem::ManuallyDrop, sync::{ Arc, Mutex, Weak, atomic::{AtomicBool, AtomicUsize, Ordering}, @@ -10,9 +11,9 @@ use std::{ use coreaudio::audio_unit::{AudioUnit, Scope}; use objc2_core_audio::{ AudioDeviceID, AudioObjectID, AudioObjectPropertyAddress, kAudioDeviceProcessorOverload, - kAudioDevicePropertyDeviceIsAlive, kAudioDevicePropertyNominalSampleRate, - kAudioHardwarePropertyDefaultOutputDevice, kAudioObjectPropertyElementMain, - kAudioObjectPropertyScopeGlobal, kAudioObjectSystemObject, + kAudioDevicePropertyBufferFrameSize, kAudioDevicePropertyDeviceIsAlive, + kAudioDevicePropertyNominalSampleRate, kAudioHardwarePropertyDefaultOutputDevice, + kAudioObjectPropertyElementMain, kAudioObjectPropertyScopeGlobal, kAudioObjectSystemObject, }; use property_listener::AudioObjectPropertyListener; @@ -25,6 +26,7 @@ use crate::{ }; mod device; +mod duplex; pub mod enumerate; mod loopback; mod property_listener; @@ -112,11 +114,17 @@ pub(super) trait Monitor: Send + Sync { /// AudioObjectPropertyListener is always created and dropped on the same thread. /// This avoids potential threading issues with CoreAudio APIs. /// -/// When a device disconnects, this manager: -/// 1. Attempts to pause the stream to stop audio I/O -/// 2. Calls the error callback with `ErrorKind::DeviceNotAvailable` +/// Always listens for: +/// - device disconnection (`kAudioDevicePropertyDeviceIsAlive`) → `DeviceNotAvailable` +/// - sample rate changes (`kAudioDevicePropertyNominalSampleRate`) → `StreamInvalidated` /// -/// The dedicated thread architecture ensures `Stream` can implement `Send`. +/// Optionally (when `listen_buffer_size = true`) also listens for: +/// - buffer-size changes (`kAudioDevicePropertyBufferFrameSize`) → `StreamInvalidated` +/// Duplex streams enable this because their input scratch buffer is sized exactly to the +/// negotiated frame count; any runtime change would invalidate the buffer. +/// +/// On any of these events the manager attempts to pause the stream and fires the error +/// callback. The dedicated thread architecture ensures `Stream` can implement `Send`. struct DisconnectManager { latch: Latch, _shutdown_tx: mpsc::Sender<()>, @@ -128,6 +136,7 @@ impl DisconnectManager { stream_weak: Weak>, error_callback: Arc>, pending_xrun: Arc, + listen_buffer_size: bool, ) -> Result { let (shutdown_tx, shutdown_rx) = mpsc::channel(); let (disconnect_tx, disconnect_rx) = mpsc::channel::(); @@ -136,7 +145,8 @@ impl DisconnectManager { // 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 disconnect_tx_rate = disconnect_tx.clone(); + let disconnect_tx_buffer = disconnect_tx; std::thread::spawn(move || { let alive_address = AudioObjectPropertyAddress { mSelector: kAudioDevicePropertyDeviceIsAlive, @@ -175,13 +185,47 @@ impl DisconnectManager { pending_xrun.store(true, Ordering::Relaxed); }); - match (alive_listener, rate_listener, overload_listener) { - (Ok(_alive), Ok(_rate), Ok(_overload)) => { + // Only registered when callers opt in. Held in an `Option` to keep the listener + // alive (or not) for the duration of the same shutdown loop below. + let buffer_size_listener = if listen_buffer_size { + let buffer_size_address = AudioObjectPropertyAddress { + mSelector: kAudioDevicePropertyBufferFrameSize, + mScope: kAudioObjectPropertyScopeGlobal, + mElement: kAudioObjectPropertyElementMain, + }; + Some(AudioObjectPropertyListener::new( + device_id, + buffer_size_address, + move || { + let _ = disconnect_tx_buffer.send(Error::with_message( + ErrorKind::StreamInvalidated, + "Device buffer size changed", + )); + }, + )) + } else { + None + }; + + // Tease out the registration results: any failure aborts. + let buffer_result: Result, Error> = match buffer_size_listener { + Some(Ok(listener)) => Ok(Some(listener)), + Some(Err(e)) => Err(e), + None => Ok(None), + }; + + match ( + alive_listener, + rate_listener, + overload_listener, + buffer_result, + ) { + (Ok(_alive), Ok(_rate), Ok(_overload), Ok(_buf)) => { 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), _, _) | (_, _, Err(e), _) | (_, _, _, Err(e)) => { let _ = ready_tx.send(Err(e)); } } @@ -379,13 +423,36 @@ impl Monitor for DefaultOutputMonitor { } } +/// Owning pointer to a duplex callback wrapper, shared with CoreAudio's render thread. +/// +/// SAFETY: The pointer originates from `Box::into_raw` in `Device::build_duplex_stream_raw` +/// and is registered with CoreAudio via `inputProcRefCon`. +/// CoreAudio dereferences it from its single render thread for the lifetime of the audio unit. +/// `StreamInner::drop` stops the audio unit (by dropping `audio_unit`) *before* reclaiming the +/// box, which guarantees the render thread cannot observe a freed pointer. There is no other +/// concurrent access — the build/drop thread never touches the pointer while the audio unit is +/// running. +struct DuplexCallbackPtr(*mut duplex::DuplexProcWrapper); + +// SAFETY: see `DuplexCallbackPtr`. The pointer is shared with CoreAudio's audio thread but is +// never accessed concurrently from another thread; the audio unit is stopped before the pointer +// is reclaimed in `StreamInner::drop`. +unsafe impl Send for DuplexCallbackPtr {} + struct StreamInner { playing: bool, - audio_unit: AudioUnit, + /// Wrapped in [`ManuallyDrop`] so `Drop for StreamInner` can stop the audio unit *before* + /// reclaiming [`duplex_callback_ptr`](Self::duplex_callback_ptr). Dropping the inner + /// `AudioUnit` stops it, which is what guarantees CoreAudio will not invoke the duplex + /// render callback after we free the boxed closure. + audio_unit: ManuallyDrop, // Track the device with which the audio unit was spawned _device_id: AudioDeviceID, /// Manage the lifetime of the aggregate device used for loopback recording _loopback_device: Option, + /// Boxed duplex render callback, owned by this `StreamInner`. `None` for simplex (input or + /// output only) streams, populated by the duplex build path. + duplex_callback_ptr: Option, } impl StreamInner { @@ -410,6 +477,29 @@ impl StreamInner { } } +impl Drop for StreamInner { + fn drop(&mut self) { + // SAFETY: This is the sole owning instance of `audio_unit` (wrapped in `ManuallyDrop` + // so we control drop order). Dropping it stops the audio unit, which guarantees + // CoreAudio will not invoke the render callback after this point. That makes it safe + // to reclaim the duplex callback box below. `audio_unit` is not accessed afterwards. + unsafe { + ManuallyDrop::drop(&mut self.audio_unit); + } + + if let Some(DuplexCallbackPtr(ptr)) = self.duplex_callback_ptr.take() { + if !ptr.is_null() { + // SAFETY: `ptr` was produced by `Box::into_raw` in the duplex build path. + // The audio unit was stopped above, so the render thread no longer references + // it. We are the sole owner, so reclaiming and dropping is sound. + unsafe { + drop(Box::from_raw(ptr)); + } + } + } + } +} + pub struct Stream { inner: Arc>, monitor: Box,