Rework sync callback mechanism

ChunMinChang · ChunMinChang · commit 579b75af21c0 · 2026-02-10T14:36:51.000-08:00
diff --git a/src/backend/aggregate_device.rs b/src/backend/aggregate_device.rs
@@ -152,9 +152,9 @@ impl AggregateDevice {
         debug_assert_running_serially();
         let waiting_time = Duration::new(5, 0);
 
-        let condvar_pair = Arc::new((Mutex::new(()), Condvar::new()));
-        let mut cloned_condvar_pair = condvar_pair.clone();
-        let data_ptr = &mut cloned_condvar_pair as *mut Arc<(Mutex<()>, Condvar)>;
+        let condvar_pair = Box::new((Mutex::new(()), Condvar::new()));
+        let data_ptr = Box::into_raw(condvar_pair);
+        sync_callback_registry_register(data_ptr as usize);
 
         let address = get_property_address(
             Property::HardwareDevices,
@@ -177,12 +177,14 @@ impl AggregateDevice {
                 data_ptr as *mut c_void,
             );
             assert_eq!(status, NO_ERR);
+            sync_callback_registry_unregister(data_ptr as usize);
+            unsafe { drop(Box::from_raw(data_ptr)) };
         });
 
         let device = Self::create_blank_device(plugin_id)?;
 
         // Wait until the aggregate is created.
-        let (lock, cvar) = &*condvar_pair;
+        let (lock, cvar) = unsafe { &*data_ptr };
         let guard = lock.lock().unwrap();
         let (_guard, timeout_res) = cvar
             .wait_timeout_while(guard, waiting_time, |()| !get_devices().contains(&device))
@@ -202,10 +204,12 @@ impl AggregateDevice {
             data: *mut c_void,
         ) -> OSStatus {
             assert_eq!(id, kAudioObjectSystemObject);
-            let pair = unsafe { &mut *(data as *mut Arc<(Mutex<()>, Condvar)>) };
-            let (lock, cvar) = &**pair;
-            let _guard = lock.lock().unwrap();
-            cvar.notify_one();
+            with_sync_callback_ptr(data, || {
+                let pair = unsafe { &*(data as *const (Mutex<()>, Condvar)) };
+                let (lock, cvar) = pair;
+                let _guard = lock.lock().unwrap();
+                cvar.notify_one();
+            });
             NO_ERR
         }
 
