Merging upstream version 126.0.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

11 files changed, 2633 insertions, 763 deletions

diff --git a/third_party/rust/cubeb-coreaudio/src/backend/aggregate_device.rs b/third_party/rust/cubeb-coreaudio/src/backend/aggregate_device.rs
index 2738631b87..782e76de2f 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/aggregate_device.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/aggregate_device.rs
@@ -69,6 +69,7 @@ impl AggregateDevice {
         input_id: AudioObjectID,
         output_id: AudioObjectID,
     ) -> std::result::Result<Self, Error> {
+        debug_assert_running_serially();
         let plugin_id = Self::get_system_plugin_id()?;
         let device_id = Self::create_blank_device_sync(plugin_id)?;
 
@@ -399,12 +400,12 @@ impl AggregateDevice {
             let sub_devices = CFArrayCreateMutable(ptr::null(), 0, &kCFTypeArrayCallBacks);
             // The order of the items in the array is significant and is used to determine the order of the streams
             // of the AudioAggregateDevice.
-            for device in output_sub_devices {
+            for device in input_sub_devices {
                 let uid = get_device_global_uid(device)?;
                 CFArrayAppendValue(sub_devices, uid.get_raw() as *const c_void);
             }
 
-            for device in input_sub_devices {
+            for device in output_sub_devices {
                 let uid = get_device_global_uid(device)?;
                 CFArrayAppendValue(sub_devices, uid.get_raw() as *const c_void);
             }
@@ -466,6 +467,28 @@ impl AggregateDevice {
         }
     }
 
+    pub fn get_master_device_uid(device_id: AudioDeviceID) -> std::result::Result<String, Error> {
+        let address = AudioObjectPropertyAddress {
+            mSelector: kAudioAggregateDevicePropertyMainSubDevice,
+            mScope: kAudioObjectPropertyScopeGlobal,
+            mElement: kAudioObjectPropertyElementMaster,
+        };
+
+        let mut master: CFStringRef = ptr::null_mut();
+        let mut size = mem::size_of::<CFStringRef>();
+        let status = audio_object_get_property_data(device_id, &address, &mut size, &mut master);
+        if status != NO_ERR {
+            return Err(Error::from(status));
+        }
+
+        if master.is_null() {
+            return Ok(String::default());
+        }
+
+        let master = StringRef::new(master as _);
+        Ok(master.into_string())
+    }
+
     pub fn set_master_device(
         device_id: AudioDeviceID,
         primary_id: AudioDeviceID,
@@ -480,12 +503,12 @@ impl AggregateDevice {
         );
 
         let address = AudioObjectPropertyAddress {
-            mSelector: kAudioAggregateDevicePropertyMasterSubDevice,
+            mSelector: kAudioAggregateDevicePropertyMainSubDevice,
             mScope: kAudioObjectPropertyScopeGlobal,
             mElement: kAudioObjectPropertyElementMaster,
         };
 
-        // Master become the 1st sub device of the primary device
+        // The master device will be the 1st sub device of the primary device.
         let output_sub_devices = Self::get_sub_devices(primary_id)?;
         assert!(!output_sub_devices.is_empty());
         let master_sub_device_uid = get_device_global_uid(output_sub_devices[0]).unwrap();
@@ -548,16 +571,23 @@ impl AggregateDevice {
             return Err(Error::from(status));
         }
 
+        let master_sub_device_uid = Self::get_master_device_uid(device_id)?;
+
         let address = AudioObjectPropertyAddress {
             mSelector: kAudioSubDevicePropertyDriftCompensation,
             mScope: kAudioObjectPropertyScopeGlobal,
             mElement: kAudioObjectPropertyElementMaster,
         };
 
-        // Start from the second device since the first is the master clock
-        for device in &sub_devices[1..] {
+        for &device in &sub_devices {
+            let uid = get_device_global_uid(device)
+                .map(|sr| sr.into_string())
+                .unwrap_or_default();
+            if uid == master_sub_device_uid {
+                continue;
+            }
             let status = audio_object_set_property_data(
-                *device,
+                device,
                 &address,
                 mem::size_of::<u32>(),
                 &DRIFT_COMPENSATION,
@@ -671,6 +701,7 @@ impl Default for AggregateDevice {
 
 impl Drop for AggregateDevice {
     fn drop(&mut self) {
+        debug_assert_running_serially();
         if self.plugin_id != kAudioObjectUnknown && self.device_id != kAudioObjectUnknown {
             if let Err(r) = Self::destroy_device(self.plugin_id, self.device_id) {
                 cubeb_log!(
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/mod.rs b/third_party/rust/cubeb-coreaudio/src/backend/mod.rs
index 61ae44fea1..e6be028a2e 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/mod.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/mod.rs
@@ -46,15 +46,15 @@ use std::mem;
 use std::os::raw::{c_uint, c_void};
 use std::ptr;
 use std::slice;
+use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
-use std::sync::{Arc, Condvar, Mutex};
-use std::time::Duration;
+use std::sync::{Arc, Condvar, Mutex, MutexGuard, Weak};
+use std::time::{Duration, Instant};
 const NO_ERR: OSStatus = 0;
 
 const AU_OUT_BUS: AudioUnitElement = 0;
 const AU_IN_BUS: AudioUnitElement = 1;
 
-const DISPATCH_QUEUE_LABEL: &str = "org.mozilla.cubeb";
 const PRIVATE_AGGREGATE_DEVICE_NAME: &str = "CubebAggregateDevice";
 const VOICEPROCESSING_AGGREGATE_DEVICE_NAME: &str = "VPAUAggregateAudioDevice";
 
@@ -65,6 +65,34 @@ const APPLE_STUDIO_DISPLAY_USB_ID: &str = "05AC:1114";
 const SAFE_MIN_LATENCY_FRAMES: u32 = 128;
 const SAFE_MAX_LATENCY_FRAMES: u32 = 512;
 
+const VPIO_IDLE_TIMEOUT: Duration = Duration::from_secs(10);
+
+const MACOS_KERNEL_MAJOR_VERSION_MONTEREY: u32 = 21;
+
+#[derive(Debug, PartialEq)]
+enum ParseMacOSKernelVersionError {
+    SysCtl,
+    Malformed,
+    Parsing,
+}
+
+fn macos_kernel_major_version() -> std::result::Result<u32, ParseMacOSKernelVersionError> {
+    let ver = whatsys::kernel_version();
+    if ver.is_none() {
+        return Err(ParseMacOSKernelVersionError::SysCtl);
+    }
+    let ver = ver.unwrap();
+    let major = ver.split('.').next();
+    if major.is_none() {
+        return Err(ParseMacOSKernelVersionError::Malformed);
+    }
+    let parsed_major = u32::from_str(major.unwrap());
+    if parsed_major.is_err() {
+        return Err(ParseMacOSKernelVersionError::Parsing);
+    }
+    Ok(parsed_major.unwrap())
+}
+
 bitflags! {
     #[allow(non_camel_case_types)]
     #[derive(Clone, Debug, PartialEq, Copy)]
@@ -191,6 +219,7 @@ fn set_notification_runloop() {
 
 fn create_device_info(devid: AudioDeviceID, devtype: DeviceType) -> Option<device_info> {
     assert_ne!(devid, kAudioObjectSystemObject);
+    debug_assert_running_serially();
 
     let mut flags = match devtype {
         DeviceType::INPUT => device_flags::DEV_INPUT,
@@ -457,7 +486,6 @@ extern "C" fn audiounit_input_callback(
 
     assert!(!user_ptr.is_null());
     let stm = unsafe { &mut *(user_ptr as *mut AudioUnitStream) };
-    let using_voice_processing_unit = stm.core_stream_data.using_voice_processing_unit();
 
     if unsafe { *flags | kAudioTimeStampHostTimeValid } != 0 {
         let now = unsafe { mach_absolute_time() };
@@ -597,17 +625,20 @@ extern "C" fn audiounit_input_callback(
         ErrorHandle::Return(NO_ERR)
     };
 
-    // If the input (input-only stream) or the output is drained (duplex stream),
-    // cancel this callback. Note that for voice processing cases (a single unit),
-    // the output callback handles stopping the unit and notifying of state.
-    if !using_voice_processing_unit && stm.draining.load(Ordering::SeqCst) {
-        let r = stop_audiounit(stm.core_stream_data.input_unit);
-        assert!(r.is_ok());
-        // Only fire state-changed callback for input-only stream.
-        // The state-changed callback for the duplex stream is fired in the output callback.
-        if stm.core_stream_data.output_unit.is_null() {
-            stm.notify_state_changed(State::Drained);
-        }
+    // If the input (input-only stream) is drained, cancel this callback. Whenever an output
+    // is involved, the output callback handles stopping all units and notifying of state.
+    if stm.core_stream_data.output_unit.is_null() && stm.draining.load(Ordering::SeqCst) {
+        stm.stopped.store(true, Ordering::SeqCst);
+        cubeb_alog!("({:p}) Input-only drained.", stm as *const AudioUnitStream);
+        stm.notify_state_changed(State::Drained);
+        let queue = stm.queue.clone();
+        // Use a new thread, through the queue, to avoid deadlock when calling
+        // AudioOutputUnitStop method from inside render callback
+        let stm_ptr = user_ptr as usize;
+        queue.run_async(move || {
+            let stm = unsafe { &mut *(stm_ptr as *mut AudioUnitStream) };
+            stm.core_stream_data.stop_audiounits();
+        });
     }
 
     match handle {
@@ -681,11 +712,17 @@ extern "C" fn audiounit_output_callback(
     }
 
     if stm.draining.load(Ordering::SeqCst) {
-        // Cancel the output callback only. For duplex stream,
-        // the input callback will be cancelled in its own callback.
-        let r = stop_audiounit(stm.core_stream_data.output_unit);
-        assert!(r.is_ok());
+        // Cancel all callbacks. For input-only streams, the input callback handles
+        // cancelling itself.
+        stm.stopped.store(true, Ordering::SeqCst);
+        cubeb_alog!("({:p}) output drained.", stm as *const AudioUnitStream);
         stm.notify_state_changed(State::Drained);
+        let queue = stm.queue.clone();
+        // Use a new thread, through the queue, to avoid deadlock when calling
+        // AudioOutputUnitStop method from inside render callback
+        queue.run_async(move || {
+            stm.core_stream_data.stop_audiounits();
+        });
         audiounit_make_silent(&buffers[0]);
         return NO_ERR;
     }
@@ -1157,38 +1194,56 @@ fn create_audiounit(device: &device_info) -> Result<AudioUnit> {
     Ok(unit)
 }
 
-fn create_voiceprocessing_audiounit(
+fn get_voiceprocessing_audiounit(
+    shared_voice_processing_unit: &mut SharedVoiceProcessingUnitManager,
     in_device: &device_info,
     out_device: &device_info,
-) -> Result<AudioUnit> {
+) -> Result<OwningHandle<VoiceProcessingUnit>> {
+    debug_assert_running_serially();
     assert!(in_device.flags.contains(device_flags::DEV_INPUT));
     assert!(!in_device.flags.contains(device_flags::DEV_OUTPUT));
     assert!(!out_device.flags.contains(device_flags::DEV_INPUT));
-    assert!(out_device.flags.contains(device_flags::DEV_OUTPUT));
-
-    let unit = create_typed_audiounit(kAudioUnitSubType_VoiceProcessingIO)?;
 
-    if let Err(e) = set_device_to_audiounit(unit, in_device.id, AU_IN_BUS) {
+    let unit_handle = shared_voice_processing_unit.take_or_create();
+    if let Err(e) = unit_handle {
         cubeb_log!(
-            "Failed to set in device {} to the created audiounit. Error: {}",
-            in_device.id,
+            "Failed to create shared voiceprocessing audiounit. Error: {}",
             e
         );
-        dispose_audio_unit(unit);
         return Err(Error::error());
     }
+    let mut unit_handle = unit_handle.unwrap();
 
-    if let Err(e) = set_device_to_audiounit(unit, out_device.id, AU_OUT_BUS) {
+    if let Err(e) = set_device_to_audiounit(unit_handle.as_mut().unit, in_device.id, AU_IN_BUS) {
         cubeb_log!(
-            "Failed to set out device {} to the created audiounit. Error: {}",
-            out_device.id,
+            "Failed to set in device {} to the created audiounit. Error: {}",
+            in_device.id,
             e
         );
-        dispose_audio_unit(unit);
         return Err(Error::error());
     }
 
-    Ok(unit)
+    let has_output = out_device.id != kAudioObjectUnknown;
+    if let Err(e) =
+        enable_audiounit_scope(unit_handle.as_mut().unit, DeviceType::OUTPUT, has_output)
+    {
+        cubeb_log!("Failed to enable audiounit input scope. Error: {}", e);
+        return Err(Error::error());
+    }
+    if has_output {
+        if let Err(e) =
+            set_device_to_audiounit(unit_handle.as_mut().unit, out_device.id, AU_OUT_BUS)
+        {
+            cubeb_log!(
+                "Failed to set out device {} to the created audiounit. Error: {}",
+                out_device.id,
+                e
+            );
+            return Err(Error::error());
+        }
+    }
+
+    Ok(unit_handle)
 }
 
 fn enable_audiounit_scope(
@@ -1276,6 +1331,31 @@ fn create_blank_audiounit() -> Result<AudioUnit> {
     return create_typed_audiounit(kAudioUnitSubType_RemoteIO);
 }
 
+fn create_voiceprocessing_audiounit() -> Result<VoiceProcessingUnit> {
+    let res = create_typed_audiounit(kAudioUnitSubType_VoiceProcessingIO);
+    if res.is_err() {
+        return Err(Error::error());
+    }
+
+    match get_default_device(DeviceType::OUTPUT) {
+        None => {
+            cubeb_log!("Could not get default output device in order to undo vpio ducking");
+        }
+        Some(id) => {
+            let r = audio_device_duck(id, 1.0, ptr::null_mut(), 0.5);
+            if r != NO_ERR {
+                cubeb_log!(
+                        "Failed to undo ducking of voiceprocessing on output device {}. Proceeding... Error: {}",
+                        id,
+                        r
+                    );
+            }
+        }
+    };
+
+    res.map(|unit| VoiceProcessingUnit { unit })
+}
+
 fn get_buffer_size(unit: AudioUnit, devtype: DeviceType) -> std::result::Result<u32, OSStatus> {
     assert!(!unit.is_null());
     let (scope, element) = match devtype {
@@ -1558,7 +1638,7 @@ fn get_range_of_sample_rates(
     if rates.is_empty() {
         return Err(String::from("No data"));
     }
-    let (mut min, mut max) = (std::f64::MAX, std::f64::MIN);
+    let (mut min, mut max) = (f64::MAX, f64::MIN);
     for rate in rates {
         if rate.mMaximum > max {
             max = rate.mMaximum;
@@ -2056,7 +2136,7 @@ struct LatencyController {
 }
 
 impl LatencyController {
-    fn add_stream(&mut self, latency: u32) -> Option<u32> {
+    fn add_stream(&mut self, latency: u32) -> u32 {
         self.streams += 1;
         // For the 1st stream set anything within safe min-max
         if self.streams == 1 {
@@ -2065,16 +2145,322 @@ impl LatencyController {
             // synthetize the clock from the callbacks, and we want the clock to update often.
             self.latency = Some(latency.clamp(SAFE_MIN_LATENCY_FRAMES, SAFE_MAX_LATENCY_FRAMES));
         }
-        self.latency
+        self.latency.unwrap_or(latency)
     }
 
-    fn subtract_stream(&mut self) -> Option<u32> {
+    fn subtract_stream(&mut self) {
         self.streams -= 1;
         if self.streams == 0 {
             assert!(self.latency.is_some());
             self.latency = None;
         }
-        self.latency
+    }
+}
+
+// SharedStorage<T> below looks generic but has evolved to be pretty tailored
+// the observed behavior of VoiceProcessingIO audio units on macOS 14.
+// Some key points are:
+// - Creating the first VoiceProcessingIO unit in a process takes a long time, often > 3s.
+// - Creating a second VoiceProcessingIO unit in a process is significantly faster, < 1s.
+// - Disposing of a VoiceProcessingIO unit when all other VoiceProcessingIO units are
+//   uninitialized will take significantly longer than disposing the remaining
+//   VoiceProcessingIO units, and will have other side effects: starting another
+//   VoiceProcessingIO unit after this is on par with creating the first one in the
+//   process, bluetooth devices will move away from the handsfree profile, etc.
+// The takeaway is that there is something internal to the VoiceProcessingIO audio unit
+// that is costly to create and dispose of and its creation is triggered by creation of
+// the first VoiceProcessingIO unit, and its disposal is triggered by the disposal of
+// the first VoiceProcessingIO unit when no other VoiceProcessingIO units are initialized.
+//
+// The intended behavior of SharedStorage<T> and SharedVoiceProcessingUnitManager is therefore:
+// - Retain ideally just one VoiceProcessingIO unit after stream destruction, so device
+//   switching is fast. The benefit of retaining more than one is unclear.
+// - Dispose of either all VoiceProcessingIO units, or none at all, such that the retained
+//   VoiceProcessingIO unit really helps speed up creating and starting the next. In practice
+//   this means we retain all VoiceProcessingIO units until they can all be disposed of.
+
+#[derive(Debug)]
+struct SharedStorageInternal<T> {
+    // Storage for shared elements.
+    elements: Vec<T>,
+    // Number of elements in use, i.e. all elements created/taken and not recycled.
+    outstanding_element_count: usize,
+    // Used for invalidation of in-flight tasks to clear elements.
+    // Incremented when something takes a shared element.
+    generation: usize,
+}
+
+#[derive(Debug)]
+struct SharedStorage<T> {
+    queue: Queue,
+    idle_timeout: Duration,
+    storage: Mutex<SharedStorageInternal<T>>,
+}
+
+impl<T: Send> SharedStorage<T> {
+    fn with_idle_timeout(queue: Queue, idle_timeout: Duration) -> Self {
+        Self {
+            queue,
+            idle_timeout,
+            storage: Mutex::new(SharedStorageInternal::<T> {
+                elements: Vec::default(),
+                outstanding_element_count: 0,
+                generation: 0,
+            }),
+        }
+    }
+
+    fn take_locked(guard: &mut MutexGuard<'_, SharedStorageInternal<T>>) -> Result<T> {
+        if let Some(e) = guard.elements.pop() {
+            cubeb_log!("Taking shared element #{}.", guard.elements.len());
+            guard.outstanding_element_count += 1;
+            guard.generation += 1;
+            return Ok(e);
+        }
+
+        Err(Error::not_supported())
+    }
+
+    fn create_with_locked<F>(
+        guard: &mut MutexGuard<'_, SharedStorageInternal<T>>,
+        f: F,
+    ) -> Result<T>
+    where
+        F: FnOnce() -> Result<T>,
+    {
+        let start = Instant::now();
+        match f() {
+            Ok(obj) => {
+                cubeb_log!(
+                    "Just created shared element #{}. Took {}s.",
+                    guard.outstanding_element_count,
+                    (Instant::now() - start).as_secs_f32()
+                );
+                guard.outstanding_element_count += 1;
+                guard.generation += 1;
+                Ok(obj)
+            }
+            Err(_) => {
+                cubeb_log!("Creating shared element failed");
+                Err(Error::error())
+            }
+        }
+    }
+
+    #[cfg(test)]
+    fn take(&self) -> Result<T> {
+        let mut guard = self.storage.lock().unwrap();
+        SharedStorage::take_locked(&mut guard)
+    }
+
+    fn take_or_create_with<F>(&self, f: F) -> Result<T>
+    where
+        F: FnOnce() -> Result<T>,
+    {
+        let mut guard = self.storage.lock().unwrap();
+        SharedStorage::take_locked(&mut guard)
+            .or_else(|_| SharedStorage::create_with_locked(&mut guard, f))
+    }
+
+    fn recycle(&self, obj: T) {
+        let mut guard = self.storage.lock().unwrap();
+        guard.outstanding_element_count -= 1;
+        cubeb_log!(
+            "Recycling shared element #{}. Nr of live elements now {}.",
+            guard.elements.len(),
+            guard.outstanding_element_count
+        );
+        guard.elements.push(obj);
+    }
+
+    fn clear_locked(guard: &mut MutexGuard<'_, SharedStorageInternal<T>>) {
+        let count = guard.elements.len();
+        let start = Instant::now();
+        guard.elements.clear();
+        cubeb_log!(
+            "Cleared {} shared element{}. Took {}s.",
+            count,
+            if count == 1 { "" } else { "s" },
+            (Instant::now() - start).as_secs_f32()
+        );
+    }
+
+    fn clear(&self) {
+        debug_assert_running_serially();
+        let mut guard = self.storage.lock().unwrap();
+        SharedStorage::clear_locked(&mut guard);
+    }
+
+    fn clear_if_all_idle_async(storage: &Arc<SharedStorage<T>>) {
+        let (queue, outstanding_element_count, generation) = {
+            let guard = storage.storage.lock().unwrap();
+            (
+                storage.queue.clone(),
+                guard.outstanding_element_count,
+                guard.generation,
+            )
+        };
+        if outstanding_element_count > 0 {
+            cubeb_log!(
+                "Not clearing shared voiceprocessing unit storage because {} elements are in use. Generation={}.",
+                outstanding_element_count,
+                generation
+            );
+            return;
+        }
+        cubeb_log!(
+            "Clearing shared voiceprocessing unit storage in {}s if still at generation {}.",
+            storage.idle_timeout.as_secs_f32(),
+            generation
+        );
+        let storage = storage.clone();
+        queue.run_after(Instant::now() + storage.idle_timeout, move || {
+            let mut guard = storage.storage.lock().unwrap();
+            if generation != guard.generation {
+                cubeb_log!(
+                    "Not clearing shared voiceprocessing unit storage for generation {} as we're now at {}.",
+                    generation,
+                    guard.generation
+                );
+                return;
+            }
+            SharedStorage::clear_locked(&mut guard);
+        });
+    }
+}
+
+#[derive(Debug)]
+struct OwningHandle<T>
+where
+    T: Send,
+{
+    storage: Weak<SharedStorage<T>>,
+    obj: Option<T>,
+}
+
+impl<T: Send> OwningHandle<T> {
+    fn new(storage: Weak<SharedStorage<T>>, obj: T) -> Self {
+        Self {
+            storage,
+            obj: Some(obj),
+        }
+    }
+}
+
+impl<T: Send> AsRef<T> for OwningHandle<T> {
+    fn as_ref(&self) -> &T {
+        self.obj.as_ref().unwrap()
+    }
+}
+
+impl<T: Send> AsMut<T> for OwningHandle<T> {
+    fn as_mut(&mut self) -> &mut T {
+        self.obj.as_mut().unwrap()
+    }
+}
+
+impl<T: Send> Drop for OwningHandle<T> {
+    fn drop(&mut self) {
+        let storage = self.storage.upgrade();
+        assert!(
+            storage.is_some(),
+            "Storage must outlive the handle, but didn't"
+        );
+        let storage = storage.unwrap();
+        if self.obj.is_none() {
+            return;
+        }
+        let obj = self.obj.take().unwrap();
+        storage.recycle(obj);
+        SharedStorage::clear_if_all_idle_async(&storage);
+    }
+}
+
+#[derive(Debug)]
+struct VoiceProcessingUnit {
+    unit: AudioUnit,
+}
+
+impl Drop for VoiceProcessingUnit {
+    fn drop(&mut self) {
+        assert!(!self.unit.is_null());
+        dispose_audio_unit(self.unit);
+    }
+}
+
+unsafe impl Send for VoiceProcessingUnit {}
+
+#[derive(Debug)]
+struct SharedVoiceProcessingUnitManager {
+    sync_storage: Mutex<Option<Arc<SharedStorage<VoiceProcessingUnit>>>>,
+    queue: Queue,
+    idle_timeout: Duration,
+}
+
+impl SharedVoiceProcessingUnitManager {
+    fn with_idle_timeout(queue: Queue, idle_timeout: Duration) -> Self {
+        Self {
+            sync_storage: Mutex::new(None),
+            queue,
+            idle_timeout,
+        }
+    }
+
+    fn new(queue: Queue) -> Self {
+        SharedVoiceProcessingUnitManager::with_idle_timeout(queue, VPIO_IDLE_TIMEOUT)
+    }
+
+    fn ensure_storage_locked(
+        &self,
+        guard: &mut MutexGuard<Option<Arc<SharedStorage<VoiceProcessingUnit>>>>,
+    ) {
+        if guard.is_some() {
+            return;
+        }
+        cubeb_log!("Creating shared voiceprocessing storage.");
+        let storage = SharedStorage::<VoiceProcessingUnit>::with_idle_timeout(
+            self.queue.clone(),
+            self.idle_timeout,
+        );
+        let old_storage = guard.replace(Arc::from(storage));
+        assert!(old_storage.is_none());
+    }
+
+    // Take an already existing, shared, vpio unit, if one is available.
+    #[cfg(test)]
+    fn take(&mut self) -> Result<OwningHandle<VoiceProcessingUnit>> {
+        debug_assert_running_serially();
+        let mut guard = self.sync_storage.lock().unwrap();
+        self.ensure_storage_locked(&mut guard);
+        let storage = guard.as_mut().unwrap();
+        let res = storage.take();
+        res.map(|u| OwningHandle::new(Arc::downgrade(storage), u))
+    }
+
+    // Take an already existing, shared, vpio unit, or create one if none are available.
+    fn take_or_create(&mut self) -> Result<OwningHandle<VoiceProcessingUnit>> {
+        debug_assert_running_serially();
+        let mut guard = self.sync_storage.lock().unwrap();
+        self.ensure_storage_locked(&mut guard);
+        let storage = guard.as_mut().unwrap();
+        let res = storage.take_or_create_with(create_voiceprocessing_audiounit);
+        res.map(|u| OwningHandle::new(Arc::downgrade(storage), u))
+    }
+}
+
+unsafe impl Send for SharedVoiceProcessingUnitManager {}
+unsafe impl Sync for SharedVoiceProcessingUnitManager {}
+
+impl Drop for SharedVoiceProcessingUnitManager {
+    fn drop(&mut self) {
+        debug_assert_not_running_serially();
+        self.queue.run_final(|| {
+            let mut guard = self.sync_storage.lock().unwrap();
+            if guard.is_none() {
+                return;
+            }
+            guard.as_mut().unwrap().clear();
+        });
     }
 }
 
@@ -2091,15 +2477,26 @@ pub struct AudioUnitContext {
     serial_queue: Queue,
     latency_controller: Mutex<LatencyController>,
     devices: Mutex<SharedDevices>,
+    // Storage for a context-global vpio unit. Duplex streams that need one will take this
+    // and return it when done.
+    shared_voice_processing_unit: SharedVoiceProcessingUnitManager,
 }
 
 impl AudioUnitContext {
     fn new() -> Self {
+        let queue_label = format!("{}.context", DISPATCH_QUEUE_LABEL);
+        let serial_queue =
+            Queue::new_with_target(queue_label.as_str(), get_serial_queue_singleton());
+        let shared_vp_queue = Queue::new_with_target(
+            format!("{}.context.shared_vpio", DISPATCH_QUEUE_LABEL).as_str(),
+            &serial_queue,
+        );
         Self {
             _ops: &OPS as *const _,
-            serial_queue: Queue::new(DISPATCH_QUEUE_LABEL),
+            serial_queue,
             latency_controller: Mutex::new(LatencyController::default()),
             devices: Mutex::new(SharedDevices::default()),
+            shared_voice_processing_unit: SharedVoiceProcessingUnitManager::new(shared_vp_queue),
         }
     }
 
@@ -2108,14 +2505,14 @@ impl AudioUnitContext {
         controller.streams
     }
 
-    fn update_latency_by_adding_stream(&self, latency_frames: u32) -> Option<u32> {
+    fn update_latency_by_adding_stream(&self, latency_frames: u32) -> u32 {
         let mut controller = self.latency_controller.lock().unwrap();
         controller.add_stream(latency_frames)
     }
 
-    fn update_latency_by_removing_stream(&self) -> Option<u32> {
+    fn update_latency_by_removing_stream(&self) {
         let mut controller = self.latency_controller.lock().unwrap();
-        controller.subtract_stream()
+        controller.subtract_stream();
     }
 
     fn add_devices_changed_listener(
@@ -2228,8 +2625,16 @@ impl AudioUnitContext {
 
 impl ContextOps for AudioUnitContext {
     fn init(_context_name: Option<&CStr>) -> Result<Context> {
-        set_notification_runloop();
-        let ctx = Box::new(AudioUnitContext::new());
+        run_serially(set_notification_runloop);
+        let mut ctx = Box::new(AudioUnitContext::new());
+        let queue_label = format!("{}.context.{:p}", DISPATCH_QUEUE_LABEL, ctx.as_ref());
+        ctx.serial_queue =
+            Queue::new_with_target(queue_label.as_str(), get_serial_queue_singleton());
+        let shared_vp_queue = Queue::new_with_target(
+            format!("{}.shared_vpio", queue_label).as_str(),
+            &ctx.serial_queue,
+        );
+        ctx.shared_voice_processing_unit = SharedVoiceProcessingUnitManager::new(shared_vp_queue);
         Ok(unsafe { Context::from_ptr(Box::into_raw(ctx) as *mut _) })
     }
 
@@ -2379,28 +2784,18 @@ impl ContextOps for AudioUnitContext {
             return Err(Error::invalid_parameter());
         }
 
-        // Latency cannot change if another stream is operating in parallel. In this case
-        // latency is set to the other stream value.
-        let global_latency_frames = self
-            .update_latency_by_adding_stream(latency_frames)
-            .unwrap();
-        if global_latency_frames != latency_frames {
-            cubeb_log!(
-                "Use global latency {} instead of the requested latency {}.",
-                global_latency_frames,
-                latency_frames
-            );
-        }
-
         let in_stm_settings = if let Some(params) = input_stream_params {
-            let in_device =
-                match create_device_info(input_device as AudioDeviceID, DeviceType::INPUT) {
-                    None => {
-                        cubeb_log!("Fail to create device info for input");
-                        return Err(Error::error());
-                    }
-                    Some(d) => d,
-                };
+            let in_device = match self
+                .serial_queue
+                .run_sync(|| create_device_info(input_device as AudioDeviceID, DeviceType::INPUT))
+                .unwrap()
+            {
+                None => {
+                    cubeb_log!("Fail to create device info for input");
+                    return Err(Error::error());
+                }
+                Some(d) => d,
+            };
             let stm_params = StreamParams::from(unsafe { *params.as_ptr() });
             Some((stm_params, in_device))
         } else {
@@ -2408,20 +2803,34 @@ impl ContextOps for AudioUnitContext {
         };
 
         let out_stm_settings = if let Some(params) = output_stream_params {
-            let out_device =
-                match create_device_info(output_device as AudioDeviceID, DeviceType::OUTPUT) {
-                    None => {
-                        cubeb_log!("Fail to create device info for output");
-                        return Err(Error::error());
-                    }
-                    Some(d) => d,
-                };
+            let out_device = match self
+                .serial_queue
+                .run_sync(|| create_device_info(output_device as AudioDeviceID, DeviceType::OUTPUT))
+                .unwrap()
+            {
+                None => {
+                    cubeb_log!("Fail to create device info for output");
+                    return Err(Error::error());
+                }
+                Some(d) => d,
+            };
             let stm_params = StreamParams::from(unsafe { *params.as_ptr() });
             Some((stm_params, out_device))
         } else {
             None
         };
 
+        // Latency cannot change if another stream is operating in parallel. In this case
+        // latency is set to the other stream value.
+        let global_latency_frames = self.update_latency_by_adding_stream(latency_frames);
+        if global_latency_frames != latency_frames {
+            cubeb_log!(
+                "Use global latency {} instead of the requested latency {}.",
+                global_latency_frames,
+                latency_frames
+            );
+        }
+
         let mut boxed_stream = Box::new(AudioUnitStream::new(
             self,
             user_ptr,
@@ -2431,16 +2840,25 @@ impl ContextOps for AudioUnitContext {
         ));
 
         // Rename the task queue to be an unique label.
-        let queue_label = format!("{}.{:p}", DISPATCH_QUEUE_LABEL, boxed_stream.as_ref());
-        boxed_stream.queue = Queue::new(queue_label.as_str());
+        let queue_label = format!(
+            "{}.stream.{:p}",
+            DISPATCH_QUEUE_LABEL,
+            boxed_stream.as_ref()
+        );
+        boxed_stream.queue = Queue::new_with_target(queue_label.as_str(), &boxed_stream.queue);
 
         boxed_stream.core_stream_data =
             CoreStreamData::new(boxed_stream.as_ref(), in_stm_settings, out_stm_settings);
 
-        let mut result = Ok(());
-        boxed_stream.queue.clone().run_sync(|| {
-            result = boxed_stream.core_stream_data.setup();
-        });
+        let result = boxed_stream
+            .queue
+            .clone()
+            .run_sync(|| {
+                boxed_stream
+                    .core_stream_data
+                    .setup(&mut boxed_stream.context.shared_voice_processing_unit)
+            })
+            .unwrap();
         if let Err(r) = result {
             cubeb_log!(
                 "({:p}) Could not setup the audiounit stream.",
@@ -2465,25 +2883,43 @@ impl ContextOps for AudioUnitContext {
         if devtype == DeviceType::UNKNOWN {
             return Err(Error::invalid_parameter());
         }
-        if collection_changed_callback.is_some() {
-            self.add_devices_changed_listener(devtype, collection_changed_callback, user_ptr)
-        } else {
-            self.remove_devices_changed_listener(devtype)
-        }
+        self.serial_queue
+            .clone()
+            .run_sync(|| {
+                if collection_changed_callback.is_some() {
+                    self.add_devices_changed_listener(
+                        devtype,
+                        collection_changed_callback,
+                        user_ptr,
+                    )
+                } else {
+                    self.remove_devices_changed_listener(devtype)
+                }
+            })
+            .unwrap()
     }
 }
 
 impl Drop for AudioUnitContext {
     fn drop(&mut self) {
-        let devices = self.devices.lock().unwrap();
-        assert!(
+        assert!({
+            let devices = self.devices.lock().unwrap();
             devices.input.changed_callback.is_none() && devices.output.changed_callback.is_none()
-        );
+        });
+
+        self.shared_voice_processing_unit =
+            SharedVoiceProcessingUnitManager::new(self.serial_queue.clone());
+
+        // Make sure all the pending (device-collection-changed-callback) tasks
+        // in queue are done, and cancel all the tasks appended after `drop` is executed.
+        let queue = self.serial_queue.clone();
+        queue.run_final(|| {});
 
         {
             let controller = self.latency_controller.lock().unwrap();
-            // Disabling this assert for bug 1083664 -- we seem to leak a stream
+            // Disabling this assert in release for bug 1083664 -- we seem to leak a stream
             // assert(controller.streams == 0);
+            debug_assert!(controller.streams == 0);
             if controller.streams > 0 {
                 cubeb_log!(
                     "({:p}) API misuse, {} streams active when context destroyed!",
@@ -2492,10 +2928,6 @@ impl Drop for AudioUnitContext {
                 );
             }
         }
-        // Make sure all the pending (device-collection-changed-callback) tasks
-        // in queue are done, and cancel all the tasks appended after `drop` is executed.
-        let queue = self.serial_queue.clone();
-        queue.run_final(|| {});
     }
 }
 
@@ -2562,6 +2994,8 @@ struct CoreStreamData<'ctx> {
     // I/O AudioUnits.
     input_unit: AudioUnit,
     output_unit: AudioUnit,
+    // Handle to shared voiceprocessing AudioUnit, if in use.
+    voiceprocessing_unit_handle: Option<OwningHandle<VoiceProcessingUnit>>,
     // Info of the I/O devices.
     input_device: device_info,
     output_device: device_info,
@@ -2603,6 +3037,7 @@ impl<'ctx> Default for CoreStreamData<'ctx> {
             output_dev_desc: AudioStreamBasicDescription::default(),
             input_unit: ptr::null_mut(),
             output_unit: ptr::null_mut(),
+            voiceprocessing_unit_handle: None,
             input_device: device_info::default(),
             output_device: device_info::default(),
             input_processing_params: InputProcessingParams::NONE,
@@ -2649,6 +3084,7 @@ impl<'ctx> CoreStreamData<'ctx> {
             output_dev_desc: AudioStreamBasicDescription::default(),
             input_unit: ptr::null_mut(),
             output_unit: ptr::null_mut(),
+            voiceprocessing_unit_handle: None,
             input_device: in_dev,
             output_device: out_dev,
             input_processing_params: InputProcessingParams::NONE,
@@ -2716,7 +3152,7 @@ impl<'ctx> CoreStreamData<'ctx> {
     }
 
     fn using_voice_processing_unit(&self) -> bool {
-        !self.input_unit.is_null() && self.input_unit == self.output_unit
+        self.voiceprocessing_unit_handle.is_some()
     }
 
     fn same_clock_domain(&self) -> bool {
@@ -2744,6 +3180,22 @@ impl<'ctx> CoreStreamData<'ctx> {
         input_domain == output_domain
     }
 
+    #[allow(non_upper_case_globals)]
+    fn should_force_vpio_for_input_device(&self, in_device: &device_info) -> bool {
+        assert!(in_device.id != kAudioObjectUnknown);
+        self.debug_assert_is_on_stream_queue();
+        match get_device_transport_type(in_device.id, DeviceType::INPUT) {
+            Ok(kAudioDeviceTransportTypeBuiltIn) => {
+                cubeb_log!(
+                    "Forcing VPIO because input device is built in, and its volume \
+                     is known to be very low without VPIO whenever VPIO is hooked up to it elsewhere."
+                );
+                true
+            }
+            _ => false,
+        }
+    }
+
     fn should_block_vpio_for_device_pair(
         &self,
         in_device: &device_info,
@@ -2751,7 +3203,10 @@ impl<'ctx> CoreStreamData<'ctx> {
     ) -> bool {
         self.debug_assert_is_on_stream_queue();
         cubeb_log!("Evaluating device pair against VPIO block list");
-        let log_device = |id, devtype| -> std::result::Result<(), OSStatus> {
+        let log_device_and_get_model_uid = |id, devtype| -> String {
+            let device_model_uid = get_device_model_uid(id, devtype)
+                .map(|s| s.into_string())
+                .unwrap_or_default();
             cubeb_log!("{} uid=\"{}\", model_uid=\"{}\", transport_type={:?}, source={:?}, source_name=\"{}\", name=\"{}\", manufacturer=\"{}\"",
                 if devtype == DeviceType::INPUT {
                     "Input"
@@ -2760,43 +3215,59 @@ impl<'ctx> CoreStreamData<'ctx> {
                     "Output"
                 },
                 get_device_uid(id, devtype).map(|s| s.into_string()).unwrap_or_default(),
-                get_device_model_uid(id, devtype).map(|s| s.into_string()).unwrap_or_default(),
+                device_model_uid,
                 convert_uint32_into_string(get_device_transport_type(id, devtype).unwrap_or(0)),
                 convert_uint32_into_string(get_device_source(id, devtype).unwrap_or(0)),
                 get_device_source_name(id, devtype).map(|s| s.into_string()).unwrap_or_default(),
                 get_device_name(id, devtype).map(|s| s.into_string()).unwrap_or_default(),
                 get_device_manufacturer(id, devtype).map(|s| s.into_string()).unwrap_or_default());
-            Ok(())
+            device_model_uid
         };
-        log_device(in_device.id, DeviceType::INPUT);
-        log_device(out_device.id, DeviceType::OUTPUT);
-        match (
-            get_device_model_uid(in_device.id, DeviceType::INPUT).map(|s| s.to_string()),
-            get_device_model_uid(out_device.id, DeviceType::OUTPUT).map(|s| s.to_string()),
-        ) {
-            (Ok(in_model_uid), Ok(out_model_uid))
-                if in_model_uid.contains(APPLE_STUDIO_DISPLAY_USB_ID)
-                    && out_model_uid.contains(APPLE_STUDIO_DISPLAY_USB_ID) =>
-            {
-                cubeb_log!("Both input and output device is an Apple Studio Display. BLOCKED");
-                true
-            }
-            _ => {
-                cubeb_log!("Device pair is not blocked");
-                false
-            }
+
+        #[allow(non_upper_case_globals)]
+        let in_id = match in_device.id {
+            kAudioObjectUnknown => None,
+            id => Some(id),
+        };
+        #[allow(non_upper_case_globals)]
+        let out_id = match out_device.id {
+            kAudioObjectUnknown => None,
+            id => Some(id),
+        };
+
+        let (in_model_uid, out_model_uid) = (
+            in_id
+                .map(|id| log_device_and_get_model_uid(id, DeviceType::INPUT))
+                .unwrap_or_default(),
+            out_id
+                .map(|id| log_device_and_get_model_uid(id, DeviceType::OUTPUT))
+                .unwrap_or_default(),
+        );
+
+        if in_model_uid.contains(APPLE_STUDIO_DISPLAY_USB_ID)
+            && out_model_uid.contains(APPLE_STUDIO_DISPLAY_USB_ID)
+        {
+            cubeb_log!("Both input and output device is an Apple Studio Display. BLOCKED");
+            return true;
         }
+
+        cubeb_log!("Device pair is not blocked");
+        false
     }
 
-    fn create_audiounits(&mut self) -> Result<(device_info, device_info)> {
+    fn create_audiounits(
+        &mut self,
+        shared_voice_processing_unit: &mut SharedVoiceProcessingUnitManager,
+    ) -> Result<(device_info, device_info)> {
         self.debug_assert_is_on_stream_queue();
         let should_use_voice_processing_unit = self.has_input()
-            && self.has_output()
-            && self
+            && (self
                 .input_stream_params
                 .prefs()
                 .contains(StreamPrefs::VOICE)
-            && !self.should_block_vpio_for_device_pair(&self.input_device, &self.output_device);
+                || self.should_force_vpio_for_input_device(&self.input_device))
+            && !self.should_block_vpio_for_device_pair(&self.input_device, &self.output_device)
+            && macos_kernel_major_version() != Ok(MACOS_KERNEL_MAJOR_VERSION_MONTEREY);
 
         let should_use_aggregate_device = {
             // It's impossible to create an aggregate device from an aggregate device, and it's
@@ -2843,16 +3314,20 @@ impl<'ctx> CoreStreamData<'ctx> {
         // - As last resort, create regular AudioUnits. This is also the normal non-duplex path.
 
         if should_use_voice_processing_unit {
-            if let Ok(au) =
-                create_voiceprocessing_audiounit(&self.input_device, &self.output_device)
-            {
-                cubeb_log!("({:p}) Using VoiceProcessingIO AudioUnit", self.stm_ptr);
-                self.input_unit = au;
-                self.output_unit = au;
+            if let Ok(mut au_handle) = get_voiceprocessing_audiounit(
+                shared_voice_processing_unit,
+                &self.input_device,
+                &self.output_device,
+            ) {
+                self.input_unit = au_handle.as_mut().unit;
+                if self.has_output() {
+                    self.output_unit = au_handle.as_mut().unit;
+                }
+                self.voiceprocessing_unit_handle = Some(au_handle);
                 return Ok((self.input_device.clone(), self.output_device.clone()));
             }
             cubeb_log!(
-                "({:p}) Failed to create VoiceProcessingIO AudioUnit. Trying a regular one.",
+                "({:p}) Failed to get VoiceProcessingIO AudioUnit. Trying a regular one.",
                 self.stm_ptr
             );
         }
@@ -2954,7 +3429,10 @@ impl<'ctx> CoreStreamData<'ctx> {
     }
 
     #[allow(clippy::cognitive_complexity)] // TODO: Refactoring.
-    fn setup(&mut self) -> Result<()> {
+    fn setup(
+        &mut self,
+        shared_voice_processing_unit: &mut SharedVoiceProcessingUnitManager,
+    ) -> Result<()> {
         self.debug_assert_is_on_stream_queue();
         if self
             .input_stream_params
@@ -2970,7 +3448,7 @@ impl<'ctx> CoreStreamData<'ctx> {
         }
 
         let same_clock_domain = self.same_clock_domain();
-        let (in_dev_info, out_dev_info) = self.create_audiounits()?;
+        let (in_dev_info, out_dev_info) = self.create_audiounits(shared_voice_processing_unit)?;
         let using_voice_processing_unit = self.using_voice_processing_unit();
 
         assert!(!self.stm_ptr.is_null());
@@ -3155,6 +3633,25 @@ impl<'ctx> CoreStreamData<'ctx> {
             );
         }
 
+        if self.has_input() && !self.has_output() && using_voice_processing_unit {
+            // We must configure the output side of VPIO to match the input side, even if we don't use it.
+            let r = audio_unit_set_property(
+                self.input_unit,
+                kAudioUnitProperty_StreamFormat,
+                kAudioUnitScope_Input,
+                AU_OUT_BUS,
+                &self.input_dev_desc,
+                mem::size_of::<AudioStreamBasicDescription>(),
+            );
+            if r != NO_ERR {
+                cubeb_log!(
+                    "AudioUnitSetProperty/output/kAudioUnitProperty_StreamFormat rv={}",
+                    r
+                );
+                return Err(Error::error());
+            }
+        }
+
         if self.has_output() {
             assert!(!self.output_unit.is_null());
 
@@ -3463,15 +3960,31 @@ impl<'ctx> CoreStreamData<'ctx> {
             // NOTE: On MacOS 14 the ducking happens on creation of the VPIO AudioUnit.
             //       On MacOS 10.15 it happens on both creation and initialization, which
             //       is why we defer the unducking until now.
-            let r = audio_device_duck(self.output_device.id, 1.0, ptr::null_mut(), 0.5);
-            if r != NO_ERR {
-                cubeb_log!(
-                    "({:p}) Failed to undo ducking of voiceprocessing on output device {}. Proceeding... Error: {}",
-                    self.stm_ptr,
-                    self.output_device.id,
-                    r
-                );
-            }
+            #[allow(non_upper_case_globals)]
+            let mut device = match self.output_device.id {
+                kAudioObjectUnknown => None,
+                id => Some(id),
+            };
+            device = device.or_else(|| get_default_device(DeviceType::OUTPUT));
+            match device {
+                None => {
+                    cubeb_log!(
+                        "({:p}) No output device to undo vpio ducking on",
+                        self.stm_ptr
+                    );
+                }
+                Some(id) => {
+                    let r = audio_device_duck(id, 1.0, ptr::null_mut(), 0.5);
+                    if r != NO_ERR {
+                        cubeb_log!(
+                            "({:p}) Failed to undo ducking of voiceprocessing on output device {}. Proceeding... Error: {}",
+                            self.stm_ptr,
+                            id,
+                            r
+                        );
+                    }
+                }
+            };
 
             // Always try to remember the applied input mute state. If it cannot be applied
             // to the new device pair, we notify the client of an error and it will have to
@@ -3558,10 +4071,16 @@ impl<'ctx> CoreStreamData<'ctx> {
         }
 
         if !self.input_unit.is_null() {
-            dispose_audio_unit(self.input_unit);
+            if !self.using_voice_processing_unit() {
+                // The VPIO unit is shared and must not be disposed.
+                dispose_audio_unit(self.input_unit);
+            }
             self.input_unit = ptr::null_mut();
         }
 
+        // Return the VPIO unit if present.
+        self.voiceprocessing_unit_handle = None;
+
         self.resampler.destroy();
         self.mixer = None;
         self.aggregate_device = None;
@@ -3873,8 +4392,6 @@ struct OutputCallbackTimingData {
 // #[repr(C)] is used to prevent any padding from being added in the beginning of the AudioUnitStream.
 #[repr(C)]
 #[derive(Debug)]
-// Allow exposing this private struct in public interfaces when running tests.
-#[cfg_attr(test, allow(private_in_public))]
 struct AudioUnitStream<'ctx> {
     context: &'ctx mut AudioUnitContext,
     user_ptr: *mut c_void,
@@ -3927,10 +4444,11 @@ impl<'ctx> AudioUnitStream<'ctx> {
             });
         let (output_callback_timing_data_write, output_callback_timing_data_read) =
             output_callback_timing_data.split();
+        let queue = context.serial_queue.clone();
         AudioUnitStream {
             context,
             user_ptr,
-            queue: Queue::new(DISPATCH_QUEUE_LABEL),
+            queue,
             data_callback,
             state_callback,
             device_changed_callback: Mutex::new(None),
@@ -4046,10 +4564,12 @@ impl<'ctx> AudioUnitStream<'ctx> {
                 }
         }
 
-        self.core_stream_data.setup().map_err(|e| {
-            cubeb_log!("({:p}) Setup failed.", self.core_stream_data.stm_ptr);
-            e
-        })?;
+        self.core_stream_data
+            .setup(&mut self.context.shared_voice_processing_unit)
+            .map_err(|e| {
+                cubeb_log!("({:p}) Setup failed.", self.core_stream_data.stm_ptr);
+                e
+            })?;
 
         if let Ok(volume) = vol_rv {
             set_volume(self.core_stream_data.output_unit, volume);
@@ -4171,7 +4691,7 @@ impl<'ctx> AudioUnitStream<'ctx> {
 impl<'ctx> Drop for AudioUnitStream<'ctx> {
     fn drop(&mut self) {
         // Execute destroy in serial queue to avoid collision with reinit when un/plug devices
-        self.queue.clone().run_final(move || {
+        self.queue.clone().run_final(|| {
             self.destroy();
             self.core_stream_data = CoreStreamData::default();
         });
@@ -4184,12 +4704,10 @@ impl<'ctx> StreamOps for AudioUnitStream<'ctx> {
         self.draining.store(false, Ordering::SeqCst);
 
         // Execute start in serial queue to avoid racing with destroy or reinit.
-        let mut result = Err(Error::error());
-        let started = &mut result;
-        let stream = &self;
-        self.queue.run_sync(move || {
-            *started = stream.core_stream_data.start_audiounits();
-        });
+        let result = self
+            .queue
+            .run_sync(|| self.core_stream_data.start_audiounits())
+            .unwrap();
 
         result?;
 
@@ -4205,10 +4723,8 @@ impl<'ctx> StreamOps for AudioUnitStream<'ctx> {
         self.stopped.store(true, Ordering::SeqCst);
 
         // Execute stop in serial queue to avoid racing with destroy or reinit.
-        let stream = &self;
-        self.queue.run_sync(move || {
-            stream.core_stream_data.stop_audiounits();
-        });
+        self.queue
+            .run_sync(|| self.core_stream_data.stop_audiounits());
 
         self.notify_state_changed(State::Stopped);
 
@@ -4285,12 +4801,10 @@ impl<'ctx> StreamOps for AudioUnitStream<'ctx> {
     }
     fn set_volume(&mut self, volume: f32) -> Result<()> {
         // Execute set_volume in serial queue to avoid racing with destroy or reinit.
-        let mut result = Err(Error::error());
-        let set = &mut result;
-        let stream = &self;
-        self.queue.run_sync(move || {
-            *set = set_volume(stream.core_stream_data.output_unit, volume);
-        });
+        let result = self
+            .queue
+            .run_sync(|| set_volume(self.core_stream_data.output_unit, volume))
+            .unwrap();
 
         result?;
 
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/aggregate_device.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/aggregate_device.rs
index 1d3c341ae8..6c5e494059 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/aggregate_device.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/aggregate_device.rs
@@ -3,6 +3,8 @@ use super::utils::{
     test_get_drift_compensations, test_get_master_device, DeviceFilter, Scope,
 };
 use super::*;
+use std::iter::zip;
+use std::panic;
 
 // AggregateDevice::set_sub_devices
 // ------------------------------------
@@ -19,21 +21,27 @@ fn test_aggregate_set_sub_devices_for_an_unknown_aggregate_device() {
     let default_input = default_input.unwrap();
     let default_output = default_output.unwrap();
     assert!(
-        AggregateDevice::set_sub_devices(kAudioObjectUnknown, default_input, default_output)
-            .is_err()
+        run_serially_forward_panics(|| AggregateDevice::set_sub_devices(
+            kAudioObjectUnknown,
+            default_input,
+            default_output
+        ))
+        .is_err()
     );
 }
 
 #[test]
 #[should_panic]
 fn test_aggregate_set_sub_devices_for_unknown_devices() {
-    // If aggregate device id is kAudioObjectUnknown, we are unable to set device list.
-    assert!(AggregateDevice::set_sub_devices(
-        kAudioObjectUnknown,
-        kAudioObjectUnknown,
-        kAudioObjectUnknown
-    )
-    .is_err());
+    run_serially_forward_panics(|| {
+        // If aggregate device id is kAudioObjectUnknown, we are unable to set device list.
+        assert!(AggregateDevice::set_sub_devices(
+            kAudioObjectUnknown,
+            kAudioObjectUnknown,
+            kAudioObjectUnknown
+        )
+        .is_err());
+    });
 }
 
 // AggregateDevice::get_sub_devices
@@ -48,7 +56,13 @@ fn test_aggregate_get_sub_devices() {
         // containing `device` itself if it's not an aggregate device. This test assumes devices
         // is not an empty aggregate device (Test will panic when calling get_sub_devices with
         // an empty aggregate device).
-        let sub_devices = AggregateDevice::get_sub_devices(device).unwrap();
+        println!(
+            "get_sub_devices({}={})",
+            device,
+            run_serially_forward_panics(|| get_device_uid(device))
+        );
+        let sub_devices =
+            run_serially_forward_panics(|| AggregateDevice::get_sub_devices(device).unwrap());
         // TODO: If the device is a blank aggregate device, then the assertion fails!
         assert!(!sub_devices.is_empty());
     }
@@ -57,8 +71,10 @@ fn test_aggregate_get_sub_devices() {
 #[test]
 #[should_panic]
 fn test_aggregate_get_sub_devices_for_a_unknown_device() {
-    let devices = AggregateDevice::get_sub_devices(kAudioObjectUnknown).unwrap();
-    assert!(devices.is_empty());
+    run_serially_forward_panics(|| {
+        let devices = AggregateDevice::get_sub_devices(kAudioObjectUnknown).unwrap();
+        assert!(devices.is_empty());
+    });
 }
 
 // AggregateDevice::set_master_device
@@ -66,7 +82,11 @@ fn test_aggregate_get_sub_devices_for_a_unknown_device() {
 #[test]
 #[should_panic]
 fn test_aggregate_set_master_device_for_an_unknown_aggregate_device() {
-    assert!(AggregateDevice::set_master_device(kAudioObjectUnknown, kAudioObjectUnknown).is_err());
+    run_serially_forward_panics(|| {
+        assert!(
+            AggregateDevice::set_master_device(kAudioObjectUnknown, kAudioObjectUnknown).is_err()
+        );
+    });
 }
 
 // AggregateDevice::activate_clock_drift_compensation
@@ -74,7 +94,9 @@ fn test_aggregate_set_master_device_for_an_unknown_aggregate_device() {
 #[test]
 #[should_panic]
 fn test_aggregate_activate_clock_drift_compensation_for_an_unknown_aggregate_device() {
-    assert!(AggregateDevice::activate_clock_drift_compensation(kAudioObjectUnknown).is_err());
+    run_serially_forward_panics(|| {
+        assert!(AggregateDevice::activate_clock_drift_compensation(kAudioObjectUnknown).is_err());
+    });
 }
 
 // AggregateDevice::destroy_device
@@ -82,60 +104,55 @@ fn test_aggregate_activate_clock_drift_compensation_for_an_unknown_aggregate_dev
 #[test]
 #[should_panic]
 fn test_aggregate_destroy_device_for_unknown_plugin_and_aggregate_devices() {
-    assert!(AggregateDevice::destroy_device(kAudioObjectUnknown, kAudioObjectUnknown).is_err())
+    run_serially_forward_panics(|| {
+        assert!(AggregateDevice::destroy_device(kAudioObjectUnknown, kAudioObjectUnknown).is_err())
+    });
 }
 
 #[test]
 #[should_panic]
 fn test_aggregate_destroy_aggregate_device_for_a_unknown_aggregate_device() {
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    assert!(AggregateDevice::destroy_device(plugin, kAudioObjectUnknown).is_err());
+    run_serially_forward_panics(|| {
+        let plugin = AggregateDevice::get_system_plugin_id().unwrap();
+        assert!(AggregateDevice::destroy_device(plugin, kAudioObjectUnknown).is_err());
+    });
 }
 
-// Default Ignored Tests
-// ================================================================================================
-// The following tests that calls `AggregateDevice::create_blank_device` are marked `ignore` by
-// default since the device-collection-changed callbacks will be fired upon
-// `AggregateDevice::create_blank_device` is called (it will plug a new device in system!).
-// Some tests rely on the device-collection-changed callbacks in a certain way. The callbacks
-// fired from a unexpected `AggregateDevice::create_blank_device` will break those tests.
-
 // AggregateDevice::create_blank_device_sync
 // ------------------------------------
 #[test]
-#[ignore]
 fn test_aggregate_create_blank_device() {
     // TODO: Test this when there is no available devices.
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
+    let plugin = run_serially(|| AggregateDevice::get_system_plugin_id()).unwrap();
+    let device = run_serially(|| AggregateDevice::create_blank_device_sync(plugin)).unwrap();
     let devices = test_get_all_devices(DeviceFilter::IncludeAll);
     let device = devices.into_iter().find(|dev| dev == &device).unwrap();
-    let uid = get_device_global_uid(device).unwrap().into_string();
+    let uid = run_serially(|| get_device_global_uid(device).unwrap().into_string());
     assert!(uid.contains(PRIVATE_AGGREGATE_DEVICE_NAME));
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    assert!(run_serially(|| AggregateDevice::destroy_device(plugin, device)).is_ok());
 }
 
 // AggregateDevice::get_sub_devices
 // ------------------------------------
 #[test]
-#[ignore]
 #[should_panic]
 fn test_aggregate_get_sub_devices_for_blank_aggregate_devices() {
-    // TODO: Test this when there is no available devices.
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
-    // There is no sub device in a blank aggregate device!
-    // AggregateDevice::get_sub_devices guarantees returning a non-empty devices vector, so
-    // the following call will panic!
-    let sub_devices = AggregateDevice::get_sub_devices(device).unwrap();
-    assert!(sub_devices.is_empty());
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    run_serially_forward_panics(|| {
+        // TODO: Test this when there is no available devices.
+        let plugin = AggregateDevice::get_system_plugin_id().unwrap();
+        let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
+        // There is no sub device in a blank aggregate device!
+        // AggregateDevice::get_sub_devices guarantees returning a non-empty devices vector, so
+        // the following call will panic!
+        let sub_devices = AggregateDevice::get_sub_devices(device).unwrap();
+        assert!(sub_devices.is_empty());
+        assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    });
 }
 
 // AggregateDevice::set_sub_devices_sync
 // ------------------------------------
 #[test]
-#[ignore]
 fn test_aggregate_set_sub_devices() {
     let input_device = test_get_default_device(Scope::Input);
     let output_device = test_get_default_device(Scope::Output);
@@ -147,13 +164,20 @@ fn test_aggregate_set_sub_devices() {
     let input_device = input_device.unwrap();
     let output_device = output_device.unwrap();
 
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
-    assert!(AggregateDevice::set_sub_devices_sync(device, input_device, output_device).is_ok());
-
-    let sub_devices = AggregateDevice::get_sub_devices(device).unwrap();
-    let input_sub_devices = AggregateDevice::get_sub_devices(input_device).unwrap();
-    let output_sub_devices = AggregateDevice::get_sub_devices(output_device).unwrap();
+    let plugin = run_serially(|| AggregateDevice::get_system_plugin_id()).unwrap();
+    let device = run_serially(|| AggregateDevice::create_blank_device_sync(plugin)).unwrap();
+    assert!(run_serially(|| AggregateDevice::set_sub_devices_sync(
+        device,
+        input_device,
+        output_device
+    ))
+    .is_ok());
+
+    let sub_devices = run_serially(|| AggregateDevice::get_sub_devices(device)).unwrap();
+    let input_sub_devices =
+        run_serially(|| AggregateDevice::get_sub_devices(input_device)).unwrap();
+    let output_sub_devices =
+        run_serially(|| AggregateDevice::get_sub_devices(output_device)).unwrap();
 
     // TODO: There may be overlapping devices between input_sub_devices and output_sub_devices,
     //       but now AggregateDevice::set_sub_devices will add them directly.
@@ -168,10 +192,10 @@ fn test_aggregate_set_sub_devices() {
         assert!(sub_devices.contains(dev));
     }
 
-    let onwed_devices = test_get_all_onwed_devices(device);
-    let onwed_device_uids = get_device_uids(&onwed_devices);
-    let input_sub_device_uids = get_device_uids(&input_sub_devices);
-    let output_sub_device_uids = get_device_uids(&output_sub_devices);
+    let onwed_devices = run_serially(|| test_get_all_onwed_devices(device));
+    let onwed_device_uids = run_serially(|| get_device_uids(&onwed_devices));
+    let input_sub_device_uids = run_serially(|| get_device_uids(&input_sub_devices));
+    let output_sub_device_uids = run_serially(|| get_device_uids(&output_sub_devices));
     for uid in &input_sub_device_uids {
         assert!(onwed_device_uids.contains(uid));
     }
@@ -179,11 +203,10 @@ fn test_aggregate_set_sub_devices() {
         assert!(onwed_device_uids.contains(uid));
     }
 
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    assert!(run_serially(|| AggregateDevice::destroy_device(plugin, device)).is_ok());
 }
 
 #[test]
-#[ignore]
 #[should_panic]
 fn test_aggregate_set_sub_devices_for_unknown_input_devices() {
     let output_device = test_get_default_device(Scope::Output);
@@ -192,16 +215,19 @@ fn test_aggregate_set_sub_devices_for_unknown_input_devices() {
     }
     let output_device = output_device.unwrap();
 
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
+    run_serially_forward_panics(|| {
+        let plugin = AggregateDevice::get_system_plugin_id().unwrap();
+        let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
 
-    assert!(AggregateDevice::set_sub_devices(device, kAudioObjectUnknown, output_device).is_err());
+        assert!(
+            AggregateDevice::set_sub_devices(device, kAudioObjectUnknown, output_device).is_err()
+        );
 
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+        assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    });
 }
 
 #[test]
-#[ignore]
 #[should_panic]
 fn test_aggregate_set_sub_devices_for_unknown_output_devices() {
     let input_device = test_get_default_device(Scope::Input);
@@ -210,12 +236,16 @@ fn test_aggregate_set_sub_devices_for_unknown_output_devices() {
     }
     let input_device = input_device.unwrap();
 
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
+    run_serially_forward_panics(|| {
+        let plugin = AggregateDevice::get_system_plugin_id().unwrap();
+        let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
 
-    assert!(AggregateDevice::set_sub_devices(device, input_device, kAudioObjectUnknown).is_err());
+        assert!(
+            AggregateDevice::set_sub_devices(device, input_device, kAudioObjectUnknown).is_err()
+        );
 
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+        assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    });
 }
 
 fn get_device_uids(devices: &Vec<AudioObjectID>) -> Vec<String> {
@@ -228,7 +258,6 @@ fn get_device_uids(devices: &Vec<AudioObjectID>) -> Vec<String> {
 // AggregateDevice::set_master_device
 // ------------------------------------
 #[test]
-#[ignore]
 fn test_aggregate_set_master_device() {
     let input_device = test_get_default_device(Scope::Input);
     let output_device = test_get_default_device(Scope::Output);
@@ -240,22 +269,28 @@ fn test_aggregate_set_master_device() {
     let input_device = input_device.unwrap();
     let output_device = output_device.unwrap();
 
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
-    assert!(AggregateDevice::set_sub_devices_sync(device, input_device, output_device).is_ok());
-    assert!(AggregateDevice::set_master_device(device, output_device).is_ok());
-
-    // Check if master is set to the first sub device of the default output device.
-    let first_output_sub_device_uid =
-        get_device_uid(AggregateDevice::get_sub_devices(device).unwrap()[0]);
-    let master_device_uid = test_get_master_device(device);
+    let plugin = run_serially(|| AggregateDevice::get_system_plugin_id()).unwrap();
+    let device = run_serially(|| AggregateDevice::create_blank_device_sync(plugin)).unwrap();
+    assert!(run_serially(|| AggregateDevice::set_sub_devices_sync(
+        device,
+        input_device,
+        output_device
+    ))
+    .is_ok());
+    assert!(run_serially(|| AggregateDevice::set_master_device(device, output_device)).is_ok());
+
+    let output_sub_devices =
+        run_serially(|| AggregateDevice::get_sub_devices(output_device)).unwrap();
+    let first_output_sub_device_uid = run_serially(|| get_device_uid(output_sub_devices[0]));
+
+    // Check that the first sub device of the output device is set as master device.
+    let master_device_uid = run_serially(|| test_get_master_device(device));
     assert_eq!(first_output_sub_device_uid, master_device_uid);
 
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    assert!(run_serially(|| AggregateDevice::destroy_device(plugin, device)).is_ok());
 }
 
 #[test]
-#[ignore]
 fn test_aggregate_set_master_device_for_a_blank_aggregate_device() {
     let output_device = test_get_default_device(Scope::Output);
     if output_device.is_none() {
@@ -263,9 +298,11 @@ fn test_aggregate_set_master_device_for_a_blank_aggregate_device() {
         return;
     }
 
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
-    assert!(AggregateDevice::set_master_device(device, output_device.unwrap()).is_ok());
+    let plugin = run_serially(|| AggregateDevice::get_system_plugin_id()).unwrap();
+    let device = run_serially(|| AggregateDevice::create_blank_device_sync(plugin)).unwrap();
+    assert!(
+        run_serially(|| AggregateDevice::set_master_device(device, output_device.unwrap())).is_ok()
+    );
 
     // TODO: it's really weird the aggregate device actually own nothing
     //       but its master device can be set successfully!
@@ -275,17 +312,16 @@ fn test_aggregate_set_master_device_for_a_blank_aggregate_device() {
     // The CFStringRef of the master device returned from `test_get_master_device` is actually
     // non-null.
 
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    assert!(run_serially(|| AggregateDevice::destroy_device(plugin, device)).is_ok());
 }
 
 fn get_device_uid(id: AudioObjectID) -> String {
-    get_device_global_uid(id).unwrap().into_string()
+    get_device_global_uid(id).map_or(String::new(), |uid| uid.into_string())
 }
 
 // AggregateDevice::activate_clock_drift_compensation
 // ------------------------------------
 #[test]
-#[ignore]
 fn test_aggregate_activate_clock_drift_compensation() {
     let input_device = test_get_default_device(Scope::Input);
     let output_device = test_get_default_device(Scope::Output);
@@ -297,27 +333,40 @@ fn test_aggregate_activate_clock_drift_compensation() {
     let input_device = input_device.unwrap();
     let output_device = output_device.unwrap();
 
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
-    assert!(AggregateDevice::set_sub_devices_sync(device, input_device, output_device).is_ok());
-    assert!(AggregateDevice::set_master_device(device, output_device).is_ok());
-    assert!(AggregateDevice::activate_clock_drift_compensation(device).is_ok());
+    let plugin = run_serially(|| AggregateDevice::get_system_plugin_id()).unwrap();
+    let device = run_serially(|| AggregateDevice::create_blank_device_sync(plugin)).unwrap();
+    assert!(run_serially(|| AggregateDevice::set_sub_devices_sync(
+        device,
+        input_device,
+        output_device
+    ))
+    .is_ok());
+    assert!(run_serially(|| AggregateDevice::set_master_device(device, output_device)).is_ok());
+    assert!(run_serially(|| AggregateDevice::activate_clock_drift_compensation(device)).is_ok());
 
     // Check the compensations.
-    let devices = test_get_all_onwed_devices(device);
-    let compensations = get_drift_compensations(&devices);
+    let devices = run_serially(|| test_get_all_onwed_devices(device));
+    let compensations = run_serially(|| get_drift_compensations(&devices));
+    let master_device_uid = run_serially(|| test_get_master_device(device));
     assert!(!compensations.is_empty());
     assert_eq!(devices.len(), compensations.len());
 
-    for (i, compensation) in compensations.iter().enumerate() {
-        assert_eq!(*compensation, if i == 0 { 0 } else { DRIFT_COMPENSATION });
+    for (device, compensation) in zip(devices, compensations) {
+        let uid = get_device_uid(device);
+        assert_eq!(
+            compensation,
+            if uid == master_device_uid {
+                0
+            } else {
+                DRIFT_COMPENSATION
+            }
+        );
     }
 
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    assert!(run_serially(|| AggregateDevice::destroy_device(plugin, device)).is_ok());
 }
 
 #[test]
-#[ignore]
 fn test_aggregate_activate_clock_drift_compensation_for_an_aggregate_device_without_master_device()
 {
     let input_device = test_get_default_device(Scope::Input);
@@ -330,25 +379,32 @@ fn test_aggregate_activate_clock_drift_compensation_for_an_aggregate_device_with
     let input_device = input_device.unwrap();
     let output_device = output_device.unwrap();
 
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
-    assert!(AggregateDevice::set_sub_devices_sync(device, input_device, output_device).is_ok());
-
-    // TODO: Is the master device the first output sub device by default if we
-    //       don't set that ? Is it because we add the output sub device list
-    //       before the input's one ? (See implementation of
-    //       AggregateDevice::set_sub_devices).
-    let first_output_sub_device_uid =
-        get_device_uid(AggregateDevice::get_sub_devices(output_device).unwrap()[0]);
-    let master_device_uid = test_get_master_device(device);
-    assert_eq!(first_output_sub_device_uid, master_device_uid);
+    let plugin = run_serially(|| AggregateDevice::get_system_plugin_id()).unwrap();
+    let device = run_serially(|| AggregateDevice::create_blank_device_sync(plugin)).unwrap();
+    assert!(run_serially(|| AggregateDevice::set_sub_devices_sync(
+        device,
+        input_device,
+        output_device
+    ))
+    .is_ok());
+
+    // The master device is by default the first sub device in the list.
+    // This happens to be the first sub device of the input device, see implementation of
+    // AggregateDevice::set_sub_devices.
+    let first_input_sub_device_uid =
+        run_serially(|| get_device_uid(AggregateDevice::get_sub_devices(input_device).unwrap()[0]));
+    let first_sub_device_uid =
+        run_serially(|| get_device_uid(AggregateDevice::get_sub_devices(device).unwrap()[0]));
+    assert_eq!(first_input_sub_device_uid, first_sub_device_uid);
+    let master_device_uid = run_serially(|| test_get_master_device(device));
+    assert_eq!(first_sub_device_uid, master_device_uid);
 
     // Compensate the drift directly without setting master device.
-    assert!(AggregateDevice::activate_clock_drift_compensation(device).is_ok());
+    assert!(run_serially(|| AggregateDevice::activate_clock_drift_compensation(device)).is_ok());
 
     // Check the compensations.
-    let devices = test_get_all_onwed_devices(device);
-    let compensations = get_drift_compensations(&devices);
+    let devices = run_serially(|| test_get_all_onwed_devices(device));
+    let compensations = run_serially(|| get_drift_compensations(&devices));
     assert!(!compensations.is_empty());
     assert_eq!(devices.len(), compensations.len());
 
@@ -356,25 +412,26 @@ fn test_aggregate_activate_clock_drift_compensation_for_an_aggregate_device_with
         assert_eq!(*compensation, if i == 0 { 0 } else { DRIFT_COMPENSATION });
     }
 
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    assert!(run_serially(|| AggregateDevice::destroy_device(plugin, device)).is_ok());
 }
 
 #[test]
 #[should_panic]
-#[ignore]
 fn test_aggregate_activate_clock_drift_compensation_for_a_blank_aggregate_device() {
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
+    run_serially_forward_panics(|| {
+        let plugin = AggregateDevice::get_system_plugin_id().unwrap();
+        let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
 
-    let sub_devices = AggregateDevice::get_sub_devices(device).unwrap();
-    assert!(sub_devices.is_empty());
-    let onwed_devices = test_get_all_onwed_devices(device);
-    assert!(onwed_devices.is_empty());
+        let sub_devices = AggregateDevice::get_sub_devices(device).unwrap();
+        assert!(sub_devices.is_empty());
+        let onwed_devices = test_get_all_onwed_devices(device);
+        assert!(onwed_devices.is_empty());
 
-    // Get a panic since no sub devices to be set compensation.
-    assert!(AggregateDevice::activate_clock_drift_compensation(device).is_err());
+        // Get a panic since no sub devices to be set compensation.
+        assert!(AggregateDevice::activate_clock_drift_compensation(device).is_err());
 
-    assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+        assert!(AggregateDevice::destroy_device(plugin, device).is_ok());
+    });
 }
 
 fn get_drift_compensations(devices: &Vec<AudioObjectID>) -> Vec<u32> {
@@ -391,10 +448,56 @@ fn get_drift_compensations(devices: &Vec<AudioObjectID>) -> Vec<u32> {
 // AggregateDevice::destroy_device
 // ------------------------------------
 #[test]
-#[ignore]
 #[should_panic]
 fn test_aggregate_destroy_aggregate_device_for_a_unknown_plugin_device() {
-    let plugin = AggregateDevice::get_system_plugin_id().unwrap();
-    let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
-    assert!(AggregateDevice::destroy_device(kAudioObjectUnknown, device).is_err());
+    run_serially_forward_panics(|| {
+        let plugin = AggregateDevice::get_system_plugin_id().unwrap();
+        let device = AggregateDevice::create_blank_device_sync(plugin).unwrap();
+        assert!(AggregateDevice::destroy_device(kAudioObjectUnknown, device).is_err());
+    });
+}
+
+// AggregateDevice::new
+// ------------------------------------
+#[test]
+fn test_aggregate_new() {
+    let input_device = test_get_default_device(Scope::Input);
+    let output_device = test_get_default_device(Scope::Output);
+    if input_device.is_none() || output_device.is_none() || input_device == output_device {
+        println!("No input or output device to create an aggregate device.");
+        return;
+    }
+
+    run_serially_forward_panics(|| {
+        let input_device = input_device.unwrap();
+        let output_device = output_device.unwrap();
+
+        let aggr = AggregateDevice::new(input_device, output_device).unwrap();
+
+        // Check main device
+        let output_sub_devices = AggregateDevice::get_sub_devices(output_device).unwrap();
+        let first_output_sub_device_uid = get_device_uid(output_sub_devices[0]);
+        let master_device_uid = test_get_master_device(aggr.get_device_id());
+        assert_eq!(first_output_sub_device_uid, master_device_uid);
+
+        // Check drift compensation
+        let devices = test_get_all_onwed_devices(aggr.get_device_id());
+        let compensations = get_drift_compensations(&devices);
+        assert!(!compensations.is_empty());
+        assert_eq!(devices.len(), compensations.len());
+
+        let device_uids = devices.iter().map(|&id| get_device_uid(id));
+        for (uid, compensation) in zip(device_uids, compensations) {
+            assert_eq!(
+                compensation,
+                if uid == master_device_uid {
+                    0
+                } else {
+                    DRIFT_COMPENSATION
+                },
+                "Unexpected drift value for device with uid {}",
+                uid
+            );
+        }
+    });
 }
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/api.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/api.rs
index 4cd86c094e..5ce2374a3e 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/api.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/api.rs
@@ -56,7 +56,7 @@ fn test_increase_and_decrease_context_streams() {
     assert_eq!(context.active_streams(), STREAMS);
     check_streams(&context, STREAMS);
 
-    check_latency(&context, latencies[0]);
+    check_latency(&context, Some(latencies[0]));
     for i in 0..latencies.len() - 1 {
         assert_eq!(latencies[i], latencies[i + 1]);
     }
@@ -149,7 +149,8 @@ fn test_minimum_resampling_input_frames_equal_input_output_rate() {
 #[test]
 fn test_create_device_info_from_unknown_input_device() {
     if let Some(default_device_id) = test_get_default_device(Scope::Input) {
-        let default_device = create_device_info(kAudioObjectUnknown, DeviceType::INPUT).unwrap();
+        let default_device =
+            run_serially(|| create_device_info(kAudioObjectUnknown, DeviceType::INPUT).unwrap());
         assert_eq!(default_device.id, default_device_id);
         assert_eq!(
             default_device.flags,
@@ -163,7 +164,8 @@ fn test_create_device_info_from_unknown_input_device() {
 #[test]
 fn test_create_device_info_from_unknown_output_device() {
     if let Some(default_device_id) = test_get_default_device(Scope::Output) {
-        let default_device = create_device_info(kAudioObjectUnknown, DeviceType::OUTPUT).unwrap();
+        let default_device =
+            run_serially(|| create_device_info(kAudioObjectUnknown, DeviceType::OUTPUT)).unwrap();
         assert_eq!(default_device.id, default_device_id);
         assert_eq!(
             default_device.flags,
@@ -177,13 +179,17 @@ fn test_create_device_info_from_unknown_output_device() {
 #[test]
 #[should_panic]
 fn test_set_device_info_to_system_input_device() {
-    let _device = create_device_info(kAudioObjectSystemObject, DeviceType::INPUT);
+    let _device = run_serially_forward_panics(|| {
+        create_device_info(kAudioObjectSystemObject, DeviceType::INPUT)
+    });
 }
 
 #[test]
 #[should_panic]
 fn test_set_device_info_to_system_output_device() {
-    let _device = create_device_info(kAudioObjectSystemObject, DeviceType::OUTPUT);
+    let _device = run_serially_forward_panics(|| {
+        create_device_info(kAudioObjectSystemObject, DeviceType::OUTPUT)
+    });
 }
 
 // FIXME: Is it ok to set input device to a nonexistent device ?
@@ -192,7 +198,8 @@ fn test_set_device_info_to_system_output_device() {
 #[should_panic]
 fn test_set_device_info_to_nonexistent_input_device() {
     let nonexistent_id = std::u32::MAX;
-    let _device = create_device_info(nonexistent_id, DeviceType::INPUT);
+    let _device =
+        run_serially_forward_panics(|| create_device_info(nonexistent_id, DeviceType::INPUT));
 }
 
 // FIXME: Is it ok to set output device to a nonexistent device ?
@@ -201,7 +208,8 @@ fn test_set_device_info_to_nonexistent_input_device() {
 #[should_panic]
 fn test_set_device_info_to_nonexistent_output_device() {
     let nonexistent_id = std::u32::MAX;
-    let _device = create_device_info(nonexistent_id, DeviceType::OUTPUT);
+    let _device =
+        run_serially_forward_panics(|| create_device_info(nonexistent_id, DeviceType::OUTPUT));
 }
 
 // add_listener (for default output device)
@@ -227,10 +235,10 @@ fn test_add_listener_unknown_device() {
             ),
             callback,
         );
-        let mut res: OSStatus = 0;
-        stream
+        let res = stream
             .queue
-            .run_sync(|| res = stream.add_device_listener(&listener));
+            .run_sync(|| stream.add_device_listener(&listener))
+            .unwrap();
         assert_eq!(res, kAudioHardwareBadObjectError as OSStatus);
     });
 }
@@ -258,14 +266,15 @@ fn test_add_listener_then_remove_system_device() {
             ),
             callback,
         );
-        let mut res: OSStatus = 0;
-        stream
+        let res = stream
             .queue
-            .run_sync(|| res = stream.add_device_listener(&listener));
+            .run_sync(|| stream.add_device_listener(&listener))
+            .unwrap();
         assert_eq!(res, NO_ERR);
-        stream
+        let res = stream
             .queue
-            .run_sync(|| res = stream.remove_device_listener(&listener));
+            .run_sync(|| stream.remove_device_listener(&listener))
+            .unwrap();
         assert_eq!(res, NO_ERR);
     });
 }
@@ -291,10 +300,10 @@ fn test_remove_listener_without_adding_any_listener_before_system_device() {
             ),
             callback,
         );
-        let mut res: OSStatus = 0;
-        stream
+        let res = stream
             .queue
-            .run_sync(|| res = stream.remove_device_listener(&listener));
+            .run_sync(|| stream.remove_device_listener(&listener))
+            .unwrap();
         assert_eq!(res, NO_ERR);
     });
 }
@@ -320,10 +329,10 @@ fn test_remove_listener_unknown_device() {
             ),
             callback,
         );
-        let mut res: OSStatus = 0;
-        stream
+        let res = stream
             .queue
-            .run_sync(|| res = stream.remove_device_listener(&listener));
+            .run_sync(|| stream.remove_device_listener(&listener))
+            .unwrap();
         assert_eq!(res, kAudioHardwareBadObjectError as OSStatus);
     });
 }
@@ -334,14 +343,14 @@ fn test_remove_listener_unknown_device() {
 fn test_get_default_device_id() {
     if test_get_default_device(Scope::Input).is_some() {
         assert_ne!(
-            get_default_device_id(DeviceType::INPUT).unwrap(),
+            run_serially(|| get_default_device_id(DeviceType::INPUT)).unwrap(),
             kAudioObjectUnknown,
         );
     }
 
     if test_get_default_device(Scope::Output).is_some() {
         assert_ne!(
-            get_default_device_id(DeviceType::OUTPUT).unwrap(),
+            run_serially(|| get_default_device_id(DeviceType::OUTPUT)).unwrap(),
             kAudioObjectUnknown,
         );
     }
@@ -350,13 +359,16 @@ fn test_get_default_device_id() {
 #[test]
 #[should_panic]
 fn test_get_default_device_id_with_unknown_type() {
-    assert!(get_default_device_id(DeviceType::UNKNOWN).is_err());
+    assert!(run_serially_forward_panics(|| get_default_device_id(DeviceType::UNKNOWN)).is_err());
 }
 
 #[test]
 #[should_panic]
 fn test_get_default_device_id_with_inout_type() {
-    assert!(get_default_device_id(DeviceType::INPUT | DeviceType::OUTPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_default_device_id(
+        DeviceType::INPUT | DeviceType::OUTPUT
+    ))
+    .is_err());
 }
 
 // convert_channel_layout
@@ -724,9 +736,11 @@ fn test_convert_channel_layout() {
 #[test]
 fn test_get_preferred_channel_layout_output() {
     match test_get_default_audiounit(Scope::Output) {
-        Some(unit) => assert!(!audiounit_get_preferred_channel_layout(unit.get_inner())
-            .unwrap()
-            .is_empty()),
+        Some(unit) => assert!(!run_serially(|| audiounit_get_preferred_channel_layout(
+            unit.get_inner()
+        ))
+        .unwrap()
+        .is_empty()),
         None => println!("No output audiounit for test."),
     }
 }
@@ -736,9 +750,15 @@ fn test_get_preferred_channel_layout_output() {
 #[test]
 fn test_get_current_channel_layout_output() {
     match test_get_default_audiounit(Scope::Output) {
-        Some(unit) => assert!(!audiounit_get_current_channel_layout(unit.get_inner())
-            .unwrap()
-            .is_empty()),
+        Some(unit) => {
+            assert!(
+                !run_serially_forward_panics(|| audiounit_get_current_channel_layout(
+                    unit.get_inner()
+                ))
+                .unwrap()
+                .is_empty()
+            )
+        }
         None => println!("No output audiounit for test."),
     }
 }
@@ -814,10 +834,30 @@ fn test_enable_audiounit_scope() {
     // for the unit whose subtype is kAudioUnitSubType_HALOutput
     // even when there is no available input or output devices.
     if let Some(unit) = test_create_audiounit(ComponentSubType::HALOutput) {
-        assert!(enable_audiounit_scope(unit.get_inner(), DeviceType::OUTPUT, true).is_ok());
-        assert!(enable_audiounit_scope(unit.get_inner(), DeviceType::OUTPUT, false).is_ok());
-        assert!(enable_audiounit_scope(unit.get_inner(), DeviceType::INPUT, true).is_ok());
-        assert!(enable_audiounit_scope(unit.get_inner(), DeviceType::INPUT, false).is_ok());
+        assert!(run_serially_forward_panics(|| enable_audiounit_scope(
+            unit.get_inner(),
+            DeviceType::OUTPUT,
+            true
+        ))
+        .is_ok());
+        assert!(run_serially_forward_panics(|| enable_audiounit_scope(
+            unit.get_inner(),
+            DeviceType::OUTPUT,
+            false
+        ))
+        .is_ok());
+        assert!(run_serially_forward_panics(|| enable_audiounit_scope(
+            unit.get_inner(),
+            DeviceType::INPUT,
+            true
+        ))
+        .is_ok());
+        assert!(run_serially_forward_panics(|| enable_audiounit_scope(
+            unit.get_inner(),
+            DeviceType::INPUT,
+            false
+        ))
+        .is_ok());
     } else {
         println!("No audiounit to perform test.");
     }
@@ -827,19 +867,23 @@ fn test_enable_audiounit_scope() {
 fn test_enable_audiounit_scope_for_default_output_unit() {
     if let Some(unit) = test_create_audiounit(ComponentSubType::DefaultOutput) {
         assert_eq!(
-            enable_audiounit_scope(unit.get_inner(), DeviceType::OUTPUT, true).unwrap_err(),
+            run_serially(|| enable_audiounit_scope(unit.get_inner(), DeviceType::OUTPUT, true))
+                .unwrap_err(),
             kAudioUnitErr_InvalidProperty
         );
         assert_eq!(
-            enable_audiounit_scope(unit.get_inner(), DeviceType::OUTPUT, false).unwrap_err(),
+            run_serially(|| enable_audiounit_scope(unit.get_inner(), DeviceType::OUTPUT, false))
+                .unwrap_err(),
             kAudioUnitErr_InvalidProperty
         );
         assert_eq!(
-            enable_audiounit_scope(unit.get_inner(), DeviceType::INPUT, true).unwrap_err(),
+            run_serially(|| enable_audiounit_scope(unit.get_inner(), DeviceType::INPUT, true))
+                .unwrap_err(),
             kAudioUnitErr_InvalidProperty
         );
         assert_eq!(
-            enable_audiounit_scope(unit.get_inner(), DeviceType::INPUT, false).unwrap_err(),
+            run_serially(|| enable_audiounit_scope(unit.get_inner(), DeviceType::INPUT, false))
+                .unwrap_err(),
             kAudioUnitErr_InvalidProperty
         );
     }
@@ -849,7 +893,10 @@ fn test_enable_audiounit_scope_for_default_output_unit() {
 #[should_panic]
 fn test_enable_audiounit_scope_with_null_unit() {
     let unit: AudioUnit = ptr::null_mut();
-    assert!(enable_audiounit_scope(unit, DeviceType::INPUT, false).is_err());
+    assert!(
+        run_serially_forward_panics(|| enable_audiounit_scope(unit, DeviceType::INPUT, false))
+            .is_err()
+    );
 }
 
 // create_audiounit
@@ -868,29 +915,29 @@ fn test_for_create_audiounit() {
         // Check the output scope is enabled.
         if device.flags.contains(device_flags::DEV_OUTPUT) && default_output.is_some() {
             device.id = default_output.unwrap();
-            let unit = create_audiounit(&device).unwrap();
+            let unit = run_serially(|| create_audiounit(&device).unwrap());
             assert!(!unit.is_null());
             assert!(test_audiounit_scope_is_enabled(unit, Scope::Output));
 
             // Destroy the AudioUnit.
-            unsafe {
+            run_serially(|| unsafe {
                 AudioUnitUninitialize(unit);
                 AudioComponentInstanceDispose(unit);
-            }
+            });
         }
 
         // Check the input scope is enabled.
         if device.flags.contains(device_flags::DEV_INPUT) && default_input.is_some() {
             let device_id = default_input.unwrap();
             device.id = device_id;
-            let unit = create_audiounit(&device).unwrap();
+            let unit = run_serially(|| create_audiounit(&device).unwrap());
             assert!(!unit.is_null());
             assert!(test_audiounit_scope_is_enabled(unit, Scope::Input));
             // Destroy the AudioUnit.
-            unsafe {
+            run_serially(|| unsafe {
                 AudioUnitUninitialize(unit);
                 AudioComponentInstanceDispose(unit);
-            }
+            });
         }
     }
 }
@@ -899,7 +946,7 @@ fn test_for_create_audiounit() {
 #[should_panic]
 fn test_create_audiounit_with_unknown_scope() {
     let device = device_info::default();
-    let _unit = create_audiounit(&device);
+    let _unit = run_serially_forward_panics(|| create_audiounit(&device));
 }
 
 // set_buffer_size_sync
@@ -927,9 +974,12 @@ fn test_set_buffer_size_sync() {
         .unwrap();
         assert_ne!(buffer_frames, 0);
         buffer_frames *= 2;
-        assert!(
-            set_buffer_size_sync(unit.get_inner(), scope.clone().into(), buffer_frames).is_ok()
-        );
+        assert!(run_serially(|| set_buffer_size_sync(
+            unit.get_inner(),
+            scope.clone().into(),
+            buffer_frames
+        ))
+        .is_ok());
         let new_buffer_frames =
             test_audiounit_get_buffer_frame_size(unit.get_inner(), scope.clone(), prop_scope)
                 .unwrap();
@@ -951,7 +1001,9 @@ fn test_set_buffer_size_sync_for_output_with_null_output_unit() {
 
 fn test_set_buffer_size_sync_by_scope_with_null_unit(scope: Scope) {
     let unit: AudioUnit = ptr::null_mut();
-    assert!(set_buffer_size_sync(unit, scope.into(), 2048).is_err());
+    assert!(
+        run_serially_forward_panics(|| set_buffer_size_sync(unit, scope.into(), 2048)).is_err()
+    );
 }
 
 // get_volume, set_volume
@@ -960,8 +1012,11 @@ fn test_set_buffer_size_sync_by_scope_with_null_unit(scope: Scope) {
 fn test_stream_get_volume() {
     if let Some(unit) = test_get_default_audiounit(Scope::Output) {
         let expected_volume: f32 = 0.5;
-        set_volume(unit.get_inner(), expected_volume);
-        assert_eq!(expected_volume, get_volume(unit.get_inner()).unwrap());
+        run_serially(|| set_volume(unit.get_inner(), expected_volume));
+        assert_eq!(
+            expected_volume,
+            run_serially(|| get_volume(unit.get_inner()).unwrap())
+        );
     } else {
         println!("No output audiounit.");
     }
@@ -988,7 +1043,9 @@ fn test_get_channel_count() {
 
     fn test_channel_count(scope: Scope) {
         if let Some(device) = test_get_default_device(scope.clone()) {
-            let channels = get_channel_count(device, DeviceType::from(scope.clone())).unwrap();
+            let channels =
+                run_serially(|| get_channel_count(device, DeviceType::from(scope.clone())))
+                    .unwrap();
             assert!(channels > 0);
             assert_eq!(
                 channels,
@@ -1008,7 +1065,7 @@ fn test_get_channel_count_of_input_for_a_output_only_deivce() {
         if test_device_in_scope(device, Scope::Input) {
             continue;
         }
-        let count = get_channel_count(device, DeviceType::INPUT).unwrap();
+        let count = run_serially(|| get_channel_count(device, DeviceType::INPUT)).unwrap();
         assert_eq!(count, 0);
     }
 }
@@ -1021,7 +1078,7 @@ fn test_get_channel_count_of_output_for_a_input_only_deivce() {
         if test_device_in_scope(device, Scope::Output) {
             continue;
         }
-        let count = get_channel_count(device, DeviceType::OUTPUT).unwrap();
+        let count = run_serially(|| get_channel_count(device, DeviceType::OUTPUT)).unwrap();
         assert_eq!(count, 0);
     }
 }
@@ -1029,7 +1086,11 @@ fn test_get_channel_count_of_output_for_a_input_only_deivce() {
 #[test]
 #[should_panic]
 fn test_get_channel_count_of_unknown_device() {
-    assert!(get_channel_count(kAudioObjectUnknown, DeviceType::OUTPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_channel_count(
+        kAudioObjectUnknown,
+        DeviceType::OUTPUT
+    ))
+    .is_err());
 }
 
 #[test]
@@ -1039,14 +1100,16 @@ fn test_get_channel_count_of_inout_type() {
 
     fn test_channel_count(scope: Scope) {
         if let Some(device) = test_get_default_device(scope.clone()) {
-            assert_eq!(
-                get_channel_count(device, DeviceType::INPUT | DeviceType::OUTPUT),
-                get_channel_count(device, DeviceType::INPUT).map(|c| c + get_channel_count(
-                    device,
-                    DeviceType::OUTPUT
-                )
-                .unwrap_or(0))
-            );
+            run_serially_forward_panics(|| {
+                assert_eq!(
+                    get_channel_count(device, DeviceType::INPUT | DeviceType::OUTPUT),
+                    get_channel_count(device, DeviceType::INPUT).map(|c| c + get_channel_count(
+                        device,
+                        DeviceType::OUTPUT
+                    )
+                    .unwrap_or(0))
+                );
+            });
         } else {
             println!("No device for {:?}.", scope);
         }
@@ -1095,7 +1158,9 @@ fn test_get_range_of_sample_rates() {
         ];
         let mut ranges = Vec::new();
         for scope in scopes.iter() {
-            ranges.push(get_range_of_sample_rates(id, *scope).unwrap());
+            ranges.push(
+                run_serially_forward_panics(|| get_range_of_sample_rates(id, *scope)).unwrap(),
+            );
         }
         ranges
     }
@@ -1117,7 +1182,7 @@ fn test_get_device_presentation_latency() {
     fn test_get_device_presentation_latencies_in_scope(scope: Scope) {
         if let Some(device) = test_get_default_device(scope.clone()) {
             // TODO: The latencies very from devices to devices. Check nothing here.
-            let latency = get_fixed_latency(device, scope.clone().into());
+            let latency = run_serially(|| get_fixed_latency(device, scope.clone().into()));
             println!(
                 "present latency on the device {} in scope {:?}: {}",
                 device, scope, latency
@@ -1133,7 +1198,7 @@ fn test_get_device_presentation_latency() {
 #[test]
 fn test_get_device_group_id() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        match get_device_group_id(device, DeviceType::INPUT) {
+        match run_serially(|| get_device_group_id(device, DeviceType::INPUT)) {
             Ok(id) => println!("input group id: {:?}", id),
             Err(e) => println!("No input group id. Error: {}", e),
         }
@@ -1142,7 +1207,7 @@ fn test_get_device_group_id() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        match get_device_group_id(device, DeviceType::OUTPUT) {
+        match run_serially(|| get_device_group_id(device, DeviceType::OUTPUT)) {
             Ok(id) => println!("output group id: {:?}", id),
             Err(e) => println!("No output group id. Error: {}", e),
         }
@@ -1165,8 +1230,8 @@ fn test_get_same_group_id_for_builtin_device_pairs() {
     let mut input_group_ids = HashMap::<u32, String>::new();
     let input_devices = test_get_devices_in_scope(Scope::Input);
     for device in input_devices.iter() {
-        match get_device_source(*device, DeviceType::INPUT) {
-            Ok(source) => match get_device_group_id(*device, DeviceType::INPUT) {
+        match run_serially(|| get_device_source(*device, DeviceType::INPUT)) {
+            Ok(source) => match run_serially(|| get_device_group_id(*device, DeviceType::INPUT)) {
                 Ok(id) => assert!(input_group_ids
                     .insert(source, id.into_string().unwrap())
                     .is_none()),
@@ -1181,8 +1246,8 @@ fn test_get_same_group_id_for_builtin_device_pairs() {
     let mut output_group_ids = HashMap::<u32, String>::new();
     let output_devices = test_get_devices_in_scope(Scope::Output);
     for device in output_devices.iter() {
-        match get_device_source(*device, DeviceType::OUTPUT) {
-            Ok(source) => match get_device_group_id(*device, DeviceType::OUTPUT) {
+        match run_serially(|| get_device_source(*device, DeviceType::OUTPUT)) {
+            Ok(source) => match run_serially(|| get_device_group_id(*device, DeviceType::OUTPUT)) {
                 Ok(id) => assert!(output_group_ids
                     .insert(source, id.into_string().unwrap())
                     .is_none()),
@@ -1210,7 +1275,11 @@ fn test_get_same_group_id_for_builtin_device_pairs() {
 #[test]
 #[should_panic]
 fn test_get_device_group_id_by_unknown_device() {
-    assert!(get_device_group_id(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_group_id(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_label
@@ -1218,14 +1287,14 @@ fn test_get_device_group_id_by_unknown_device() {
 #[test]
 fn test_get_device_label() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let name = get_device_label(device, DeviceType::INPUT).unwrap();
+        let name = run_serially(|| get_device_label(device, DeviceType::INPUT)).unwrap();
         println!("input device label: {}", name.into_string());
     } else {
         println!("No input device.");
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let name = get_device_label(device, DeviceType::OUTPUT).unwrap();
+        let name = run_serially(|| get_device_label(device, DeviceType::OUTPUT)).unwrap();
         println!("output device label: {}", name.into_string());
     } else {
         println!("No output device.");
@@ -1235,7 +1304,11 @@ fn test_get_device_label() {
 #[test]
 #[should_panic]
 fn test_get_device_label_by_unknown_device() {
-    assert!(get_device_label(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_label(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_global_uid
@@ -1244,14 +1317,14 @@ fn test_get_device_label_by_unknown_device() {
 fn test_get_device_global_uid() {
     // Input device.
     if let Some(input) = test_get_default_device(Scope::Input) {
-        let uid = get_device_global_uid(input).unwrap();
+        let uid = run_serially(|| get_device_global_uid(input)).unwrap();
         let uid = uid.into_string();
         assert!(!uid.is_empty());
     }
 
     // Output device.
     if let Some(output) = test_get_default_device(Scope::Output) {
-        let uid = get_device_global_uid(output).unwrap();
+        let uid = run_serially(|| get_device_global_uid(output)).unwrap();
         let uid = uid.into_string();
         assert!(!uid.is_empty());
     }
@@ -1285,7 +1358,7 @@ fn test_create_cubeb_device_info() {
             if is_input {
                 let mut input_device_info = input_result.unwrap();
                 check_device_info_by_device(&input_device_info, device, Scope::Input);
-                destroy_cubeb_device_info(&mut input_device_info);
+                run_serially(|| destroy_cubeb_device_info(&mut input_device_info));
             } else {
                 assert_eq!(input_result.unwrap_err(), Error::error());
             }
@@ -1294,7 +1367,7 @@ fn test_create_cubeb_device_info() {
             if is_output {
                 let mut output_device_info = output_result.unwrap();
                 check_device_info_by_device(&output_device_info, device, Scope::Output);
-                destroy_cubeb_device_info(&mut output_device_info);
+                run_serially(|| destroy_cubeb_device_info(&mut output_device_info));
             } else {
                 assert_eq!(output_result.unwrap_err(), Error::error());
             }
@@ -1309,7 +1382,7 @@ fn test_create_cubeb_device_info() {
         let dev_types = [DeviceType::INPUT, DeviceType::OUTPUT];
         let mut results = VecDeque::new();
         for dev_type in dev_types.iter() {
-            results.push_back(create_cubeb_device_info(id, *dev_type));
+            results.push_back(run_serially(|| create_cubeb_device_info(id, *dev_type)));
         }
         results
     }
@@ -1369,7 +1442,9 @@ fn test_create_device_info_with_unknown_type() {
 
     fn test_create_device_info_with_unknown_type_by_scope(scope: Scope) {
         if let Some(device) = test_get_default_device(scope.clone()) {
-            assert!(create_cubeb_device_info(device, DeviceType::UNKNOWN).is_err());
+            assert!(
+                run_serially(|| create_cubeb_device_info(device, DeviceType::UNKNOWN)).is_err()
+            );
         }
     }
 }
@@ -1401,9 +1476,11 @@ fn test_create_device_from_hwdev_with_inout_type() {
     fn test_create_device_from_hwdev_with_inout_type_by_scope(scope: Scope) {
         if let Some(device) = test_get_default_device(scope.clone()) {
             // Get a kAudioHardwareUnknownPropertyError in get_channel_count actually.
-            assert!(
-                create_cubeb_device_info(device, DeviceType::INPUT | DeviceType::OUTPUT).is_err()
-            );
+            assert!(run_serially(|| create_cubeb_device_info(
+                device,
+                DeviceType::INPUT | DeviceType::OUTPUT
+            ))
+            .is_err());
         } else {
             println!("No device for {:?}.", scope);
         }
@@ -1416,9 +1493,10 @@ fn test_create_device_from_hwdev_with_inout_type() {
 fn test_get_devices_of_type() {
     use std::collections::HashSet;
 
-    let all_devices = audiounit_get_devices_of_type(DeviceType::INPUT | DeviceType::OUTPUT);
-    let input_devices = audiounit_get_devices_of_type(DeviceType::INPUT);
-    let output_devices = audiounit_get_devices_of_type(DeviceType::OUTPUT);
+    let all_devices =
+        run_serially(|| audiounit_get_devices_of_type(DeviceType::INPUT | DeviceType::OUTPUT));
+    let input_devices = run_serially(|| audiounit_get_devices_of_type(DeviceType::INPUT));
+    let output_devices = run_serially(|| audiounit_get_devices_of_type(DeviceType::OUTPUT));
 
     let mut expected_all = test_get_all_devices(DeviceFilter::ExcludeCubebAggregateAndVPIO);
     expected_all.sort();
@@ -1443,8 +1521,10 @@ fn test_get_devices_of_type() {
 #[test]
 #[should_panic]
 fn test_get_devices_of_type_unknown() {
-    let no_devs = audiounit_get_devices_of_type(DeviceType::UNKNOWN);
-    assert!(no_devs.is_empty());
+    run_serially_forward_panics(|| {
+        let no_devs = audiounit_get_devices_of_type(DeviceType::UNKNOWN);
+        assert!(no_devs.is_empty());
+    });
 }
 
 // add_devices_changed_listener
@@ -1468,9 +1548,10 @@ fn test_add_devices_changed_listener() {
             assert!(get_devices_changed_callback(context, Scope::Output).is_none());
 
             // Register a callback within a specific scope.
-            assert!(context
-                .add_devices_changed_listener(*devtype, Some(*callback), ptr::null_mut())
-                .is_ok());
+            assert!(run_serially(|| {
+                context.add_devices_changed_listener(*devtype, Some(*callback), ptr::null_mut())
+            })
+            .is_ok());
 
             if devtype.contains(DeviceType::INPUT) {
                 let cb = get_devices_changed_callback(context, Scope::Input);
@@ -1491,9 +1572,10 @@ fn test_add_devices_changed_listener() {
             }
 
             // Unregister the callbacks within all scopes.
-            assert!(context
-                .remove_devices_changed_listener(DeviceType::INPUT | DeviceType::OUTPUT)
-                .is_ok());
+            assert!(run_serially(|| {
+                context.remove_devices_changed_listener(DeviceType::INPUT | DeviceType::OUTPUT)
+            })
+            .is_ok());
 
             assert!(get_devices_changed_callback(context, Scope::Input).is_none());
             assert!(get_devices_changed_callback(context, Scope::Output).is_none());
@@ -1547,9 +1629,12 @@ fn test_remove_devices_changed_listener() {
 
             // Register callbacks within all scopes.
             for (scope, listener) in map.iter() {
-                assert!(context
-                    .add_devices_changed_listener(*scope, Some(*listener), ptr::null_mut())
-                    .is_ok());
+                assert!(run_serially(|| context.add_devices_changed_listener(
+                    *scope,
+                    Some(*listener),
+                    ptr::null_mut()
+                ))
+                .is_ok());
             }
 
             let input_callback = get_devices_changed_callback(context, Scope::Input);
@@ -1566,7 +1651,7 @@ fn test_remove_devices_changed_listener() {
             );
 
             // Unregister the callbacks within one specific scopes.
-            assert!(context.remove_devices_changed_listener(*devtype).is_ok());
+            assert!(run_serially(|| context.remove_devices_changed_listener(*devtype)).is_ok());
 
             if devtype.contains(DeviceType::INPUT) {
                 let cb = get_devices_changed_callback(context, Scope::Input);
@@ -1587,9 +1672,10 @@ fn test_remove_devices_changed_listener() {
             }
 
             // Unregister the callbacks within all scopes.
-            assert!(context
-                .remove_devices_changed_listener(DeviceType::INPUT | DeviceType::OUTPUT)
-                .is_ok());
+            assert!(run_serially(
+                || context.remove_devices_changed_listener(DeviceType::INPUT | DeviceType::OUTPUT)
+            )
+            .is_ok());
         }
     });
 }
@@ -1602,7 +1688,7 @@ fn test_remove_devices_changed_listener_without_adding_listeners() {
             DeviceType::OUTPUT,
             DeviceType::INPUT | DeviceType::OUTPUT,
         ] {
-            assert!(context.remove_devices_changed_listener(*devtype).is_ok());
+            assert!(run_serially(|| context.remove_devices_changed_listener(*devtype)).is_ok());
         }
     });
 }
@@ -1625,9 +1711,12 @@ fn test_remove_devices_changed_listener_within_all_scopes() {
             assert!(get_devices_changed_callback(context, Scope::Input).is_none());
             assert!(get_devices_changed_callback(context, Scope::Output).is_none());
 
-            assert!(context
-                .add_devices_changed_listener(*devtype, Some(*callback), ptr::null_mut())
-                .is_ok());
+            assert!(run_serially(|| context.add_devices_changed_listener(
+                *devtype,
+                Some(*callback),
+                ptr::null_mut()
+            ))
+            .is_ok());
 
             if devtype.contains(DeviceType::INPUT) {
                 let cb = get_devices_changed_callback(context, Scope::Input);
@@ -1641,9 +1730,10 @@ fn test_remove_devices_changed_listener_within_all_scopes() {
                 assert_eq!(cb.unwrap(), *callback);
             }
 
-            assert!(context
-                .remove_devices_changed_listener(DeviceType::INPUT | DeviceType::OUTPUT)
-                .is_ok());
+            assert!(run_serially(
+                || context.remove_devices_changed_listener(DeviceType::INPUT | DeviceType::OUTPUT)
+            )
+            .is_ok());
 
             assert!(get_devices_changed_callback(context, Scope::Input).is_none());
             assert!(get_devices_changed_callback(context, Scope::Output).is_none());
@@ -1661,3 +1751,135 @@ fn get_devices_changed_callback(
         Scope::Output => devices_guard.output.changed_callback,
     }
 }
+
+// SharedVoiceProcessingUnitManager
+// ------------------------------------
+#[test]
+fn test_shared_voice_processing_unit() {
+    let queue = Queue::new_with_target(
+        "test_shared_voice_processing_unit",
+        get_serial_queue_singleton(),
+    );
+    let mut shared = SharedVoiceProcessingUnitManager::new(queue.clone());
+    let r1 = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r1.is_err());
+    let r2 = queue.run_sync(|| shared.take_or_create()).unwrap();
+    assert!(r2.is_ok());
+    {
+        let _handle = r2.unwrap();
+        let r3 = queue.run_sync(|| shared.take()).unwrap();
+        assert!(r3.is_err());
+    }
+    let r4 = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r4.is_ok());
+}
+
+#[test]
+#[should_panic]
+fn test_shared_voice_processing_unit_bad_release_order() {
+    let queue = Queue::new_with_target(
+        "test_shared_voice_processing_unit_bad_release_order",
+        get_serial_queue_singleton(),
+    );
+    let mut shared = SharedVoiceProcessingUnitManager::new(queue.clone());
+    let r1 = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r1.is_ok());
+    drop(shared);
+    run_serially_forward_panics(|| drop(r1));
+}
+
+#[test]
+fn test_shared_voice_processing_multiple_units() {
+    let queue = Queue::new_with_target(
+        "test_shared_voice_processing_multiple_units",
+        get_serial_queue_singleton(),
+    );
+    let mut shared = SharedVoiceProcessingUnitManager::new(queue.clone());
+    let r1 = queue.run_sync(|| shared.take_or_create()).unwrap();
+    assert!(r1.is_ok());
+    let r2 = queue.run_sync(|| shared.take_or_create()).unwrap();
+    assert!(r2.is_ok());
+    {
+        let _handle1 = r1.unwrap();
+        let _handle2 = r2.unwrap();
+        let r3 = queue.run_sync(|| shared.take()).unwrap();
+        assert!(r3.is_err());
+    }
+    let r1 = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r1.is_ok());
+    let r2 = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r2.is_ok());
+    let r3 = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r3.is_err());
+}
+
+#[test]
+fn test_shared_voice_processing_release_on_idle() {
+    let queue = Queue::new_with_target(
+        "test_shared_voice_processing_release_on_idle",
+        get_serial_queue_singleton(),
+    );
+    let mut shared = SharedVoiceProcessingUnitManager::with_idle_timeout(
+        queue.clone(),
+        Duration::from_millis(0),
+    );
+    let r = queue.run_sync(|| shared.take_or_create()).unwrap();
+    assert!(r.is_ok());
+    {
+        let _handle = r.unwrap();
+    }
+    queue.run_sync(|| {});
+    let r = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r.is_err());
+}
+
+#[test]
+fn test_shared_voice_processing_no_release_on_outstanding() {
+    let queue = Queue::new_with_target(
+        "test_shared_voice_processing_no_release_on_outstanding",
+        get_serial_queue_singleton(),
+    );
+    let mut shared = SharedVoiceProcessingUnitManager::with_idle_timeout(
+        queue.clone(),
+        Duration::from_millis(0),
+    );
+    let r1 = queue.run_sync(|| shared.take_or_create()).unwrap();
+    assert!(r1.is_ok());
+    let r2 = queue.run_sync(|| shared.take_or_create()).unwrap();
+    assert!(r2.is_ok());
+    {
+        let _handle1 = r1.unwrap();
+    }
+    queue.run_sync(|| {});
+    let r1 = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r1.is_ok());
+}
+
+#[test]
+fn test_shared_voice_processing_release_on_idle_cancel_on_take() {
+    let queue = Queue::new_with_target(
+        "test_shared_voice_processing_release_on_idle_cancel_on_take",
+        get_serial_queue_singleton(),
+    );
+    let mut shared = SharedVoiceProcessingUnitManager::with_idle_timeout(
+        queue.clone(),
+        Duration::from_millis(0),
+    );
+    let r1 = queue.run_sync(|| shared.take_or_create()).unwrap();
+    assert!(r1.is_ok());
+    let r2 = queue.run_sync(|| shared.take_or_create()).unwrap();
+    assert!(r2.is_ok());
+    let r1 = queue
+        .run_sync(|| {
+            {
+                let _handle1 = r1.unwrap();
+                let _handle2 = r2.unwrap();
+            }
+            shared.take()
+        })
+        .unwrap();
+    assert!(r1.is_ok());
+    queue.run_sync(|| {});
+    let r2 = queue.run_sync(|| shared.take()).unwrap();
+    assert!(r2.is_ok());
+}
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/device_change.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/device_change.rs
index c27dada7ad..1201c446ea 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/device_change.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/device_change.rs
@@ -21,7 +21,7 @@ use super::utils::{
     test_set_default_device, Scope, StreamType, TestDevicePlugger, TestDeviceSwitcher,
 };
 use super::*;
-use std::sync::{LockResult, MutexGuard, WaitTimeoutResult};
+use std::sync::{LockResult, WaitTimeoutResult};
 
 // Switch default devices used by the active streams, to test stream reinitialization
 // ================================================================================================
@@ -49,19 +49,21 @@ fn test_switch_device_in_scope(scope: Scope) {
 
     let notifier = Arc::new(Notifier::new(0));
     let also_notifier = notifier.clone();
-    let listener = test_create_device_change_listener(scope.clone(), move |_addresses| {
-        let mut cnt = notifier.lock().unwrap();
-        *cnt += 1;
-        notifier.notify(cnt);
-        NO_ERR
+    let listener = run_serially(|| {
+        test_create_device_change_listener(scope.clone(), move |_addresses| {
+            let mut cnt = notifier.lock().unwrap();
+            *cnt += 1;
+            notifier.notify(cnt);
+            NO_ERR
+        })
     });
-    listener.start();
+    run_serially(|| listener.start());
 
     let changed_watcher = Watcher::new(&also_notifier);
     test_get_started_stream_in_scope(scope.clone(), move |_stream| loop {
-        let mut guard = changed_watcher.lock().unwrap();
-        let start_cnt = guard.clone();
+        let start_cnt = changed_watcher.lock().unwrap().clone();
         device_switcher.next();
+        let mut guard = changed_watcher.lock().unwrap();
         guard = changed_watcher
             .wait_while(guard, |cnt| *cnt == start_cnt)
             .unwrap();
@@ -709,11 +711,7 @@ fn test_unplug_a_device_on_an_active_stream(
         state_callback,
         device_changed_callback,
         |stream| {
-            stream.start();
-
-            let changed_watcher = Watcher::new(&notifier);
-            let mut data_guard = notifier.lock().unwrap();
-            assert_eq!(data_guard.states.last().unwrap(), &ffi::CUBEB_STATE_STARTED);
+            assert_eq!(stream.start(), Ok(()));
 
             println!(
                 "Stream runs on the device {} for {:?}",
@@ -722,13 +720,20 @@ fn test_unplug_a_device_on_an_active_stream(
             );
 
             let dev = plugger.get_device_id();
-            let start_changed_count = data_guard.changed_count.clone();
+            let start_changed_count = {
+                let guard = notifier.lock().unwrap();
+                assert_eq!(guard.states.last().unwrap(), &ffi::CUBEB_STATE_STARTED);
+                guard.changed_count.clone()
+            };
 
             assert!(plugger.unplug().is_ok());
 
+            let changed_watcher = Watcher::new(&notifier);
+
             if set_device_to_default {
                 // The stream will be reinitialized if it follows the default input or output device.
                 println!("Waiting for default device to change and reinit");
+                let mut data_guard = notifier.lock().unwrap();
                 data_guard = changed_watcher
                     .wait_while(data_guard, |data| {
                         data.changed_count == start_changed_count
@@ -740,6 +745,7 @@ fn test_unplug_a_device_on_an_active_stream(
                 // stream can be dropped immediately before device-changed callback
                 // so we only check the states if we wait for it explicitly.
                 println!("Waiting for non-default device to enter error state");
+                let mut data_guard = notifier.lock().unwrap();
                 let (new_guard, timeout_res) = changed_watcher
                     .wait_timeout_while(data_guard, Duration::from_millis(wait_up_to_ms), |data| {
                         data.states.last().unwrap_or(&ffi::CUBEB_STATE_STARTED)
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/device_property.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/device_property.rs
index 8277a7642d..a974aee64b 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/device_property.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/device_property.rs
@@ -7,14 +7,14 @@ use super::*;
 fn test_get_device_uid() {
     // Input device.
     if let Some(input) = test_get_default_device(Scope::Input) {
-        let uid = get_device_uid(input, DeviceType::INPUT).unwrap();
+        let uid = run_serially(|| get_device_uid(input, DeviceType::INPUT)).unwrap();
         let uid = uid.into_string();
         assert!(!uid.is_empty());
     }
 
     // Output device.
     if let Some(output) = test_get_default_device(Scope::Output) {
-        let uid = get_device_uid(output, DeviceType::OUTPUT).unwrap();
+        let uid = run_serially(|| get_device_uid(output, DeviceType::OUTPUT)).unwrap();
         let uid = uid.into_string();
         assert!(!uid.is_empty());
     }
@@ -24,7 +24,10 @@ fn test_get_device_uid() {
 #[should_panic]
 fn test_get_device_uid_by_unknwon_device() {
     // Unknown device.
-    assert!(get_device_uid(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(
+        run_serially_forward_panics(|| get_device_uid(kAudioObjectUnknown, DeviceType::INPUT))
+            .is_err()
+    );
 }
 
 // get_device_model_uid
@@ -33,7 +36,7 @@ fn test_get_device_uid_by_unknwon_device() {
 #[test]
 fn test_get_device_model_uid() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        match get_device_model_uid(device, DeviceType::INPUT) {
+        match run_serially(|| get_device_model_uid(device, DeviceType::INPUT)) {
             Ok(uid) => println!("input model uid: {}", uid.into_string()),
             Err(e) => println!("No input model uid. Error: {}", e),
         }
@@ -42,7 +45,7 @@ fn test_get_device_model_uid() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        match get_device_model_uid(device, DeviceType::OUTPUT) {
+        match run_serially(|| get_device_model_uid(device, DeviceType::OUTPUT)) {
             Ok(uid) => println!("output model uid: {}", uid.into_string()),
             Err(e) => println!("No output model uid. Error: {}", e),
         }
@@ -54,7 +57,11 @@ fn test_get_device_model_uid() {
 #[test]
 #[should_panic]
 fn test_get_device_model_uid_by_unknown_device() {
-    assert!(get_device_model_uid(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_model_uid(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_transport_type
@@ -62,7 +69,7 @@ fn test_get_device_model_uid_by_unknown_device() {
 #[test]
 fn test_get_device_transport_type() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        match get_device_transport_type(device, DeviceType::INPUT) {
+        match run_serially(|| get_device_transport_type(device, DeviceType::INPUT)) {
             Ok(trans_type) => println!(
                 "input transport type: {:X}, {:?}",
                 trans_type,
@@ -75,7 +82,7 @@ fn test_get_device_transport_type() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        match get_device_transport_type(device, DeviceType::OUTPUT) {
+        match run_serially(|| get_device_transport_type(device, DeviceType::OUTPUT)) {
             Ok(trans_type) => println!(
                 "output transport type: {:X}, {:?}",
                 trans_type,
@@ -91,7 +98,11 @@ fn test_get_device_transport_type() {
 #[test]
 #[should_panic]
 fn test_get_device_transport_type_by_unknown_device() {
-    assert!(get_device_transport_type(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_transport_type(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_source
@@ -100,7 +111,7 @@ fn test_get_device_transport_type_by_unknown_device() {
 #[test]
 fn test_get_device_source() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        match get_device_source(device, DeviceType::INPUT) {
+        match run_serially(|| get_device_source(device, DeviceType::INPUT)) {
             Ok(source) => println!(
                 "input source: {:X}, {:?}",
                 source,
@@ -113,7 +124,7 @@ fn test_get_device_source() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        match get_device_source(device, DeviceType::OUTPUT) {
+        match run_serially(|| get_device_source(device, DeviceType::OUTPUT)) {
             Ok(source) => println!(
                 "output source: {:X}, {:?}",
                 source,
@@ -129,7 +140,11 @@ fn test_get_device_source() {
 #[test]
 #[should_panic]
 fn test_get_device_source_by_unknown_device() {
-    assert!(get_device_source(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_source(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_source_name
@@ -137,7 +152,7 @@ fn test_get_device_source_by_unknown_device() {
 #[test]
 fn test_get_device_source_name() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        match get_device_source_name(device, DeviceType::INPUT) {
+        match run_serially(|| get_device_source_name(device, DeviceType::INPUT)) {
             Ok(name) => println!("input: {}", name.into_string()),
             Err(e) => println!("No input data source name. Error: {}", e),
         }
@@ -146,7 +161,7 @@ fn test_get_device_source_name() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        match get_device_source_name(device, DeviceType::OUTPUT) {
+        match run_serially(|| get_device_source_name(device, DeviceType::OUTPUT)) {
             Ok(name) => println!("output: {}", name.into_string()),
             Err(e) => println!("No output data source name. Error: {}", e),
         }
@@ -158,7 +173,11 @@ fn test_get_device_source_name() {
 #[test]
 #[should_panic]
 fn test_get_device_source_name_by_unknown_device() {
-    assert!(get_device_source_name(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_source_name(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_name
@@ -166,14 +185,14 @@ fn test_get_device_source_name_by_unknown_device() {
 #[test]
 fn test_get_device_name() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let name = get_device_name(device, DeviceType::INPUT).unwrap();
+        let name = run_serially(|| get_device_name(device, DeviceType::INPUT)).unwrap();
         println!("input device name: {}", name.into_string());
     } else {
         println!("No input device.");
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let name = get_device_name(device, DeviceType::OUTPUT).unwrap();
+        let name = run_serially(|| get_device_name(device, DeviceType::OUTPUT).unwrap());
         println!("output device name: {}", name.into_string());
     } else {
         println!("No output device.");
@@ -183,7 +202,11 @@ fn test_get_device_name() {
 #[test]
 #[should_panic]
 fn test_get_device_name_by_unknown_device() {
-    assert!(get_device_name(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_name(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_manufacturer
@@ -193,7 +216,7 @@ fn test_get_device_manufacturer() {
     if let Some(device) = test_get_default_device(Scope::Input) {
         // Some devices like AirPods cannot get the vendor info so we print the error directly.
         // TODO: Replace `map` and `unwrap_or_else` by `map_or_else`
-        let name = get_device_manufacturer(device, DeviceType::INPUT)
+        let name = run_serially(|| get_device_manufacturer(device, DeviceType::INPUT))
             .map(|name| name.into_string())
             .unwrap_or_else(|e| format!("Error: {}", e));
         println!("input device vendor: {}", name);
@@ -204,9 +227,10 @@ fn test_get_device_manufacturer() {
     if let Some(device) = test_get_default_device(Scope::Output) {
         // Some devices like AirPods cannot get the vendor info so we print the error directly.
         // TODO: Replace `map` and `unwrap_or_else` by `map_or_else`
-        let name = get_device_manufacturer(device, DeviceType::OUTPUT)
-            .map(|name| name.into_string())
-            .unwrap_or_else(|e| format!("Error: {}", e));
+        let name =
+            run_serially_forward_panics(|| get_device_manufacturer(device, DeviceType::OUTPUT))
+                .map(|name| name.into_string())
+                .unwrap_or_else(|e| format!("Error: {}", e));
         println!("output device vendor: {}", name);
     } else {
         println!("No output device.");
@@ -216,7 +240,11 @@ fn test_get_device_manufacturer() {
 #[test]
 #[should_panic]
 fn test_get_device_manufacturer_by_unknown_device() {
-    assert!(get_device_manufacturer(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_manufacturer(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_buffer_frame_size_range
@@ -224,7 +252,8 @@ fn test_get_device_manufacturer_by_unknown_device() {
 #[test]
 fn test_get_device_buffer_frame_size_range() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let range = get_device_buffer_frame_size_range(device, DeviceType::INPUT).unwrap();
+        let range =
+            run_serially(|| get_device_buffer_frame_size_range(device, DeviceType::INPUT)).unwrap();
         println!(
             "range of input buffer frame size: {}-{}",
             range.mMinimum, range.mMaximum
@@ -234,7 +263,8 @@ fn test_get_device_buffer_frame_size_range() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let range = get_device_buffer_frame_size_range(device, DeviceType::OUTPUT).unwrap();
+        let range = run_serially(|| get_device_buffer_frame_size_range(device, DeviceType::OUTPUT))
+            .unwrap();
         println!(
             "range of output buffer frame size: {}-{}",
             range.mMinimum, range.mMaximum
@@ -247,7 +277,13 @@ fn test_get_device_buffer_frame_size_range() {
 #[test]
 #[should_panic]
 fn test_get_device_buffer_frame_size_range_by_unknown_device() {
-    assert!(get_device_buffer_frame_size_range(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(
+        run_serially_forward_panics(|| get_device_buffer_frame_size_range(
+            kAudioObjectUnknown,
+            DeviceType::INPUT
+        ))
+        .is_err()
+    );
 }
 
 // get_device_latency
@@ -255,14 +291,14 @@ fn test_get_device_buffer_frame_size_range_by_unknown_device() {
 #[test]
 fn test_get_device_latency() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let latency = get_device_latency(device, DeviceType::INPUT).unwrap();
+        let latency = run_serially(|| get_device_latency(device, DeviceType::INPUT)).unwrap();
         println!("latency of input device: {}", latency);
     } else {
         println!("No input device.");
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let latency = get_device_latency(device, DeviceType::OUTPUT).unwrap();
+        let latency = run_serially(|| get_device_latency(device, DeviceType::OUTPUT)).unwrap();
         println!("latency of output device: {}", latency);
     } else {
         println!("No output device.");
@@ -272,7 +308,11 @@ fn test_get_device_latency() {
 #[test]
 #[should_panic]
 fn test_get_device_latency_by_unknown_device() {
-    assert!(get_device_latency(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_latency(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_streams
@@ -280,7 +320,7 @@ fn test_get_device_latency_by_unknown_device() {
 #[test]
 fn test_get_device_streams() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let streams = get_device_streams(device, DeviceType::INPUT).unwrap();
+        let streams = run_serially(|| get_device_streams(device, DeviceType::INPUT)).unwrap();
         println!("streams on the input device: {:?}", streams);
         assert!(!streams.is_empty());
     } else {
@@ -288,7 +328,7 @@ fn test_get_device_streams() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let streams = get_device_streams(device, DeviceType::OUTPUT).unwrap();
+        let streams = run_serially(|| get_device_streams(device, DeviceType::OUTPUT)).unwrap();
         println!("streams on the output device: {:?}", streams);
         assert!(!streams.is_empty());
     } else {
@@ -299,7 +339,11 @@ fn test_get_device_streams() {
 #[test]
 #[should_panic]
 fn test_get_device_streams_by_unknown_device() {
-    assert!(get_device_streams(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_streams(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_device_sample_rate
@@ -307,14 +351,14 @@ fn test_get_device_streams_by_unknown_device() {
 #[test]
 fn test_get_device_sample_rate() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let rate = get_device_sample_rate(device, DeviceType::INPUT).unwrap();
+        let rate = run_serially(|| get_device_sample_rate(device, DeviceType::INPUT)).unwrap();
         println!("input sample rate: {}", rate);
     } else {
         println!("No input device.");
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let rate = get_device_sample_rate(device, DeviceType::OUTPUT).unwrap();
+        let rate = run_serially(|| get_device_sample_rate(device, DeviceType::OUTPUT).unwrap());
         println!("output sample rate: {}", rate);
     } else {
         println!("No output device.");
@@ -324,7 +368,11 @@ fn test_get_device_sample_rate() {
 #[test]
 #[should_panic]
 fn test_get_device_sample_rate_by_unknown_device() {
-    assert!(get_device_sample_rate(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(run_serially_forward_panics(|| get_device_sample_rate(
+        kAudioObjectUnknown,
+        DeviceType::INPUT
+    ))
+    .is_err());
 }
 
 // get_ranges_of_device_sample_rate
@@ -332,14 +380,16 @@ fn test_get_device_sample_rate_by_unknown_device() {
 #[test]
 fn test_get_ranges_of_device_sample_rate() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let ranges = get_ranges_of_device_sample_rate(device, DeviceType::INPUT).unwrap();
+        let ranges =
+            run_serially(|| get_ranges_of_device_sample_rate(device, DeviceType::INPUT)).unwrap();
         println!("ranges of input sample rate: {:?}", ranges);
     } else {
         println!("No input device.");
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let ranges = get_ranges_of_device_sample_rate(device, DeviceType::OUTPUT).unwrap();
+        let ranges =
+            run_serially(|| get_ranges_of_device_sample_rate(device, DeviceType::OUTPUT)).unwrap();
         println!("ranges of output sample rate: {:?}", ranges);
     } else {
         println!("No output device.");
@@ -349,7 +399,13 @@ fn test_get_ranges_of_device_sample_rate() {
 #[test]
 #[should_panic]
 fn test_get_ranges_of_device_sample_rate_by_unknown_device() {
-    assert!(get_ranges_of_device_sample_rate(kAudioObjectUnknown, DeviceType::INPUT).is_err());
+    assert!(
+        run_serially_forward_panics(|| get_ranges_of_device_sample_rate(
+            kAudioObjectUnknown,
+            DeviceType::INPUT
+        ))
+        .is_err()
+    );
 }
 
 // get_stream_latency
@@ -357,9 +413,9 @@ fn test_get_ranges_of_device_sample_rate_by_unknown_device() {
 #[test]
 fn test_get_stream_latency() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let streams = get_device_streams(device, DeviceType::INPUT).unwrap();
+        let streams = run_serially(|| get_device_streams(device, DeviceType::INPUT)).unwrap();
         for stream in streams {
-            let latency = get_stream_latency(stream).unwrap();
+            let latency = run_serially(|| get_stream_latency(stream)).unwrap();
             println!("latency of the input stream {} is {}", stream, latency);
         }
     } else {
@@ -367,9 +423,9 @@ fn test_get_stream_latency() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let streams = get_device_streams(device, DeviceType::OUTPUT).unwrap();
+        let streams = run_serially(|| get_device_streams(device, DeviceType::OUTPUT)).unwrap();
         for stream in streams {
-            let latency = get_stream_latency(stream).unwrap();
+            let latency = run_serially(|| get_stream_latency(stream)).unwrap();
             println!("latency of the output stream {} is {}", stream, latency);
         }
     } else {
@@ -388,10 +444,10 @@ fn test_get_stream_latency_by_unknown_device() {
 #[test]
 fn test_get_stream_virtual_format() {
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let streams = get_device_streams(device, DeviceType::INPUT).unwrap();
+        let streams = run_serially(|| get_device_streams(device, DeviceType::INPUT)).unwrap();
         let formats = streams
             .iter()
-            .map(|s| get_stream_virtual_format(*s))
+            .map(|s| run_serially(|| get_stream_virtual_format(*s)))
             .collect::<Vec<std::result::Result<AudioStreamBasicDescription, OSStatus>>>();
         println!("input stream formats: {:?}", formats);
         assert!(!formats.is_empty());
@@ -400,10 +456,10 @@ fn test_get_stream_virtual_format() {
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let streams = get_device_streams(device, DeviceType::OUTPUT).unwrap();
+        let streams = run_serially(|| get_device_streams(device, DeviceType::OUTPUT)).unwrap();
         let formats = streams
             .iter()
-            .map(|s| get_stream_virtual_format(*s))
+            .map(|s| run_serially(|| get_stream_virtual_format(*s)))
             .collect::<Vec<std::result::Result<AudioStreamBasicDescription, OSStatus>>>();
         println!("output stream formats: {:?}", formats);
         assert!(!formats.is_empty());
@@ -415,7 +471,9 @@ fn test_get_stream_virtual_format() {
 #[test]
 #[should_panic]
 fn test_get_stream_virtual_format_by_unknown_stream() {
-    assert!(get_stream_virtual_format(kAudioObjectUnknown).is_err());
+    assert!(
+        run_serially_forward_panics(|| get_stream_virtual_format(kAudioObjectUnknown)).is_err()
+    );
 }
 
 // get_stream_terminal_type
@@ -442,25 +500,21 @@ fn test_get_stream_terminal_type() {
         }
     }
     if let Some(device) = test_get_default_device(Scope::Input) {
-        let streams = get_device_streams(device, DeviceType::INPUT).unwrap();
-        for stream in streams {
-            assert_eq!(
-                terminal_type_to_device_type(get_stream_terminal_type(stream).unwrap()),
-                Some(DeviceType::INPUT)
-            );
-        }
+        let streams = run_serially(|| get_device_streams(device, DeviceType::INPUT)).unwrap();
+        assert!(streams.iter().any(|&s| {
+            terminal_type_to_device_type(run_serially(|| get_stream_terminal_type(s)).unwrap())
+                == Some(DeviceType::INPUT)
+        }));
     } else {
         println!("No input device.");
     }
 
     if let Some(device) = test_get_default_device(Scope::Output) {
-        let streams = get_device_streams(device, DeviceType::OUTPUT).unwrap();
-        for stream in streams {
-            assert_eq!(
-                terminal_type_to_device_type(get_stream_terminal_type(stream).unwrap()),
-                Some(DeviceType::OUTPUT)
-            );
-        }
+        let streams = run_serially(|| get_device_streams(device, DeviceType::OUTPUT)).unwrap();
+        assert!(streams.iter().any(|&s| {
+            terminal_type_to_device_type(run_serially(|| get_stream_terminal_type(s)).unwrap())
+                == Some(DeviceType::OUTPUT)
+        }));
     } else {
         println!("No output device.");
     }
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/interfaces.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/interfaces.rs
index 340fec002d..aa15b7428b 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/interfaces.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/interfaces.rs
@@ -2,10 +2,12 @@ extern crate itertools;
 
 use self::itertools::iproduct;
 use super::utils::{
-    get_devices_info_in_scope, noop_data_callback, test_device_channels_in_scope,
-    test_get_default_device, test_ops_context_operation, test_ops_stream_operation, Scope,
+    draining_data_callback, get_devices_info_in_scope, noop_data_callback,
+    test_device_channels_in_scope, test_get_default_device, test_ops_context_operation,
+    test_ops_stream_operation, test_ops_stream_operation_on_context, Scope,
 };
 use super::*;
+use std::thread;
 
 // Context Operations
 // ------------------------------------------------------------------------------------------------
@@ -368,9 +370,6 @@ fn test_ops_context_register_device_collection_changed() {
 
 #[test]
 fn test_ops_context_register_device_collection_changed_with_a_duplex_stream() {
-    use std::thread;
-    use std::time::Duration;
-
     extern "C" fn callback(_: *mut ffi::cubeb, got_called_ptr: *mut c_void) {
         let got_called = unsafe { &mut *(got_called_ptr as *mut bool) };
         *got_called = true;
@@ -667,14 +666,20 @@ fn test_ops_context_stream_init_channel_rate_combinations() {
                 ffi::CUBEB_OK
             );
             assert!(!stream.is_null());
+
+            unsafe { OPS.stream_destroy.unwrap()(stream) };
         }
     });
 }
 
 // Stream Operations
 // ------------------------------------------------------------------------------------------------
-fn test_default_output_stream_operation<F>(name: &'static str, operation: F)
-where
+fn test_default_output_stream_operation_on_context_with_callback<F>(
+    name: &'static str,
+    context_ptr: *mut ffi::cubeb,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
     F: FnOnce(*mut ffi::cubeb_stream),
 {
     // Make sure the parameters meet the requirements of AudioUnitContext::stream_init
@@ -686,24 +691,53 @@ where
     output_params.layout = ffi::CUBEB_LAYOUT_UNDEFINED;
     output_params.prefs = ffi::CUBEB_STREAM_PREF_NONE;
 
-    test_ops_stream_operation(
+    test_ops_stream_operation_on_context(
         name,
+        context_ptr,
         ptr::null_mut(), // Use default input device.
         ptr::null_mut(), // No input parameters.
         ptr::null_mut(), // Use default output device.
         &mut output_params,
         4096, // TODO: Get latency by get_min_latency instead ?
-        Some(noop_data_callback),
+        data_callback,
         None,            // No state callback.
         ptr::null_mut(), // No user data pointer.
         operation,
     );
 }
 
-fn test_default_duplex_stream_operation<F>(name: &'static str, operation: F)
+fn test_default_output_stream_operation_with_callback<F>(
+    name: &'static str,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_ops_context_operation("context: default output stream operation", |context_ptr| {
+        test_default_output_stream_operation_on_context_with_callback(
+            name,
+            context_ptr,
+            data_callback,
+            operation,
+        );
+    });
+}
+
+fn test_default_output_stream_operation<F>(name: &'static str, operation: F)
 where
     F: FnOnce(*mut ffi::cubeb_stream),
 {
+    test_default_output_stream_operation_with_callback(name, Some(noop_data_callback), operation);
+}
+
+fn test_default_duplex_stream_operation_on_context_with_callback<F>(
+    name: &'static str,
+    context_ptr: *mut ffi::cubeb,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
     // Make sure the parameters meet the requirements of AudioUnitContext::stream_init
     // (in the comments).
     let mut input_params = ffi::cubeb_stream_params::default();
@@ -720,24 +754,53 @@ where
     output_params.layout = ffi::CUBEB_LAYOUT_UNDEFINED;
     output_params.prefs = ffi::CUBEB_STREAM_PREF_NONE;
 
-    test_ops_stream_operation(
+    test_ops_stream_operation_on_context(
         name,
+        context_ptr,
         ptr::null_mut(), // Use default input device.
         &mut input_params,
         ptr::null_mut(), // Use default output device.
         &mut output_params,
         4096, // TODO: Get latency by get_min_latency instead ?
-        Some(noop_data_callback),
+        data_callback,
         None,            // No state callback.
         ptr::null_mut(), // No user data pointer.
         operation,
     );
 }
 
-fn test_stereo_input_duplex_stream_operation<F>(name: &'static str, operation: F)
+fn test_default_duplex_stream_operation_with_callback<F>(
+    name: &'static str,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_ops_context_operation("context: default duplex stream operation", |context_ptr| {
+        test_default_duplex_stream_operation_on_context_with_callback(
+            name,
+            context_ptr,
+            data_callback,
+            operation,
+        );
+    });
+}
+
+fn test_default_duplex_stream_operation<F>(name: &'static str, operation: F)
 where
     F: FnOnce(*mut ffi::cubeb_stream),
 {
+    test_default_duplex_stream_operation_with_callback(name, Some(noop_data_callback), operation);
+}
+
+fn test_stereo_input_duplex_stream_operation_on_context_with_callback<F>(
+    name: &'static str,
+    context_ptr: *mut ffi::cubeb,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
     let mut input_devices = get_devices_info_in_scope(Scope::Input);
     input_devices.retain(|d| test_device_channels_in_scope(d.id, Scope::Input).unwrap_or(0) >= 2);
     if input_devices.is_empty() {
@@ -759,24 +822,138 @@ where
     output_params.layout = ffi::CUBEB_LAYOUT_UNDEFINED;
     output_params.prefs = ffi::CUBEB_STREAM_PREF_NONE;
 
-    test_ops_stream_operation(
+    test_ops_stream_operation_on_context(
         name,
+        context_ptr,
         input_devices[0].id as ffi::cubeb_devid,
         &mut input_params,
         ptr::null_mut(), // Use default output device.
         &mut output_params,
         4096, // TODO: Get latency by get_min_latency instead ?
+        data_callback,
+        None,            // No state callback.
+        ptr::null_mut(), // No user data pointer.
+        operation,
+    );
+}
+
+fn test_stereo_input_duplex_stream_operation_with_callback<F>(
+    name: &'static str,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_ops_context_operation(
+        "context: stereo input duplex stream operation",
+        |context_ptr| {
+            test_stereo_input_duplex_stream_operation_on_context_with_callback(
+                name,
+                context_ptr,
+                data_callback,
+                operation,
+            );
+        },
+    );
+}
+
+fn test_stereo_input_duplex_stream_operation<F>(name: &'static str, operation: F)
+where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_stereo_input_duplex_stream_operation_with_callback(
+        name,
         Some(noop_data_callback),
+        operation,
+    );
+}
+
+fn test_default_input_voice_stream_operation_on_context_with_callback<F>(
+    name: &'static str,
+    context_ptr: *mut ffi::cubeb,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    // Make sure the parameters meet the requirements of AudioUnitContext::stream_init
+    // (in the comments).
+    let mut input_params = ffi::cubeb_stream_params::default();
+    input_params.format = ffi::CUBEB_SAMPLE_FLOAT32NE;
+    input_params.rate = 44100;
+    input_params.channels = 1;
+    input_params.layout = ffi::CUBEB_LAYOUT_UNDEFINED;
+    input_params.prefs = ffi::CUBEB_STREAM_PREF_VOICE;
+
+    test_ops_stream_operation_on_context(
+        name,
+        context_ptr,
+        ptr::null_mut(), // Use default input device.
+        &mut input_params,
+        ptr::null_mut(), // Use default output device.
+        ptr::null_mut(), // No output parameters.
+        4096,            // TODO: Get latency by get_min_latency instead ?
+        data_callback,
         None,            // No state callback.
         ptr::null_mut(), // No user data pointer.
         operation,
     );
 }
 
-fn test_default_duplex_voice_stream_operation<F>(name: &'static str, operation: F)
+fn test_default_input_voice_stream_operation_on_context<F>(
+    name: &'static str,
+    context_ptr: *mut ffi::cubeb,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_default_input_voice_stream_operation_on_context_with_callback(
+        name,
+        context_ptr,
+        Some(noop_data_callback),
+        operation,
+    );
+}
+
+fn test_default_input_voice_stream_operation_with_callback<F>(
+    name: &'static str,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_ops_context_operation(
+        "context: default input voice stream operation",
+        |context_ptr| {
+            test_default_input_voice_stream_operation_on_context_with_callback(
+                name,
+                context_ptr,
+                data_callback,
+                operation,
+            );
+        },
+    );
+}
+
+fn test_default_input_voice_stream_operation<F>(name: &'static str, operation: F)
 where
     F: FnOnce(*mut ffi::cubeb_stream),
 {
+    test_default_input_voice_stream_operation_with_callback(
+        name,
+        Some(noop_data_callback),
+        operation,
+    );
+}
+
+fn test_default_duplex_voice_stream_operation_on_context_with_callback<F>(
+    name: &'static str,
+    context_ptr: *mut ffi::cubeb,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
     // Make sure the parameters meet the requirements of AudioUnitContext::stream_init
     // (in the comments).
     let mut input_params = ffi::cubeb_stream_params::default();
@@ -793,20 +970,64 @@ where
     output_params.layout = ffi::CUBEB_LAYOUT_UNDEFINED;
     output_params.prefs = ffi::CUBEB_STREAM_PREF_VOICE;
 
-    test_ops_stream_operation(
+    test_ops_stream_operation_on_context(
         name,
+        context_ptr,
         ptr::null_mut(), // Use default input device.
         &mut input_params,
         ptr::null_mut(), // Use default output device.
         &mut output_params,
         4096, // TODO: Get latency by get_min_latency instead ?
-        Some(noop_data_callback),
+        data_callback,
         None,            // No state callback.
         ptr::null_mut(), // No user data pointer.
         operation,
     );
 }
 
+fn test_default_duplex_voice_stream_operation_on_context<F>(
+    name: &'static str,
+    context_ptr: *mut ffi::cubeb,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_default_duplex_voice_stream_operation_on_context_with_callback(
+        name,
+        context_ptr,
+        Some(noop_data_callback),
+        operation,
+    );
+}
+
+fn test_default_duplex_voice_stream_operation_with_callback<F>(
+    name: &'static str,
+    data_callback: ffi::cubeb_data_callback,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_ops_context_operation("context: duplex voice stream operation", |context_ptr| {
+        test_default_duplex_voice_stream_operation_on_context_with_callback(
+            name,
+            context_ptr,
+            data_callback,
+            operation,
+        );
+    });
+}
+
+fn test_default_duplex_voice_stream_operation<F>(name: &'static str, operation: F)
+where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_default_duplex_voice_stream_operation_with_callback(
+        name,
+        Some(noop_data_callback),
+        operation,
+    );
+}
+
 fn test_stereo_input_duplex_voice_stream_operation<F>(name: &'static str, operation: F)
 where
     F: FnOnce(*mut ffi::cubeb_stream),
@@ -866,6 +1087,18 @@ fn test_ops_stream_stop() {
 }
 
 #[test]
+fn test_ops_stream_drain() {
+    test_default_output_stream_operation_with_callback(
+        "stream: drain",
+        Some(draining_data_callback),
+        |stream| {
+            assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+            thread::sleep(Duration::from_millis(10));
+        },
+    );
+}
+
+#[test]
 fn test_ops_stream_position() {
     test_default_output_stream_operation("stream: position", |stream| {
         let mut position = u64::max_value();
@@ -987,17 +1220,72 @@ fn test_ops_stereo_input_duplex_stream_stop() {
 }
 
 #[test]
-fn test_ops_duplex_voice_stream_init_and_destroy() {
-    test_default_duplex_voice_stream_operation(
-        "duplex voice stream: init and destroy",
-        |_stream| {},
+fn test_ops_stereo_input_duplex_stream_drain() {
+    test_stereo_input_duplex_stream_operation_with_callback(
+        "stereo-input duplex stream: drain",
+        Some(draining_data_callback),
+        |stream| {
+            assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+            thread::sleep(Duration::from_millis(10));
+        },
     );
 }
 
 #[test]
+fn test_ops_input_voice_stream_init_and_destroy() {
+    test_default_input_voice_stream_operation("input voice stream: init and destroy", |stream| {
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert_eq!(
+            stm.core_stream_data.using_voice_processing_unit(),
+            macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+        );
+    });
+}
+
+#[test]
+fn test_ops_input_voice_stream_start() {
+    test_default_input_voice_stream_operation("input voice stream: start", |stream| {
+        assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert_eq!(
+            stm.core_stream_data.using_voice_processing_unit(),
+            macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+        );
+    });
+}
+
+#[test]
+fn test_ops_input_voice_stream_stop() {
+    test_default_input_voice_stream_operation("input voice stream: stop", |stream| {
+        assert_eq!(unsafe { OPS.stream_stop.unwrap()(stream) }, ffi::CUBEB_OK);
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert_eq!(
+            stm.core_stream_data.using_voice_processing_unit(),
+            macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+        );
+    });
+}
+
+#[test]
+fn test_ops_duplex_voice_stream_init_and_destroy() {
+    test_default_duplex_voice_stream_operation("duplex voice stream: init and destroy", |stream| {
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert_eq!(
+            stm.core_stream_data.using_voice_processing_unit(),
+            macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+        );
+    });
+}
+
+#[test]
 fn test_ops_duplex_voice_stream_start() {
     test_default_duplex_voice_stream_operation("duplex voice stream: start", |stream| {
         assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert_eq!(
+            stm.core_stream_data.using_voice_processing_unit(),
+            macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+        );
     });
 }
 
@@ -1005,21 +1293,334 @@ fn test_ops_duplex_voice_stream_start() {
 fn test_ops_duplex_voice_stream_stop() {
     test_default_duplex_voice_stream_operation("duplex voice stream: stop", |stream| {
         assert_eq!(unsafe { OPS.stream_stop.unwrap()(stream) }, ffi::CUBEB_OK);
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert_eq!(
+            stm.core_stream_data.using_voice_processing_unit(),
+            macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+        );
+    });
+}
+
+#[test]
+fn test_ops_duplex_voice_stream_drain() {
+    test_default_duplex_voice_stream_operation_with_callback(
+        "duplex voice stream: drain",
+        Some(draining_data_callback),
+        |stream| {
+            assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+            let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert_eq!(
+                stm.core_stream_data.using_voice_processing_unit(),
+                macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+            );
+            thread::sleep(Duration::from_millis(10));
+        },
+    );
+}
+
+#[test]
+#[ignore]
+fn test_ops_timing_sensitive_multiple_voice_stream_init_and_destroy() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
+    let start = Instant::now();
+    let mut t1 = start;
+    let mut t2 = start;
+    let mut t3 = start;
+    let mut t4 = start;
+    let mut t5 = start;
+    let mut t6 = start;
+    let mut t7 = start;
+    let mut t8 = start;
+    let mut t9 = start;
+    let mut t10 = start;
+    test_ops_context_operation("multiple duplex voice streams", |context_ptr| {
+        // First stream uses vpio, creates the shared vpio unit.
+        test_default_duplex_voice_stream_operation_on_context(
+            "multiple voice streams: stream 1, duplex",
+            context_ptr,
+            |stream| {
+                let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                assert!(stm.core_stream_data.using_voice_processing_unit());
+
+                // Two concurrent vpio streams are supported.
+                test_default_input_voice_stream_operation_on_context(
+                    "multiple voice streams: stream 2, input-only",
+                    context_ptr,
+                    |stream| {
+                        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                        assert!(stm.core_stream_data.using_voice_processing_unit());
+
+                        // Three concurrent vpio streams are supported.
+                        test_default_duplex_voice_stream_operation_on_context(
+                            "multiple voice streams: stream 3, duplex",
+                            context_ptr,
+                            |stream| {
+                                let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                                assert!(stm.core_stream_data.using_voice_processing_unit());
+                            },
+                        );
+                    },
+                );
+            },
+        );
+        t1 = Instant::now();
+        // Fourth stream uses vpio, allows reuse of one already created.
+        test_default_duplex_voice_stream_operation_on_context(
+            "multiple voice streams: stream 4, duplex",
+            context_ptr,
+            |stream| {
+                let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                assert!(stm.core_stream_data.using_voice_processing_unit());
+                t2 = Instant::now();
+
+                // Fifth stream uses vpio, allows reuse of one already created.
+                test_default_duplex_voice_stream_operation_on_context(
+                    "multiple voice streams: stream 5, duplex",
+                    context_ptr,
+                    |stream| {
+                        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                        assert!(stm.core_stream_data.using_voice_processing_unit());
+                        t3 = Instant::now();
+
+                        // Sixth stream uses vpio, allows reuse of one already created.
+                        test_default_input_voice_stream_operation_on_context(
+                            "multiple voice streams: stream 6, input-only",
+                            context_ptr,
+                            |stream| {
+                                let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                                assert!(stm.core_stream_data.using_voice_processing_unit());
+                                t4 = Instant::now();
+
+                                // Seventh stream uses vpio, but is created anew.
+                                test_default_input_voice_stream_operation_on_context(
+                                    "multiple voice streams: stream 7, input-only",
+                                    context_ptr,
+                                    |stream| {
+                                        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                                        assert!(stm.core_stream_data.using_voice_processing_unit());
+                                        t5 = Instant::now();
+                                    },
+                                );
+                                t6 = Instant::now();
+                            },
+                        );
+                        t7 = Instant::now();
+                    },
+                );
+                t8 = Instant::now();
+            },
+        );
+        t9 = Instant::now();
+    });
+    t10 = Instant::now();
+
+    let reuse_vpio_1 = t2 - t1;
+    let reuse_vpio_2 = t3 - t2;
+    let reuse_vpio_3 = t4 - t3;
+    let create_standalone_vpio = t5 - t4;
+    assert!(
+        create_standalone_vpio > reuse_vpio_1 * 2,
+        "Failed create_standalone_vpio={}s > reuse_vpio_1={}s * 2",
+        create_standalone_vpio.as_secs_f32(),
+        reuse_vpio_1.as_secs_f32()
+    );
+    assert!(
+        create_standalone_vpio > reuse_vpio_2 * 2,
+        "Failed create_standalone_vpio={}s > reuse_vpio_2={}s * 2",
+        create_standalone_vpio.as_secs_f32(),
+        reuse_vpio_2.as_secs_f32()
+    );
+    assert!(
+        create_standalone_vpio > reuse_vpio_3 * 2,
+        "Failed create_standalone_vpio={}s > reuse_vpio_3={}s * 2",
+        create_standalone_vpio.as_secs_f32(),
+        reuse_vpio_3.as_secs_f32()
+    );
+
+    let recycle_vpio_1 = t6 - t5;
+    let recycle_vpio_2 = t7 - t6;
+    let recycle_vpio_3 = t8 - t7;
+    let recycle_vpio_4 = t9 - t8;
+    let dispose_vpios = t10 - t9;
+    assert!(
+        dispose_vpios > recycle_vpio_1 * 2,
+        "Failed dispose_vpios={}s > recycle_vpio_1 ={}s * 2",
+        dispose_vpios.as_secs_f32(),
+        recycle_vpio_1.as_secs_f32()
+    );
+    assert!(
+        dispose_vpios > recycle_vpio_2 * 2,
+        "Failed dispose_vpios={}s > recycle_vpio_2 ={}s * 2",
+        dispose_vpios.as_secs_f32(),
+        recycle_vpio_2.as_secs_f32()
+    );
+    assert!(
+        dispose_vpios > recycle_vpio_3 * 2,
+        "Failed dispose_vpios={}s > recycle_vpio_3 ={}s * 2",
+        dispose_vpios.as_secs_f32(),
+        recycle_vpio_3.as_secs_f32()
+    );
+    assert!(
+        dispose_vpios > recycle_vpio_4 * 2,
+        "Failed dispose_vpios={}s > recycle_vpio_4 ={}s * 2",
+        dispose_vpios.as_secs_f32(),
+        recycle_vpio_4.as_secs_f32()
+    );
+}
+
+#[test]
+#[ignore]
+fn test_ops_timing_sensitive_multiple_duplex_voice_stream_start() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
+    test_ops_context_operation("multiple duplex voice streams", |context_ptr| {
+        let start = Instant::now();
+        // First stream uses vpio, creates the shared vpio unit.
+        test_default_duplex_voice_stream_operation_on_context(
+            "multiple duplex voice streams: stream 1",
+            context_ptr,
+            |stream| {
+                let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                assert!(stm.core_stream_data.using_voice_processing_unit());
+                assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+            },
+        );
+        let d1 = start.elapsed();
+        // Second stream uses vpio, allows reuse of the one already created.
+        test_default_duplex_voice_stream_operation_on_context(
+            "multiple duplex voice streams: stream 2",
+            context_ptr,
+            |stream| {
+                let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                assert!(stm.core_stream_data.using_voice_processing_unit());
+                assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+            },
+        );
+        let d2 = start.elapsed() - d1;
+        // d1 being significantly longer than d2 is proof we reuse vpio.
+        assert!(
+            d1 > d2 * 2,
+            "Failed d1={}s > d2={}s * s",
+            d1.as_secs_f32(),
+            d2.as_secs_f32()
+        );
+    });
+}
+
+#[test]
+#[ignore]
+fn test_ops_timing_sensitive_multiple_duplex_voice_stream_params() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
+    test_ops_context_operation("multiple duplex voice streams with params", |context_ptr| {
+        let start = Instant::now();
+        // First stream uses vpio, creates the shared vpio unit.
+        test_default_duplex_voice_stream_operation_on_context(
+            "multiple duplex voice streams: stream 1",
+            context_ptr,
+            |stream| {
+                let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                assert!(stm.core_stream_data.using_voice_processing_unit());
+                assert_eq!(
+                    unsafe {
+                        OPS.stream_set_input_processing_params.unwrap()(
+                            stream,
+                            ffi::CUBEB_INPUT_PROCESSING_PARAM_ECHO_CANCELLATION
+                                | ffi::CUBEB_INPUT_PROCESSING_PARAM_NOISE_SUPPRESSION,
+                        )
+                    },
+                    ffi::CUBEB_OK
+                );
+                assert_eq!(
+                    unsafe { OPS.stream_set_input_mute.unwrap()(stream, 1) },
+                    ffi::CUBEB_OK
+                );
+            },
+        );
+        let d1 = start.elapsed();
+        // Second stream uses vpio, allows reuse of the one already created.
+        test_default_duplex_voice_stream_operation_on_context(
+            "multiple duplex voice streams: stream 2",
+            context_ptr,
+            |stream| {
+                let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+                assert!(stm.core_stream_data.using_voice_processing_unit());
+                let queue = stm.queue.clone();
+                // Test that input processing params does not carry over when reusing vpio.
+                let mut bypass: u32 = 0;
+                let r = queue
+                    .run_sync(|| {
+                        audio_unit_get_property(
+                            stm.core_stream_data.input_unit,
+                            kAUVoiceIOProperty_BypassVoiceProcessing,
+                            kAudioUnitScope_Global,
+                            AU_IN_BUS,
+                            &mut bypass,
+                            &mut mem::size_of::<u32>(),
+                        )
+                    })
+                    .unwrap();
+                assert_eq!(r, NO_ERR);
+                assert_eq!(bypass, 1);
+
+                // Test that input mute state does not carry over when reusing vpio.
+                let mut mute: u32 = 0;
+                let r = queue
+                    .run_sync(|| {
+                        audio_unit_get_property(
+                            stm.core_stream_data.input_unit,
+                            kAUVoiceIOProperty_MuteOutput,
+                            kAudioUnitScope_Global,
+                            AU_IN_BUS,
+                            &mut mute,
+                            &mut mem::size_of::<u32>(),
+                        )
+                    })
+                    .unwrap();
+                assert_eq!(r, NO_ERR);
+                assert_eq!(mute, 0);
+            },
+        );
+        let d2 = start.elapsed() - d1;
+        // d1 being significantly longer than d2 is proof we reuse vpio.
+        assert!(
+            d1 > d2 * 2,
+            "Failed d1={}s > d2={}s * 2",
+            d1.as_secs_f32(),
+            d2.as_secs_f32()
+        );
     });
 }
 
 #[test]
 fn test_ops_duplex_voice_stream_set_input_mute() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
     test_default_duplex_voice_stream_operation("duplex voice stream: mute", |stream| {
         assert_eq!(
             unsafe { OPS.stream_set_input_mute.unwrap()(stream, 1) },
             ffi::CUBEB_OK
         );
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert!(stm.core_stream_data.using_voice_processing_unit());
     });
 }
 
 #[test]
 fn test_ops_duplex_voice_stream_set_input_mute_before_start() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
     test_default_duplex_voice_stream_operation(
         "duplex voice stream: mute before start",
         |stream| {
@@ -1028,12 +1629,18 @@ fn test_ops_duplex_voice_stream_set_input_mute_before_start() {
                 ffi::CUBEB_OK
             );
             assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+            let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert!(stm.core_stream_data.using_voice_processing_unit());
         },
     );
 }
 
 #[test]
 fn test_ops_duplex_voice_stream_set_input_mute_before_start_with_reinit() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
     test_default_duplex_voice_stream_operation(
         "duplex voice stream: mute before start with reinit",
         |stream| {
@@ -1045,6 +1652,7 @@ fn test_ops_duplex_voice_stream_set_input_mute_before_start_with_reinit() {
 
             // Hacky cast, but testing this here was simplest for now.
             let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert!(stm.core_stream_data.using_voice_processing_unit());
             stm.reinit_async();
             let queue = stm.queue.clone();
             let mut mute_after_reinit = false;
@@ -1068,17 +1676,27 @@ fn test_ops_duplex_voice_stream_set_input_mute_before_start_with_reinit() {
 
 #[test]
 fn test_ops_duplex_voice_stream_set_input_mute_after_start() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
     test_default_duplex_voice_stream_operation("duplex voice stream: mute after start", |stream| {
         assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
         assert_eq!(
             unsafe { OPS.stream_set_input_mute.unwrap()(stream, 1) },
             ffi::CUBEB_OK
         );
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert!(stm.core_stream_data.using_voice_processing_unit());
     });
 }
 
 #[test]
 fn test_ops_duplex_voice_stream_set_input_processing_params() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
     test_default_duplex_voice_stream_operation("duplex voice stream: processing", |stream| {
         let params: ffi::cubeb_input_processing_params =
             ffi::CUBEB_INPUT_PROCESSING_PARAM_ECHO_CANCELLATION
@@ -1088,11 +1706,17 @@ fn test_ops_duplex_voice_stream_set_input_processing_params() {
             unsafe { OPS.stream_set_input_processing_params.unwrap()(stream, params) },
             ffi::CUBEB_OK
         );
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        assert!(stm.core_stream_data.using_voice_processing_unit());
     });
 }
 
 #[test]
 fn test_ops_duplex_voice_stream_set_input_processing_params_before_start() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
     test_default_duplex_voice_stream_operation(
         "duplex voice stream: processing before start",
         |stream| {
@@ -1105,12 +1729,18 @@ fn test_ops_duplex_voice_stream_set_input_processing_params_before_start() {
                 ffi::CUBEB_OK
             );
             assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
+            let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert!(stm.core_stream_data.using_voice_processing_unit());
         },
     );
 }
 
 #[test]
 fn test_ops_duplex_voice_stream_set_input_processing_params_before_start_with_reinit() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
     test_default_duplex_voice_stream_operation(
         "duplex voice stream: processing before start with reinit",
         |stream| {
@@ -1126,6 +1756,7 @@ fn test_ops_duplex_voice_stream_set_input_processing_params_before_start_with_re
 
             // Hacky cast, but testing this here was simplest for now.
             let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert!(stm.core_stream_data.using_voice_processing_unit());
             stm.reinit_async();
             let queue = stm.queue.clone();
             let mut params_after_reinit: ffi::cubeb_input_processing_params =
@@ -1170,9 +1801,15 @@ fn test_ops_duplex_voice_stream_set_input_processing_params_before_start_with_re
 
 #[test]
 fn test_ops_duplex_voice_stream_set_input_processing_params_after_start() {
+    if macos_kernel_major_version().unwrap() == MACOS_KERNEL_MAJOR_VERSION_MONTEREY {
+        // We disable VPIO on Monterey.
+        return;
+    }
     test_default_duplex_voice_stream_operation(
         "duplex voice stream: processing after start",
         |stream| {
+            let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert!(stm.core_stream_data.using_voice_processing_unit());
             assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
             let params: ffi::cubeb_input_processing_params =
                 ffi::CUBEB_INPUT_PROCESSING_PARAM_ECHO_CANCELLATION
@@ -1190,7 +1827,13 @@ fn test_ops_duplex_voice_stream_set_input_processing_params_after_start() {
 fn test_ops_stereo_input_duplex_voice_stream_init_and_destroy() {
     test_stereo_input_duplex_voice_stream_operation(
         "stereo-input duplex voice stream: init and destroy",
-        |_stream| {},
+        |stream| {
+            let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert_eq!(
+                stm.core_stream_data.using_voice_processing_unit(),
+                macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+            );
+        },
     );
 }
 
@@ -1199,6 +1842,11 @@ fn test_ops_stereo_input_duplex_voice_stream_start() {
     test_stereo_input_duplex_voice_stream_operation(
         "stereo-input duplex voice stream: start",
         |stream| {
+            let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert_eq!(
+                stm.core_stream_data.using_voice_processing_unit(),
+                macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+            );
             assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
         },
     );
@@ -1209,6 +1857,11 @@ fn test_ops_stereo_input_duplex_voice_stream_stop() {
     test_stereo_input_duplex_voice_stream_operation(
         "stereo-input duplex voice stream: stop",
         |stream| {
+            let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+            assert_eq!(
+                stm.core_stream_data.using_voice_processing_unit(),
+                macos_kernel_major_version().unwrap() != MACOS_KERNEL_MAJOR_VERSION_MONTEREY
+            );
             assert_eq!(unsafe { OPS.stream_stop.unwrap()(stream) }, ffi::CUBEB_OK);
         },
     );
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/manual.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/manual.rs
index b2b2241cc9..3bf96c65c6 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/manual.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/manual.rs
@@ -4,7 +4,6 @@ use super::utils::{
 };
 use super::*;
 use std::io;
-use std::sync::atomic::AtomicBool;
 
 #[ignore]
 #[test]
@@ -153,25 +152,84 @@ fn test_device_collection_change() {
     let _ = std::io::stdin().read_line(&mut input);
 }
 
+struct StreamData {
+    stream_ptr: *mut ffi::cubeb_stream,
+    enable_loopback: AtomicBool,
+}
+
+impl StreamData {
+    fn new() -> Self {
+        Self {
+            stream_ptr: ptr::null_mut(),
+            enable_loopback: AtomicBool::new(false),
+        }
+    }
+}
+
+struct StreamsData {
+    streams: Vec<StreamData>,
+    current_idx: Option<usize>,
+}
+
+impl StreamsData {
+    fn new() -> Self {
+        Self {
+            streams: Vec::new(),
+            current_idx: None,
+        }
+    }
+
+    fn len(&self) -> usize {
+        self.streams.len()
+    }
+
+    fn current_mut(&mut self) -> &mut StreamData {
+        &mut self.streams[self.current_idx.unwrap()]
+    }
+
+    fn current(&self) -> &StreamData {
+        &self.streams[self.current_idx.unwrap()]
+    }
+
+    fn select(&mut self, idx: usize) {
+        assert!(idx < self.len());
+        self.current_idx = Some(idx);
+    }
+
+    fn push(&mut self, stream: StreamData) {
+        self.streams.push(stream)
+    }
+}
+
 #[ignore]
 #[test]
 fn test_stream_tester() {
     test_ops_context_operation("context: stream tester", |context_ptr| {
-        let mut stream_ptr: *mut ffi::cubeb_stream = ptr::null_mut();
-        let enable_loopback = AtomicBool::new(false);
+        let mut input_prefs = StreamPrefs::NONE;
+        let mut output_prefs = StreamPrefs::NONE;
+        let mut streams = StreamsData::new();
         loop {
             println!(
-                "commands:\n\
+                "Current stream: {} (of {}). Commands:\n\
                  \t'q': quit\n\
+                 \t'b': change current stream\n\
+                 \t'i': set input stream prefs to be used for creating streams\n\
+                 \t'o': set output stream prefs to be used for creating streams\n\
                  \t'c': create a stream\n\
-                 \t'd': destroy a stream\n\
-                 \t's': start the created stream\n\
-                 \t't': stop the created stream\n\
+                 Commands on the current stream:\n\
+                 \t'd': destroy\n\
+                 \t's': start\n\
+                 \t't': stop\n\
                  \t'r': register a device changed callback\n\
                  \t'l': set loopback (DUPLEX-only)\n\
                  \t'v': set volume\n\
                  \t'm': set input mute\n\
-                 \t'p': set input processing"
+                 \t'p': set input processing",
+                streams
+                    .current_idx
+                    .map(|i| format!("{}", i + 1 as usize))
+                    .unwrap_or(String::from("N/A")),
+                streams.len(),
             );
 
             let mut command = String::new();
@@ -181,66 +239,130 @@ fn test_stream_tester() {
             match command.as_str() {
                 "q" => {
                     println!("Quit.");
-                    destroy_stream(&mut stream_ptr);
+                    for mut stream in streams.streams {
+                        if !stream.stream_ptr.is_null() {
+                            destroy_stream(&mut stream);
+                        }
+                    }
                     break;
                 }
-                "c" => create_stream(&mut stream_ptr, context_ptr, &enable_loopback),
-                "d" => destroy_stream(&mut stream_ptr),
-                "s" => start_stream(stream_ptr),
-                "t" => stop_stream(stream_ptr),
-                "r" => register_device_change_callback(stream_ptr),
-                "l" => set_loopback(stream_ptr, &enable_loopback),
-                "v" => set_volume(stream_ptr),
-                "m" => set_input_mute(stream_ptr),
-                "p" => set_input_processing(stream_ptr),
-                x => println!("Unknown command: {}", x),
+                "i" => set_prefs(&mut input_prefs),
+                "o" => set_prefs(&mut output_prefs),
+                "c" => create_stream(context_ptr, &mut streams, input_prefs, output_prefs),
+                _ if streams.current_idx.is_none() => {
+                    println!("There are no streams! Create a stream first.")
+                }
+                cmd => match cmd {
+                    "b" => select_stream(&mut streams),
+                    "d" => destroy_stream(streams.current_mut()),
+                    "s" => start_stream(streams.current()),
+                    "t" => stop_stream(streams.current()),
+                    "r" => register_device_change_callback(streams.current()),
+                    "l" => set_loopback(streams.current()),
+                    "v" => set_volume(streams.current()),
+                    "m" => set_input_mute(streams.current()),
+                    "p" => set_input_processing(streams.current()),
+                    x => println!("Unknown command: {}", x),
+                },
             }
         }
     });
 
-    fn start_stream(stream_ptr: *mut ffi::cubeb_stream) {
-        if stream_ptr.is_null() {
+    fn set_prefs(prefs: &mut StreamPrefs) {
+        let mut done = false;
+        while !done {
+            println!(
+                "Current prefs: {:?}\nSelect action:\n\
+                 \t1) Set None\n\
+                 \t2) Toggle Loopback\n\
+                 \t3) Toggle Disable Device Switching\n\
+                 \t4) Toggle Voice\n\
+                 \t5) Set All\n\
+                 \t0) Done",
+                prefs
+            );
+            let mut input = String::new();
+            let _ = io::stdin().read_line(&mut input);
+            assert_eq!(input.pop().unwrap(), '\n');
+            match input.as_str() {
+                "1" => *prefs = StreamPrefs::NONE,
+                "2" => prefs.toggle(StreamPrefs::LOOPBACK),
+                "3" => prefs.toggle(StreamPrefs::DISABLE_DEVICE_SWITCHING),
+                "4" => prefs.toggle(StreamPrefs::VOICE),
+                "5" => *prefs = StreamPrefs::all(),
+                "0" => done = true,
+                _ => println!("Invalid action. Select again.\n"),
+            }
+        }
+    }
+
+    fn select_stream(streams: &mut StreamsData) {
+        let num_streams = streams.len();
+        let current_idx = streams.current_idx.unwrap();
+        println!(
+            "Current stream is {}. Select stream 1 to {} on which to apply commands:",
+            current_idx + 1 as usize,
+            num_streams
+        );
+        let mut selection: Option<usize> = None;
+        while selection.is_none() {
+            let mut input = String::new();
+            let _ = io::stdin().read_line(&mut input);
+            assert_eq!(input.pop().unwrap(), '\n');
+            selection = match input.parse::<usize>() {
+                Ok(i) if (1..=num_streams).contains((&i).into()) => Some(i),
+                _ => {
+                    println!("Invalid stream. Select again.\n");
+                    None
+                }
+            }
+        }
+        streams.select(selection.unwrap() - 1)
+    }
+
+    fn start_stream(stream: &StreamData) {
+        if stream.stream_ptr.is_null() {
             println!("No stream can start.");
             return;
         }
         assert_eq!(
-            unsafe { OPS.stream_start.unwrap()(stream_ptr) },
+            unsafe { OPS.stream_start.unwrap()(stream.stream_ptr) },
             ffi::CUBEB_OK
         );
-        println!("Stream {:p} started.", stream_ptr);
+        println!("Stream {:p} started.", stream.stream_ptr);
     }
 
-    fn stop_stream(stream_ptr: *mut ffi::cubeb_stream) {
-        if stream_ptr.is_null() {
+    fn stop_stream(stream: &StreamData) {
+        if stream.stream_ptr.is_null() {
             println!("No stream can stop.");
             return;
         }
         assert_eq!(
-            unsafe { OPS.stream_stop.unwrap()(stream_ptr) },
+            unsafe { OPS.stream_stop.unwrap()(stream.stream_ptr) },
             ffi::CUBEB_OK
         );
-        println!("Stream {:p} stopped.", stream_ptr);
+        println!("Stream {:p} stopped.", stream.stream_ptr);
     }
 
-    fn set_volume(stream_ptr: *mut ffi::cubeb_stream) {
-        if stream_ptr.is_null() {
+    fn set_volume(stream: &StreamData) {
+        if stream.stream_ptr.is_null() {
             println!("No stream can set volume.");
             return;
         }
         const VOL: f32 = 0.5;
         assert_eq!(
-            unsafe { OPS.stream_set_volume.unwrap()(stream_ptr, VOL) },
+            unsafe { OPS.stream_set_volume.unwrap()(stream.stream_ptr, VOL) },
             ffi::CUBEB_OK
         );
-        println!("Set stream {:p} volume to {}", stream_ptr, VOL);
+        println!("Set stream {:p} volume to {}", stream.stream_ptr, VOL);
     }
 
-    fn set_loopback(stream_ptr: *mut ffi::cubeb_stream, enable_loopback: &AtomicBool) {
-        if stream_ptr.is_null() {
+    fn set_loopback(stream: &StreamData) {
+        if stream.stream_ptr.is_null() {
             println!("No stream can set loopback.");
             return;
         }
-        let stm = unsafe { &mut *(stream_ptr as *mut AudioUnitStream) };
+        let stm = unsafe { &mut *(stream.stream_ptr as *mut AudioUnitStream) };
         if !stm.core_stream_data.has_input() || !stm.core_stream_data.has_output() {
             println!("Duplex stream needed to set loopback");
             return;
@@ -261,20 +383,20 @@ fn test_stream_tester() {
             }
         }
         let loopback = loopback.unwrap();
-        enable_loopback.store(loopback, Ordering::SeqCst);
+        stream.enable_loopback.store(loopback, Ordering::SeqCst);
         println!(
             "Loopback {} for stream {:p}",
             if loopback { "enabled" } else { "disabled" },
-            stream_ptr
+            stream.stream_ptr
         );
     }
 
-    fn set_input_mute(stream_ptr: *mut ffi::cubeb_stream) {
-        if stream_ptr.is_null() {
+    fn set_input_mute(stream: &StreamData) {
+        if stream.stream_ptr.is_null() {
             println!("No stream can set input mute.");
             return;
         }
-        let stm = unsafe { &mut *(stream_ptr as *mut AudioUnitStream) };
+        let stm = unsafe { &mut *(stream.stream_ptr as *mut AudioUnitStream) };
         if !stm.core_stream_data.has_input() {
             println!("Input stream needed to set loopback");
             return;
@@ -295,7 +417,7 @@ fn test_stream_tester() {
             }
         }
         let mute = mute.unwrap();
-        let res = unsafe { OPS.stream_set_input_mute.unwrap()(stream_ptr, mute.into()) };
+        let res = unsafe { OPS.stream_set_input_mute.unwrap()(stream.stream_ptr, mute.into()) };
         println!(
             "{} set stream {:p} input {}",
             if res == ffi::CUBEB_OK {
@@ -303,17 +425,17 @@ fn test_stream_tester() {
             } else {
                 "Failed to"
             },
-            stream_ptr,
+            stream.stream_ptr,
             if mute { "mute" } else { "unmute" }
         );
     }
 
-    fn set_input_processing(stream_ptr: *mut ffi::cubeb_stream) {
-        if stream_ptr.is_null() {
+    fn set_input_processing(stream: &StreamData) {
+        if stream.stream_ptr.is_null() {
             println!("No stream can set input processing.");
             return;
         }
-        let stm = unsafe { &mut *(stream_ptr as *mut AudioUnitStream) };
+        let stm = unsafe { &mut *(stream.stream_ptr as *mut AudioUnitStream) };
         if !stm.core_stream_data.using_voice_processing_unit() {
             println!("Duplex stream with voice processing needed to set input processing params");
             return;
@@ -338,6 +460,23 @@ fn test_stream_tester() {
                 params.set(InputProcessingParams::ECHO_CANCELLATION, true);
                 params.set(InputProcessingParams::NOISE_SUPPRESSION, true);
             }
+            let mut agc = u32::from(false);
+            let mut size: usize = mem::size_of::<u32>();
+            assert_eq!(
+                audio_unit_get_property(
+                    stm.core_stream_data.input_unit,
+                    kAUVoiceIOProperty_VoiceProcessingEnableAGC,
+                    kAudioUnitScope_Global,
+                    AU_IN_BUS,
+                    &mut agc,
+                    &mut size,
+                ),
+                NO_ERR
+            );
+            assert_eq!(size, mem::size_of::<u32>());
+            if agc == 1 {
+                params.set(InputProcessingParams::AUTOMATIC_GAIN_CONTROL, true);
+            }
         }
         let mut done = false;
         while !done {
@@ -367,8 +506,9 @@ fn test_stream_tester() {
                 _ => println!("Invalid action. Select again.\n"),
             }
         }
-        let res =
-            unsafe { OPS.stream_set_input_processing_params.unwrap()(stream_ptr, params.bits()) };
+        let res = unsafe {
+            OPS.stream_set_input_processing_params.unwrap()(stream.stream_ptr, params.bits())
+        };
         println!(
             "{} set stream {:p} input processing params to {:?}",
             if res == ffi::CUBEB_OK {
@@ -376,48 +516,62 @@ fn test_stream_tester() {
             } else {
                 "Failed to"
             },
-            stream_ptr,
+            stream.stream_ptr,
             params,
         );
     }
 
-    fn register_device_change_callback(stream_ptr: *mut ffi::cubeb_stream) {
+    fn register_device_change_callback(stream: &StreamData) {
         extern "C" fn callback(user_ptr: *mut c_void) {
             println!("user pointer @ {:p}", user_ptr);
             assert!(user_ptr.is_null());
         }
 
-        if stream_ptr.is_null() {
+        if stream.stream_ptr.is_null() {
             println!("No stream for registering the callback.");
             return;
         }
         assert_eq!(
             unsafe {
-                OPS.stream_register_device_changed_callback.unwrap()(stream_ptr, Some(callback))
+                OPS.stream_register_device_changed_callback.unwrap()(
+                    stream.stream_ptr,
+                    Some(callback),
+                )
             },
             ffi::CUBEB_OK
         );
-        println!("Stream {:p} now has a device change callback.", stream_ptr);
+        println!(
+            "Stream {:p} now has a device change callback.",
+            stream.stream_ptr
+        );
     }
 
-    fn destroy_stream(stream_ptr: &mut *mut ffi::cubeb_stream) {
-        if stream_ptr.is_null() {
+    fn destroy_stream(stream: &mut StreamData) {
+        if stream.stream_ptr.is_null() {
             println!("No need to destroy stream.");
             return;
         }
         unsafe {
-            OPS.stream_destroy.unwrap()(*stream_ptr);
+            OPS.stream_destroy.unwrap()((*stream).stream_ptr);
         }
-        println!("Stream {:p} destroyed.", *stream_ptr);
-        *stream_ptr = ptr::null_mut();
+        println!("Stream {:p} destroyed.", stream.stream_ptr);
+        stream.stream_ptr = ptr::null_mut();
     }
 
     fn create_stream(
-        stream_ptr: &mut *mut ffi::cubeb_stream,
         context_ptr: *mut ffi::cubeb,
-        enable_loopback: &AtomicBool,
+        streams: &mut StreamsData,
+        input_prefs: StreamPrefs,
+        output_prefs: StreamPrefs,
     ) {
-        if !stream_ptr.is_null() {
+        if streams.len() == 0 || !streams.current().stream_ptr.is_null() {
+            println!("Allocating stream {}.", streams.len() + 1);
+            streams.push(StreamData::new());
+            streams.select(streams.len() - 1);
+        }
+
+        let stream = streams.current_mut();
+        if !stream.stream_ptr.is_null() {
             println!("Stream has been created.");
             return;
         }
@@ -486,8 +640,8 @@ fn test_stream_tester() {
             }
         };
 
-        let mut input_params = get_dummy_stream_params(Scope::Input);
-        let mut output_params = get_dummy_stream_params(Scope::Output);
+        let mut input_params = get_dummy_stream_params(Scope::Input, input_prefs);
+        let mut output_params = get_dummy_stream_params(Scope::Output, output_prefs);
 
         let (input_device, input_stream_params) = if stream_type.contains(StreamType::INPUT) {
             (
@@ -519,7 +673,7 @@ fn test_stream_tester() {
             unsafe {
                 OPS.stream_init.unwrap()(
                     context_ptr,
-                    stream_ptr,
+                    &mut stream.stream_ptr,
                     stream_name.as_ptr(),
                     input_device as ffi::cubeb_devid,
                     input_stream_params,
@@ -528,13 +682,13 @@ fn test_stream_tester() {
                     4096, // latency
                     Some(data_callback),
                     Some(state_callback),
-                    enable_loopback as *const AtomicBool as *mut c_void, // user pointer
+                    &stream.enable_loopback as *const AtomicBool as *mut c_void, // user pointer
                 )
             },
             ffi::CUBEB_OK
         );
-        assert!(!stream_ptr.is_null());
-        println!("Stream {:p} created.", *stream_ptr);
+        assert!(!stream.stream_ptr.is_null());
+        println!("Stream {:p} created.", stream.stream_ptr);
 
         extern "C" fn state_callback(
             stream: *mut ffi::cubeb_stream,
@@ -592,14 +746,14 @@ fn test_stream_tester() {
             nframes
         }
 
-        fn get_dummy_stream_params(scope: Scope) -> ffi::cubeb_stream_params {
+        fn get_dummy_stream_params(scope: Scope, prefs: StreamPrefs) -> ffi::cubeb_stream_params {
             // The stream format for input and output must be same.
             const STREAM_FORMAT: u32 = ffi::CUBEB_SAMPLE_FLOAT32NE;
 
             // Make sure the parameters meet the requirements of AudioUnitContext::stream_init
             // (in the comments).
             let mut stream_params = ffi::cubeb_stream_params::default();
-            stream_params.prefs = ffi::CUBEB_STREAM_PREF_VOICE;
+            stream_params.prefs = prefs.bits();
             let (format, rate, channels, layout) = match scope {
                 Scope::Input => (STREAM_FORMAT, 48000, 1, ffi::CUBEB_LAYOUT_MONO),
                 Scope::Output => (STREAM_FORMAT, 44100, 2, ffi::CUBEB_LAYOUT_STEREO),
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/parallel.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/parallel.rs
index 16063d0011..be5edb1919 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/parallel.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/parallel.rs
@@ -537,14 +537,16 @@ fn test_set_buffer_frame_size_in_parallel_in_scope(scope: Scope) {
         units.push(test_get_default_audiounit(scope.clone()).unwrap());
         let unit_value = units.last().unwrap().get_inner() as usize;
         join_handles.push(thread::spawn(move || {
-            let status = audio_unit_set_property(
-                unit_value as AudioUnit,
-                kAudioDevicePropertyBufferFrameSize,
-                unit_scope,
-                unit_element,
-                &latency_frames,
-                mem::size_of::<u32>(),
-            );
+            let status = run_serially(|| {
+                audio_unit_set_property(
+                    unit_value as AudioUnit,
+                    kAudioDevicePropertyBufferFrameSize,
+                    unit_scope,
+                    unit_element,
+                    &latency_frames,
+                    mem::size_of::<u32>(),
+                )
+            });
             (latency_frames, status)
         }));
     }
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/tone.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/tone.rs
index 42cb9ee997..5f54286153 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/tone.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/tone.rs
@@ -1,6 +1,6 @@
 use super::utils::{test_get_default_device, test_ops_stream_operation, Scope};
 use super::*;
-use std::sync::atomic::{AtomicI64, Ordering};
+use std::sync::atomic::AtomicI64;
 
 #[test]
 fn test_dial_tone() {
@@ -202,7 +202,7 @@ fn test_dial_tone() {
         for data in buffer.iter_mut() {
             let t1 = (2.0 * PI * 350.0 * (closure.phase) as f32 / SAMPLE_FREQUENCY as f32).sin();
             let t2 = (2.0 * PI * 440.0 * (closure.phase) as f32 / SAMPLE_FREQUENCY as f32).sin();
-            *data = f32_to_i16_sample(0.5 * (t1 + t2));
+            *data = f32_to_i16_sample(0.45 * (t1 + t2));
             closure.phase += 1;
         }
 
diff --git a/third_party/rust/cubeb-coreaudio/src/backend/tests/utils.rs b/third_party/rust/cubeb-coreaudio/src/backend/tests/utils.rs
index ef07aeeeb4..6fb6d38fb3 100644
--- a/third_party/rust/cubeb-coreaudio/src/backend/tests/utils.rs
+++ b/third_party/rust/cubeb-coreaudio/src/backend/tests/utils.rs
@@ -25,6 +25,29 @@ pub extern "C" fn noop_data_callback(
     nframes
 }
 
+pub extern "C" fn draining_data_callback(
+    stream: *mut ffi::cubeb_stream,
+    _user_ptr: *mut c_void,
+    _input_buffer: *const c_void,
+    output_buffer: *mut c_void,
+    nframes: i64,
+) -> i64 {
+    assert!(!stream.is_null());
+
+    // Feed silence data to output buffer
+    if !output_buffer.is_null() {
+        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
+        let channels = stm.core_stream_data.output_stream_params.channels();
+        let samples = nframes as usize * channels as usize;
+        let sample_size = cubeb_sample_size(stm.core_stream_data.output_stream_params.format());
+        unsafe {
+            ptr::write_bytes(output_buffer, 0, samples * sample_size);
+        }
+    }
+
+    nframes - 1
+}
+
 #[derive(Clone, Debug, PartialEq)]
 pub enum Scope {
     Input,
@@ -47,31 +70,35 @@ pub enum PropertyScope {
 }
 
 pub fn test_get_default_device(scope: Scope) -> Option<AudioObjectID> {
-    let address = AudioObjectPropertyAddress {
-        mSelector: match scope {
-            Scope::Input => kAudioHardwarePropertyDefaultInputDevice,
-            Scope::Output => kAudioHardwarePropertyDefaultOutputDevice,
-        },
-        mScope: kAudioObjectPropertyScopeGlobal,
-        mElement: kAudioObjectPropertyElementMaster,
-    };
+    debug_assert_not_running_serially();
+    run_serially_forward_panics(|| {
+        let address = AudioObjectPropertyAddress {
+            mSelector: match scope {
+                Scope::Input => kAudioHardwarePropertyDefaultInputDevice,
+                Scope::Output => kAudioHardwarePropertyDefaultOutputDevice,
+            },
+            mScope: kAudioObjectPropertyScopeGlobal,
+            mElement: kAudioObjectPropertyElementMaster,
+        };
 
-    let mut devid: AudioObjectID = kAudioObjectUnknown;
-    let mut size = mem::size_of::<AudioObjectID>();
-    let status = unsafe {
-        AudioObjectGetPropertyData(
-            kAudioObjectSystemObject,
-            &address,
-            0,
-            ptr::null(),
-            &mut size as *mut usize as *mut UInt32,
-            &mut devid as *mut AudioObjectID as *mut c_void,
-        )
-    };
-    if status != NO_ERR || devid == kAudioObjectUnknown {
-        return None;
-    }
-    Some(devid)
+        let mut devid: AudioObjectID = kAudioObjectUnknown;
+        let mut size = mem::size_of::<AudioObjectID>();
+        let status = unsafe {
+            AudioObjectGetPropertyData(
+                kAudioObjectSystemObject,
+                &address,
+                0,
+                ptr::null(),
+                &mut size as *mut usize as *mut UInt32,
+                &mut devid as *mut AudioObjectID as *mut c_void,
+            )
+        };
+
+        if status != NO_ERR || devid == kAudioObjectUnknown {
+            return None;
+        }
+        Some(devid)
+    })
 }
 
 // TODO: Create a GetProperty trait and add a default implementation for it, then implement it
@@ -99,16 +126,17 @@ impl TestAudioUnit {
 
 impl Drop for TestAudioUnit {
     fn drop(&mut self) {
-        unsafe {
+        run_serially_forward_panics(|| unsafe {
             AudioUnitUninitialize(self.0);
             AudioComponentInstanceDispose(self.0);
-        }
+        });
     }
 }
 
 // TODO: 1. Return Result with custom errors.
 //       2. Allow to create a in-out unit.
 pub fn test_get_default_audiounit(scope: Scope) -> Option<TestAudioUnit> {
+    debug_assert_not_running_serially();
     let device = test_get_default_device(scope.clone());
     let unit = test_create_audiounit(ComponentSubType::HALOutput);
     if device.is_none() || unit.is_none() {
@@ -133,7 +161,7 @@ pub fn test_get_default_audiounit(scope: Scope) -> Option<TestAudioUnit> {
         }
     }
 
-    let status = unsafe {
+    let status = run_serially(|| unsafe {
         AudioUnitSetProperty(
             unit.get_inner(),
             kAudioOutputUnitProperty_CurrentDevice,
@@ -142,7 +170,7 @@ pub fn test_get_default_audiounit(scope: Scope) -> Option<TestAudioUnit> {
             &device as *const AudioObjectID as *const c_void,
             mem::size_of::<AudioObjectID>() as u32,
         )
-    };
+    });
     if status == NO_ERR {
         Some(unit)
     } else {
@@ -159,6 +187,7 @@ pub enum ComponentSubType {
 // Surprisingly the AudioUnit can be created even when there is no any device on the platform,
 // no matter its subtype is HALOutput or DefaultOutput.
 pub fn test_create_audiounit(unit_type: ComponentSubType) -> Option<TestAudioUnit> {
+    debug_assert_not_running_serially();
     let desc = AudioComponentDescription {
         componentType: kAudioUnitType_Output,
         componentSubType: match unit_type {
@@ -169,12 +198,12 @@ pub fn test_create_audiounit(unit_type: ComponentSubType) -> Option<TestAudioUni
         componentFlags: 0,
         componentFlagsMask: 0,
     };
-    let comp = unsafe { AudioComponentFindNext(ptr::null_mut(), &desc) };
+    let comp = run_serially(|| unsafe { AudioComponentFindNext(ptr::null_mut(), &desc) });
     if comp.is_null() {
         return None;
     }
     let mut unit: AudioUnit = ptr::null_mut();
-    let status = unsafe { AudioComponentInstanceNew(comp, &mut unit) };
+    let status = run_serially(|| unsafe { AudioComponentInstanceNew(comp, &mut unit) });
     // TODO: Is unit possible to be null when no error returns ?
     if status != NO_ERR || unit.is_null() {
         None
@@ -188,13 +217,14 @@ fn test_enable_audiounit_in_scope(
     scope: Scope,
     enable: bool,
 ) -> std::result::Result<(), OSStatus> {
+    debug_assert_not_running_serially();
     assert!(!unit.is_null());
     let (scope, element) = match scope {
         Scope::Input => (kAudioUnitScope_Input, AU_IN_BUS),
         Scope::Output => (kAudioUnitScope_Output, AU_OUT_BUS),
     };
     let on_off: u32 = if enable { 1 } else { 0 };
-    let status = unsafe {
+    let status = run_serially(|| unsafe {
         AudioUnitSetProperty(
             unit,
             kAudioOutputUnitProperty_EnableIO,
@@ -203,7 +233,7 @@ fn test_enable_audiounit_in_scope(
             &on_off as *const u32 as *const c_void,
             mem::size_of::<u32>() as u32,
         )
-    };
+    });
     if status == NO_ERR {
         Ok(())
     } else {
@@ -216,72 +246,80 @@ pub enum DeviceFilter {
     IncludeAll,
 }
 pub fn test_get_all_devices(filter: DeviceFilter) -> Vec<AudioObjectID> {
-    let mut devices = Vec::new();
-    let address = AudioObjectPropertyAddress {
-        mSelector: kAudioHardwarePropertyDevices,
-        mScope: kAudioObjectPropertyScopeGlobal,
-        mElement: kAudioObjectPropertyElementMaster,
-    };
-    let mut size: usize = 0;
-    let status = unsafe {
-        AudioObjectGetPropertyDataSize(
-            kAudioObjectSystemObject,
-            &address,
-            0,
-            ptr::null(),
-            &mut size as *mut usize as *mut u32,
-        )
-    };
-    // size will be 0 if there is no device at all.
-    if status != NO_ERR || size == 0 {
-        return devices;
-    }
-    assert_eq!(size % mem::size_of::<AudioObjectID>(), 0);
-    let elements = size / mem::size_of::<AudioObjectID>();
-    devices.resize(elements, kAudioObjectUnknown);
-    let status = unsafe {
-        AudioObjectGetPropertyData(
-            kAudioObjectSystemObject,
-            &address,
-            0,
-            ptr::null(),
-            &mut size as *mut usize as *mut u32,
-            devices.as_mut_ptr() as *mut c_void,
-        )
-    };
-    if status != NO_ERR {
-        devices.clear();
-        return devices;
-    }
-    for device in devices.iter() {
-        assert_ne!(*device, kAudioObjectUnknown);
-    }
+    debug_assert_not_running_serially();
+    // To avoid races, the devices getter and the device name filtering have
+    // to run in the same serial task. If not, a device may exist when the
+    // getter runs but not when getting its uid.
+    run_serially_forward_panics(|| {
+        let mut devices = Vec::new();
+        let address = AudioObjectPropertyAddress {
+            mSelector: kAudioHardwarePropertyDevices,
+            mScope: kAudioObjectPropertyScopeGlobal,
+            mElement: kAudioObjectPropertyElementMaster,
+        };
+        let mut size: usize = 0;
+        let status = unsafe {
+            AudioObjectGetPropertyDataSize(
+                kAudioObjectSystemObject,
+                &address,
+                0,
+                ptr::null(),
+                &mut size as *mut usize as *mut u32,
+            )
+        };
+        // size will be 0 if there is no device at all.
+        if status != NO_ERR || size == 0 {
+            return devices;
+        }
+        assert_eq!(size % mem::size_of::<AudioObjectID>(), 0);
+        let elements = size / mem::size_of::<AudioObjectID>();
+        devices.resize(elements, kAudioObjectUnknown);
+        let status = unsafe {
+            AudioObjectGetPropertyData(
+                kAudioObjectSystemObject,
+                &address,
+                0,
+                ptr::null(),
+                &mut size as *mut usize as *mut u32,
+                devices.as_mut_ptr() as *mut c_void,
+            )
+        };
+        if status != NO_ERR {
+            devices.clear();
+            return devices;
+        }
+        for device in devices.iter() {
+            assert_ne!(*device, kAudioObjectUnknown);
+        }
 
-    match filter {
-        DeviceFilter::ExcludeCubebAggregateAndVPIO => {
-            devices.retain(|&device| {
-                if let Ok(uid) = get_device_global_uid(device) {
-                    let uid = uid.into_string();
-                    !uid.contains(PRIVATE_AGGREGATE_DEVICE_NAME)
-                        && !uid.contains(VOICEPROCESSING_AGGREGATE_DEVICE_NAME)
-                } else {
-                    true
-                }
-            });
+        match filter {
+            DeviceFilter::ExcludeCubebAggregateAndVPIO => {
+                devices.retain(|&device| {
+                    if let Ok(uid) = get_device_global_uid(device) {
+                        let uid = uid.into_string();
+                        !uid.contains(PRIVATE_AGGREGATE_DEVICE_NAME)
+                            && !uid.contains(VOICEPROCESSING_AGGREGATE_DEVICE_NAME)
+                    } else {
+                        true
+                    }
+                });
+            }
+            _ => {}
         }
-        _ => {}
-    }
 
-    devices
+        devices
+    })
 }
 
 pub fn test_get_devices_in_scope(scope: Scope) -> Vec<AudioObjectID> {
+    debug_assert_not_running_serially();
     let mut devices = test_get_all_devices(DeviceFilter::ExcludeCubebAggregateAndVPIO);
     devices.retain(|device| test_device_in_scope(*device, scope.clone()));
     devices
 }
 
 pub fn get_devices_info_in_scope(scope: Scope) -> Vec<TestDeviceInfo> {
+    debug_assert_not_running_serially();
     fn print_info(info: &TestDeviceInfo) {
         println!("{:>4}: {}\n\tuid: {}", info.id, info.label, info.uid);
     }
@@ -337,8 +375,9 @@ pub fn test_device_channels_in_scope(
     id: AudioObjectID,
     scope: Scope,
 ) -> std::result::Result<u32, OSStatus> {
+    debug_assert_not_running_serially();
     let address = AudioObjectPropertyAddress {
-        mSelector: kAudioDevicePropertyStreamConfiguration,
+        mSelector: kAudioDevicePropertyStreams,
         mScope: match scope {
             Scope::Input => kAudioDevicePropertyScopeInput,
             Scope::Output => kAudioDevicePropertyScopeOutput,
@@ -346,7 +385,7 @@ pub fn test_device_channels_in_scope(
         mElement: kAudioObjectPropertyElementMaster,
     };
     let mut size: usize = 0;
-    let status = unsafe {
+    let status = run_serially(|| unsafe {
         AudioObjectGetPropertyDataSize(
             id,
             &address,
@@ -354,49 +393,90 @@ pub fn test_device_channels_in_scope(
             ptr::null(),
             &mut size as *mut usize as *mut u32,
         )
-    };
+    });
     if status != NO_ERR {
         return Err(status);
     }
     if size == 0 {
         return Ok(0);
     }
-    let byte_len = size / mem::size_of::<u8>();
-    let mut bytes = vec![0u8; byte_len];
-    let status = unsafe {
+    let mut stream_list = vec![0, (size / mem::size_of::<AudioObjectID>()) as u32];
+    let status = run_serially(|| unsafe {
         AudioObjectGetPropertyData(
             id,
             &address,
             0,
             ptr::null(),
             &mut size as *mut usize as *mut u32,
-            bytes.as_mut_ptr() as *mut c_void,
+            stream_list.as_mut_ptr() as *mut c_void,
         )
-    };
+    });
     if status != NO_ERR {
         return Err(status);
     }
-    let buf_list = unsafe { &*(bytes.as_mut_ptr() as *mut AudioBufferList) };
-    let buf_len = buf_list.mNumberBuffers as usize;
-    if buf_len == 0 {
-        return Ok(0);
-    }
-    let buf_ptr = buf_list.mBuffers.as_ptr() as *const AudioBuffer;
-    let buffers = unsafe { slice::from_raw_parts(buf_ptr, buf_len) };
-    let mut channels: u32 = 0;
-    for buffer in buffers {
-        channels += buffer.mNumberChannels;
-    }
+    let channels = stream_list
+        .iter()
+        .filter(|s: &&AudioObjectID| {
+            if scope != Scope::Input {
+                return true;
+            }
+            let address = AudioObjectPropertyAddress {
+                mSelector: kAudioStreamPropertyTerminalType,
+                mScope: kAudioObjectPropertyScopeGlobal,
+                mElement: kAudioObjectPropertyElementMaster,
+            };
+            let mut ttype: u32 = 0;
+            let status = unsafe {
+                AudioObjectGetPropertyData(
+                    **s,
+                    &address,
+                    0,
+                    ptr::null(),
+                    &mut mem::size_of::<u32>() as *mut usize as *mut u32,
+                    &mut ttype as *mut u32 as *mut c_void,
+                )
+            };
+            if status != NO_ERR {
+                return false;
+            }
+            ttype == kAudioStreamTerminalTypeMicrophone
+                || (INPUT_MICROPHONE..OUTPUT_UNDEFINED).contains(&ttype)
+        })
+        .map(|s: &AudioObjectID| {
+            let address = AudioObjectPropertyAddress {
+                mSelector: kAudioStreamPropertyVirtualFormat,
+                mScope: kAudioObjectPropertyScopeGlobal,
+                mElement: kAudioObjectPropertyElementMaster,
+            };
+            let mut format = AudioStreamBasicDescription::default();
+            let status = unsafe {
+                AudioObjectGetPropertyData(
+                    *s,
+                    &address,
+                    0,
+                    ptr::null(),
+                    &mut mem::size_of::<AudioStreamBasicDescription>() as *mut usize as *mut u32,
+                    &mut format as *mut AudioStreamBasicDescription as *mut c_void,
+                )
+            };
+            if status != NO_ERR {
+                return 0;
+            }
+            format.mChannelsPerFrame
+        })
+        .sum();
     Ok(channels)
 }
 
 pub fn test_device_in_scope(id: AudioObjectID, scope: Scope) -> bool {
+    debug_assert_not_running_serially();
     let channels = test_device_channels_in_scope(id, scope);
     channels.is_ok() && channels.unwrap() > 0
 }
 
 pub fn test_get_all_onwed_devices(id: AudioDeviceID) -> Vec<AudioObjectID> {
     assert_ne!(id, kAudioObjectUnknown);
+    debug_assert_running_serially();
 
     let address = AudioObjectPropertyAddress {
         mSelector: kAudioObjectPropertyOwnedObjects,
@@ -409,45 +489,46 @@ pub fn test_get_all_onwed_devices(id: AudioDeviceID) -> Vec<AudioObjectID> {
     let qualifier_data = &class_id;
     let mut size: usize = 0;
 
-    unsafe {
-        assert_eq!(
+    assert_eq!(
+        unsafe {
             AudioObjectGetPropertyDataSize(
                 id,
                 &address,
                 qualifier_data_size as u32,
                 qualifier_data as *const u32 as *const c_void,
-                &mut size as *mut usize as *mut u32
-            ),
-            NO_ERR
-        );
-    }
+                &mut size as *mut usize as *mut u32,
+            )
+        },
+        NO_ERR
+    );
     assert_ne!(size, 0);
 
     let elements = size / mem::size_of::<AudioObjectID>();
     let mut devices: Vec<AudioObjectID> = allocate_array(elements);
 
-    unsafe {
-        assert_eq!(
+    assert_eq!(
+        unsafe {
             AudioObjectGetPropertyData(
                 id,
                 &address,
                 qualifier_data_size as u32,
                 qualifier_data as *const u32 as *const c_void,
                 &mut size as *mut usize as *mut u32,
-                devices.as_mut_ptr() as *mut c_void
-            ),
-            NO_ERR
-        );
-    }
+                devices.as_mut_ptr() as *mut c_void,
+            )
+        },
+        NO_ERR
+    );
 
     devices
 }
 
 pub fn test_get_master_device(id: AudioObjectID) -> String {
     assert_ne!(id, kAudioObjectUnknown);
+    debug_assert_running_serially();
 
     let address = AudioObjectPropertyAddress {
-        mSelector: kAudioAggregateDevicePropertyMasterSubDevice,
+        mSelector: kAudioAggregateDevicePropertyMainSubDevice,
         mScope: kAudioObjectPropertyScopeGlobal,
         mElement: kAudioObjectPropertyElementMaster,
     };
@@ -465,6 +546,7 @@ pub fn test_get_master_device(id: AudioObjectID) -> String {
 }
 
 pub fn test_get_drift_compensations(id: AudioObjectID) -> std::result::Result<u32, OSStatus> {
+    debug_assert_running_serially();
     let address = AudioObjectPropertyAddress {
         mSelector: kAudioSubDevicePropertyDriftCompensation,
         mScope: kAudioObjectPropertyScopeGlobal,
@@ -491,6 +573,7 @@ pub fn test_get_drift_compensations(id: AudioObjectID) -> std::result::Result<u3
 
 pub fn test_audiounit_scope_is_enabled(unit: AudioUnit, scope: Scope) -> bool {
     assert!(!unit.is_null());
+    debug_assert_not_running_serially();
     let mut has_io: UInt32 = 0;
     let (scope, element) = match scope {
         Scope::Input => (kAudioUnitScope_Input, AU_IN_BUS),
@@ -498,14 +581,14 @@ pub fn test_audiounit_scope_is_enabled(unit: AudioUnit, scope: Scope) -> bool {
     };
     let mut size = mem::size_of::<UInt32>();
     assert_eq!(
-        audio_unit_get_property(
+        run_serially(|| audio_unit_get_property(
             unit,
             kAudioOutputUnitProperty_HasIO,
             scope,
             element,
             &mut has_io,
             &mut size
-        ),
+        )),
         NO_ERR
     );
     has_io != 0
@@ -516,6 +599,7 @@ pub fn test_audiounit_get_buffer_frame_size(
     scope: Scope,
     prop_scope: PropertyScope,
 ) -> std::result::Result<u32, OSStatus> {
+    debug_assert_not_running_serially();
     let element = match scope {
         Scope::Input => AU_IN_BUS,
         Scope::Output => AU_OUT_BUS,
@@ -526,7 +610,7 @@ pub fn test_audiounit_get_buffer_frame_size(
     };
     let mut buffer_frames: u32 = 0;
     let mut size = mem::size_of::<u32>();
-    let status = unsafe {
+    let status = run_serially(|| unsafe {
         AudioUnitGetProperty(
             unit,
             kAudioDevicePropertyBufferFrameSize,
@@ -535,7 +619,7 @@ pub fn test_audiounit_get_buffer_frame_size(
             &mut buffer_frames as *mut u32 as *mut c_void,
             &mut size as *mut usize as *mut u32,
         )
-    };
+    });
     if status == NO_ERR {
         Ok(buffer_frames)
     } else {
@@ -554,6 +638,7 @@ pub fn test_set_default_device(
     device: AudioObjectID,
     scope: Scope,
 ) -> std::result::Result<AudioObjectID, OSStatus> {
+    debug_assert_not_running_serially();
     assert!(test_device_in_scope(device, scope.clone()));
     let default = test_get_default_device(scope.clone()).unwrap();
     if default == device {
@@ -570,7 +655,7 @@ pub fn test_set_default_device(
         mElement: kAudioObjectPropertyElementMaster,
     };
     let size = mem::size_of::<AudioObjectID>();
-    let status = unsafe {
+    let status = run_serially(|| unsafe {
         AudioObjectSetPropertyData(
             kAudioObjectSystemObject,
             &address,
@@ -579,7 +664,7 @@ pub fn test_set_default_device(
             size as u32,
             &device as *const AudioObjectID as *const c_void,
         )
-    };
+    });
     let new_default = test_get_default_device(scope.clone()).unwrap();
     if new_default == default {
         Err(-1)
@@ -644,6 +729,7 @@ pub fn test_create_device_change_listener<F>(scope: Scope, listener: F) -> TestP
 where
     F: Fn(&[AudioObjectPropertyAddress]) -> OSStatus,
 {
+    debug_assert_running_serially();
     let address = AudioObjectPropertyAddress {
         mSelector: match scope {
             Scope::Input => kAudioHardwarePropertyDefaultInputDevice,
@@ -652,6 +738,7 @@ where
         mScope: kAudioObjectPropertyScopeGlobal,
         mElement: kAudioObjectPropertyElementMaster,
     };
+
     TestPropertyListener::new(kAudioObjectSystemObject, address, listener)
 }
 
@@ -677,6 +764,7 @@ where
     }
 
     pub fn start(&self) -> std::result::Result<(), OSStatus> {
+        debug_assert_running_serially();
         let status = unsafe {
             AudioObjectAddPropertyListener(
                 self.device,
@@ -693,6 +781,7 @@ where
     }
 
     pub fn stop(&self) -> std::result::Result<(), OSStatus> {
+        debug_assert_running_serially();
         let status = unsafe {
             AudioObjectRemovePropertyListener(
                 self.device,
@@ -726,7 +815,7 @@ where
     F: Fn(&[AudioObjectPropertyAddress]) -> OSStatus,
 {
     fn drop(&mut self) {
-        self.stop();
+        run_serially(|| self.stop());
     }
 }
 
@@ -767,6 +856,7 @@ impl TestDevicePlugger {
     }
 
     fn destroy_aggregate_device(&mut self) -> std::result::Result<(), OSStatus> {
+        debug_assert_not_running_serially();
         assert_ne!(self.plugin_id, kAudioObjectUnknown);
         assert_ne!(self.device_id, kAudioObjectUnknown);
 
@@ -777,7 +867,7 @@ impl TestDevicePlugger {
         };
 
         let mut size: usize = 0;
-        let status = unsafe {
+        let status = run_serially(|| unsafe {
             AudioObjectGetPropertyDataSize(
                 self.plugin_id,
                 &address,
@@ -785,13 +875,13 @@ impl TestDevicePlugger {
                 ptr::null(),
                 &mut size as *mut usize as *mut u32,
             )
-        };
+        });
         if status != NO_ERR {
             return Err(status);
         }
         assert_ne!(size, 0);
 
-        let status = unsafe {
+        let status = run_serially(|| unsafe {
             // This call can simulate removing a device.
             AudioObjectGetPropertyData(
                 self.plugin_id,
@@ -801,7 +891,7 @@ impl TestDevicePlugger {
                 &mut size as *mut usize as *mut u32,
                 &mut self.device_id as *mut AudioDeviceID as *mut c_void,
             )
-        };
+        });
         if status == NO_ERR {
             self.device_id = kAudioObjectUnknown;
             Ok(())
@@ -811,6 +901,7 @@ impl TestDevicePlugger {
     }
 
     fn create_aggregate_device(&self) -> std::result::Result<AudioObjectID, OSStatus> {
+        debug_assert_not_running_serially();
         use std::time::{SystemTime, UNIX_EPOCH};
 
         const TEST_AGGREGATE_DEVICE_NAME: &str = "TestAggregateDevice";
@@ -830,7 +921,7 @@ impl TestDevicePlugger {
         };
 
         let mut size: usize = 0;
-        let status = unsafe {
+        let status = run_serially(|| unsafe {
             AudioObjectGetPropertyDataSize(
                 self.plugin_id,
                 &address,
@@ -838,7 +929,7 @@ impl TestDevicePlugger {
                 ptr::null(),
                 &mut size as *mut usize as *mut u32,
             )
-        };
+        });
         if status != NO_ERR {
             return Err(status);
         }
@@ -915,14 +1006,18 @@ impl TestDevicePlugger {
             CFRelease(sub_devices as *const c_void);
 
             // This call can simulate adding a device.
-            let status = AudioObjectGetPropertyData(
-                self.plugin_id,
-                &address,
-                mem::size_of_val(&device_dict) as u32,
-                &device_dict as *const CFMutableDictionaryRef as *const c_void,
-                &mut size as *mut usize as *mut u32,
-                &mut device_id as *mut AudioDeviceID as *mut c_void,
-            );
+            let status = {
+                run_serially(|| {
+                    AudioObjectGetPropertyData(
+                        self.plugin_id,
+                        &address,
+                        mem::size_of_val(&device_dict) as u32,
+                        &device_dict as *const CFMutableDictionaryRef as *const c_void,
+                        &mut size as *mut usize as *mut u32,
+                        &mut device_id as *mut AudioDeviceID as *mut c_void,
+                    )
+                })
+            };
             CFRelease(device_dict as *const c_void);
             status
         };
@@ -935,6 +1030,7 @@ impl TestDevicePlugger {
     }
 
     fn get_system_plugin_id() -> std::result::Result<AudioObjectID, OSStatus> {
+        debug_assert_not_running_serially();
         let address = AudioObjectPropertyAddress {
             mSelector: kAudioHardwarePropertyPlugInForBundleID,
             mScope: kAudioObjectPropertyScopeGlobal,
@@ -942,7 +1038,7 @@ impl TestDevicePlugger {
         };
 
         let mut size: usize = 0;
-        let status = unsafe {
+        let status = run_serially(|| unsafe {
             AudioObjectGetPropertyDataSize(
                 kAudioObjectSystemObject,
                 &address,
@@ -950,7 +1046,7 @@ impl TestDevicePlugger {
                 ptr::null(),
                 &mut size as *mut usize as *mut u32,
             )
-        };
+        });
         if status != NO_ERR {
             return Err(status);
         }
@@ -967,14 +1063,16 @@ impl TestDevicePlugger {
         assert_eq!(size, mem::size_of_val(&translation_value));
 
         let status = unsafe {
-            let status = AudioObjectGetPropertyData(
-                kAudioObjectSystemObject,
-                &address,
-                0,
-                ptr::null(),
-                &mut size as *mut usize as *mut u32,
-                &mut translation_value as *mut AudioValueTranslation as *mut c_void,
-            );
+            let status = run_serially(|| {
+                AudioObjectGetPropertyData(
+                    kAudioObjectSystemObject,
+                    &address,
+                    0,
+                    ptr::null(),
+                    &mut size as *mut usize as *mut u32,
+                    &mut translation_value as *mut AudioValueTranslation as *mut c_void,
+                )
+            });
             CFRelease(in_bundle_ref as *const c_void);
             status
         };
@@ -991,10 +1089,11 @@ impl TestDevicePlugger {
     //       them into the array, if the device is an aggregate device. See the code in
     //       AggregateDevice::get_sub_devices and audiounit_set_aggregate_sub_device_list.
     fn get_sub_devices(scope: Scope) -> Option<CFArrayRef> {
+        debug_assert_not_running_serially();
         let device = test_get_default_device(scope);
         device?;
         let device = device.unwrap();
-        let uid = get_device_global_uid(device);
+        let uid = run_serially(|| get_device_global_uid(device));
         if uid.is_err() {
             return None;
         }
@@ -1033,6 +1132,7 @@ pub fn test_ops_context_operation<F>(name: &'static str, operation: F)
 where
     F: FnOnce(*mut ffi::cubeb),
 {
+    debug_assert_not_running_serially();
     let name_c_string = CString::new(name).expect("Failed to create context name");
     let mut context = ptr::null_mut::<ffi::cubeb>();
     assert_eq!(
@@ -1047,8 +1147,9 @@ where
 // The in-out stream initializeed with different device will create an aggregate_device and
 // result in firing device-collection-changed callbacks. Run in-out streams with tests
 // capturing device-collection-changed callbacks may cause troubles.
-pub fn test_ops_stream_operation<F>(
+pub fn test_ops_stream_operation_on_context<F>(
     name: &'static str,
+    context_ptr: *mut ffi::cubeb,
     input_device: ffi::cubeb_devid,
     input_stream_params: *mut ffi::cubeb_stream_params,
     output_device: ffi::cubeb_devid,
@@ -1061,43 +1162,72 @@ pub fn test_ops_stream_operation<F>(
 ) where
     F: FnOnce(*mut ffi::cubeb_stream),
 {
-    test_ops_context_operation("context: stream operation", |context_ptr| {
-        // Do nothing if there is no input/output device to perform input/output tests.
-        if !input_stream_params.is_null() && test_get_default_device(Scope::Input).is_none() {
-            println!("No input device to perform input tests for \"{}\".", name);
-            return;
-        }
+    // Do nothing if there is no input/output device to perform input/output tests.
+    if !input_stream_params.is_null() && test_get_default_device(Scope::Input).is_none() {
+        println!("No input device to perform input tests for \"{}\".", name);
+        return;
+    }
 
-        if !output_stream_params.is_null() && test_get_default_device(Scope::Output).is_none() {
-            println!("No output device to perform output tests for \"{}\".", name);
-            return;
-        }
+    if !output_stream_params.is_null() && test_get_default_device(Scope::Output).is_none() {
+        println!("No output device to perform output tests for \"{}\".", name);
+        return;
+    }
 
-        let mut stream: *mut ffi::cubeb_stream = ptr::null_mut();
-        let stream_name = CString::new(name).expect("Failed to create stream name");
-        assert_eq!(
-            unsafe {
-                OPS.stream_init.unwrap()(
-                    context_ptr,
-                    &mut stream,
-                    stream_name.as_ptr(),
-                    input_device,
-                    input_stream_params,
-                    output_device,
-                    output_stream_params,
-                    latency_frames,
-                    data_callback,
-                    state_callback,
-                    user_ptr,
-                )
-            },
-            ffi::CUBEB_OK
-        );
-        assert!(!stream.is_null());
-        operation(stream);
+    let mut stream: *mut ffi::cubeb_stream = ptr::null_mut();
+    let stream_name = CString::new(name).expect("Failed to create stream name");
+    assert_eq!(
         unsafe {
-            OPS.stream_destroy.unwrap()(stream);
-        }
+            OPS.stream_init.unwrap()(
+                context_ptr,
+                &mut stream,
+                stream_name.as_ptr(),
+                input_device,
+                input_stream_params,
+                output_device,
+                output_stream_params,
+                latency_frames,
+                data_callback,
+                state_callback,
+                user_ptr,
+            )
+        },
+        ffi::CUBEB_OK
+    );
+    assert!(!stream.is_null());
+    operation(stream);
+    unsafe {
+        OPS.stream_destroy.unwrap()(stream);
+    }
+}
+
+pub fn test_ops_stream_operation<F>(
+    name: &'static str,
+    input_device: ffi::cubeb_devid,
+    input_stream_params: *mut ffi::cubeb_stream_params,
+    output_device: ffi::cubeb_devid,
+    output_stream_params: *mut ffi::cubeb_stream_params,
+    latency_frames: u32,
+    data_callback: ffi::cubeb_data_callback,
+    state_callback: ffi::cubeb_state_callback,
+    user_ptr: *mut c_void,
+    operation: F,
+) where
+    F: FnOnce(*mut ffi::cubeb_stream),
+{
+    test_ops_context_operation("context: stream operation", |context_ptr| {
+        test_ops_stream_operation_on_context(
+            name,
+            context_ptr,
+            input_device,
+            input_stream_params,
+            output_device,
+            output_stream_params,
+            latency_frames,
+            data_callback,
+            state_callback,
+            user_ptr,
+            operation,
+        );
     });
 }
 
@@ -1136,7 +1266,7 @@ fn test_get_raw_stream<F>(
         user_ptr,
         data_callback,
         state_callback,
-        global_latency_frames.unwrap(),
+        global_latency_frames,
     );
     stream.core_stream_data = CoreStreamData::new(&stream, None, None);
 
@@ -1154,6 +1284,7 @@ pub fn test_get_stream_with_default_data_callback_by_type<F>(
 ) where
     F: FnOnce(&mut AudioUnitStream),
 {
+    debug_assert_not_running_serially();
     let mut input_params = get_dummy_stream_params(Scope::Input);
     let mut output_params = get_dummy_stream_params(Scope::Output);