From e675886440a9e49c2646c2314f037cee5bcd9578 Mon Sep 17 00:00:00 2001 From: Kaileshwar16 Date: Mon, 12 Oct 2026 14:33:14 +0530 Subject: [PATCH] feat(rwlock): add an upgradable readers-writer lock --- CHANGELOG.md | 1 + asyncband/src/rwlock/mod.rs | 11 +- asyncband/src/rwlock/upgradable.rs | 379 +++++++++++ asyncband/src/rwlock/write_guard.rs | 5 + tests-integration/tests/traits_test.rs | 13 + .../tests/upgradable_rwlock_test.rs | 643 ++++++++++++++++++ xtask/src/main.rs | 4 + 7 files changed, 1054 insertions(+), 2 deletions(-) create mode 100644 asyncband/src/rwlock/upgradable.rs create mode 100644 tests-integration/tests/upgradable_rwlock_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index bd5e84c9..56f6c078 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ All notable changes to this project will be documented in this file. ### New features +* Add `rwlock::UpgradableRwLock` with a single upgradable reader, atomic upgrade and downgrade, and cancellation that releases a pending upgrade's guard; writers and upgradable readers share a FIFO reservation while ordinary readers use the data-permit queue. * Add bounded MPMC `reserve` and `try_reserve` methods returning a borrowed `Permit`, so callers can wait for capacity before constructing a value; sends and reservations receive capacity in wait-queue order, and unused permits release it. ### Bug fixes diff --git a/asyncband/src/rwlock/mod.rs b/asyncband/src/rwlock/mod.rs index a806d036..5bb718e4 100644 --- a/asyncband/src/rwlock/mod.rs +++ b/asyncband/src/rwlock/mod.rs @@ -29,12 +29,16 @@ //! A write guard allows mutation and excludes every other guard. Both release their access on //! drop, including during unwinding; a panic does not poison the lock. //! -//! Waiting requests are served in queue order. A queued writer blocks readers behind it, even -//! while earlier readers still hold the lock. Consequently, keeping a read guard while waiting +//! [`RwLock`] serves waiting requests in queue order. A queued writer blocks readers behind it, +//! even while earlier readers still hold the lock. Consequently, keeping a read guard while waiting //! for a write guard, or for another read behind a queued writer, can deadlock. The `try_` methods //! never wait or reserve a queue position. Dropping a pending acquisition cancels its request; //! a subsequent acquisition starts again at the back of the queue. //! +//! [`UpgradableRwLock`] is a separate lock for inspecting a value before deciding to mutate it. +//! It reserves one reader's ability to upgrade and has its own [ordering +//! policy](UpgradableRwLock#ordering). +//! //! # Updating and inspecting //! //! ``` @@ -114,6 +118,7 @@ mod owned_mapped_write_guard; mod owned_read_guard; mod owned_write_guard; mod read_guard; +mod upgradable; mod write_guard; pub use self::mapped_read_guard::MappedRwLockReadGuard; @@ -123,6 +128,8 @@ pub use self::owned_mapped_write_guard::OwnedMappedRwLockWriteGuard; pub use self::owned_read_guard::OwnedRwLockReadGuard; pub use self::owned_write_guard::OwnedRwLockWriteGuard; pub use self::read_guard::RwLockReadGuard; +pub use self::upgradable::RwLockUpgradableReadGuard; +pub use self::upgradable::UpgradableRwLock; pub use self::write_guard::RwLockWriteGuard; /// A value with fair, asynchronous shared or exclusive access. diff --git a/asyncband/src/rwlock/upgradable.rs b/asyncband/src/rwlock/upgradable.rs new file mode 100644 index 00000000..85f60712 --- /dev/null +++ b/asyncband/src/rwlock/upgradable.rs @@ -0,0 +1,379 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::fmt; +use std::mem; +use std::num::NonZeroUsize; +use std::ops::Deref; + +use crate::internal::semaphore::Semaphore; +use crate::rwlock::RwLock; +use crate::rwlock::RwLockReadGuard; +use crate::rwlock::RwLockWriteGuard; +use crate::rwlock::access::ReadAccess; +use crate::rwlock::access::WriteAccess; +use crate::rwlock::write_guard; + +/// Shared and exclusive access with a single reader that can reserve an atomic upgrade. +/// +/// Ordinary readers and one upgradable reader may coexist, up to the configured reader limit. +/// Writers exclude all readers. An upgrade retains its read permit until it owns every permit, +/// so no writer can modify the value between inspection and mutation. Panics do not poison the +/// lock. Ordinary read and write guards are the same types returned by [`RwLock`]. +/// +/// # Ordering +/// +/// Writers and upgradable readers acquire a shared reservation in FIFO order, before requesting +/// data permits. An upgradable reader retains this reservation until upgrade, downgrade, or drop. +/// A writer releases it once it has exclusive access. Thus a writer waiting behind an upgradable +/// reader cannot enter the data-permit queue ahead of that reader's upgrade and deadlock it. +/// A writer already waiting for the reservation precedes a later upgradable acquisition. +/// +/// Ordinary readers bypass the reservation. Data permits are assigned in FIFO order: once a +/// writer or upgrade queues for them, later readers cannot pass it. Readers already queued before +/// an upgrade may finish first. With a reader limit of one, upgrading needs no additional permits +/// and completes immediately, even if readers are queued. +/// +/// There is no single FIFO order across both queues. Readers may pass writers still waiting for +/// the reservation, including while an upgradable guard is held. Progress assumes holders release +/// their guards and granted futures are polled or cancelled; an indefinitely held upgradable guard +/// can indefinitely delay writers. Like [`RwLock`], recursive acquisition can deadlock: do not +/// retain another read guard while awaiting an upgrade, or await a writer while retaining the +/// upgradable guard. An ordinary read guard cannot be upgraded. +/// +/// # Synchronization +/// +/// Writes made under an exclusive guard happen-before subsequent successful read, upgradable-read, +/// or write acquisitions after that guard is dropped or downgraded. A successful upgrade also +/// synchronizes with the readers whose release allows it to complete. These guarantees apply to +/// both asynchronous and successful `try_` acquisitions; failed `try_` calls grant no access. +/// +/// # Cancellation +/// +/// Dropping a pending acquisition releases its reservation and any assigned permits, losing its +/// queue position. [`upgrade`](RwLockUpgradableReadGuard::upgrade) consumes its guard; cancelling +/// that future releases the guard as well, including before its first poll. +/// +/// # Example +/// +/// ``` +/// use asyncband::rwlock::UpgradableRwLock; +/// +/// # #[tokio::main] +/// # async fn main() { +/// let routes = UpgradableRwLock::new(vec!["/health"]); +/// let snapshot = routes.upgradable_read().await; +/// let snapshot = if !snapshot.contains(&"/metrics") { +/// let mut edit = snapshot.upgrade().await; +/// edit.push("/metrics"); +/// edit.downgrade() +/// } else { +/// snapshot.downgrade() +/// }; +/// assert_eq!(*snapshot, ["/health", "/metrics"]); +/// # } +/// ``` +pub struct UpgradableRwLock { + // Every operation taking more than one data permit must first hold this reservation. + // It is never exposed, nor is the inner lock, so writers cannot bypass it. + reservation: Semaphore, + inner: RwLock, +} + +impl UpgradableRwLock { + /// Wraps a value with a reader limit of `usize::MAX >> 1`. + pub const fn new(value: T) -> Self { + Self::with_max_readers(value, NonZeroUsize::new(usize::MAX >> 1).unwrap()) + } + + /// Wraps a value with an explicit nonzero limit on simultaneously held read guards. + /// + /// An upgradable or downgraded guard occupies one reader slot. Every `NonZeroUsize` is + /// accepted. + pub const fn with_max_readers(value: T, max_readers: NonZeroUsize) -> Self { + Self { + reservation: Semaphore::new(1), + inner: RwLock::with_max_readers(value, max_readers), + } + } + + /// Unwraps the value by consuming its lock. + pub fn into_inner(self) -> T { + self.inner.into_inner() + } +} + +impl UpgradableRwLock { + /// Borrows the value exclusively through an exclusive borrow of the lock itself. + pub fn get_mut(&mut self) -> &mut T { + self.inner.get_mut() + } + + /// Waits for a reader slot, bypassing the reservation queue. + /// + /// Writers and upgrades already queued for data permits must finish first. + /// + /// # Cancel safety + /// + /// Cancellation releases any assigned permit and loses the request's queue position. + pub async fn read(&self) -> RwLockReadGuard<'_, T> { + self.inner.read().await + } + + /// Returns a read guard immediately, or `None` when a reader slot cannot be acquired. + pub fn try_read(&self) -> Option> { + self.inner.try_read() + } + + /// Waits for the reservation, then for exclusive access. + /// + /// See [ordering](Self#ordering) for how ordinary readers may pass this request. + /// + /// # Cancel safety + /// + /// Cancellation releases the reservation and any assigned permits and loses both queue + /// positions. + pub async fn write(&self) -> RwLockWriteGuard<'_, T> { + self.reservation.acquire(1).await; + let reservation = ReadAccess::new(&self.reservation); + let write = self.inner.write().await; + drop(reservation); + write + } + + /// Returns a write guard immediately, or `None` when the reservation or exclusive access is + /// unavailable. Failure releases any reservation acquired by this call. + pub fn try_write(&self) -> Option> { + if !self.reservation.try_acquire(1) { + return None; + } + let reservation = ReadAccess::new(&self.reservation); + let write = self.inner.try_write(); + drop(reservation); + write + } + + /// Waits for the reservation, then for one reader slot. + /// + /// The returned guard excludes other upgradable readers and reserves priority over later + /// writers. Ordinary readers can coexist with it. + /// + /// # Cancel safety + /// + /// Cancellation releases the reservation and any assigned permit and loses both queue + /// positions. + pub async fn upgradable_read(&self) -> RwLockUpgradableReadGuard<'_, T> { + self.reservation.acquire(1).await; + let reservation = ReadAccess::new(&self.reservation); + self.inner.s.acquire(1).await; + RwLockUpgradableReadGuard { + access: ReadAccess::new(&self.inner), + reservation, + } + } + + /// Returns an upgradable guard immediately, or `None` when the reservation or a reader slot + /// is unavailable. Failure releases any reservation acquired by this call. + pub fn try_upgradable_read(&self) -> Option> { + if !self.reservation.try_acquire(1) { + return None; + } + let reservation = ReadAccess::new(&self.reservation); + if !self.inner.s.try_acquire(1) { + return None; + } + Some(RwLockUpgradableReadGuard { + access: ReadAccess::new(&self.inner), + reservation, + }) + } +} + +impl From for UpgradableRwLock { + fn from(value: T) -> Self { + Self::new(value) + } +} + +impl Default for UpgradableRwLock { + fn default() -> Self { + Self::new(T::default()) + } +} + +impl fmt::Debug for UpgradableRwLock { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut d = f.debug_struct("UpgradableRwLock"); + match self.try_read() { + Some(inner) => d.field("data", &&*inner), + None => d.field("data", &format_args!("")), + }; + d.finish() + } +} + +/// Shared access with the unique reservation to upgrade to exclusive access. +/// +/// Created by [`UpgradableRwLock::upgradable_read`]. Dropping it releases its reader slot and +/// reservation. This guard cannot be mapped: promotion always protects the original value. +#[must_use = "dropping the guard releases its read access and upgrade reservation immediately"] +pub struct RwLockUpgradableReadGuard<'a, T: ?Sized> { + // Unlike an ordinary read guard, this token must retain T: Send + Sync for Send: moving the + // guard to another thread also moves the ability to acquire mutable access by upgrading. + access: ReadAccess<&'a RwLock>, + reservation: ReadAccess<&'a Semaphore>, +} + +impl Deref for RwLockUpgradableReadGuard<'_, T> { + type Target = T; + + fn deref(&self) -> &T { + // SAFETY: the token owns one data permit, excluding writers for this borrow's lifetime. + // Upgrade consumes the guard, so it cannot overlap a shared reference borrowed from it. + unsafe { &*self.access.owner().c.get() } + } +} + +impl fmt::Debug for RwLockUpgradableReadGuard<'_, T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Debug::fmt(&**self, f) + } +} + +impl fmt::Display for RwLockUpgradableReadGuard<'_, T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&**self, f) + } +} + +impl<'a, T: ?Sized> RwLockUpgradableReadGuard<'a, T> { + /// Waits for exclusive access without releasing the held read permit. + /// + /// The reservation prevents writers from queuing ahead of this upgrade. Ordinary readers + /// already ahead in the data-permit queue may finish first; later readers cannot pass it. + /// + /// # Cancel safety + /// + /// This consumes the guard. Dropping the future releases the original reader slot, any + /// partially or fully assigned upgrade permits, and the reservation. This also holds if the + /// future has never been polled. The original guard is not returned on cancellation. + pub async fn upgrade(self) -> RwLockWriteGuard<'a, T> { + let lock = *self.access.owner(); + // Keep both RAII tokens alive across the await. The semaphore's acquisition future owns + // additional permits until completion and returns them if cancelled, even after a grant. + lock.s.acquire(lock.max_readers - 1).await; + self.into_write() + } + + /// Converts immediately to exclusive access, or returns the unchanged guard on failure. + /// + /// Failure neither consumes additional permits nor reserves a queue position. With a reader + /// limit of one, this always succeeds. + /// + /// ``` + /// use asyncband::rwlock::UpgradableRwLock; + /// + /// let lock = UpgradableRwLock::new(1); + /// let upgradable = lock.try_upgradable_read().unwrap(); + /// let reader = lock.try_read().unwrap(); + /// let upgradable = upgradable.try_upgrade().unwrap_err(); + /// drop(reader); + /// let mut writer = upgradable.try_upgrade().unwrap(); + /// *writer += 1; + /// ``` + pub fn try_upgrade(self) -> Result, Self> { + let lock = *self.access.owner(); + if lock.s.try_acquire(lock.max_readers - 1) { + Ok(self.into_write()) + } else { + Err(self) + } + } + + /// Releases the upgrade reservation while retaining an ordinary reader slot. + /// + /// There is no unlocked interval. A queued writer may prevent new readers from joining, but + /// cannot modify the value until the returned guard is dropped. + pub fn downgrade(self) -> RwLockReadGuard<'a, T> { + let Self { + access, + reservation, + } = self; + let read = RwLockReadGuard { access }; + drop(reservation); + read + } + + // Called only after acquiring all max_readers - 1 additional data permits. No await or + // callback occurs during the transfer. The borrowed token has no owner allocation to free. + fn into_write(self) -> RwLockWriteGuard<'a, T> { + let Self { + access, + reservation, + } = self; + let lock = *access.owner(); + mem::forget(access); + let write = write_guard::from_access(WriteAccess::new(lock, lock.max_readers)); + drop(reservation); + write + } +} + +#[cfg(doctest)] +mod compile_fail_tests { + /// ```compile_fail + /// use asyncband::rwlock::RwLockUpgradableReadGuard; + /// + /// fn shorten<'lock, 'short: 'lock>( + /// guard: RwLockUpgradableReadGuard<'lock, &'static str>, + /// ) -> RwLockUpgradableReadGuard<'lock, &'short str> { + /// guard + /// } + /// ``` + struct UpgradableGuardIsInvariant; + + /// ```compile_fail + /// use std::sync::MutexGuard; + /// use asyncband::rwlock::RwLockUpgradableReadGuard; + /// + /// fn assert_send() {} + /// // MutexGuard is Sync but not Send: promotion could move its value on another thread. + /// assert_send::>>(); + /// ``` + struct UpgradableGuardRequiresSend; + + /// ```compile_fail + /// use std::cell::Cell; + /// use asyncband::rwlock::RwLockUpgradableReadGuard; + /// + /// fn assert_send() {} + /// assert_send::>>(); + /// ``` + struct UpgradableGuardRequiresSync; + + /// ```compile_fail + /// use asyncband::rwlock::UpgradableRwLock; + /// + /// let lock = UpgradableRwLock::new(0); + /// let guard = lock.try_upgradable_read().unwrap(); + /// let borrowed = &*guard; + /// let mut writer = guard.try_upgrade().unwrap(); + /// *writer = 1; + /// assert_eq!(*borrowed, 0); + /// ``` + struct UpgradeExcludesOutstandingBorrows; +} diff --git a/asyncband/src/rwlock/write_guard.rs b/asyncband/src/rwlock/write_guard.rs index 965adfbd..d070611d 100644 --- a/asyncband/src/rwlock/write_guard.rs +++ b/asyncband/src/rwlock/write_guard.rs @@ -32,6 +32,11 @@ use crate::rwlock::RwLock; use crate::rwlock::RwLockReadGuard; use crate::rwlock::access::WriteAccess; +/// Transfers an already acquired exclusive access token into a borrowed write guard. +pub fn from_access(access: WriteAccess<&RwLock>) -> RwLockWriteGuard<'_, T> { + RwLockWriteGuard { access } +} + impl RwLock { /// Waits for exclusive access and returns a guard borrowing this lock. /// diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 2d7c88c6..71a6124b 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -52,7 +52,9 @@ use asyncband::rwlock::OwnedMappedRwLockWriteGuard; use asyncband::rwlock::OwnedRwLockReadGuard; use asyncband::rwlock::RwLock; use asyncband::rwlock::RwLockReadGuard; +use asyncband::rwlock::RwLockUpgradableReadGuard; use asyncband::rwlock::RwLockWriteGuard; +use asyncband::rwlock::UpgradableRwLock; use asyncband::semaphore::Semaphore; use asyncband::shutdown::Shutdown; use asyncband::shutdown::ShutdownGuard; @@ -107,6 +109,8 @@ fn public_types_are_send_and_sync() { assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); + assert_send_and_sync::>(); + assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); @@ -153,6 +157,7 @@ fn movable_public_types_are_send() { fn assert_send() {} fn assert_send_value(_: T) {} + assert_send::>>(); assert_send::>>(); assert_send::>>(); assert_send::>>(); @@ -161,6 +166,12 @@ fn movable_public_types_are_send() { assert_send::>(); assert_send::>>(); + let lock = UpgradableRwLock::new(0); + assert_send_value(lock.read()); + assert_send_value(lock.write()); + assert_send_value(lock.upgradable_read()); + assert_send_value(lock.try_upgradable_read().unwrap().upgrade()); + let (_tx, mut rx) = watch::channel(0); assert_send_value(rx.changed()); assert_send_value(rx.recv()); @@ -203,6 +214,8 @@ fn public_types_are_unpin() { assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); + assert_unpin::>(); + assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); diff --git a/tests-integration/tests/upgradable_rwlock_test.rs b/tests-integration/tests/upgradable_rwlock_test.rs new file mode 100644 index 00000000..85f380df --- /dev/null +++ b/tests-integration/tests/upgradable_rwlock_test.rs @@ -0,0 +1,643 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::num::NonZeroUsize; +use std::sync::Arc; +use std::sync::Barrier; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::thread; + +use asyncband::blocking::FutureExt; +use asyncband::rwlock::UpgradableRwLock; +use tests_integration::assert_completes_without_deadlock; +use tests_integration::poll_once; +use tests_integration::poll_with; +use tests_integration::waker_on_drop; +use tests_integration::waker_on_wake; +use tokio_test::assert_pending; +use tokio_test::assert_ready; +use tokio_test::task::spawn; + +fn lock(limit: usize) -> UpgradableRwLock { + UpgradableRwLock::with_max_readers(0, NonZeroUsize::new(limit).unwrap()) +} + +#[test] +fn upgrade_mutate_and_downgrade_without_unlocking() { + let lock = lock(3); + let upgradable = FutureExt::block_on(lock.upgradable_read()); + let first = lock.try_read().unwrap(); + let second = FutureExt::block_on(lock.read()); + assert!(lock.try_read().is_none()); + assert!(lock.try_upgradable_read().is_none()); + assert!(lock.try_write().is_none()); + drop((first, second)); + + let mut write = FutureExt::block_on(upgradable.upgrade()); + *write = 42; + assert!(lock.try_read().is_none()); + assert!(lock.try_upgradable_read().is_none()); + assert!(lock.try_write().is_none()); + let read = write.downgrade(); + assert_eq!(*read, 42); + assert_eq!(*lock.try_read().unwrap(), 42); + assert!(lock.try_write().is_none()); + drop(read); + assert!(lock.try_write().is_some()); +} + +#[test] +fn reservation_is_unique_and_drop_wakes_the_next_upgradable_reader() { + let lock = lock(2); + let first = lock.try_upgradable_read().unwrap(); + let mut second = spawn(lock.upgradable_read()); + assert_pending!(second.poll()); + assert!(lock.try_upgradable_read().is_none()); + assert!(lock.try_read().is_some()); + drop(first); + assert!(second.is_woken()); + // A granted reservation belongs to the queued future even before it is polled again. + assert!(lock.try_upgradable_read().is_none()); + let second = assert_ready!(second.poll()); + assert!(lock.try_upgradable_read().is_none()); + drop(second); + assert!(lock.try_write().is_some()); +} + +#[test] +fn writer_queued_before_upgradable_acquisition_goes_first() { + let lock = lock(3); + let read = lock.try_read().unwrap(); + let mut writer = spawn(lock.write()); + let mut upgradable = spawn(lock.upgradable_read()); + assert_pending!(writer.poll()); + assert_pending!(upgradable.poll()); + assert!(lock.try_read().is_none()); + drop(read); + assert!(writer.is_woken()); + let mut write = assert_ready!(writer.poll()); + *write = 1; + assert!(upgradable.is_woken()); + assert_pending!(upgradable.poll()); + drop(write); + assert!(upgradable.is_woken()); + let upgradable = assert_ready!(upgradable.poll()); + assert_eq!(*upgradable, 1); + assert_eq!(*FutureExt::block_on(upgradable.upgrade()), 1); +} + +#[test] +fn upgrade_passes_a_writer_queued_after_its_reservation() { + let lock = lock(3); + let upgradable = lock.try_upgradable_read().unwrap(); + let reader = lock.try_read().unwrap(); + let mut writer = spawn(lock.write()); + assert_pending!(writer.poll()); + // A writer waiting only for the reservation does not block ordinary readers. + assert!(lock.try_read().is_some()); + let mut upgrade = spawn(upgradable.upgrade()); + assert_pending!(upgrade.poll()); + let mut later_reader = spawn(lock.read()); + assert_pending!(later_reader.poll()); + assert!(lock.try_read().is_none()); + drop(reader); + assert!(upgrade.is_woken()); + assert_pending!(writer.poll()); + let mut upgraded = assert_ready!(upgrade.poll()); + *upgraded = 1; + assert_pending!(writer.poll()); + assert_pending!(later_reader.poll()); + drop(upgraded); + // This reader queued for data permits before the writer obtained the reservation. + let later_reader = assert_ready!(later_reader.poll()); + assert_eq!(*later_reader, 1); + assert_pending!(writer.poll()); + drop(later_reader); + let mut writer = assert_ready!(writer.poll()); + *writer = 2; + drop(writer); + assert_eq!(*lock.try_read().unwrap(), 2); +} + +#[test] +fn writer_queued_after_pending_upgrade_waits_for_exclusive_guard() { + let lock = lock(2); + let upgradable = lock.try_upgradable_read().unwrap(); + let read = lock.try_read().unwrap(); + let mut upgrade = spawn(upgradable.upgrade()); + assert_pending!(upgrade.poll()); + let mut writer = spawn(lock.write()); + assert_pending!(writer.poll()); + drop(read); + let upgraded = assert_ready!(upgrade.poll()); + assert_pending!(writer.poll()); + drop(upgraded); + assert!(writer.is_woken()); + drop(assert_ready!(writer.poll())); +} + +#[test] +fn readers_queued_before_upgrade_finish_first() { + let lock = lock(2); + let upgradable = lock.try_upgradable_read().unwrap(); + let held = lock.try_read().unwrap(); + let mut earlier = spawn(lock.read()); + assert_pending!(earlier.poll()); + let mut upgrade = spawn(upgradable.upgrade()); + assert_pending!(upgrade.poll()); + drop(held); + assert!(earlier.is_woken()); + assert_pending!(upgrade.poll()); + let earlier = assert_ready!(earlier.poll()); + drop(earlier); + assert!(upgrade.is_woken()); + drop(assert_ready!(upgrade.poll())); + assert!(lock.try_write().is_some()); +} + +#[test] +fn cancel_upgrade_before_poll_while_partial_and_after_grant() { + for stage in 0..3 { + let lock = lock(4); + let upgradable = lock.try_upgradable_read().unwrap(); + let mut reader = Some(lock.try_read().unwrap()); + let mut writer = spawn(lock.write()); + assert_pending!(writer.poll()); + let mut upgrade = spawn(upgradable.upgrade()); + if stage > 0 { + // Two of the three additional permits are assigned before returning Pending. + assert_pending!(upgrade.poll()); + } + if stage == 2 { + drop(reader.take()); + assert!(upgrade.is_woken()); + } + drop(upgrade); + assert!(writer.is_woken()); + if stage < 2 { + assert_pending!(writer.poll()); + } + drop(reader); + drop(assert_ready!(writer.poll())); + let upgradable = lock.try_upgradable_read().unwrap(); + let readers = (0..3).map(|_| lock.try_read().unwrap()).collect::>(); + assert!(lock.try_read().is_none()); + drop((readers, upgradable)); + assert!(lock.try_write().is_some()); + } +} + +#[test] +fn cancelling_upgrade_wakes_queued_reader_and_next_reservation() { + let lock = lock(3); + let upgradable = lock.try_upgradable_read().unwrap(); + let held = lock.try_read().unwrap(); + let mut upgrade = spawn(upgradable.upgrade()); + assert_pending!(upgrade.poll()); + let mut reader = spawn(lock.read()); + let mut next = spawn(lock.upgradable_read()); + assert_pending!(reader.poll()); + assert_pending!(next.poll()); + drop(upgrade); + assert!(reader.is_woken()); + assert!(next.is_woken()); + let reader = assert_ready!(reader.poll()); + let next = assert_ready!(next.poll()); + assert!(lock.try_read().is_none()); + drop((held, reader, next)); + assert!(lock.try_write().is_some()); +} + +#[test] +fn cancellation_at_either_acquisition_queue_returns_all_resources() { + for upgradable_request in [false, true] { + for stage in 0..3 { + let lock = lock(2); + let held = lock.try_upgradable_read().unwrap(); + let mut request = Box::pin(async { + if upgradable_request { + drop(lock.upgradable_read().await); + } else { + drop(lock.write().await); + } + }); + assert_pending!(poll_once(request.as_mut())); + // Release the reservation but retain one data permit. + let read = held.downgrade(); + if stage > 0 { + // Fill the other slot so even an upgradable acquisition has to wait for data. + let other = lock.try_read().unwrap(); + assert_pending!(poll_once(request.as_mut())); + if stage == 2 { + drop((read, other)); + } else { + drop(request); + drop((read, other)); + assert!(lock.try_write().is_some()); + continue; + } + } else { + drop(read); + } + // Drop while the reservation or data permits have been granted, before repolling. + drop(request); + assert!(lock.try_write().is_some()); + assert!(lock.try_upgradable_read().is_some()); + } + } +} + +#[test] +fn dropping_and_downgrading_upgradable_guards_wake_writers() { + for downgrade in [false, true] { + let lock = lock(2); + let upgradable = lock.try_upgradable_read().unwrap(); + let mut writer = spawn(lock.write()); + assert_pending!(writer.poll()); + if downgrade { + let read = upgradable.downgrade(); + assert!(writer.is_woken()); + assert_pending!(writer.poll()); + assert!(lock.try_read().is_none()); + assert_eq!(*read, 0); + drop(read); + } else { + drop(upgradable); + } + assert!(writer.is_woken()); + drop(assert_ready!(writer.poll())); + assert!(lock.try_upgradable_read().is_some()); + } +} + +#[test] +fn try_upgrade_failure_preserves_guard_and_reservation() { + let lock = lock(3); + let upgradable = lock.try_upgradable_read().unwrap(); + let read = lock.try_read().unwrap(); + let mut writer = spawn(lock.write()); + assert_pending!(writer.poll()); + let upgradable = upgradable.try_upgrade().unwrap_err(); + assert_eq!(*upgradable, 0); + assert!(lock.try_upgradable_read().is_none()); + assert!(lock.try_read().is_some()); + assert_pending!(writer.poll()); + drop(read); + let write = upgradable.try_upgrade().unwrap(); + assert_pending!(writer.poll()); + drop(write); + drop(assert_ready!(writer.poll())); +} + +#[test] +fn reader_limits_and_repeated_cycles_include_one_and_usize_max() { + for limit in [1, 2, 5, usize::MAX] { + let lock = lock(limit); + for value in 1..=20 { + let upgradable = FutureExt::block_on(lock.upgradable_read()); + let mut write = if value % 2 == 0 { + upgradable.try_upgrade().unwrap() + } else { + FutureExt::block_on(upgradable.upgrade()) + }; + assert_eq!(*write, value - 1); + *write = value; + let read = write.downgrade(); + assert!(lock.try_write().is_none()); + if limit <= 5 { + let others = (1..limit) + .map(|_| lock.try_read().unwrap()) + .collect::>(); + assert!(lock.try_read().is_none()); + assert!(lock.try_upgradable_read().is_none()); + drop(others); + } + drop(read); + let upgradable = lock.try_upgradable_read().unwrap(); + assert_eq!(*upgradable.downgrade(), value); + } + } +} + +#[test] +fn one_reader_upgrade_does_not_wait_behind_queued_readers() { + for try_upgrade in [false, true] { + let lock = lock(1); + let upgradable = lock.try_upgradable_read().unwrap(); + let mut reader = spawn(lock.read()); + let mut writer = spawn(lock.write()); + assert_pending!(reader.poll()); + assert_pending!(writer.poll()); + let mut write = if try_upgrade { + upgradable.try_upgrade().unwrap() + } else { + FutureExt::block_on(upgradable.upgrade()) + }; + *write = 7; + assert_pending!(reader.poll()); + let retained = write.downgrade(); + assert_pending!(reader.poll()); + assert_pending!(writer.poll()); + drop(retained); + let read = assert_ready!(reader.poll()); + assert_eq!(*read, 7); + drop(read); + drop(assert_ready!(writer.poll())); + } +} + +#[test] +fn reservation_requests_keep_fifo_order_without_upgradable_barging() { + let lock = lock(2); + let held = lock.try_upgradable_read().unwrap(); + let mut first = spawn(lock.write()); + let mut second = spawn(lock.upgradable_read()); + let mut third = spawn(lock.write()); + assert_pending!(first.poll()); + assert_pending!(second.poll()); + assert_pending!(third.poll()); + drop(held); + assert!(lock.try_write().is_none()); + assert!(lock.try_upgradable_read().is_none()); + assert_pending!(third.poll()); + let first = assert_ready!(first.poll()); + assert_pending!(second.poll()); + drop(first); + let second = assert_ready!(second.poll()); + assert_pending!(third.poll()); + drop(second); + drop(assert_ready!(third.poll())); +} + +#[test] +fn ordinary_writer_blocks_later_readers_once_waiting_for_data() { + let lock = lock(3); + let held = lock.try_read().unwrap(); + let mut writer = spawn(lock.write()); + let mut later = spawn(lock.read()); + assert_pending!(writer.poll()); + assert_pending!(later.poll()); + assert!(lock.try_read().is_none()); + drop(held); + let writer = assert_ready!(writer.poll()); + assert_pending!(later.poll()); + drop(writer); + drop(assert_ready!(later.poll())); +} + +#[test] +fn failed_try_acquisitions_release_the_reservation() { + let lock = lock(1); + let held = lock.try_read().unwrap(); + assert!(lock.try_write().is_none()); + assert!(lock.try_upgradable_read().is_none()); + drop(held); + drop(lock.try_upgradable_read().unwrap()); + let write = lock.try_write().unwrap(); + assert!(lock.try_upgradable_read().is_none()); + drop(write); + assert!(lock.try_upgradable_read().is_some()); +} + +#[test] +fn cross_thread_guard_transfer() { + let lock = UpgradableRwLock::new((0, 0)); + let upgradable = lock.try_upgradable_read().unwrap(); + thread::scope(|scope| { + scope + .spawn(move || { + let mut write = FutureExt::block_on(upgradable.upgrade()); + *write = (1, 1); + }) + .join() + .unwrap(); + let upgradable = lock.try_upgradable_read().unwrap(); + assert_eq!(*upgradable, (1, 1)); + let read = upgradable.downgrade(); + let mut writer = spawn(lock.write()); + assert_pending!(writer.poll()); + scope.spawn(move || drop(read)).join().unwrap(); + let mut writer = assert_ready!(writer.poll()); + *writer = (2, 2); + }); + assert_eq!(*lock.try_read().unwrap(), (2, 2)); +} + +#[test] +fn concurrent_readers_writers_and_upgrades_preserve_values() { + let lock = Arc::new(UpgradableRwLock::new((0, 0))); + let rounds = if cfg!(miri) { 3 } else { 100 }; + let start = Barrier::new(6); + thread::scope(|scope| { + for role in 0..6 { + let lock = lock.clone(); + let start = &start; + scope.spawn(move || { + start.wait(); + for _ in 0..rounds { + match role % 3 { + 0 => { + let read = FutureExt::block_on(lock.read()); + assert_eq!(read.0, read.1); + thread::yield_now(); + } + 1 => { + let mut write = FutureExt::block_on(lock.write()); + write.0 += 1; + thread::yield_now(); + write.1 += 1; + } + _ => { + let read = FutureExt::block_on(lock.upgradable_read()); + assert_eq!(read.0, read.1); + thread::yield_now(); + let mut write = FutureExt::block_on(read.upgrade()); + write.0 += 1; + thread::yield_now(); + write.1 += 1; + let read = write.downgrade(); + assert_eq!(read.0, read.1); + } + } + } + }); + } + }); + assert_eq!(*lock.try_read().unwrap(), (4 * rounds, 4 * rounds)); +} + +#[test] +fn constructors_unsized_values_and_exclusive_borrows() { + let mut lock = const { UpgradableRwLock::new(1) }; + *lock.get_mut() = 2; + assert_eq!(lock.into_inner(), 2); + assert_eq!(UpgradableRwLock::::default().into_inner(), 0); + assert_eq!(UpgradableRwLock::from(3).into_inner(), 3); + let lock: Box> = Box::new(UpgradableRwLock::new([0, 1])); + let mut write = lock.try_upgradable_read().unwrap().try_upgrade().unwrap(); + write[0] = 2; + assert_eq!(&*write.downgrade(), &[2, 1]); +} + +#[test] +fn cancelled_reservation_waiter_does_not_block_its_successor() { + let lock = lock(2); + let held = lock.try_upgradable_read().unwrap(); + let mut cancelled = spawn(lock.write()); + let mut successor = spawn(lock.upgradable_read()); + assert_pending!(cancelled.poll()); + assert_pending!(successor.poll()); + drop(cancelled); + assert_pending!(successor.poll()); + drop(held); + assert!(successor.is_woken()); + drop(assert_ready!(successor.poll())); + assert!(lock.try_write().is_some()); +} + +#[test] +fn cancelling_an_upgrade_races_with_the_last_readers_release() { + let rounds = if cfg!(miri) { 3 } else { 100 }; + for _ in 0..rounds { + let lock = lock(3); + let upgradable = lock.try_upgradable_read().unwrap(); + let reader = lock.try_read().unwrap(); + let mut upgrade = spawn(upgradable.upgrade()); + let mut writer = spawn(lock.write()); + assert_pending!(upgrade.poll()); + assert_pending!(writer.poll()); + let start = Barrier::new(2); + thread::scope(|scope| { + scope.spawn(|| { + start.wait(); + drop(reader); + }); + start.wait(); + drop(upgrade); + }); + assert!(writer.is_woken()); + drop(assert_ready!(writer.poll())); + let readers = (0..3).map(|_| lock.try_read().unwrap()).collect::>(); + assert!(lock.try_read().is_none()); + drop(readers); + assert!(lock.try_upgradable_read().is_some()); + } +} + +#[test] +fn upgrade_observes_reader_writes_without_a_thread_join() { + assert_completes_without_deadlock(|| { + for queued in [false, true] { + let lock = UpgradableRwLock::with_max_readers( + AtomicUsize::new(0), + NonZeroUsize::new(2).unwrap(), + ); + let upgradable = lock.try_upgradable_read().unwrap(); + let reader = lock.try_read().unwrap(); + let released = AtomicBool::new(false); + let mut upgrade = Box::pin(upgradable.upgrade()); + if queued { + assert_pending!(poll_once(upgrade.as_mut())); + } + thread::scope(|scope| { + scope.spawn(|| { + reader.store(42, Ordering::Relaxed); + drop(reader); + released.store(true, Ordering::Relaxed); + }); + // The flag only controls the schedule. Relaxed ordering and the no-op waker + // cannot publish the reader's write; the lock must provide that synchronization. + while !released.load(Ordering::Relaxed) { + thread::yield_now(); + } + let mut write = assert_ready!(poll_once(upgrade.as_mut())); + // A non-atomic read lets Miri verify that the upgrade acquired the reader's + // release, both when permits were immediately available and when queued. + assert_eq!(*write.get_mut(), 42); + }); + } + }); +} + +#[test] +fn downgraded_writer_publishes_to_readers_without_a_thread_join() { + assert_completes_without_deadlock(|| { + for upgradable_reader in [false, true] { + let lock = UpgradableRwLock::new(0); + let mut write = lock.try_upgradable_read().unwrap().try_upgrade().unwrap(); + thread::scope(|scope| { + let holder = scope.spawn(move || { + *write = 42; + write.downgrade() + }); + // A try-acquisition may observe the old unavailable state. Retry until it + // succeeds, without using another synchronization primitive to publish the value. + let value = loop { + let value = if upgradable_reader { + lock.try_upgradable_read().map(|read| *read) + } else { + lock.try_read().map(|read| *read) + }; + if let Some(value) = value { + break value; + } + thread::yield_now(); + }; + assert_eq!(value, 42); + // The thread's result owns the downgraded guard until this join, so dropping + // that guard cannot supply the synchronization missing from a downgrade. + drop(holder.join().unwrap()); + }); + } + }); +} + +#[test] +fn upgrade_waker_replacement_and_grant_allow_reentry() { + assert_completes_without_deadlock(|| { + let lock = Arc::new(lock(2)); + let callbacks = Arc::new(AtomicUsize::new(0)); + let reenter = || { + let lock = lock.clone(); + let callbacks = callbacks.clone(); + move || { + // All permits are assigned to the upgrader and the existing reader, or to the + // granted upgrade. Registering and cancelling a reader reenters the data queue. + let mut read = Box::pin(lock.read()); + assert_pending!(poll_once(read.as_mut())); + drop(read); + callbacks.fetch_add(1, Ordering::Relaxed); + } + }; + let upgradable = lock.try_upgradable_read().unwrap(); + let reader = lock.try_read().unwrap(); + let mut upgrade = Box::pin(upgradable.upgrade()); + let old_waker = waker_on_drop(reenter()); + assert_pending!(poll_with(upgrade.as_mut(), &old_waker)); + drop(old_waker); + let new_waker = waker_on_wake(reenter()); + assert_pending!(poll_with(upgrade.as_mut(), &new_waker)); + assert_eq!(callbacks.load(Ordering::Relaxed), 1); + drop(reader); + assert_eq!(callbacks.load(Ordering::Relaxed), 2); + // Cancellation after the wake must also reclaim every assigned permit. + drop(upgrade); + assert!(lock.try_write().is_some()); + }); +} diff --git a/xtask/src/main.rs b/xtask/src/main.rs index a047a4f4..4fbce906 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -134,6 +134,10 @@ impl CommandMiri { "tests-integration", &["--test", "unsafe_paths_test"], )); + run_command(make_miri_cmd( + "tests-integration", + &["--test", "upgradable_rwlock_test"], + )); run_command(make_miri_cmd("tests-integration", &["--test", "mpsc_test"])); run_command(make_miri_cmd("tests-integration", &["--test", "mpmc_test"])); run_command(make_miri_cmd("tests-integration", &["--test", "spmc_test"]));