fix(asio): paused streams, draining timeouts and in-use drivers

CHANGELOG.md

@@ -49,6 +49,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **ASIO**: Fix a deadlock when dropping a `Stream` that owns another ASIO `Stream`.
 - **ASIO**: Fix loading a driver while a previous driver was still unloading.
 - **ASIO**: `Stream` no longer risks blocking or panicking in the driver callback while another stream is being created or destroyed.
+- **ASIO**: A paused output stream now goes silent instead of looping the last buffered audio.
+- **ASIO**: A device in use by an existing stream is no longer missing from device enumeration.
 - **AudioWorklet**: Fix processor construction failures not being reported to `error_callback`.
 - **AudioWorklet**: Fix dropouts in output streams when the callback buffer grows.
 - **CoreAudio**: Fix the device running at a different sample rate from the stream on hardware that reports a continuous rate range.

asio-sys/src/bindings/errors.rs

@@ -49,7 +49,10 @@ impl fmt::Display for LoadDriverError {
                 write!(f, "{err}")
             }
             LoadDriverError::DriverAlreadyExists => {
-                write!(f, "ASIO only supports loading one driver at a time")
+                write!(
+                    f,
+                    "ASIO supports only one driver at a time and a different one is already loaded"
+                )
             }
         }
     }

src/host/asio/device.rs

@@ -225,9 +225,9 @@ impl Iterator for Devices {
                         current_callback_flag: Arc::new(AtomicU32::new(u32::MAX)),
                     });
                 }
-                // A different driver is already loaded (e.g. an active Stream holds it). Stop
-                // cleanly rather than spinning through the rest of the list.
-                Err(sys::LoadDriverError::DriverAlreadyExists) => return None,
+                // Another driver is already loaded (e.g. an active Stream holds it). Only that
+                // driver's own entry can load now, so keep going to reach it instead of stopping.
+                Err(sys::LoadDriverError::DriverAlreadyExists) => continue,
                 // Driver failed to load for its own reasons; skip and try the next.
                 Err(_) => continue,
             }

src/host/asio/mod.rs

@@ -154,8 +154,8 @@ impl StreamTrait for Stream {
         Stream::pause(self)
     }
 