@@ -314,9 +318,9 @@ impl AggregateDevice {
 
         let waiting_time = Duration::new(5, 0);
 
-        let condvar_pair = Arc::new((Mutex::new(AudioObjectID::default()), Condvar::new()));
-        let mut cloned_condvar_pair = condvar_pair.clone();
-        let data_ptr = &mut cloned_condvar_pair as *mut Arc<(Mutex<AudioObjectID>, Condvar)>;
+        let condvar_pair = Box::new((Mutex::new(AudioObjectID::default()), Condvar::new()));
+        let data_ptr = Box::into_raw(condvar_pair);
+        sync_callback_registry_register(data_ptr as usize);
 
         let status = audio_object_add_property_listener(
             device_id,
@@ -325,22 +329,27 @@ impl AggregateDevice {
             data_ptr as *mut c_void,
         );
         if status != NO_ERR {
+            sync_callback_registry_unregister(data_ptr as usize);
+            unsafe { drop(Box::from_raw(data_ptr)) };
             return Err(Error::from(status));
         }
 
-        let remove_listener = || -> OSStatus {
-            audio_object_remove_property_listener(
+        let cleanup = || {
+            let status = audio_object_remove_property_listener(
                 device_id,
                 &address,
                 devices_changed_callback,
                 data_ptr as *mut c_void,
-            )
+            );
+            sync_callback_registry_unregister(data_ptr as usize);
+            unsafe { drop(Box::from_raw(data_ptr)) };
+            status
         };
 
         Self::set_sub_devices(device_id, input_id, output_id)?;
 
         // Wait until the sub devices are added.
-        let (lock, cvar) = &*condvar_pair;
+        let (lock, cvar) = unsafe { &*data_ptr };
         let device = lock.lock().unwrap();
         if *device != device_id {
             let (dev, timeout_res) = cvar.wait_timeout(device, waiting_time).unwrap();
@@ -353,13 +362,10 @@ impl AggregateDevice {
                 );
             }
             if *dev != device_id {
-                let status = remove_listener();
+                let status = cleanup();
                 // If the error is kAudioHardwareBadObjectError, it implies `device_id` is somehow
                 // dead, so its listener should receive nothing. It's ok to leave here.
                 assert!(status == NO_ERR || status == (kAudioHardwareBadObjectError as OSStatus));
-                // TODO: Destroy the aggregate device immediately if error is not
-                // kAudioHardwareBadObjectError. Otherwise the `devices_changed_callback` is able
-                // to touch the `cloned_condvar_pair` after it's freed.
                 return Err(Error::from(waiting_time));
             }
         }
@@ -370,15 +376,17 @@ impl AggregateDevice {
             _addresses: *const AudioObjectPropertyAddress,
             data: *mut c_void,
         ) -> OSStatus {
-            let pair = unsafe { &mut *(data as *mut Arc<(Mutex<AudioObjectID>, Condvar)>) };
-            let (lock, cvar) = &**pair;
-            let mut device = lock.lock().unwrap();
-            *device = id;
-            cvar.notify_one();
+            with_sync_callback_ptr(data, || {
+                let pair = unsafe { &*(data as *const (Mutex<AudioObjectID>, Condvar)) };
+                let (lock, cvar) = pair;
+                let mut device = lock.lock().unwrap();
+                *device = id;
+                cvar.notify_one();
+            });
             NO_ERR
         }
 
-        let status = remove_listener();
+        let status = cleanup();
         assert_eq!(status, NO_ERR);
         Ok(())
     }
diff --git a/src/backend/mod.rs b/src/backend/mod.rs
@@ -41,6 +41,7 @@ use cubeb_backend::{
 };
 use mach::mach_time::{mach_absolute_time, mach_timebase_info};
 use std::cmp;
+use std::collections::HashSet;
 use std::ffi::{CStr, CString};
 use std::fmt;
 use std::mem;
@@ -49,7 +50,7 @@ use std::ptr;
 use std::slice;
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
-use std::sync::{Arc, Condvar, Mutex, MutexGuard, Weak};
+use std::sync::{Arc, Condvar, LazyLock, Mutex, MutexGuard, Weak};
 use std::time::{Duration, Instant};
 const NO_ERR: OSStatus = 0;
 
@@ -70,6 +71,28 @@ const VPIO_IDLE_TIMEOUT: Duration = Duration::from_secs(10);
 
 const MACOS_KERNEL_MAJOR_VERSION_MONTEREY: u32 = 21;
 
+// Global registry for tracking valid sync callback pointers.
+static SYNC_CALLBACK_REGISTRY: LazyLock<Mutex<HashSet<usize>>> =
+    LazyLock::new(|| Mutex::new(HashSet::new()));
+
+fn sync_callback_registry_register(ptr: usize) {
+    SYNC_CALLBACK_REGISTRY.lock().unwrap().insert(ptr);
+}
+
+fn sync_callback_registry_unregister(ptr: usize) {
+    SYNC_CALLBACK_REGISTRY.lock().unwrap().remove(&ptr);
+}
+
+fn with_sync_callback_ptr<F>(ptr: *mut c_void, f: F)
+where
+    F: FnOnce(),
+{
+    let guard = SYNC_CALLBACK_REGISTRY.lock().unwrap();
+    if guard.contains(&(ptr as usize)) {
+        f();
+    }
+}
+
 #[derive(Debug, PartialEq)]
 enum ParseMacOSKernelVersionError {
     SysCtl,
@@ -1508,16 +1531,16 @@ fn set_buffer_size_sync(unit: AudioUnit, devtype: DeviceType, frames: u32) -> Re
     }
 
     let waiting_time = Duration::from_millis(100);
-    let pair = Arc::new((Mutex::new(false), Condvar::new()));
-    let mut pair2 = pair.clone();
-    let pair_ptr = &mut pair2;
+    let pair = Box::new((Mutex::new(false), Condvar::new()));
+    let pair_ptr = Box::into_raw(pair);
+    sync_callback_registry_register(pair_ptr as usize);
 
     assert_eq!(
         audio_unit_add_property_listener(
             unit,
             kAudioDevicePropertyBufferFrameSize,
             buffer_size_changed_callback,
-            pair_ptr,
+            pair_ptr as *mut c_void,
         ),
         NO_ERR
     );
@@ -1528,10 +1551,14 @@ fn set_buffer_size_sync(unit: AudioUnit, devtype: DeviceType, frames: u32) -> Re
                 unit,
                 kAudioDevicePropertyBufferFrameSize,
                 buffer_size_changed_callback,
-                pair_ptr,
+                pair_ptr as *mut c_void,
             ),
             NO_ERR
         );
