From 33ddf749548d87bf54ce18eb342f954cec1465b2 Mon Sep 17 00:00:00 2001 From: mitchmindtree Date: Sun, 5 Jan 2020 17:43:14 +0100 Subject: [PATCH 1/3] Explicitly make dynamically dispatched API !Send + !Sync This is in order to ensure consistent restrictions across platforms in a manner that ensures thread safety across each of the supported platforms. Please see added comments in the diff for details on which platforms impose these restrictions. --- src/platform/mod.rs | 89 ++++++++++++++++++++++++++++++++++++++------- 1 file changed, 75 insertions(+), 14 deletions(-) diff --git a/src/platform/mod.rs b/src/platform/mod.rs index d28e58e..e6a927d 100644 --- a/src/platform/mod.rs +++ b/src/platform/mod.rs @@ -48,11 +48,18 @@ macro_rules! impl_platform_host { /// /// This type may be constructed via the **host_from_id** function. **HostId**s may /// be acquired via the **ALL_HOSTS** const and the **available_hosts** function. - pub struct Host(HostInner); + // `Host` and `Device` cannot assume to be `Sync` as we have not yet been able to confirm + // whether or not the ASIO API is thread-safe. + // + // TODO: Try to contact ASIO to get more information. Review the existing implementation of + // the `asio` backend's `Host` and `Driver` types and see if the existing `Arc`s and + // `Mutex`s don't already make the API `Sync`. + pub struct Host(HostInner, crate::platform::NotSyncAcrossAllPlatforms); /// The **Device** implementation associated with the platform's dynamically dispatched /// **Host** type. - pub struct Device(DeviceInner); + // See comment above `Host` for reasoning behind `NotSyncAcrossAllPlatforms`. + pub struct Device(DeviceInner, crate::platform::NotSyncAcrossAllPlatforms); /// The **Devices** iterator associated with the platform's dynamically dispatched **Host** /// type. @@ -60,7 +67,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 +154,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 +268,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 +279,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 +297,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 +307,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 +317,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 +346,52 @@ macro_rules! impl_platform_host { } } + impl From for Device { + fn from(d: DeviceInner) -> Self { + Device(d, Default::default()) + } + } + + impl From for Devices { + fn from(d: DevicesInner) -> Self { + Devices(d) + } + } + + impl From for Host { + fn from(h: HostInner) -> Self { + Host(h, Default::default()) + } + } + + impl From for Stream { + fn from(s: StreamInner) -> Self { + Stream(s, Default::default()) + } + } + $( impl From for Device { fn from(h: crate::host::$host_mod::Device) -> Self { - Device(DeviceInner::$HostVariant(h)) + DeviceInner::$HostVariant(h).into() } } impl From for Devices { fn from(h: crate::host::$host_mod::Devices) -> Self { - Devices(DevicesInner::$HostVariant(h)) + DevicesInner::$HostVariant(h).into() } } impl From for Host { fn from(h: crate::host::$host_mod::Host) -> Self { - Host(HostInner::$HostVariant(h)) + HostInner::$HostVariant(h).into() } } impl From for Stream { fn from(h: crate::host::$host_mod::Stream) -> Self { - Stream(StreamInner::$HostVariant(h)) + StreamInner::$HostVariant(h).into() } } )* @@ -378,7 +414,7 @@ macro_rules! impl_platform_host { HostId::$HostVariant => { crate::host::$host_mod::Host::new() .map(HostInner::$HostVariant) - .map(Host) + .map(Host::from) } )* } @@ -507,3 +543,28 @@ 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 ()>); + +// A marker used to remove the `Sync` traits. +struct NotSyncAcrossAllPlatforms(std::marker::PhantomData>); + +impl Default for NotSyncAcrossAllPlatforms { + fn default() -> Self { + NotSyncAcrossAllPlatforms(std::marker::PhantomData) + } +} + +impl Default for NotSendSyncAcrossAllPlatforms { + fn default() -> Self { + NotSendSyncAcrossAllPlatforms(std::marker::PhantomData) + } +} From ca2aceb5369aa3a3d4ffa2c0b4857842c34a66ca Mon Sep 17 00:00:00 2001 From: mitchmindtree Date: Sun, 12 Jan 2020 19:30:04 +0100 Subject: [PATCH 2/3] Fix state transition synchronisation in ASIO This makes some tweaks to the ASIO backend in order to fix some cases where races may have occured. This should allow us to remove the `Sync` bound on the `Device` and `Host` types. --- asio-sys/src/bindings/mod.rs | 89 +++++++++++++++++++++--------------- 1 file changed, 53 insertions(+), 36 deletions(-) diff --git a/asio-sys/src/bindings/mod.rs b/asio-sys/src/bindings/mod.rs index eee5c8b..0d38801 100644 --- a/asio-sys/src/bindings/mod.rs +++ b/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); +struct BufferCallback(Box); /// 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() { - self.stop()?; + if let DriverState::Running = *state { + state.stop()?; } - if let DriverState::Prepared = self.inner.state() { - self.dispose_buffers()?; + if let DriverState::Prepared = *state { + 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 { - fn state(&self) -> DriverState { - *self.state.lock().expect("failed to lock `DriverState`") - } - - 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() { +impl DriverState { + fn stop(&mut self) -> Result<(), AsioError> { + if let DriverState::Running = *self { unsafe { asio_result!(ai::ASIOStop())?; } - self.set_state(DriverState::Prepared); + *self = DriverState::Prepared; } Ok(()) } - fn dispose_buffers_inner(&self) -> Result<(), AsioError> { - if let DriverState::Initialized = self.state() { + fn dispose_buffers(&mut self) -> Result<(), AsioError> { + if let DriverState::Initialized = *self { return Ok(()); } - if let DriverState::Running = self.state() { - self.stop_inner()?; + if let DriverState::Running = *self { + self.stop()?; } unsafe { asio_result!(ai::ASIODisposeBuffers())?; } - self.set_state(DriverState::Initialized); + *self = DriverState::Initialized; Ok(()) } - fn destroy_inner(&mut self) -> Result<(), AsioError> { - // Drop back through the driver state machine one state at a time. - if let DriverState::Running = self.state() { - self.stop_inner()?; + fn destroy(&mut self) -> Result<(), AsioError> { + if let DriverState::Running = *self { + self.stop()?; } - if let DriverState::Prepared = self.state() { - self.dispose_buffers_inner()?; + if let DriverState::Prepared = *self { + self.dispose_buffers()?; } unsafe { asio_result!(ai::ASIOExit())?; ai::remove_current_driver(); } + Ok(()) + } +} - // Clear any existing stream callbacks. - if let Ok(mut bcs) = BUFFER_CALLBACK.lock() { - bcs.clear(); +impl DriverInner { + fn lock_state(&self) -> MutexGuard { + 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. From 32d39bcfd34681215e97077f302d999daeb373ce Mon Sep 17 00:00:00 2001 From: mitchmindtree Date: Sun, 12 Jan 2020 22:43:05 +0100 Subject: [PATCH 3/3] Relax `Sync` restriction on `Device` and `Host` Originally this restriction was placed due to uncertainty around the thread safety of the ASIO API. While the ASIO API itself makes no thread-safety guarantees that we are aware of, the `asio-sys` high-level bindings enforce synchronised access to the API and state transitions via a mutex. --- src/platform/mod.rs | 24 ++++-------------------- 1 file changed, 4 insertions(+), 20 deletions(-) diff --git a/src/platform/mod.rs b/src/platform/mod.rs index e6a927d..6447f15 100644 --- a/src/platform/mod.rs +++ b/src/platform/mod.rs @@ -48,18 +48,11 @@ macro_rules! impl_platform_host { /// /// This type may be constructed via the **host_from_id** function. **HostId**s may /// be acquired via the **ALL_HOSTS** const and the **available_hosts** function. - // `Host` and `Device` cannot assume to be `Sync` as we have not yet been able to confirm - // whether or not the ASIO API is thread-safe. - // - // TODO: Try to contact ASIO to get more information. Review the existing implementation of - // the `asio` backend's `Host` and `Driver` types and see if the existing `Arc`s and - // `Mutex`s don't already make the API `Sync`. - pub struct Host(HostInner, crate::platform::NotSyncAcrossAllPlatforms); + pub struct Host(HostInner); /// The **Device** implementation associated with the platform's dynamically dispatched /// **Host** type. - // See comment above `Host` for reasoning behind `NotSyncAcrossAllPlatforms`. - pub struct Device(DeviceInner, crate::platform::NotSyncAcrossAllPlatforms); + pub struct Device(DeviceInner); /// The **Devices** iterator associated with the platform's dynamically dispatched **Host** /// type. @@ -348,7 +341,7 @@ macro_rules! impl_platform_host { impl From for Device { fn from(d: DeviceInner) -> Self { - Device(d, Default::default()) + Device(d) } } @@ -360,7 +353,7 @@ macro_rules! impl_platform_host { impl From for Host { fn from(h: HostInner) -> Self { - Host(h, Default::default()) + Host(h) } } @@ -554,15 +547,6 @@ mod platform_impl { // A marker used to remove the `Send` and `Sync` traits. struct NotSendSyncAcrossAllPlatforms(std::marker::PhantomData<*mut ()>); -// A marker used to remove the `Sync` traits. -struct NotSyncAcrossAllPlatforms(std::marker::PhantomData>); - -impl Default for NotSyncAcrossAllPlatforms { - fn default() -> Self { - NotSyncAcrossAllPlatforms(std::marker::PhantomData) - } -} - impl Default for NotSendSyncAcrossAllPlatforms { fn default() -> Self { NotSendSyncAcrossAllPlatforms(std::marker::PhantomData)