|
| 1 | +//! An asynchronous mutual exclusion primitive. |
| 2 | +
|
| 3 | +// This implementation is adapted from the one in shuttle-tokio |
| 4 | + |
| 5 | +use shuttle::future::batch_semaphore::{BatchSemaphore, Fairness, TryAcquireError}; |
| 6 | +use std::cell::UnsafeCell; |
| 7 | +use std::error::Error; |
| 8 | +use std::fmt::{self, Debug, Display}; |
| 9 | +use std::ops::{Deref, DerefMut}; |
| 10 | +use std::thread; |
| 11 | +use tracing::trace; |
| 12 | + |
| 13 | +/// An asynchronous mutex |
| 14 | +pub struct Mutex<T: ?Sized> { |
| 15 | + semaphore: BatchSemaphore, |
| 16 | + inner: UnsafeCell<T>, |
| 17 | +} |
| 18 | + |
| 19 | +/// A handle to a held `Mutex`. The guard can be held across any `.await` point |
| 20 | +/// as it is [`Send`]. |
| 21 | +pub struct MutexGuard<'a, T: ?Sized> { |
| 22 | + mutex: &'a Mutex<T>, |
| 23 | +} |
| 24 | + |
| 25 | +impl<T: ?Sized + Display> Display for MutexGuard<'_, T> { |
| 26 | + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { |
| 27 | + Display::fmt(&**self, f) |
| 28 | + } |
| 29 | +} |
| 30 | + |
| 31 | +/// Error returned from the [`Mutex::try_lock`], `RwLock::try_read` and |
| 32 | +/// `RwLock::try_write` functions. |
| 33 | +#[derive(Debug)] |
| 34 | +pub struct TryLockError(pub(super) ()); |
| 35 | + |
| 36 | +impl Display for TryLockError { |
| 37 | + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { |
| 38 | + write!(fmt, "operation would block") |
| 39 | + } |
| 40 | +} |
| 41 | + |
| 42 | +impl Error for TryLockError {} |
| 43 | + |
| 44 | +// As long as T: Send, it's fine to send and share Mutex<T> between threads. |
| 45 | +// If T was not Send, sending and sharing a Mutex<T> would be bad, since you can |
| 46 | +// access T through Mutex<T>. |
| 47 | +unsafe impl<T> Send for Mutex<T> where T: ?Sized + Send {} |
| 48 | +unsafe impl<T> Sync for Mutex<T> where T: ?Sized + Send {} |
| 49 | +unsafe impl<T> Sync for MutexGuard<'_, T> where T: ?Sized + Send + Sync {} |
| 50 | + |
| 51 | +impl<T> Mutex<T> { |
| 52 | + /// Creates a new lock in an unlocked state ready for use. |
| 53 | + pub const fn new(t: T) -> Self { |
| 54 | + Self { |
| 55 | + semaphore: BatchSemaphore::const_new(1, Fairness::StrictlyFair), |
| 56 | + inner: UnsafeCell::new(t), |
| 57 | + } |
| 58 | + } |
| 59 | + |
| 60 | + /// Consumes the mutex, returning the underlying data. |
| 61 | + pub fn into_inner(self) -> T { |
| 62 | + self.inner.into_inner() |
| 63 | + } |
| 64 | +} |
| 65 | + |
| 66 | +impl<T: ?Sized> Mutex<T> { |
| 67 | + fn acquire(&self) { |
| 68 | + trace!("acquiring parking_lot lock {:p}", self); |
| 69 | + self.semaphore.acquire_blocking(1).unwrap_or_else(|_| { |
| 70 | + // The semaphore was closed. but, we never explicitly close it, and |
| 71 | + // we own it exclusively, which means that this can never happen. |
| 72 | + if !thread::panicking() { |
| 73 | + unreachable!() |
| 74 | + } |
| 75 | + }); |
| 76 | + trace!("acquired parking_lot lock {:p}", self); |
| 77 | + } |
| 78 | + |
| 79 | + /// Acquires a mutex, blocking the current thread until it is able to do so. |
| 80 | + pub fn lock(&self) -> MutexGuard<'_, T> { |
| 81 | + self.acquire(); |
| 82 | + |
| 83 | + MutexGuard { mutex: self } |
| 84 | + } |
| 85 | + |
| 86 | + fn try_acquire(&self) -> Result<(), TryAcquireError> { |
| 87 | + self.semaphore.try_acquire(1) |
| 88 | + } |
| 89 | + |
| 90 | + /// Attempts to acquire this lock. |
| 91 | + pub fn try_lock(&self) -> Option<MutexGuard<'_, T>> { |
| 92 | + match self.try_acquire() { |
| 93 | + Ok(()) => Some(MutexGuard { mutex: self }), |
| 94 | + Err(_) => None, |
| 95 | + } |
| 96 | + } |
| 97 | +} |
| 98 | + |
| 99 | +impl<T: ?Sized + Debug> Debug for Mutex<T> { |
| 100 | + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 101 | + // SAFETY: Shuttle is running single-threaded, only we are able to access `inner` at the time of this call. |
| 102 | + Debug::fmt(&unsafe { &*self.inner.get() }, f) |
| 103 | + } |
| 104 | +} |
| 105 | + |
| 106 | +impl<T: ?Sized> Deref for MutexGuard<'_, T> { |
| 107 | + type Target = T; |
| 108 | + |
| 109 | + fn deref(&self) -> &Self::Target { |
| 110 | + unsafe { &*self.mutex.inner.get() } |
| 111 | + } |
| 112 | +} |
| 113 | + |
| 114 | +impl<T: ?Sized> DerefMut for MutexGuard<'_, T> { |
| 115 | + fn deref_mut(&mut self) -> &mut Self::Target { |
| 116 | + unsafe { &mut *self.mutex.inner.get() } |
| 117 | + } |
| 118 | +} |
| 119 | + |
| 120 | +impl<T: ?Sized> Drop for MutexGuard<'_, T> { |
| 121 | + fn drop(&mut self) { |
| 122 | + trace!("releasing lock {:p}", self); |
| 123 | + self.mutex.semaphore.release(1); |
| 124 | + } |
| 125 | +} |
| 126 | + |
| 127 | +impl<T: ?Sized + Debug> Debug for MutexGuard<'_, T> { |
| 128 | + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { |
| 129 | + Debug::fmt(&self.mutex, f) |
| 130 | + } |
| 131 | +} |
| 132 | + |
| 133 | +impl<T> From<T> for Mutex<T> { |
| 134 | + fn from(s: T) -> Self { |
| 135 | + Self::new(s) |
| 136 | + } |
| 137 | +} |
| 138 | + |
| 139 | +impl<T> Default for Mutex<T> |
| 140 | +where |
| 141 | + T: Default, |
| 142 | +{ |
| 143 | + fn default() -> Self { |
| 144 | + Self::new(T::default()) |
| 145 | + } |
| 146 | +} |
0 commit comments