+        // Unregister blocks if callback is in-flight (holds registry lock)
+        sync_callback_registry_unregister(pair_ptr as usize);
+        // Safe to drop: callback has either completed or will exit early
+        unsafe { drop(Box::from_raw(pair_ptr)) };
     });
 
     set_buffer_size(unit, devtype, frames).map_err(|e| {
@@ -1544,7 +1571,7 @@ fn set_buffer_size_sync(unit: AudioUnit, devtype: DeviceType, frames: u32) -> Re
         Error::error()
     })?;
 
-    let (lock, cvar) = &*pair;
+    let (lock, cvar) = unsafe { &*pair_ptr };
     let changed = lock.lock().unwrap();
     if !*changed {
         let (chg, timeout_res) = cvar.wait_timeout(changed, waiting_time).unwrap();
@@ -1584,16 +1611,17 @@ fn set_buffer_size_sync(unit: AudioUnit, devtype: DeviceType, frames: u32) -> Re
         in_element: AudioUnitElement,
     ) {
         if in_scope == 0 {
-            // filter out the callback for global scope.
             return;
         }
-        assert!(in_element == AU_IN_BUS || in_element == AU_OUT_BUS);
-        assert_eq!(in_property_id, kAudioDevicePropertyBufferFrameSize);
-        let pair = unsafe { &mut *(in_client_data as *mut Arc<(Mutex<bool>, Condvar)>) };
-        let (lock, cvar) = &**pair;
-        let mut changed = lock.lock().unwrap();
-        *changed = true;
-        cvar.notify_one();
+        with_sync_callback_ptr(in_client_data, || {
+            assert!(in_element == AU_IN_BUS || in_element == AU_OUT_BUS);
+            assert_eq!(in_property_id, kAudioDevicePropertyBufferFrameSize);
+            let pair = unsafe { &*(in_client_data as *const (Mutex<bool>, Condvar)) };
+            let (lock, cvar) = pair;
+            let mut changed = lock.lock().unwrap();
+            *changed = true;
+            cvar.notify_one();
+        });
     }
 
     Ok(())
diff --git a/src/backend/tests/mod.rs b/src/backend/tests/mod.rs
@@ -8,5 +8,6 @@ mod device_property;
 mod interfaces;
 mod manual;
 mod parallel;
+mod sync_callback;
 mod tone;
 mod utils;
