From 18dffe8b438238bf8a9966d488e575bc2c15c700 Mon Sep 17 00:00:00 2001 From: dignifiedquire Date: Sun, 21 Jun 2020 21:23:21 +0200 Subject: [PATCH 1/2] refactor: switch to async-mutex for Mutex implementation --- Cargo.toml | 4 +- src/sync/condvar.rs | 8 +- src/sync/mod.rs | 5 +- src/sync/mutex.rs | 294 -------------------------------------------- 4 files changed, 10 insertions(+), 301 deletions(-) delete mode 100644 src/sync/mutex.rs diff --git a/Cargo.toml b/Cargo.toml index 933023341..e98700539 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,6 +48,7 @@ std = [ "slab", "wasm-bindgen-futures", "futures-channel", + "async-mutex", ] alloc = [ "futures-core/alloc", @@ -58,6 +59,7 @@ tokio02 = ["smol/tokio02"] [dependencies] async-attributes = { version = "1.1.1", optional = true } async-task = { version = "3.0.0", optional = true } +async-mutex = { version = "1.1.3", optional = true } crossbeam-utils = { version = "0.7.2", optional = true } futures-core = { version = "0.3.4", optional = true, default-features = false } futures-io = { version = "0.3.4", optional = true } @@ -75,7 +77,7 @@ futures-timer = { version = "3.0.2", optional = true } surf = { version = "1.0.3", optional = true } [target.'cfg(not(target_os = "unknown"))'.dependencies] -smol = { version = "0.1.14", optional = true } +smol = { version = "0.1.17", optional = true } [target.'cfg(target_arch = "wasm32")'.dependencies] futures-timer = { version = "3.0.2", optional = true, features = ["wasm-bindgen"] } diff --git a/src/sync/condvar.rs b/src/sync/condvar.rs index 67507f384..09aea3a1a 100644 --- a/src/sync/condvar.rs +++ b/src/sync/condvar.rs @@ -2,7 +2,7 @@ use std::fmt; use std::pin::Pin; use std::time::Duration; -use super::mutex::{guard_lock, MutexGuard}; +use super::MutexGuard; use crate::future::{timeout, Future}; use crate::sync::WakerSet; use crate::task::{Context, Poll}; @@ -120,7 +120,7 @@ impl Condvar { /// ``` #[allow(clippy::needless_lifetimes)] pub async fn wait<'a, T>(&self, guard: MutexGuard<'a, T>) -> MutexGuard<'a, T> { - let mutex = guard_lock(&guard); + let mutex = MutexGuard::source(&guard); self.await_notify(guard).await; @@ -228,7 +228,7 @@ impl Condvar { guard: MutexGuard<'a, T>, dur: Duration, ) -> (MutexGuard<'a, T>, WaitTimeoutResult) { - let mutex = guard_lock(&guard); + let mutex = MutexGuard::source(&guard); match timeout(dur, self.wait(guard)).await { Ok(guard) => (guard, WaitTimeoutResult(false)), Err(_) => (mutex.lock().await, WaitTimeoutResult(true)), @@ -281,7 +281,7 @@ impl Condvar { where F: FnMut(&mut T) -> bool, { - let mutex = guard_lock(&guard); + let mutex = MutexGuard::source(&guard); match timeout(dur, self.wait_until(guard, condition)).await { Ok(guard) => (guard, WaitTimeoutResult(false)), Err(_) => (mutex.lock().await, WaitTimeoutResult(true)), diff --git a/src/sync/mod.rs b/src/sync/mod.rs index bccc6ec87..8b7fe3102 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -176,10 +176,11 @@ #[doc(inline)] pub use std::sync::{Arc, Weak}; -pub use mutex::{Mutex, MutexGuard}; +#[doc(inline)] +pub use async_mutex::{Mutex, MutexGuard}; + pub use rwlock::{RwLock, RwLockReadGuard, RwLockWriteGuard}; -mod mutex; mod rwlock; cfg_unstable! { diff --git a/src/sync/mutex.rs b/src/sync/mutex.rs deleted file mode 100644 index ae953fd82..000000000 --- a/src/sync/mutex.rs +++ /dev/null @@ -1,294 +0,0 @@ -use std::cell::UnsafeCell; -use std::fmt; -use std::ops::{Deref, DerefMut}; -use std::pin::Pin; -use std::sync::atomic::{AtomicBool, Ordering}; -use std::future::Future; - -use crate::sync::WakerSet; -use crate::task::{Context, Poll}; - -/// A mutual exclusion primitive for protecting shared data. -/// -/// This type is an async version of [`std::sync::Mutex`]. -/// -/// [`std::sync::Mutex`]: https://doc.rust-lang.org/std/sync/struct.Mutex.html -/// -/// # Examples -/// -/// ``` -/// # async_std::task::block_on(async { -/// # -/// use async_std::sync::{Arc, Mutex}; -/// use async_std::task; -/// -/// let m = Arc::new(Mutex::new(0)); -/// let mut tasks = vec![]; -/// -/// for _ in 0..10 { -/// let m = m.clone(); -/// tasks.push(task::spawn(async move { -/// *m.lock().await += 1; -/// })); -/// } -/// -/// for t in tasks { -/// t.await; -/// } -/// assert_eq!(*m.lock().await, 10); -/// # -/// # }) -/// ``` -pub struct Mutex { - locked: AtomicBool, - wakers: WakerSet, - value: UnsafeCell, -} - -unsafe impl Send for Mutex {} -unsafe impl Sync for Mutex {} - -impl Mutex { - /// Creates a new mutex. - /// - /// # Examples - /// - /// ``` - /// use async_std::sync::Mutex; - /// - /// let mutex = Mutex::new(0); - /// ``` - pub fn new(t: T) -> Mutex { - Mutex { - locked: AtomicBool::new(false), - wakers: WakerSet::new(), - value: UnsafeCell::new(t), - } - } -} - -impl Mutex { - /// Acquires the lock. - /// - /// Returns a guard that releases the lock when dropped. - /// - /// # Examples - /// - /// ``` - /// # async_std::task::block_on(async { - /// # - /// use async_std::sync::{Arc, Mutex}; - /// use async_std::task; - /// - /// let m1 = Arc::new(Mutex::new(10)); - /// let m2 = m1.clone(); - /// - /// task::spawn(async move { - /// *m1.lock().await = 20; - /// }) - /// .await; - /// - /// assert_eq!(*m2.lock().await, 20); - /// # - /// # }) - /// ``` - pub async fn lock(&self) -> MutexGuard<'_, T> { - pub struct LockFuture<'a, T: ?Sized> { - mutex: &'a Mutex, - opt_key: Option, - } - - impl<'a, T: ?Sized> Future for LockFuture<'a, T> { - type Output = MutexGuard<'a, T>; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - loop { - // If the current task is in the set, remove it. - if let Some(key) = self.opt_key.take() { - self.mutex.wakers.remove(key); - } - - // Try acquiring the lock. - match self.mutex.try_lock() { - Some(guard) => return Poll::Ready(guard), - None => { - // Insert this lock operation. - self.opt_key = Some(self.mutex.wakers.insert(cx)); - - // If the mutex is still locked, return. - if self.mutex.locked.load(Ordering::SeqCst) { - return Poll::Pending; - } - } - } - } - } - } - - impl Drop for LockFuture<'_, T> { - fn drop(&mut self) { - // If the current task is still in the set, that means it is being cancelled now. - if let Some(key) = self.opt_key { - self.mutex.wakers.cancel(key); - } - } - } - - LockFuture { - mutex: self, - opt_key: None, - } - .await - } - - /// Attempts to acquire the lock. - /// - /// If the lock could not be acquired at this time, then [`None`] is returned. Otherwise, a - /// guard is returned that releases the lock when dropped. - /// - /// [`None`]: https://doc.rust-lang.org/std/option/enum.Option.html#variant.None - /// - /// # Examples - /// - /// ``` - /// # async_std::task::block_on(async { - /// # - /// use async_std::sync::{Arc, Mutex}; - /// use async_std::task; - /// - /// let m1 = Arc::new(Mutex::new(10)); - /// let m2 = m1.clone(); - /// - /// task::spawn(async move { - /// if let Some(mut guard) = m1.try_lock() { - /// *guard = 20; - /// } else { - /// println!("try_lock failed"); - /// } - /// }) - /// .await; - /// - /// assert_eq!(*m2.lock().await, 20); - /// # - /// # }) - /// ``` - #[inline] - pub fn try_lock(&self) -> Option> { - if !self.locked.swap(true, Ordering::SeqCst) { - Some(MutexGuard(self)) - } else { - None - } - } - - /// Consumes the mutex, returning the underlying data. - /// - /// # Examples - /// - /// ``` - /// use async_std::sync::Mutex; - /// - /// let mutex = Mutex::new(10); - /// assert_eq!(mutex.into_inner(), 10); - /// ``` - pub fn into_inner(self) -> T where T: Sized { - self.value.into_inner() - } - - /// Returns a mutable reference to the underlying data. - /// - /// Since this call borrows the mutex mutably, no actual locking takes place -- the mutable - /// borrow statically guarantees no locks exist. - /// - /// # Examples - /// - /// ``` - /// # async_std::task::block_on(async { - /// # - /// use async_std::sync::Mutex; - /// - /// let mut mutex = Mutex::new(0); - /// *mutex.get_mut() = 10; - /// assert_eq!(*mutex.lock().await, 10); - /// # - /// # }) - /// ``` - pub fn get_mut(&mut self) -> &mut T { - unsafe { &mut *self.value.get() } - } -} - -impl fmt::Debug for Mutex { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - struct Locked; - impl fmt::Debug for Locked { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.write_str("") - } - } - - match self.try_lock() { - None => f.debug_struct("Mutex").field("data", &Locked).finish(), - Some(guard) => f.debug_struct("Mutex").field("data", &&*guard).finish(), - } - } -} - -impl From for Mutex { - fn from(val: T) -> Mutex { - Mutex::new(val) - } -} - -impl Default for Mutex { - fn default() -> Mutex { - Mutex::new(Default::default()) - } -} - -/// A guard that releases the lock when dropped. -pub struct MutexGuard<'a, T: ?Sized>(&'a Mutex); - -unsafe impl Send for MutexGuard<'_, T> {} -unsafe impl Sync for MutexGuard<'_, T> {} - -impl Drop for MutexGuard<'_, T> { - fn drop(&mut self) { - // Use `SeqCst` ordering to synchronize with `WakerSet::insert()` and `WakerSet::update()`. - self.0.locked.store(false, Ordering::SeqCst); - - // Notify a blocked `lock()` operation if none were notified already. - self.0.wakers.notify_any(); - } -} - -impl fmt::Debug for MutexGuard<'_, T> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt::Debug::fmt(&**self, f) - } -} - -impl fmt::Display for MutexGuard<'_, T> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - (**self).fmt(f) - } -} - -impl Deref for MutexGuard<'_, T> { - type Target = T; - - fn deref(&self) -> &T { - unsafe { &*self.0.value.get() } - } -} - -impl DerefMut for MutexGuard<'_, T> { - fn deref_mut(&mut self) -> &mut T { - unsafe { &mut *self.0.value.get() } - } -} - -#[cfg(feature = "unstable")] -pub fn guard_lock<'a, T>(guard: &MutexGuard<'a, T>) -> &'a Mutex { - guard.0 -} From 8f17e9275ba3f6470f5dad8945d97870602a79f1 Mon Sep 17 00:00:00 2001 From: dignifiedquire Date: Fri, 26 Jun 2020 12:48:23 +0200 Subject: [PATCH 2/2] test: try to stabilize CI --- tests/timeout.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/timeout.rs b/tests/timeout.rs index e09acdfe4..60594d06b 100644 --- a/tests/timeout.rs +++ b/tests/timeout.rs @@ -10,9 +10,9 @@ wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser); #[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)] fn timeout_future_many() { task::block_on(async { - let futures = (0..100) + let futures = (0..10) .map(|i| { - timeout(Duration::from_millis(i * 20), async move { + timeout(Duration::from_millis(i * 50), async move { task::sleep(Duration::from_millis(i)).await; Ok::<(), async_std::future::TimeoutError>(()) })