fix(asio): share duplex buffer state across Device handles for the same driver

CHANGELOG.md

@@ -28,6 +28,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Timestamps now stay monotonic across device and graph changes.
 - **ALSA**: A nonzero but sub-millisecond stream timeout is no longer treated as a non-blocking poll.
 - **ALSA**: Fix compilation on FreeBSD, DragonFly, and NetBSD.
+- **ASIO**: Fix duplex streams silently dropping audio when built from separate `Device` handles.
 - **AudioWorklet**: Fix `Stream` operations to work when called from any thread.
 - **AudioWorklet**: Fix stale audio output when a data callback wrote a partial buffer.
 - **CoreAudio**: Default-output streams now report xrun status.

asio-sys/src/bindings/mod.rs

@@ -63,6 +63,8 @@ pub struct Driver {
 #[derive(Debug)]
 struct DriverInner {
     state: Mutex<DriverState>,
+    // Input/output buffer state, shared across every `Driver` handle for this driver.
+    streams: Arc<Mutex<AsioStreams>>,
     // The unique name associated with this driver.
     name: String,
     // Track whether or not the driver has been destroyed.
@@ -131,6 +133,7 @@ struct BufferCallback(Box<dyn FnMut(&CallbackInfo) + Send>);
 /// There is only ever max one input and one output.
 ///
 /// Only one is required.
+#[derive(Debug)]
 pub struct AsioStreams {
     pub input: Option<AsioStream>,
     pub output: Option<AsioStream>,
@@ -139,6 +142,7 @@ pub struct AsioStreams {
 /// A stream to ASIO.
 ///
 /// Contains the buffers.
+#[derive(Debug)]
 pub struct AsioStream {
     /// A Double buffer per channel
     pub buffer_infos: Vec<AsioBufferInfo>,
@@ -471,11 +475,16 @@ impl Asio {
                         CURRENT_SAMPLE_RATE.store(rate.to_bits(), Ordering::Release);
                     }
                     let state = Mutex::new(DriverState::Initialized);
+                    let streams = Arc::new(Mutex::new(AsioStreams {
+                        input: None,
+                        output: None,
+                    }));
                     let name = driver_name.to_string();
                     let destroyed = false;
                     let inner = Arc::new(DriverInner {
                         name,
                         state,
+                        streams,
                         destroyed,
                     });
                     *loaded = Arc::downgrade(&inner);
@@ -501,6 +510,11 @@ impl Driver {
         &self.inner.name
     }
 
+    /// The shared input/output buffer state for this driver.
+    pub fn streams(&self) -> Arc<Mutex<AsioStreams>> {
+        self.inner.streams.clone()
+    }
+
     /// Returns the number of input and output channels available on the driver.
     pub fn channels(&self) -> Result<Channels, AsioError> {
         let _guard = self.inner.lock_state();

examples/feedback.rs

@@ -1,15 +1,16 @@
 //! Feeds back the input stream directly into the output stream.
 //!
-//! Assumes that the input and output devices can use the same stream configuration and that they
-//! support the f32 sample format.
+//! Assumes that the input and output devices can use the same stream configuration and share a
+//! sample format.
 //!
 //! Uses a delay of `LATENCY_MS` milliseconds in case the default input and output streams are not
 //! precisely synchronised.
 
 use clap::Parser;
 use cpal::{
     traits::{DeviceTrait, HostTrait, StreamTrait},
-    Error, ErrorKind, HostId, InputCallbackInfo, OutputCallbackInfo, Sample, StreamConfig,
+    Device, Error, ErrorKind, HostId, InputCallbackInfo, OutputCallbackInfo, SampleFormat,
+    SizedSample, StreamConfig,
 };
 use ringbuf::{
     traits::{Consumer, Producer, Split},
@@ -38,17 +39,23 @@ struct Opt {
     /// Use the PulseAudio host. Requires `--features pulseaudio`.
     #[arg(long, default_value_t = false)]
     pulseaudio: bool,
+
+    /// Use the ASIO host. Requires `--features asio`.
+    #[arg(long, default_value_t = false)]
+    asio: bool,
 }
 
 fn main() -> anyhow::Result<()> {
     let opt = Opt::parse();
 
-    // JACK/PulseAudio support must be enabled at compile time, and is
+    // JACK/PulseAudio/ASIO support must be enabled at compile time, and is
     // only available on some platforms.
     #[allow(unused_mut, unused_assignments)]
     let mut jack_host_id: Result<HostId, Error> = Err(ErrorKind::HostUnavailable.into());
     #[allow(unused_mut, unused_assignments)]
     let mut pulseaudio_host_id: Result<HostId, Error> = Err(ErrorKind::HostUnavailable.into());
+    #[allow(unused_mut, unused_assignments)]
+    let mut asio_host_id: Result<HostId, Error> = Err(ErrorKind::HostUnavailable.into());
 
     #[cfg(any(
         target_os = "linux",
@@ -68,6 +75,14 @@ fn main() -> anyhow::Result<()> {
         }
     }
 
+    #[cfg(target_os = "windows")]
+    {
+        #[cfg(feature = "asio")]
+        {
+            asio_host_id = Ok(HostId::Asio);
+        }
+    }
+
     // Manually check for flags. Can be passed through cargo with -- e.g.
     // cargo run --release --example beep --features jack -- --jack
     let host = if opt.jack {
@@ -78,6 +93,10 @@ fn main() -> anyhow::Result<()> {
         pulseaudio_host_id
             .and_then(cpal::host_from_id)
             .expect("make sure `--features pulseaudio` is specified, and the platform is supported")
+    } else if opt.asio {
+        asio_host_id
+            .and_then(cpal::host_from_id)
+            .expect("make sure `--features asio` is specified, and the platform is supported")
     } else {
         cpal::default_host()
     };
@@ -103,48 +122,128 @@ fn main() -> anyhow::Result<()> {
     println!("Using output device: \"{}\"", output_device.id()?);
 
     // We'll try and use the same configuration between streams to keep it simple.
-    let config: StreamConfig = input_device.default_input_config()?.into();
+    let input_config = input_device.default_input_config()?;
+    let output_config = output_device.default_output_config()?;
+    assert_eq!(
+        input_config.sample_format(),
+        output_config.sample_format(),
+        "input and output devices must share a sample format for this example"
+    );
+
+    match input_config.sample_format() {
+        SampleFormat::I8 => run::<i8>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::I16 => run::<i16>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::I32 => run::<i32>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::I64 => run::<i64>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::U8 => run::<u8>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::U16 => run::<u16>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::U32 => run::<u32>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::U64 => run::<u64>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::F32 => run::<f32>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        SampleFormat::F64 => run::<f64>(
+            &input_device,
+            &output_device,
+            input_config.into(),
+            opt.latency,
+        ),
+        sample_format => panic!("Unsupported sample format '{sample_format}'"),
+    }
+}
 
+fn run<T>(
+    input_device: &Device,
+    output_device: &Device,
+    config: StreamConfig,
+    latency_ms: f32,
+) -> anyhow::Result<()>
+where
+    T: SizedSample + std::fmt::Debug + Send + 'static,
+{
     // Create a delay in case the input and output devices aren't synced.
-    let latency_frames = (opt.latency / 1_000.0) * config.sample_rate as f32;
+    let latency_frames = (latency_ms / 1_000.0) * config.sample_rate as f32;
     let latency_samples = latency_frames as usize * config.channels as usize;
 
     // The buffer to share samples
-    let ring = HeapRb::<f32>::new(latency_samples * 2);
+    let ring = HeapRb::<T>::new(latency_samples * 2);
     let (mut producer, mut consumer) = ring.split();
 
     // Pre-fill with silence equal to the length of the delay.
     for _ in 0..latency_samples {
         // The ring buffer has twice as much space as necessary to add latency here,
         // so this should never fail
-        producer.try_push(f32::EQUILIBRIUM).unwrap();
+        producer.try_push(T::EQUILIBRIUM).unwrap();
     }
 
-    let input_data_fn = move |data: &[f32], _: &InputCallbackInfo| {
+    let input_data_fn = move |data: &[T], _: &InputCallbackInfo| {
         if producer.push_slice(data) < data.len() {
             eprintln!("output stream fell behind: try increasing latency");
         }
     };
 
-    let output_data_fn = move |data: &mut [f32], _: &OutputCallbackInfo| {
+    let output_data_fn = move |data: &mut [T], _: &OutputCallbackInfo| {
         let read = consumer.pop_slice(data);
         if read < data.len() {
-            data[read..].fill(f32::EQUILIBRIUM);
+            data[read..].fill(T::EQUILIBRIUM);
             eprintln!("input stream fell behind: try increasing latency");
         }
     };
 
     // Build streams.
-    println!("Attempting to build both streams with f32 samples and `{config:?}`.");
+    println!(
+        "Attempting to build both streams with {} samples and `{config:?}`.",
+        T::FORMAT
+    );
     let input_stream = input_device.build_input_stream(config, input_data_fn, err_fn, None)?;
     let output_stream = output_device.build_output_stream(config, output_data_fn, err_fn, None)?;
     println!("Successfully built streams.");
 
     // Play the streams.
-    println!(
-        "Starting the input and output streams with `{}` milliseconds of latency.",
-        opt.latency
-    );
+    println!("Starting the input and output streams with `{latency_ms}` milliseconds of latency.");
     input_stream.play()?;
     output_stream.play()?;
 

src/host/asio/device.rs

@@ -213,13 +213,9 @@ impl Iterator for Devices {
                         .filter(|&r| driver.can_sample_rate(r.into()).unwrap_or(false))
                         .collect();
 
+                    let asio_streams = driver.streams();
                     self.current_driver = Some(driver);
 
-                    let asio_streams = Arc::new(Mutex::new(sys::AsioStreams {
-                        input: None,
-                        output: None,
-                    }));
-
                     return Some(Device {
                         name,
                         channels_in: channels.ins as ChannelCount,