tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-15 00:00:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

sync: Make Lock more similar to std::sync::Mutex (#1573)

Browse Source 
This renames `Lock` to `Mutex`, and brings the API more in line with `std::sync::Mutex`.

In partcular, locking now only takes `&self`, with the expectation that you place the `Mutex` in an `Arc` (or something similar) to share it between threads.

Fixes #1544.
Part of #1210.
This commit is contained in:
Jon Gjengset
2019-09-19 11:46:52 -04:00
committed by GitHub
parent d1f60ac4c6
commit e3415d8d61
6 changed files with 168 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-sync/src/lib.rs
+2 -2
View File
@@ -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;
tokio-sync/src/lock.rs
-173
View File
@@ -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<T> {
inner: Arc<State<T>>,
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<T>(Lock<T>);
// As long as T: Send, it's fine to send and share Lock<T> between threads.
// If T was not Send, sending and sharing a Lock<T> would be bad, since you can access T through
// Lock<T>.
unsafe impl<T> Send for Lock<T> where T: Send {}
unsafe impl<T> Sync for Lock<T> where T: Send {}
unsafe impl<T> Sync for LockGuard<T> where T: Send + Sync {}
#[derive(Debug)]
struct State<T> {
c: UnsafeCell<T>,
s: semaphore::Semaphore,
}
#[test]
fn bounds() {
fn check<T: Send>() {}
check::<LockGuard<u32>>();
}
impl<T> Lock<T> {
/// 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<LockGuard<T>> {
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<T> {
poll_fn(|cx| self.poll_lock(cx)).await
}
}
impl<T> Drop for LockGuard<T> {
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<T> From<T> for Lock<T> {
fn from(s: T) -> Self {
Self::new(s)
}
}
impl<T> Clone for Lock<T> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
permit: semaphore::Permit::new(),
}
}
}
impl<T> Default for Lock<T>
where
T: Default,
{
fn default() -> Self {
Self::new(T::default())
}
}
impl<T> Deref for LockGuard<T> {
type Target = T;
fn deref(&self) -> &Self::Target {
assert!(self.0.permit.is_acquired());
unsafe { &*self.0.inner.c.get() }
}
}
impl<T> DerefMut for LockGuard<T> {
fn deref_mut(&mut self) -> &mut Self::Target {
assert!(self.0.permit.is_acquired());
unsafe { &mut *self.0.inner.c.get() }
}
}
impl<T: fmt::Display> fmt::Display for LockGuard<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt::Display::fmt(&**self, f)
}
}
tokio-sync/src/mutex.rs
+148
View File
@@ -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<T> {
c: UnsafeCell<T>,
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<T>,
permit: semaphore::Permit,
}
// As long as T: Send, it's fine to send and share Mutex<T> between threads.
// If T was not Send, sending and sharing a Mutex<T> would be bad, since you can access T through
// Mutex<T>.
unsafe impl<T> Send for Mutex<T> where T: Send {}
unsafe impl<T> Sync for Mutex<T> where T: Send {}
unsafe impl<'a, T> Sync for MutexGuard<'a, T> where T: Send + Sync {}
#[test]
fn bounds() {
fn check<T: Send>() {}
check::<MutexGuard<'_, u32>>();
}
impl<T> Mutex<T> {
/// 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<T> From<T> for Mutex<T> {
fn from(s: T) -> Self {
Self::new(s)
}
}
impl<T> Default for Mutex<T>
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)
}
}
tokio-sync/tests/lock.rs → tokio-sync/tests/mutex.rs
+7 -12
View File
@@ -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 || {
tokio/examples/chat.rs
+9 -6
View File
@@ -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<dyn Error>> {
// 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<dyn Error>> {
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<Shared>,
state: Arc<Mutex<Shared>>,
lines: Framed<TcpStream, LinesCodec>,
) -> io::Result<Peer> {
// Get the client socket address
@@ -184,7 +187,7 @@ impl Stream for Peer {
/// Process an individual chat client
async fn process(
mut state: Lock<Shared>,
state: Arc<Mutex<Shared>>,
stream: TcpStream,
addr: SocketAddr,
) -> Result<(), Box<dyn Error>> {
tokio/src/sync.rs
+2 -2
View File
@@ -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};