-    fn stop(&self, _timeout: Option<Duration>) -> Result<(), Error> {
-        Stream::pause(self)
+    fn stop(&self, timeout: Option<Duration>) -> Result<(), Error> {
+        Stream::stop(self, timeout)
     }
 
     fn now(&self) -> StreamInstant {

src/host/asio/stream.rs

@@ -15,7 +15,10 @@ use super::Device;
 use crate::{
     BufferSize, CallbackInfo, Data, Error, ErrorKind, FrameCount, I24, Sample, SampleFormat,
     SampleRate, StreamConfig, StreamInstant, StreamTimestamp,
-    host::{com, equilibrium::fill_equilibrium, error_emit::emit_error, frames_to_duration},
+    host::{
+        com, equilibrium::fill_equilibrium, error_emit::emit_error, frames_to_duration,
+        wait_for_drain,
+    },
 };
 
 /// Shared state for extending the 32-bit `timeGetTime()` millisecond counter into a
@@ -38,11 +41,11 @@ impl TimeBase {
         let epoch = if ns < prev {
             self.epoch_ns
                 .fetch_add(TIMEGETIME_WRAP_NS, Ordering::Relaxed)
-                + TIMEGETIME_WRAP_NS
+                .wrapping_add(TIMEGETIME_WRAP_NS)
         } else {
             self.epoch_ns.load(Ordering::Relaxed)
         };
-        StreamInstant::from_nanos(epoch + ns)
+        StreamInstant::from_nanos(epoch.wrapping_add(ns))
     }
 }
 
@@ -78,6 +81,8 @@ pub struct Stream {
     callback_id: sys::BufferCallbackId,
     driver_event_callback_id: sys::DriverEventCallbackId,
     time_base: Arc<TimeBase>,
+    drain_frames: Arc<AtomicU32>,
+    sample_rate: SampleRate,
 }
 
 impl Stream {
@@ -99,6 +104,15 @@ impl Stream {
         Ok(())
     }
 
+    pub fn stop(&self, timeout: Option<Duration>) -> Result<(), Error> {
+        self.pause()?;
+        wait_for_drain(
+            frames_to_duration(self.drain_frames.load(Ordering::Relaxed), self.sample_rate),
+            timeout,
+        );
+        Ok(())
+    }
+
     pub fn buffer_size(&self) -> Result<FrameCount, Error> {
         let streams = self.asio_streams.lock().map_err(|_| {
             Error::with_message(ErrorKind::StreamInvalidated, "Stream lock poisoned")
@@ -466,6 +480,8 @@ impl Device {
             callback_id,
             driver_event_callback_id,
             time_base: Arc::clone(&time_base),
+            drain_frames: Arc::new(AtomicU32::new(0)),
+            sample_rate: config.sample_rate,
         })
     }
 
@@ -538,7 +554,16 @@ impl Device {
         ));
 
         let playback_state = Arc::new(AtomicU8::new(StreamState::Starting as u8));
+        let playback_state_wrapper = Arc::clone(&playback_state);
+        let mut data_callback = move |data: &mut Data, info: &CallbackInfo| {
+            if StreamState::load(&playback_state_wrapper, Ordering::Relaxed) == StreamState::Playing
+            {
+                data_callback(data, info);
+            }
+        };
+
         let pending_xrun = Arc::new(AtomicBool::new(false));
+        let drain_frames = Arc::clone(&hardware_output_latency);
         let driver_event_callback_id = self
             .add_event_callback(
                 &driver,
@@ -564,12 +589,10 @@ impl Device {
         let time_base = Arc::new(TimeBase::default());
         let time_base_cb = Arc::clone(&time_base);
 
+        // Runs whether or not the stream is playing: the driver plays the buffers back as it finds
+        // them, so returning early here would loop the last cycle's audio. When not Playing, the
+        // user callback is suppressed above and the write below is silence instead.
         let callback_id = driver.add_callback(move |callback_info| unsafe {
-            // If not playing, return early.
-            if StreamState::load(&playback_state_cb, Ordering::Relaxed) != StreamState::Playing {
-                return;
-            }
-
             // Guard against non-conformant drivers (e.g. Focusrite USB ASIO, ReaRoute) that
             // fire the buffer callback multiple times per buffer cycle with the same buffer
             // index.
@@ -601,7 +624,11 @@ impl Device {
             let hardware_output_latency = hardware_output_latency.load(Ordering::Relaxed) as usize;
 
             let callback_instant = time_base_cb.to_stream_instant(callback_info.system_time);
-            let xrun = pending_xrun_cb.swap(false, Ordering::Relaxed);
+            // Only consume it when the user callback will actually run, so a pause does not
+            // swallow the notice.
+            let xrun = StreamState::load(&playback_state_cb, Ordering::Relaxed)
+                == StreamState::Playing
+                && pending_xrun_cb.swap(false, Ordering::Relaxed);
 
             // Silence the ASIO buffer that is about to be used.
             //
@@ -873,6 +900,8 @@ impl Device {
             callback_id,
             driver_event_callback_id,
             time_base: Arc::clone(&time_base),
+            drain_frames,
+            sample_rate: config.sample_rate,
         })
     }
 

src/host/mod.rs

@@ -270,6 +270,21 @@ pub(crate) fn frames_to_duration(
     std::time::Duration::new(secs, nanos as u32)
 }
 
+/// Waits out `window` of buffered audio, cut short by `timeout` when it is the smaller of the two.
+///
+/// Implements [`StreamTrait::stop`]'s timeout contract for the backends that approximate a drain
+/// by sleeping, rather than blocking on a native drain primitive: `None` waits the full window and
+/// `Some(Duration::ZERO)` returns immediately.
+///
+/// [`StreamTrait::stop`]: crate::traits::StreamTrait::stop
+#[cfg(windows)]
+pub(crate) fn wait_for_drain(window: std::time::Duration, timeout: Option<std::time::Duration>) {
+    let wait = timeout.map_or(window, |t| window.min(t));
+    if !wait.is_zero() {
+        std::thread::sleep(wait);
+    }
+}
+
 /// Clamps a timestamp so it never precedes one we've already returned.
 #[allow(dead_code)]
 fn non_decreasing(floor: &mut u64, instant: crate::StreamInstant) -> crate::StreamInstant {