feat(coreaudio): synchronized duplex impl on macOS

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.

Cargo.toml

@@ -226,6 +226,9 @@ name = "enumerate"
 [[example]]
 name = "feedback"
 
+[[example]]
+name = "duplex_feedback"
+
 [[example]]
 name = "record_wav"
 

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<String>,
+
+        /// 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<FrameCount>,
+    }
+
+    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(|_| "<unknown>".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::<f32, f32, _, _>(
+            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(|_| "<unknown>".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(|| "<no id>".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");
+    }
+}

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<Duration>,
@@ -285,6 +285,51 @@ fn get_io_buffer_frame_size_range(device_id: AudioDeviceID) -> Result<SupportedB
     })
 }
 
+/// Compute the capture-side timestamp from a callback instant and the input latency.
+///
+/// Falls back to `callback` and reports a [`ErrorKind::BackendError`] via `error_callback` if
+/// `callback - delay` would underflow. `callback` is a monotonic clock that starts before the
+/// stream opens, so underflow indicates a pathological latency value and should not happen in
+/// practice.
+pub(super) fn estimate_capture_instant(
+    callback: StreamInstant,
+    delay: Duration,
+    error_callback: &ErrorCallbackArc,
+) -> 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<D, E>(
+        &self,
+        config: DuplexStreamConfig,
+        input_sample_format: SampleFormat,
+        output_sample_format: SampleFormat,
+        data_callback: D,
+        error_callback: E,
+        timeout: Option<Duration>,
+    ) -> Result<Self::Stream, Error>
+    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,16 +891,18 @@ 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<dyn Monitor> = Box::new(DisconnectManager::new(
             self.audio_device_id,
             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<dyn Monitor> = 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);