nuclei/sync/
rwlock.rs

1//! An asynchronous reader-writer lock with an upgradable read; how it
2//! admits waiters is described on its core, `raw_rwlock`.
3
4#![allow(unsafe_code)]
5
6use core::fmt;
7use core::future::Future;
8use core::ops::Deref;
9use core::pin::Pin;
10use core::task::{Context, Poll};
11use std::sync::Arc;
12
13use super::internal;
14use super::mutex::TryLockError;
15use super::raw_rwlock::{
16    MAX_READERS, READ, RawRwLock, RwWait, UPGRADABLE_READ, UpgradeWait, WRITE,
17};
18
19/// The most readers [`RwLock::with_max_readers`] accepts: `u32::MAX >> 3`,
20/// or the reader field of the lock where a word is too narrow for that.
21const MAX_READS: usize = if (u32::MAX >> 3) as usize <= MAX_READERS {
22    (u32::MAX >> 3) as usize
23} else {
24    MAX_READERS
25};
26
27/// A reader-writer lock whose waits are futures.
28///
29/// Uncontended, a read or write lock and its release are one atomic
30/// read-modify-write each and allocate nothing. Under contention a caller
31/// takes the lock whenever it is free to it, even past queued waiters, so
32/// a release never waits for a suspended task; a waiter passed over
33/// [`MAX_BYPASS`](super::MAX_BYPASS) times is handed the lock next, and
34/// nobody gets in ahead of it meanwhile, so neither readers nor writers
35/// starve.
36///
37/// [`upgradable_read`](Self::upgradable_read) takes a read that can later
38/// become the write without letting a writer in between. One upgradable
39/// read exists at a time; it shares the lock with plain readers, and plain
40/// read guards cannot upgrade. A write guard becomes a read guard, with no
41/// writer in between, through
42/// [`RwLockWriteGuard::downgrade`].
43///
44/// ```
45/// use nuclei::sync::RwLock;
46///
47/// let lock = RwLock::new(5);
48/// let a = lock.try_read().expect("free");
49/// let b = lock.try_read().expect("readers share");
50/// assert!(lock.try_write().is_err());
51/// drop((a, b));
52/// *lock.try_write().expect("free again") += 1;
53/// assert_eq!(lock.into_inner(), 6);
54/// ```
55pub struct RwLock<T: ?Sized> {
56    raw: RawRwLock,
57    value: core::cell::UnsafeCell<T>,
58}
59
60// SAFETY: the value moves with the lock.
61unsafe impl<T: ?Sized + Send> Send for RwLock<T> {}
62// SAFETY: readers share `&T` across threads and a writer gets `&mut T`
63// alone, the same contract as `std::sync::RwLock`.
64unsafe impl<T: ?Sized + Send + Sync> Sync for RwLock<T> {}
65
66impl<T> RwLock<T> {
67    loom_const_fn! {
68        /// An unlocked lock holding `value`.
69        pub fn new(value: T) -> Self {
70            Self { raw: RawRwLock::new(), value: core::cell::UnsafeCell::new(value) }
71        }
72    }
73
74    loom_const_fn! {
75        /// An unlocked lock holding `value`, usable in a `static`.
76        pub fn const_new(value: T) -> Self {
77            Self { raw: RawRwLock::new(), value: core::cell::UnsafeCell::new(value) }
78        }
79    }
80
81    loom_const_fn! {
82        /// An unlocked lock holding `value` that admits at most `max_reads`
83        /// readers at once: a reader that finds them all inside waits for
84        /// one to leave, as one held back by a writer does.
85        ///
86        /// # Panics
87        ///
88        /// When `max_reads` is `0` or is bigger than `u32::MAX >> 3`. (A
89        /// target whose word is narrower than 64 bits caps it lower, at the
90        /// readers its state word can count: `2^23 - 1` on 32 bits.)
91        pub fn with_max_readers(value: T, max_reads: u32) -> Self {
92            assert!(max_reads != 0, "a RwLock may not be created with 0 readers");
93            assert!(
94                max_reads as usize <= MAX_READS,
95                "a RwLock may not be created with more than u32::MAX >> 3 readers"
96            );
97            Self {
98                raw: RawRwLock::with_max_readers(max_reads as usize),
99                value: core::cell::UnsafeCell::new(value),
100            }
101        }
102    }
103
104    loom_const_fn! {
105        /// [`with_max_readers`](Self::with_max_readers), usable in a
106        /// `static`.
107        ///
108        /// # Panics
109        ///
110        /// As for [`with_max_readers`](Self::with_max_readers).
111        pub fn const_with_max_readers(value: T, max_reads: u32) -> Self {
112            assert!(max_reads != 0, "a RwLock may not be created with 0 readers");
113            assert!(
114                max_reads as usize <= MAX_READS,
115                "a RwLock may not be created with more than u32::MAX >> 3 readers"
116            );
117            Self {
118                raw: RawRwLock::with_max_readers(max_reads as usize),
119                value: core::cell::UnsafeCell::new(value),
120            }
121        }
122    }
123
124    loom_const_fn! {
125        /// An unlocked lock holding `value` whose contended acquires never
126        /// spin: they queue at once. For a lock its holders keep across long
127        /// waits of their own (I/O, say), where a spin could only burn the
128        /// caller's time.
129        #[cfg(test)]
130        pub(crate) fn new_without_spin(value: T) -> Self {
131            Self { raw: RawRwLock::without_spin(), value: core::cell::UnsafeCell::new(value) }
132        }
133    }
134
135    /// Consumes the lock and returns its value.
136    pub fn into_inner(self) -> T {
137        self.value.into_inner()
138    }
139}
140
141#[cfg(all(test, not(loom)))]
142impl<T: ?Sized> RwLock<T> {
143    /// The rounds a contended acquire of this lock would spin now.
144    pub(crate) fn spin_rounds(&self) -> u32 {
145        self.raw.spin_rounds()
146    }
147}
148
149impl<T: ?Sized> RwLock<T> {
150    /// Shares the lock if no writer holds it, no upgrade is under way, the
151    /// readers have room and no handoff is owed.
152    ///
153    /// # Errors
154    ///
155    /// [`TryLockError`] when any of those fails.
156    pub fn try_read(&self) -> Result<RwLockReadGuard<'_, T>, TryLockError> {
157        if self.raw.try_kind(READ) {
158            Ok(RwLockReadGuard::new(self))
159        } else {
160            Err(TryLockError(()))
161        }
162    }
163
164    /// [`try_read`](Self::try_read) through an `Arc`, so the guard can
165    /// outlive the borrow.
166    ///
167    /// # Errors
168    ///
169    /// [`TryLockError`] when the lock cannot be shared now.
170    pub fn try_read_owned(self: Arc<Self>) -> Result<OwnedRwLockReadGuard<T>, TryLockError> {
171        if self.raw.try_kind(READ) {
172            Ok(OwnedRwLockReadGuard::new(self))
173        } else {
174            Err(TryLockError(()))
175        }
176    }
177
178    /// Takes the lock alone if nobody holds it and no handoff is owed.
179    ///
180    /// # Errors
181    ///
182    /// [`TryLockError`] when any reader or writer holds the lock, or a
183    /// waiter is owed it.
184    pub fn try_write(&self) -> Result<RwLockWriteGuard<'_, T>, TryLockError> {
185        if self.raw.try_kind(WRITE) {
186            Ok(RwLockWriteGuard::new(self))
187        } else {
188            Err(TryLockError(()))
189        }
190    }
191
192    /// [`try_write`](Self::try_write) through an `Arc`, so the guard can
193    /// outlive the borrow.
194    ///
195    /// # Errors
196    ///
197    /// [`TryLockError`] when the lock cannot be taken alone now.
198    pub fn try_write_owned(self: Arc<Self>) -> Result<OwnedRwLockWriteGuard<T>, TryLockError> {
199        if self.raw.try_kind(WRITE) {
200            Ok(OwnedRwLockWriteGuard::new(self))
201        } else {
202            Err(TryLockError(()))
203        }
204    }
205
206    /// Takes the upgradable read if no writer and no other upgradable read
207    /// holds the lock and no handoff is owed.
208    pub fn try_upgradable_read(&self) -> Option<RwLockUpgradableReadGuard<'_, T>> {
209        self.raw
210            .try_kind(UPGRADABLE_READ)
211            .then(|| RwLockUpgradableReadGuard { lock: self })
212    }
213
214    /// Waits to share the lock. Uncontended, the first poll is one atomic
215    /// read-modify-write. Contended, it is lock-free, not wait-free: see
216    /// [Progress](super#progress).
217    ///
218    /// Dropping the future withdraws the wait, and passes on a lock handed
219    /// to it meanwhile.
220    pub fn read(&self) -> Read<'_, T> {
221        Read {
222            lock: self,
223            wait: RwWait::new(),
224        }
225    }
226
227    /// [`read`](Self::read) through an `Arc`, so the guard can outlive the
228    /// borrow.
229    pub fn read_owned(self: Arc<Self>) -> ReadOwned<T> {
230        ReadOwned {
231            lock: self,
232            wait: RwWait::new(),
233        }
234    }
235
236    /// Waits to hold the lock alone. Uncontended, the first poll is one atomic
237    /// read-modify-write. Contended, it is lock-free, not wait-free: see
238    /// [Progress](super#progress).
239    ///
240    /// Dropping the future withdraws the wait, and passes on a lock handed
241    /// to it meanwhile.
242    pub fn write(&self) -> Write<'_, T> {
243        Write {
244            lock: self,
245            wait: RwWait::new(),
246        }
247    }
248
249    /// Waits to hold the lock alone, with a guard that keeps the lock alive
250    /// itself, so the future and the guard can outlive the borrow that made
251    /// them: a wait kept across calls, or a guard returned to a caller.
252    pub fn write_owned(self: Arc<Self>) -> WriteOwned<T> {
253        WriteOwned {
254            lock: self,
255            wait: RwWait::new(),
256        }
257    }
258
259    /// [`read`](Self::read) for a thread that may block: it parks the
260    /// calling thread until the lock can be shared. On a nuclei shard (inside one of its tasks) it does not park the thread: it runs a nested `block_on`, so the shard keeps running its other tasks meanwhile.
261    /// Inside another runtime's task it blocks that task's executor thread,
262    /// so do not: the task that would release may be the one that thread was
263    /// running.
264    ///
265    /// # Panics
266    ///
267    /// On `wasm32`, where no other thread can release, when the lock cannot
268    /// be shared at once: the call could never end.
269    pub fn blocking_read(&self) -> RwLockReadGuard<'_, T> {
270        internal::blocking(self.read())
271    }
272
273    /// [`write`](Self::write) for a thread that may block: see
274    /// [`blocking_read`](Self::blocking_read).
275    ///
276    /// # Panics
277    ///
278    /// On `wasm32`, where no other thread can release, when the lock is held
279    /// at all: the call could never end.
280    pub fn blocking_write(&self) -> RwLockWriteGuard<'_, T> {
281        internal::blocking(self.write())
282    }
283
284    /// Waits for the upgradable read. It makes the progress
285    /// [`read`](Self::read) does.
286    pub fn upgradable_read(&self) -> UpgradableRead<'_, T> {
287        UpgradableRead {
288            lock: self,
289            wait: RwWait::new(),
290        }
291    }
292
293    /// The value, borrowed mutably; `&mut self` proves no guard exists.
294    pub fn get_mut(&mut self) -> &mut T {
295        self.value.get_mut()
296    }
297}
298
299impl<T> From<T> for RwLock<T> {
300    fn from(value: T) -> Self {
301        Self::new(value)
302    }
303}
304
305impl<T: Default> Default for RwLock<T> {
306    fn default() -> Self {
307        Self::new(T::default())
308    }
309}
310
311impl<T: ?Sized + fmt::Debug> fmt::Debug for RwLock<T> {
312    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
313        let mut d = f.debug_struct("RwLock");
314        match self.try_read() {
315            Ok(inner) => d.field("data", &&*inner),
316            Err(_) => d.field("data", &format_args!("<locked>")),
317        };
318        d.finish()
319    }
320}
321
322/// The upgradable read of a [`RwLock`]: shared access that can become
323/// exclusive without a writer getting in between. Releases on drop, with
324/// the progress [`RwLockReadGuard`]'s release makes.
325///
326/// It moves to another thread only when `T` is `Send` as well as `Sync`:
327/// an upgrade turns it into `&mut T`, so moving it can move the value. A
328/// value that may be shared but not moved, such as a
329/// [`std::sync::MutexGuard`], keeps the guard on its thread:
330///
331/// ```compile_fail,E0277
332/// use nuclei::sync::RwLockUpgradableReadGuard;
333/// use std::sync::MutexGuard;
334///
335/// fn moves<T: Send>() {}
336/// moves::<RwLockUpgradableReadGuard<'static, Option<MutexGuard<'static, ()>>>>();
337/// ```
338#[must_use = "dropping the guard releases the lock at once"]
339pub struct RwLockUpgradableReadGuard<'a, T: ?Sized> {
340    lock: &'a RwLock<T>,
341}
342
343// SAFETY: moving it moves `&T` access and, through `upgrade` and
344// `try_upgrade`, `&mut T` access, as moving the `Upgrade` future does.
345unsafe impl<T: ?Sized + Send + Sync> Send for RwLockUpgradableReadGuard<'_, T> {}
346// SAFETY: `&RwLockUpgradableReadGuard` reaches `&T` only: both upgrades
347// take the guard by value.
348unsafe impl<T: ?Sized + Sync> Sync for RwLockUpgradableReadGuard<'_, T> {}
349
350impl<'a, T: ?Sized> RwLockUpgradableReadGuard<'a, T> {
351    /// Waits for the readers inside to leave, holding new readers back
352    /// meanwhile, then holds the write.
353    pub fn upgrade(self) -> Upgrade<'a, T> {
354        let lock = self.lock;
355        core::mem::forget(self);
356        Upgrade {
357            lock,
358            wait: UpgradeWait::new(),
359            done: false,
360        }
361    }
362
363    /// Holds the write at once if no reader is inside; otherwise hands the
364    /// upgradable read back.
365    pub fn try_upgrade(self) -> Result<RwLockWriteGuard<'a, T>, Self> {
366        if !self.lock.raw.try_upgrade() {
367            return Err(self);
368        }
369        let lock = self.lock;
370        core::mem::forget(self);
371        Ok(RwLockWriteGuard::new(lock))
372    }
373}
374
375impl<T: ?Sized> Deref for RwLockUpgradableReadGuard<'_, T> {
376    type Target = T;
377
378    fn deref(&self) -> &T {
379        // SAFETY: the upgradable read excludes every writer.
380        unsafe { &*self.lock.value.get() }
381    }
382}
383
384impl<T: ?Sized> Drop for RwLockUpgradableReadGuard<'_, T> {
385    fn drop(&mut self) {
386        self.lock.raw.upgradable_unlock();
387    }
388}
389
390impl<T: ?Sized + fmt::Debug> fmt::Debug for RwLockUpgradableReadGuard<'_, T> {
391    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
392        fmt::Debug::fmt(&**self, f)
393    }
394}
395
396/// The future [`RwLock::read`] returns.
397#[must_use = "futures do nothing unless polled"]
398pub struct Read<'a, T: ?Sized> {
399    lock: &'a RwLock<T>,
400    wait: RwWait,
401}
402
403impl<'a, T: ?Sized> Future for Read<'a, T> {
404    type Output = RwLockReadGuard<'a, T>;
405
406    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<RwLockReadGuard<'a, T>> {
407        let this = self.get_mut();
408        this.wait
409            .poll(&this.lock.raw, READ, cx.waker())
410            .map(|()| RwLockReadGuard::new(this.lock))
411    }
412}
413
414impl<T: ?Sized> Drop for Read<'_, T> {
415    fn drop(&mut self) {
416        self.wait.cancel(&self.lock.raw, READ);
417    }
418}
419
420impl<T: ?Sized> fmt::Debug for Read<'_, T> {
421    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
422        f.debug_struct("Read").finish_non_exhaustive()
423    }
424}
425
426/// The future [`RwLock::read_owned`] returns. It keeps its place in the
427/// queue, and the bypasses it counted, for as long as it lives; dropping it
428/// withdraws the wait and passes on a lock handed to it meanwhile.
429#[must_use = "futures do nothing unless polled"]
430pub struct ReadOwned<T: ?Sized> {
431    lock: Arc<RwLock<T>>,
432    wait: RwWait,
433}
434
435impl<T: ?Sized> Future for ReadOwned<T> {
436    type Output = OwnedRwLockReadGuard<T>;
437
438    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<OwnedRwLockReadGuard<T>> {
439        let this = self.get_mut();
440        this.wait
441            .poll(&this.lock.raw, READ, cx.waker())
442            .map(|()| OwnedRwLockReadGuard::new(Arc::clone(&this.lock)))
443    }
444}
445
446impl<T: ?Sized> Drop for ReadOwned<T> {
447    fn drop(&mut self) {
448        self.wait.cancel(&self.lock.raw, READ);
449    }
450}
451
452impl<T: ?Sized> fmt::Debug for ReadOwned<T> {
453    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
454        f.debug_struct("ReadOwned").finish_non_exhaustive()
455    }
456}
457
458/// The future [`RwLock::write`] returns.
459#[must_use = "futures do nothing unless polled"]
460pub struct Write<'a, T: ?Sized> {
461    lock: &'a RwLock<T>,
462    wait: RwWait,
463}
464
465impl<'a, T: ?Sized> Future for Write<'a, T> {
466    type Output = RwLockWriteGuard<'a, T>;
467
468    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<RwLockWriteGuard<'a, T>> {
469        let this = self.get_mut();
470        this.wait
471            .poll(&this.lock.raw, WRITE, cx.waker())
472            .map(|()| RwLockWriteGuard::new(this.lock))
473    }
474}
475
476impl<T: ?Sized> Drop for Write<'_, T> {
477    fn drop(&mut self) {
478        self.wait.cancel(&self.lock.raw, WRITE);
479    }
480}
481
482impl<T: ?Sized> fmt::Debug for Write<'_, T> {
483    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
484        f.debug_struct("Write").finish_non_exhaustive()
485    }
486}
487
488/// The future [`RwLock::write_owned`] returns. It keeps its place in the
489/// queue, and the bypasses it counted, for as long as it lives; dropping it
490/// withdraws the wait and passes on a handoff made to it meanwhile.
491#[must_use = "futures do nothing unless polled"]
492pub struct WriteOwned<T: ?Sized> {
493    lock: Arc<RwLock<T>>,
494    wait: RwWait,
495}
496
497impl<T: ?Sized> Future for WriteOwned<T> {
498    type Output = OwnedRwLockWriteGuard<T>;
499
500    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<OwnedRwLockWriteGuard<T>> {
501        let this = self.get_mut();
502        this.wait
503            .poll(&this.lock.raw, WRITE, cx.waker())
504            .map(|()| OwnedRwLockWriteGuard::new(Arc::clone(&this.lock)))
505    }
506}
507
508impl<T: ?Sized> Drop for WriteOwned<T> {
509    fn drop(&mut self) {
510        self.wait.cancel(&self.lock.raw, WRITE);
511    }
512}
513
514impl<T: ?Sized> fmt::Debug for WriteOwned<T> {
515    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
516        f.debug_struct("WriteOwned").finish_non_exhaustive()
517    }
518}
519
520/// The future [`RwLock::upgradable_read`] returns.
521#[must_use = "futures do nothing unless polled"]
522pub struct UpgradableRead<'a, T: ?Sized> {
523    lock: &'a RwLock<T>,
524    wait: RwWait,
525}
526
527impl<'a, T: ?Sized> Future for UpgradableRead<'a, T> {
528    type Output = RwLockUpgradableReadGuard<'a, T>;
529
530    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<RwLockUpgradableReadGuard<'a, T>> {
531        let this = self.get_mut();
532        this.wait
533            .poll(&this.lock.raw, UPGRADABLE_READ, cx.waker())
534            .map(|()| RwLockUpgradableReadGuard { lock: this.lock })
535    }
536}
537
538impl<T: ?Sized> Drop for UpgradableRead<'_, T> {
539    fn drop(&mut self) {
540        self.wait.cancel(&self.lock.raw, UPGRADABLE_READ);
541    }
542}
543
544impl<T: ?Sized> fmt::Debug for UpgradableRead<'_, T> {
545    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
546        f.debug_struct("UpgradableRead").finish_non_exhaustive()
547    }
548}
549
550/// The future [`RwLockUpgradableReadGuard::upgrade`] returns. Dropping it
551/// before it completes releases the upgradable read.
552#[must_use = "futures do nothing unless polled"]
553pub struct Upgrade<'a, T: ?Sized> {
554    lock: &'a RwLock<T>,
555    wait: UpgradeWait,
556    done: bool,
557}
558
559// SAFETY: it holds the upgradable read and yields the write: moving it
560// moves `&mut T` access, sharing it shares nothing.
561unsafe impl<T: ?Sized + Send + Sync> Send for Upgrade<'_, T> {}
562// SAFETY: `&Upgrade` exposes nothing.
563unsafe impl<T: ?Sized + Send + Sync> Sync for Upgrade<'_, T> {}
564
565impl<'a, T: ?Sized> Future for Upgrade<'a, T> {
566    type Output = RwLockWriteGuard<'a, T>;
567
568    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<RwLockWriteGuard<'a, T>> {
569        let this = self.get_mut();
570        assert!(
571            !this.done,
572            "an Upgrade future was polled after it completed"
573        );
574        this.wait.poll(&this.lock.raw, cx.waker()).map(|()| {
575            this.done = true;
576            RwLockWriteGuard::new(this.lock)
577        })
578    }
579}
580
581impl<T: ?Sized> Drop for Upgrade<'_, T> {
582    fn drop(&mut self) {
583        if !self.done {
584            self.wait.cancel(&self.lock.raw);
585        }
586    }
587}
588
589impl<T: ?Sized> fmt::Debug for Upgrade<'_, T> {
590    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
591        f.debug_struct("Upgrade")
592            .field("done", &self.done)
593            .finish_non_exhaustive()
594    }
595}
596
597mod guards;
598pub use guards::{
599    OwnedRwLockMappedWriteGuard, OwnedRwLockReadGuard, OwnedRwLockWriteGuard,
600    RwLockMappedWriteGuard, RwLockReadGuard, RwLockWriteGuard,
601};
602
603#[cfg(all(test, not(loom)))]
604mod tests;
nuclei/sync/rwlock.rs

nuclei/sync/
rwlock.rs
