diff --git a/tokio-sync/src/lib.rs b/tokio-sync/src/lib.rs index 28d1ce748..868ed71a9 100644 --- a/tokio-sync/src/lib.rs +++ b/tokio-sync/src/lib.rs @@ -29,13 +29,13 @@ macro_rules! if_fuzz { }} } -mod lock; mod loom; pub mod mpsc; +mod mutex; pub mod oneshot; pub mod semaphore; mod task; pub mod watch; -pub use lock::{Lock, LockGuard}; +pub use mutex::{Mutex, MutexGuard}; pub use task::AtomicWaker; diff --git a/tokio-sync/src/lock.rs b/tokio-sync/src/lock.rs deleted file mode 100644 index c17855bc7..000000000 --- a/tokio-sync/src/lock.rs +++ /dev/null @@ -1,173 +0,0 @@ -//! An asynchronous `Mutex`-like type. -//! -//! This module provides [`Lock`], a type that acts similarly to an asynchronous `Mutex`, with one -//! major difference: the [`LockGuard`] returned by `lock` is not tied to the lifetime of the -//! `Mutex`. This enables you to acquire a lock, and then pass that guard into a future, and then -//! release it at some later point in time. -//! -//! This allows you to do something along the lines of: -//! -//! ```rust,no_run -//! use tokio::sync::Lock; -//! -//! #[tokio::main] -//! async fn main() { -//! let mut data1 = Lock::new(0); -//! let mut data2 = data1.clone(); -//! -//! tokio::spawn(async move { -//! let mut lock = data2.lock().await; -//! *lock += 1; -//! }); -//! -//! let mut lock = data1.lock().await; -//! *lock += 1; -//! } -//! ``` -//! -//! [`Lock`]: struct.Lock.html -//! [`LockGuard`]: struct.LockGuard.html - -use crate::semaphore; - -use futures_core::ready; -use futures_util::future::poll_fn; -use std::cell::UnsafeCell; -use std::fmt; -use std::ops::{Deref, DerefMut}; -use std::sync::Arc; -use std::task::Poll::Ready; -use std::task::{Context, Poll}; - -/// An asynchronous mutual exclusion primitive useful for protecting shared data -/// -/// Each mutex has a type parameter (`T`) which represents the data that it is protecting. The data -/// can only be accessed through the RAII guards returned from `lock`, which -/// guarantees that the data is only ever accessed when the mutex is locked. -#[derive(Debug)] -pub struct Lock { - inner: Arc>, - permit: semaphore::Permit, -} - -/// A handle to a held `Lock`. -/// -/// As long as you have this guard, you have exclusive access to the underlying `T`. The guard -/// internally keeps a reference-couned pointer to the original `Lock`, so even if the lock goes -/// away, the guard remains valid. -/// -/// The lock is automatically released whenever the guard is dropped, at which point `lock` -/// will succeed yet again. -#[derive(Debug)] -pub struct LockGuard(Lock); - -// As long as T: Send, it's fine to send and share Lock between threads. -// If T was not Send, sending and sharing a Lock would be bad, since you can access T through -// Lock. -unsafe impl Send for Lock where T: Send {} -unsafe impl Sync for Lock where T: Send {} -unsafe impl Sync for LockGuard where T: Send + Sync {} - -#[derive(Debug)] -struct State { - c: UnsafeCell, - s: semaphore::Semaphore, -} - -#[test] -fn bounds() { - fn check() {} - check::>(); -} - -impl Lock { - /// Creates a new lock in an unlocked state ready for use. - pub fn new(t: T) -> Self { - Self { - inner: Arc::new(State { - c: UnsafeCell::new(t), - s: semaphore::Semaphore::new(1), - }), - permit: semaphore::Permit::new(), - } - } - - fn poll_lock(&mut self, cx: &mut Context<'_>) -> Poll> { - ready!(self.permit.poll_acquire(cx, &self.inner.s)).unwrap_or_else(|_| { - // The semaphore was closed. but, we never explicitly close it, and we have a - // handle to it through the Arc, which means that this can never happen. - unreachable!() - }); - - // We want to move the acquired permit into the guard, - // and leave an unacquired one in self. - let acquired = Self { - inner: self.inner.clone(), - permit: ::std::mem::replace(&mut self.permit, semaphore::Permit::new()), - }; - Ready(LockGuard(acquired)) - } - - /// A future that resolves on acquiring the lock and returns the `LockGuard`. - pub async fn lock(&mut self) -> LockGuard { - poll_fn(|cx| self.poll_lock(cx)).await - } -} - -impl Drop for LockGuard { - fn drop(&mut self) { - if self.0.permit.is_acquired() { - self.0.permit.release(&self.0.inner.s); - } else if ::std::thread::panicking() { - // A guard _should_ always hold its permit, but if the thread is already panicking, - // we don't want to generate a panic-while-panicing, since that's just unhelpful! - } else { - unreachable!("Permit not held when LockGuard was dropped") - } - } -} - -impl From for Lock { - fn from(s: T) -> Self { - Self::new(s) - } -} - -impl Clone for Lock { - fn clone(&self) -> Self { - Self { - inner: self.inner.clone(), - permit: semaphore::Permit::new(), - } - } -} - -impl Default for Lock -where - T: Default, -{ - fn default() -> Self { - Self::new(T::default()) - } -} - -impl Deref for LockGuard { - type Target = T; - fn deref(&self) -> &Self::Target { - assert!(self.0.permit.is_acquired()); - unsafe { &*self.0.inner.c.get() } - } -} - -impl DerefMut for LockGuard { - fn deref_mut(&mut self) -> &mut Self::Target { - assert!(self.0.permit.is_acquired()); - unsafe { &mut *self.0.inner.c.get() } - } -} - -impl fmt::Display for LockGuard { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt::Display::fmt(&**self, f) - } -} diff --git a/tokio-sync/src/mutex.rs b/tokio-sync/src/mutex.rs new file mode 100644 index 000000000..c0c75776d --- /dev/null +++ b/tokio-sync/src/mutex.rs @@ -0,0 +1,148 @@ +//! An asynchronous `Mutex`-like type. +//! +//! This module provides [`Mutex`], a type that acts similarly to an asynchronous `Mutex`, with one +//! major difference: the [`MutexGuard`] returned by `lock` is not tied to the lifetime of the +//! `Mutex`. This enables you to acquire a lock, and then pass that guard into a future, and then +//! release it at some later point in time. +//! +//! This allows you to do something along the lines of: +//! +//! ```rust,no_run +//! use tokio::sync::Mutex; +//! use std::sync::Arc; +//! +//! #[tokio::main] +//! async fn main() { +//! let data1 = Arc::new(Mutex::new(0)); +//! let data2 = Arc::clone(&data1); +//! +//! tokio::spawn(async move { +//! let mut lock = data2.lock().await; +//! *lock += 1; +//! }); +//! +//! let mut lock = data1.lock().await; +//! *lock += 1; +//! } +//! ``` +//! +//! [`Mutex`]: struct.Mutex.html +//! [`MutexGuard`]: struct.MutexGuard.html + +use crate::semaphore; + +use futures_util::future::poll_fn; +use std::cell::UnsafeCell; +use std::fmt; +use std::ops::{Deref, DerefMut}; + +/// An asynchronous mutual exclusion primitive useful for protecting shared data +/// +/// Each mutex has a type parameter (`T`) which represents the data that it is protecting. The data +/// can only be accessed through the RAII guards returned from `lock`, which +/// guarantees that the data is only ever accessed when the mutex is locked. +#[derive(Debug)] +pub struct Mutex { + c: UnsafeCell, + s: semaphore::Semaphore, +} + +/// A handle to a held `Mutex`. +/// +/// As long as you have this guard, you have exclusive access to the underlying `T`. The guard +/// internally keeps a reference-couned pointer to the original `Mutex`, so even if the lock goes +/// away, the guard remains valid. +/// +/// The lock is automatically released whenever the guard is dropped, at which point `lock` +/// will succeed yet again. +#[derive(Debug)] +pub struct MutexGuard<'a, T> { + lock: &'a Mutex, + permit: semaphore::Permit, +} + +// As long as T: Send, it's fine to send and share Mutex between threads. +// If T was not Send, sending and sharing a Mutex would be bad, since you can access T through +// Mutex. +unsafe impl Send for Mutex where T: Send {} +unsafe impl Sync for Mutex where T: Send {} +unsafe impl<'a, T> Sync for MutexGuard<'a, T> where T: Send + Sync {} + +#[test] +fn bounds() { + fn check() {} + check::>(); +} + +impl Mutex { + /// Creates a new lock in an unlocked state ready for use. + pub fn new(t: T) -> Self { + Self { + c: UnsafeCell::new(t), + s: semaphore::Semaphore::new(1), + } + } + + /// A future that resolves on acquiring the lock and returns the `MutexGuard`. + pub async fn lock(&self) -> MutexGuard<'_, T> { + let mut permit = semaphore::Permit::new(); + poll_fn(|cx| permit.poll_acquire(cx, &self.s)) + .await + .unwrap_or_else(|_| { + // The semaphore was closed. but, we never explicitly close it, and we have a + // handle to it through the Arc, which means that this can never happen. + unreachable!() + }); + + MutexGuard { lock: self, permit } + } +} + +impl<'a, T> Drop for MutexGuard<'a, T> { + fn drop(&mut self) { + if self.permit.is_acquired() { + self.permit.release(&self.lock.s); + } else if ::std::thread::panicking() { + // A guard _should_ always hold its permit, but if the thread is already panicking, + // we don't want to generate a panic-while-panicing, since that's just unhelpful! + } else { + unreachable!("Permit not held when MutexGuard was dropped") + } + } +} + +impl From for Mutex { + fn from(s: T) -> Self { + Self::new(s) + } +} + +impl Default for Mutex +where + T: Default, +{ + fn default() -> Self { + Self::new(T::default()) + } +} + +impl<'a, T> Deref for MutexGuard<'a, T> { + type Target = T; + fn deref(&self) -> &Self::Target { + assert!(self.permit.is_acquired()); + unsafe { &*self.lock.c.get() } + } +} + +impl<'a, T> DerefMut for MutexGuard<'a, T> { + fn deref_mut(&mut self) -> &mut Self::Target { + assert!(self.permit.is_acquired()); + unsafe { &mut *self.lock.c.get() } + } +} + +impl<'a, T: fmt::Display> fmt::Display for MutexGuard<'a, T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&**self, f) + } +} diff --git a/tokio-sync/tests/lock.rs b/tokio-sync/tests/mutex.rs similarity index 81% rename from tokio-sync/tests/lock.rs rename to tokio-sync/tests/mutex.rs index 65ecf142f..47fcb449d 100644 --- a/tokio-sync/tests/lock.rs +++ b/tokio-sync/tests/mutex.rs @@ -1,12 +1,13 @@ #![warn(rust_2018_idioms)] -use tokio_sync::Lock; +use std::sync::Arc; +use tokio_sync::Mutex; use tokio_test::task::spawn; use tokio_test::{assert_pending, assert_ready}; #[test] fn straight_execution() { - let mut l = Lock::new(100); + let l = Mutex::new(100); { let mut t = spawn(l.lock()); @@ -22,21 +23,15 @@ fn straight_execution() { } { let mut t = spawn(l.lock()); - let mut g = assert_ready!(t.poll()); + let g = assert_ready!(t.poll()); assert_eq!(&*g, &98); - - // We can continue to access the guard even if the lock is dropped - drop(t); - drop(l); - *g = 97; - assert_eq!(&*g, &97); } } #[test] fn readiness() { - let mut l1 = Lock::new(100); - let mut l2 = l1.clone(); + let l1 = Arc::new(Mutex::new(100)); + let l2 = Arc::clone(&l1); let mut t1 = spawn(l1.lock()); let mut t2 = spawn(l2.lock()); @@ -55,7 +50,7 @@ fn readiness() { #[test] #[ignore] fn lock() { - let mut lock = Lock::new(false); + let mut lock = Mutex::new(false); let mut lock2 = lock.clone(); std::thread::spawn(move || { diff --git a/tokio/examples/chat.rs b/tokio/examples/chat.rs index 719a25309..7fc317d84 100644 --- a/tokio/examples/chat.rs +++ b/tokio/examples/chat.rs @@ -27,12 +27,15 @@ #![warn(rust_2018_idioms)] use futures::{Poll, SinkExt, Stream, StreamExt}; -use std::{collections::HashMap, env, error::Error, io, net::SocketAddr, pin::Pin, task::Context}; +use std::{ + collections::HashMap, env, error::Error, io, net::SocketAddr, pin::Pin, sync::Arc, + task::Context, +}; use tokio::{ self, codec::{Framed, LinesCodec, LinesCodecError}, net::{TcpListener, TcpStream}, - sync::{mpsc, Lock}, + sync::{mpsc, Mutex}, }; #[tokio::main] @@ -42,7 +45,7 @@ async fn main() -> Result<(), Box> { // The server task will hold a handle to this. For every new client, the // `state` handle is cloned and passed into the task that processes the // client connection. - let state = Lock::new(Shared::new()); + let state = Arc::new(Mutex::new(Shared::new())); let addr = env::args().nth(1).unwrap_or("127.0.0.1:6142".to_string()); @@ -58,7 +61,7 @@ async fn main() -> Result<(), Box> { let (stream, addr) = listener.accept().await?; // Clone a handle to the `Shared` state for the new connection. - let state = state.clone(); + let state = Arc::clone(&state); // Spawn our handler to be run asynchronously. tokio::spawn(async move { @@ -129,7 +132,7 @@ impl Shared { impl Peer { /// Create a new instance of `Peer`. async fn new( - mut state: Lock, + state: Arc>, lines: Framed, ) -> io::Result { // Get the client socket address @@ -184,7 +187,7 @@ impl Stream for Peer { /// Process an individual chat client async fn process( - mut state: Lock, + state: Arc>, stream: TcpStream, addr: SocketAddr, ) -> Result<(), Box> { diff --git a/tokio/src/sync.rs b/tokio/src/sync.rs index d563d2147..30471325f 100644 --- a/tokio/src/sync.rs +++ b/tokio/src/sync.rs @@ -9,9 +9,9 @@ //! from one task to another. //! - [mpsc](mpsc/index.html), a multi-producer, single-consumer channel for //! sending values between tasks. -//! - [lock](lock/index.html), an asynchronous `Mutex`-like type. +//! - [`Mutex`](struct.Mutex.html), an asynchronous `Mutex`-like type. //! - [watch](watch/index.html), a single-producer, multi-consumer channel that //! only stores the **most recently** sent value. pub use tokio_sync::{mpsc, oneshot, watch}; -pub use tokio_sync::{Lock, LockGuard}; +pub use tokio_sync::{Mutex, MutexGuard};