WIP SharedVoiceProcessingUnit

Author: Pehrsons

src/backend/mod.rs

@@ -35,7 +35,7 @@ use self::utils::*;
 use backend::ringbuf::RingBuffer;
 use cubeb_backend::{
     ffi, ChannelLayout, Context, ContextOps, DeviceCollectionRef, DeviceId, DeviceRef, DeviceType,
-    Error, InputProcessingParams, Ops, Result, SampleFormat, State, Stream, StreamOps,
+    Error, ErrorCode, InputProcessingParams, Ops, Result, SampleFormat, State, Stream, StreamOps,
     StreamParams, StreamParamsRef, StreamPrefs,
 };
 use mach::mach_time::{mach_absolute_time, mach_timebase_info};
@@ -1162,6 +1162,7 @@ fn create_voiceprocessing_audiounit(
     in_device: &device_info,
     out_device: &device_info,
 ) -> Result<OwningHandle<AudioUnit>> {
+    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));
@@ -2134,44 +2135,29 @@ struct SharedVoiceProcessingUnitStorage {
 #[derive(Debug)]
 struct SharedVoiceProcessingUnit {
     sync_storage: Mutex<SharedVoiceProcessingUnitStorage>,
-    condvar: Condvar,
     queue: Queue,
 }
 
 impl SharedVoiceProcessingUnit {
     fn new(queue: Queue) -> Self {
         Self {
             sync_storage: Mutex::new(SharedVoiceProcessingUnitStorage::default()),
-            condvar: Condvar::new(),
             queue,
         }
     }
 
-    fn prime(&mut self) {
+    fn ensure_unit(&mut self) -> Result<()> {
         debug_assert_running_serially();
-        {
-            let mut guard = self.sync_storage.lock().unwrap();
-            if guard.priming {
-                return;
-            }
-            if guard.storage.is_some() {
-                return;
-            }
-            cubeb_log!("Priming the shared voiceprocessing unit");
-            guard.priming = true;
-        }
-        let unit = create_typed_audiounit(kAudioUnitSubType_VoiceProcessingIO);
         let mut guard = self.sync_storage.lock().unwrap();
+        if guard.storage.is_some() {
+            // No need to create vpio unit as one already exists.
+            return Err(Error::not_supported());
+        }
         guard.priming = false;
-        self.condvar.notify_all();
-        if let Err(e) = unit {
-            cubeb_log!(
-                "Failed creating voiceprocessing unit while priming async. r={}",
-                e
-            );
-            return;
+        let unit = create_typed_audiounit(kAudioUnitSubType_VoiceProcessingIO);
+        if unit.is_err() {
+            return Err(Error::error());
         }
-        cubeb_log!("Shared voiceprocessing unit primed and ready");
 
         match get_default_device(DeviceType::OUTPUT) {
             None => {
@@ -2191,31 +2177,54 @@ impl SharedVoiceProcessingUnit {
 
         let old_storage = guard.storage.replace(Arc::from(Mutex::new(unit.ok())));
         assert!(old_storage.is_none());
+        Ok(())
+    }
+
+    fn prime(&mut self) {
+        {
+            let mut guard = self.sync_storage.lock().unwrap();
+            if guard.storage.is_some() {
+                // No need to prime as unit already exists.
+                return;
+            }
+            if guard.priming {
+                // No need to prime async as another call is already doing it.
+                return;
+            }
+            cubeb_log!("Priming the shared voiceprocessing unit");
+            guard.priming = true;
+        }
+        self.queue
+            .clone()
+            .run_async(move || match self.ensure_unit() {
+                Ok(()) => {
+                    cubeb_log!("Shared voiceprocessing unit primed and ready");
+                }
+                Err(e) => match e.code() {
+                    ErrorCode::Error => {
+                        cubeb_log!("Error creating shared voiceprocessing unit for priming");
+                    }
+                    ErrorCode::NotSupported => {
+                        cubeb_log!("Shared voiceprocessing unit was created before async priming");
+                    }
+                    _ => unreachable!("Unexpected error"),
+                },
+            });
     }
 
     fn take(&mut self) -> Result<OwningHandle<AudioUnit>> {
         debug_assert_running_serially();
-        let guard = self.sync_storage.lock().unwrap();
-        let mut guard = self
-            .condvar
-            .wait_while(guard, |storage| storage.priming)
-            .unwrap();
-        if guard.storage.is_none() {
-            cubeb_log!("Creating synchronously the shared voiceprocessing unit");
-            let unit = create_typed_audiounit(kAudioUnitSubType_VoiceProcessingIO)?;
-            guard.storage.replace(Arc::from(Mutex::new(Some(unit))));
+        if let Err(e) = self.ensure_unit() {
+            if e.code() == ErrorCode::Error {
+                return Err(e);
+            }
         }
+        let mut guard = self.sync_storage.lock().unwrap();
         let storage: Arc<SharedStorage<AudioUnit>> = guard.storage.as_mut().unwrap().clone();
         let mut guard = (*storage).lock().unwrap();
         match guard.take() {
-            Some(unit) => {
-                cubeb_log!("voiceprocessing audiounit now taken");
-                Ok(OwningHandle::new(Arc::downgrade(&storage), unit))
-            }
-            None => {
-                cubeb_log!("voiceprocessing audiounit was requested but already taken");
-                Err(Error::device_unavailable())
-            }
+            Some(unit) => Ok(OwningHandle::new(Arc::downgrade(&storage), unit)),
+            None => Err(Error::device_unavailable()),
         }
     }
 }
@@ -2225,6 +2234,7 @@ unsafe impl Sync for SharedVoiceProcessingUnit {}
 
 impl Drop for SharedVoiceProcessingUnit {
     fn drop(&mut self) {
+        debug_assert_not_running_serially();
         self.queue.run_final(|| {
             let mut guard = self.sync_storage.lock().unwrap();
             assert!(!guard.priming);
@@ -2418,6 +2428,11 @@ impl ContextOps for AudioUnitContext {
         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 = SharedVoiceProcessingUnit::new(shared_vp_queue);
         Ok(unsafe { Context::from_ptr(Box::into_raw(ctx) as *mut _) })
     }
 
@@ -3071,14 +3086,17 @@ impl<'ctx> CoreStreamData<'ctx> {
                 &self.input_device,
                 &self.output_device,
             ) {
-                cubeb_log!("({:p}) Using VoiceProcessingIO AudioUnit", self.stm_ptr);
+                cubeb_log!(
+                    "({:p}) Took shared VoiceProcessingIO AudioUnit",
+                    self.stm_ptr
+                );
                 self.input_unit = *au_handle.as_mut();
                 self.output_unit = *au_handle.as_mut();
                 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 take VoiceProcessingIO AudioUnit. Trying a regular one.",
                 self.stm_ptr
             );
         }

src/backend/tests/api.rs

@@ -1776,8 +1776,7 @@ fn test_shared_voice_processing_unit_after_priming() {
         get_serial_queue_singleton(),
     );
     let mut shared = SharedVoiceProcessingUnit::new(queue.clone());
-    let queue = shared.queue.clone();
-    queue.run_async(|| shared.prime());
+    shared.prime();
     assert!(queue.run_sync(|| shared.take()).unwrap().is_ok());
 }