diff --git a/src/backend/tests/sync_callback.rs b/src/backend/tests/sync_callback.rs
@@ -0,0 +1,136 @@
+// Copyright © 2026 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+
+use super::*;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
+// Test basic registration and unregistration
+#[test]
+fn test_sync_callback_register_unregister() {
+    let dummy_data = Box::new(42u32);
+    let ptr = Box::into_raw(dummy_data);
+    let ptr_usize = ptr as usize;
+
+    // Register the pointer
+    sync_callback_registry_register(ptr_usize);
+
+    // Verify the callback is called when registered
+    let called = Arc::new(AtomicBool::new(false));
+    let called_clone = called.clone();
+    with_sync_callback_ptr(ptr as *mut c_void, || {
+        called_clone.store(true, Ordering::SeqCst);
+    });
+    assert!(
+        called.load(Ordering::SeqCst),
+        "Callback should be called when pointer is registered"
+    );
+
+    // Unregister the pointer
+    sync_callback_registry_unregister(ptr_usize);
+
+    // Verify the callback is NOT called after unregistration
+    let called2 = Arc::new(AtomicBool::new(false));
+    let called2_clone = called2.clone();
+    with_sync_callback_ptr(ptr as *mut c_void, || {
+        called2_clone.store(true, Ordering::SeqCst);
+    });
+    assert!(
+        !called2.load(Ordering::SeqCst),
+        "Callback should NOT be called after unregistration"
+    );
+
+    // Clean up
+    unsafe { drop(Box::from_raw(ptr)) };
+}
+
+// Test that unregistered pointers don't trigger callbacks
+#[test]
+fn test_sync_callback_unregistered_pointer() {
+    let dummy_data = Box::new(100u32);
+    let ptr = Box::into_raw(dummy_data);
+
+    // Try to use callback without registration
+    let called = Arc::new(AtomicBool::new(false));
+    let called_clone = called.clone();
+    with_sync_callback_ptr(ptr as *mut c_void, || {
+        called_clone.store(true, Ordering::SeqCst);
+    });
+    assert!(
+        !called.load(Ordering::SeqCst),
+        "Callback should NOT be called for unregistered pointer"
+    );
+
+    // Clean up
+    unsafe { drop(Box::from_raw(ptr)) };
+}
+
+// Test multiple registrations don't cause issues
+#[test]
+fn test_sync_callback_multiple_pointers() {
+    let data1 = Box::new(1u32);
+    let ptr1 = Box::into_raw(data1);
+    let ptr1_usize = ptr1 as usize;
+
+    let data2 = Box::new(2u32);
+    let ptr2 = Box::into_raw(data2);
+    let ptr2_usize = ptr2 as usize;
+
+    // Register both pointers
+    sync_callback_registry_register(ptr1_usize);
+    sync_callback_registry_register(ptr2_usize);
+
+    // Both should work
+    let called1 = Arc::new(AtomicBool::new(false));
+    let called1_clone = called1.clone();
+    with_sync_callback_ptr(ptr1 as *mut c_void, || {
+        called1_clone.store(true, Ordering::SeqCst);
+    });
+    assert!(
+        called1.load(Ordering::SeqCst),
+        "First pointer callback should be called"
+    );
+
+    let called2 = Arc::new(AtomicBool::new(false));
+    let called2_clone = called2.clone();
+    with_sync_callback_ptr(ptr2 as *mut c_void, || {
+        called2_clone.store(true, Ordering::SeqCst);
+    });
+    assert!(
+        called2.load(Ordering::SeqCst),
+        "Second pointer callback should be called"
+    );
+
+    // Unregister first pointer
+    sync_callback_registry_unregister(ptr1_usize);
+
+    // First should not work, second should still work
+    let called1_after = Arc::new(AtomicBool::new(false));
+    let called1_after_clone = called1_after.clone();
+    with_sync_callback_ptr(ptr1 as *mut c_void, || {
+        called1_after_clone.store(true, Ordering::SeqCst);
+    });
+    assert!(
+        !called1_after.load(Ordering::SeqCst),
+        "First pointer callback should NOT be called after unregister"
+    );
+
+    let called2_after = Arc::new(AtomicBool::new(false));
+    let called2_after_clone = called2_after.clone();
+    with_sync_callback_ptr(ptr2 as *mut c_void, || {
+        called2_after_clone.store(true, Ordering::SeqCst);
+    });
+    assert!(
+        called2_after.load(Ordering::SeqCst),
+        "Second pointer callback should still be called"
+    );
+
+    // Clean up
+    sync_callback_registry_unregister(ptr2_usize);
+    unsafe {
+        drop(Box::from_raw(ptr1));
+        drop(Box::from_raw(ptr2));
+    };
+}
