Merge pull request #2 from mitchmindtree/no-eventloop-send-sync

Explicitly make dynamically dispatched API !Send + !Sync

asio-sys/src/bindings/mod.rs

@@ -7,7 +7,7 @@ use self::errors::{AsioError, AsioErrorWrapper, LoadDriverError};
 use std::ffi::CStr;
 use std::ffi::CString;
 use std::os::raw::{c_char, c_double, c_long, c_void};
-use std::sync::{Arc, Mutex, Weak};
+use std::sync::{Arc, Mutex, MutexGuard, Weak};
 
 // Bindings import
 use self::asio_import as ai;
@@ -85,7 +85,7 @@ pub struct SampleRate {
 }
 
 /// Holds the pointer to the callbacks that come from cpal
-struct BufferCallback(Box<FnMut(i32) + Send>);
+struct BufferCallback(Box<dyn FnMut(i32) + Send>);
 
 /// Input and Output streams.
 ///
@@ -422,6 +422,8 @@ impl Driver {
         // To pass as ai::ASIOCallbacks
         let mut callbacks = create_asio_callbacks();
 
+        let mut state = self.inner.lock_state();
+
         // Retrieve the available buffer sizes.
         let buffer_sizes = asio_get_buffer_sizes()?;
         if buffer_sizes.pref <= 0 {
@@ -432,13 +434,12 @@ impl Driver {
         }
 
         // Ensure the driver is in the `Initialized` state.
-        if let DriverState::Running = self.inner.state() {
+        if let DriverState::Running = *state {
-            self.stop()?;
+            state.stop()?;
         }
-        if let DriverState::Prepared = self.inner.state() {
+        if let DriverState::Prepared = *state {
-            self.dispose_buffers()?;
+            state.dispose_buffers()?;
         }
-
         unsafe {
             asio_result!(ai::ASIOCreateBuffers(
                 buffer_infos.as_mut_ptr() as *mut _,
@@ -447,8 +448,8 @@ impl Driver {
                 &mut callbacks as *mut _ as *mut _,
             ))?;
         }
+        *state = DriverState::Prepared;
 
-        self.inner.set_state(DriverState::Prepared);
         Ok(buffer_sizes.pref)
     }
 
@@ -569,13 +570,14 @@ impl Driver {
     ///
     /// No-op if already `Running`.
     pub fn start(&self) -> Result<(), AsioError> {
-        if let DriverState::Running = self.inner.state() {
+        let mut state = self.inner.lock_state();
+        if let DriverState::Running = *state {
             return Ok(());
         }
         unsafe {
             asio_result!(ai::ASIOStart())?;
         }
-        self.inner.set_state(DriverState::Running);
+        *state = DriverState::Running;
         Ok(())
     }
 
@@ -635,55 +637,70 @@ impl Driver {
     }
 }
 
-impl DriverInner {
+impl DriverState {
-    fn state(&self) -> DriverState {
+    fn stop(&mut self) -> Result<(), AsioError> {
-        *self.state.lock().expect("failed to lock `DriverState`")
+        if let DriverState::Running = *self {
-    }
-
-    fn set_state(&self, state: DriverState) {
-        *self.state.lock().expect("failed to lock `DriverState`") = state;
-    }
-
-    fn stop_inner(&self) -> Result<(), AsioError> {
-        if let DriverState::Running = self.state() {
             unsafe {
                 asio_result!(ai::ASIOStop())?;
             }
-            self.set_state(DriverState::Prepared);
+            *self = DriverState::Prepared;
         }
         Ok(())
     }
 
-    fn dispose_buffers_inner(&self) -> Result<(), AsioError> {
+    fn dispose_buffers(&mut self) -> Result<(), AsioError> {
-        if let DriverState::Initialized = self.state() {
+        if let DriverState::Initialized = *self {
             return Ok(());
         }
-        if let DriverState::Running = self.state() {
+        if let DriverState::Running = *self {
-            self.stop_inner()?;
+            self.stop()?;
         }
         unsafe {
             asio_result!(ai::ASIODisposeBuffers())?;
         }
-        self.set_state(DriverState::Initialized);
+        *self = DriverState::Initialized;
         Ok(())
     }
 
-    fn destroy_inner(&mut self) -> Result<(), AsioError> {
+    fn destroy(&mut self) -> Result<(), AsioError> {
-        // Drop back through the driver state machine one state at a time.
+        if let DriverState::Running = *self {
-        if let DriverState::Running = self.state() {
+            self.stop()?;
-            self.stop_inner()?;
         }
-        if let DriverState::Prepared = self.state() {
+        if let DriverState::Prepared = *self {
-            self.dispose_buffers_inner()?;
+            self.dispose_buffers()?;
         }
         unsafe {
             asio_result!(ai::ASIOExit())?;
             ai::remove_current_driver();
         }
+        Ok(())
+    }
+}
 
-        // Clear any existing stream callbacks.
+impl DriverInner {
-        if let Ok(mut bcs) = BUFFER_CALLBACK.lock() {
+    fn lock_state(&self) -> MutexGuard<DriverState> {
-            bcs.clear();
+        self.state.lock().expect("failed to lock `DriverState`")
+    }
+
+    fn stop_inner(&self) -> Result<(), AsioError> {
+        let mut state = self.lock_state();
+        state.stop()
+    }
+
+    fn dispose_buffers_inner(&self) -> Result<(), AsioError> {
+        let mut state = self.lock_state();
+        state.dispose_buffers()
+    }
+
+    fn destroy_inner(&mut self) -> Result<(), AsioError> {
+        {
+            let mut state = self.lock_state();
+            state.destroy()?;
+
+            // Clear any existing stream callbacks.
+            if let Ok(mut bcs) = BUFFER_CALLBACK.lock() {
+                bcs.clear();
+            }
         }
 
         // Signal that the driver has been destroyed.

src/platform/mod.rs

@@ -60,7 +60,12 @@ macro_rules! impl_platform_host {
 
         /// The **Stream** implementation associated with the platform's dynamically dispatched
         /// **Host** type.
-        pub struct Stream(StreamInner);
+        // Streams cannot be `Send` or `Sync` if we plan to support Android's AAudio API. This is
+        // because the stream API is not thread-safe, and the API prohibits calling certain
+        // functions within the callback.
+        //
+        // TODO: Confirm this and add more specific detail and references.
+        pub struct Stream(StreamInner, crate::platform::NotSendSyncAcrossAllPlatforms);
 
         /// The **SupportedInputFormats** iterator associated with the platform's dynamically
         /// dispatched **Host** type.
@@ -142,7 +147,7 @@ macro_rules! impl_platform_host {
                 match self.0 {
                     $(
                         DevicesInner::$HostVariant(ref mut d) => {
-                            d.next().map(DeviceInner::$HostVariant).map(Device)
+                            d.next().map(DeviceInner::$HostVariant).map(Device::from)
                         }
                     )*
                 }
@@ -256,7 +261,7 @@ macro_rules! impl_platform_host {
                     $(
                         DeviceInner::$HostVariant(ref d) => d.build_input_stream(format, data_callback, error_callback)
                             .map(StreamInner::$HostVariant)
-                            .map(Stream),
+                            .map(Stream::from),
                     )*
                 }
             }
@@ -267,7 +272,7 @@ macro_rules! impl_platform_host {
                     $(
                         DeviceInner::$HostVariant(ref d) => d.build_output_stream(format, data_callback, error_callback)
                             .map(StreamInner::$HostVariant)
-                            .map(Stream),
+                            .map(Stream::from),
                     )*
                 }
             }
@@ -285,7 +290,7 @@ macro_rules! impl_platform_host {
                 match self.0 {
                     $(
                         HostInner::$HostVariant(ref h) => {
-                            h.devices().map(DevicesInner::$HostVariant).map(Devices)
+                            h.devices().map(DevicesInner::$HostVariant).map(Devices::from)
                         }
                     )*
                 }
@@ -295,7 +300,7 @@ macro_rules! impl_platform_host {
                 match self.0 {
                     $(
                         HostInner::$HostVariant(ref h) => {
-                            h.default_input_device().map(DeviceInner::$HostVariant).map(Device)
+                            h.default_input_device().map(DeviceInner::$HostVariant).map(Device::from)
                         }
                     )*
                 }
@@ -305,7 +310,7 @@ macro_rules! impl_platform_host {
                 match self.0 {
                     $(
                         HostInner::$HostVariant(ref h) => {
-                            h.default_output_device().map(DeviceInner::$HostVariant).map(Device)
+                            h.default_output_device().map(DeviceInner::$HostVariant).map(Device::from)
                         }
                     )*
                 }
@@ -334,28 +339,52 @@ macro_rules! impl_platform_host {
             }
         }
 
+        impl From<DeviceInner> for Device {
+            fn from(d: DeviceInner) -> Self {
+                Device(d)
+            }
+        }
+
+        impl From<DevicesInner> for Devices {
+            fn from(d: DevicesInner) -> Self {
+                Devices(d)
+            }
+        }
+
+        impl From<HostInner> for Host {
+            fn from(h: HostInner) -> Self {
+                Host(h)
+            }
+        }
+
+        impl From<StreamInner> for Stream {
+            fn from(s: StreamInner) -> Self {
+                Stream(s, Default::default())
+            }
+        }
+
         $(
             impl From<crate::host::$host_mod::Device> for Device {
                 fn from(h: crate::host::$host_mod::Device) -> Self {
-                    Device(DeviceInner::$HostVariant(h))
+                    DeviceInner::$HostVariant(h).into()
                 }
             }
 
             impl From<crate::host::$host_mod::Devices> for Devices {
                 fn from(h: crate::host::$host_mod::Devices) -> Self {
-                    Devices(DevicesInner::$HostVariant(h))
+                    DevicesInner::$HostVariant(h).into()
                 }
             }
 
             impl From<crate::host::$host_mod::Host> for Host {
                 fn from(h: crate::host::$host_mod::Host) -> Self {
-                    Host(HostInner::$HostVariant(h))
+                    HostInner::$HostVariant(h).into()
                 }
             }
 
             impl From<crate::host::$host_mod::Stream> for Stream {
                 fn from(h: crate::host::$host_mod::Stream) -> Self {
-                    Stream(StreamInner::$HostVariant(h))
+                    StreamInner::$HostVariant(h).into()
                 }
             }
         )*
@@ -378,7 +407,7 @@ macro_rules! impl_platform_host {
                     HostId::$HostVariant => {
                         crate::host::$host_mod::Host::new()
                             .map(HostInner::$HostVariant)
-                            .map(Host)
+                            .map(Host::from)
                     }
                 )*
             }
@@ -507,3 +536,19 @@ mod platform_impl {
             .into()
     }
 }
+
+// The following zero-sized types are for applying Send/Sync restrictions to ensure
+// consistent behaviour across different platforms. These verbosely named types are used
+// (rather than using the markers directly) in the hope of making the compile errors
+// slightly more helpful.
+//
+// TODO: Remove these in favour of using negative trait bounds if they stabilise.
+
+// A marker used to remove the `Send` and `Sync` traits.
+struct NotSendSyncAcrossAllPlatforms(std::marker::PhantomData<*mut ()>);
+
+impl Default for NotSendSyncAcrossAllPlatforms {
+    fn default() -> Self {
+        NotSendSyncAcrossAllPlatforms(std::marker::PhantomData)
+    }
+}