From 8ab7b5878b57cdff29a76cf83489ecfb4d9be03a Mon Sep 17 00:00:00 2001 From: Kaileshwar16 Date: Mon, 12 Oct 2026 14:33:14 +0530 Subject: [PATCH 1/2] 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 0e9f7ea3..008722d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,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 08f0f058..1837ebb3 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"])); From 4c117c6795b2dabfc8db29c7c24a16efcde7be6e Mon Sep 17 00:00:00 2001 From: Kaileshwar16 Date: Sat, 24 Oct 2026 15:23:39 +0530 Subject: [PATCH 2/2] fix(rwlock): address upgradable lock review comments --- CHANGELOG.md | 5 +- README.md | 55 ++++++++++--------- asyncband/src/lib.rs | 55 ++++++++++--------- asyncband/src/rwlock/upgradable.rs | 24 ++++---- tests-integration/tests/traits_test.rs | 4 +- .../tests/upgradable_rwlock_test.rs | 48 +++++++++------- 6 files changed, 104 insertions(+), 87 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 008722d1..07d4df12 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,10 @@ All notable changes to this project will be documented in this file. ## Unreleased +### 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. + ### Improvements * Reduce the cost of polling a pending oneshot receive again from the same task, which now keeps its registered waker instead of cloning and dropping it. @@ -35,7 +39,6 @@ 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/README.md b/README.md index 17ee4426..5d221ad3 100644 --- a/README.md +++ b/README.md @@ -81,33 +81,34 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon ## API map -| Area | API | Feature | Use | -|----------------------|------------------------------------------------------------------------------------------------|----------------|-----------------------------------------------------------------------------------------------------------------| -| Locks and conditions | [`Mutex`](https://docs.rs/asyncband/*/asyncband/mutex/struct.Mutex.html) | `mutex` | Protect shared data with asynchronous mutual exclusion. | -| | [`RwLock`](https://docs.rs/asyncband/*/asyncband/rwlock/struct.RwLock.html) | `rwlock` | Allow multiple readers or one writer. | -| | [`Condvar`](https://docs.rs/asyncband/*/asyncband/condvar/struct.Condvar.html) | `condvar` | Wait for notifications while releasing a mutex. | -| Coordination | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Limit concurrent work by acquiring permits. | -| | [`Barrier`](https://docs.rs/asyncband/*/asyncband/barrier/struct.Barrier.html) | `barrier` | Synchronize a fixed number of participants at a reusable rendezvous. | -| | [`ManualResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.ManualResetEvent.html) | `event` | Signal current and future waits until explicitly reset. | -| | [`AutoResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.AutoResetEvent.html) | `event` | Retain one signal and release one waiter per consumed signal. | -| | [`Latch`](https://docs.rs/asyncband/*/asyncband/latch/struct.Latch.html) | `latch` | Wait until a fixed one-way countdown reaches zero. | -| | [`Phaser`](https://docs.rs/asyncband/*/asyncband/phaser/struct.Phaser.html) | `phaser` | Coordinate repeated phases with a dynamic participant set. | -| | [`WaitGroup`](https://docs.rs/asyncband/*/asyncband/waitgroup/struct.WaitGroup.html) | `waitgroup` | Dynamically register participants and wait until all have completed. | -| | [`Shutdown`](https://docs.rs/asyncband/*/asyncband/shutdown/struct.Shutdown.html) | `shutdown` | Request shutdown and wait until all completion guards are dropped. | -| Work coalescing | [`Once`](https://docs.rs/asyncband/*/asyncband/once/struct.Once.html) | `once` | Complete one asynchronous initialization; cancelled or panicked attempts may be retried. | -| | [`OnceCell`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceCell.html) | `once-cell` | Store one value from an access-time initializer; failed, cancelled, or panicked attempts may be retried. | -| | [`LazyCell`](https://docs.rs/asyncband/*/asyncband/once/struct.LazyCell.html) | `lazy-cell` | Initialize one value with a stored function and resume the same in-flight future after caller cancellation. | -| | [`OnceMap`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceMap.html) | `once-map` | Coalesce work per key and retain each successful value until explicitly removed. | -| | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html) | `singleflight` | Coalesce overlapping work per key without retaining completed values. | -| Communication | [`Completion`](https://docs.rs/asyncband/*/asyncband/completion/struct.Completion.html) | `completion` | Publish one shared result to any number of current and future observers. | -| | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/) | `oneshot` | Send one value from one sender to one receiver. | -| | [`mpmc`](https://docs.rs/asyncband/*/asyncband/mpmc/) | `mpmc` | Distribute each value to exactly one of multiple competing receivers. | -| | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/) | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. | -| | [`spmc`](https://docs.rs/asyncband/*/asyncband/spmc/) | `spmc` | Distribute work from one exclusive sender to multiple competing receivers, with bounded or unbounded storage. | -| | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/) | `broadcast` | Deliver every value to active receivers with bounded backpressure or unbounded retention. | -| | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/) | `watch` | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. | -| Object reuse | [`pool`](https://docs.rs/asyncband/*/asyncband/pool/) | `pool` | Reuse objects through bounded or unbounded pool variants. | -| Sync interop | [`FutureExt`](https://docs.rs/asyncband/*/asyncband/blocking/trait.FutureExt.html) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | +| Area | API | Feature | Use | +|----------------------|-------------------------------------------------------------------------------------------------|----------------|-----------------------------------------------------------------------------------------------------------------| +| Locks and conditions | [`Mutex`](https://docs.rs/asyncband/*/asyncband/mutex/struct.Mutex.html) | `mutex` | Protect shared data with asynchronous mutual exclusion. | +| | [`RwLock`](https://docs.rs/asyncband/*/asyncband/rwlock/struct.RwLock.html) | `rwlock` | Allow multiple readers or one writer. | +| | [`UpgradableRwLock`](https://docs.rs/asyncband/*/asyncband/rwlock/struct.UpgradableRwLock.html) | `rwlock` | Allow multiple readers or one writer, with one upgradable reader. | +| | [`Condvar`](https://docs.rs/asyncband/*/asyncband/condvar/struct.Condvar.html) | `condvar` | Wait for notifications while releasing a mutex. | +| Coordination | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Limit concurrent work by acquiring permits. | +| | [`Barrier`](https://docs.rs/asyncband/*/asyncband/barrier/struct.Barrier.html) | `barrier` | Synchronize a fixed number of participants at a reusable rendezvous. | +| | [`ManualResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.ManualResetEvent.html) | `event` | Signal current and future waits until explicitly reset. | +| | [`AutoResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.AutoResetEvent.html) | `event` | Retain one signal and release one waiter per consumed signal. | +| | [`Latch`](https://docs.rs/asyncband/*/asyncband/latch/struct.Latch.html) | `latch` | Wait until a fixed one-way countdown reaches zero. | +| | [`Phaser`](https://docs.rs/asyncband/*/asyncband/phaser/struct.Phaser.html) | `phaser` | Coordinate repeated phases with a dynamic participant set. | +| | [`WaitGroup`](https://docs.rs/asyncband/*/asyncband/waitgroup/struct.WaitGroup.html) | `waitgroup` | Dynamically register participants and wait until all have completed. | +| | [`Shutdown`](https://docs.rs/asyncband/*/asyncband/shutdown/struct.Shutdown.html) | `shutdown` | Request shutdown and wait until all completion guards are dropped. | +| Work coalescing | [`Once`](https://docs.rs/asyncband/*/asyncband/once/struct.Once.html) | `once` | Complete one asynchronous initialization; cancelled or panicked attempts may be retried. | +| | [`OnceCell`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceCell.html) | `once-cell` | Store one value from an access-time initializer; failed, cancelled, or panicked attempts may be retried. | +| | [`LazyCell`](https://docs.rs/asyncband/*/asyncband/once/struct.LazyCell.html) | `lazy-cell` | Initialize one value with a stored function and resume the same in-flight future after caller cancellation. | +| | [`OnceMap`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceMap.html) | `once-map` | Coalesce work per key and retain each successful value until explicitly removed. | +| | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html) | `singleflight` | Coalesce overlapping work per key without retaining completed values. | +| Communication | [`Completion`](https://docs.rs/asyncband/*/asyncband/completion/struct.Completion.html) | `completion` | Publish one shared result to any number of current and future observers. | +| | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/) | `oneshot` | Send one value from one sender to one receiver. | +| | [`mpmc`](https://docs.rs/asyncband/*/asyncband/mpmc/) | `mpmc` | Distribute each value to exactly one of multiple competing receivers. | +| | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/) | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. | +| | [`spmc`](https://docs.rs/asyncband/*/asyncband/spmc/) | `spmc` | Distribute work from one exclusive sender to multiple competing receivers, with bounded or unbounded storage. | +| | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/) | `broadcast` | Deliver every value to active receivers with bounded backpressure or unbounded retention. | +| | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/) | `watch` | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. | +| Object reuse | [`pool`](https://docs.rs/asyncband/*/asyncband/pool/) | `pool` | Reuse objects through bounded or unbounded pool variants. | +| Sync interop | [`FutureExt`](https://docs.rs/asyncband/*/asyncband/blocking/trait.FutureExt.html) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | ## Synchronous interoperability diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index f2003d85..8126eb96 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -54,33 +54,34 @@ //! //! # API map //! -//! | Area | API | Feature | Use | -//! |----------------------|-----------------------------------------------|----------------|-----------------------------------------------------------------------------------------------------------------| -//! | Locks and conditions | [`Mutex`](mutex::Mutex) | `mutex` | Protect shared data with asynchronous mutual exclusion. | -//! | | [`RwLock`](rwlock::RwLock) | `rwlock` | Allow multiple readers or one writer. | -//! | | [`Condvar`](condvar::Condvar) | `condvar` | Wait for notifications while releasing a mutex. | -//! | Coordination | [`Semaphore`](semaphore::Semaphore) | `semaphore` | Limit concurrent work by acquiring permits. | -//! | | [`Barrier`](barrier::Barrier) | `barrier` | Synchronize a fixed number of participants at a reusable rendezvous. | -//! | | [`ManualResetEvent`](event::ManualResetEvent) | `event` | Signal current and future waits until explicitly reset. | -//! | | [`AutoResetEvent`](event::AutoResetEvent) | `event` | Retain one signal and release one waiter per consumed signal. | -//! | | [`Latch`](latch::Latch) | `latch` | Wait until a fixed one-way countdown reaches zero. | -//! | | [`Phaser`](phaser::Phaser) | `phaser` | Coordinate repeated phases with a dynamic participant set. | -//! | | [`WaitGroup`](waitgroup::WaitGroup) | `waitgroup` | Dynamically register participants and wait until all have completed. | -//! | | [`Shutdown`](shutdown::Shutdown) | `shutdown` | Request shutdown and wait until all completion guards are dropped. | -//! | Work coalescing | [`Once`](once::Once) | `once` | Complete one asynchronous initialization; cancelled or panicked attempts may be retried. | -//! | | [`OnceCell`](once::OnceCell) | `once-cell` | Store one value from an access-time initializer; failed, cancelled, or panicked attempts may be retried. | -//! | | [`LazyCell`](once::LazyCell) | `lazy-cell` | Initialize one value with a stored function and resume the same in-flight future after caller cancellation. | -//! | | [`OnceMap`](once::OnceMap) | `once-map` | Coalesce work per key and retain each successful value until explicitly removed. | -//! | | [`Group`](singleflight::Group) | `singleflight` | Coalesce overlapping work per key without retaining completed values. | -//! | Communication | [`Completion`](completion::Completion) | `completion` | Publish one shared result to any number of current and future observers. | -//! | | [`oneshot`] | `oneshot` | Send one value from one sender to one receiver. | -//! | | [`mpmc`] | `mpmc` | Distribute each value to exactly one of multiple competing receivers. | -//! | | [`mpsc`] | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. | -//! | | [`spmc`] | `spmc` | Distribute work from one exclusive sender to multiple competing receivers, with bounded or unbounded storage. | -//! | | [`broadcast`] | `broadcast` | Deliver every value to active receivers with bounded backpressure or unbounded retention. | -//! | | [`watch`] | `watch` | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. | -//! | Object reuse | [`pool`] | `pool` | Reuse objects through bounded or unbounded pool variants. | -//! | Sync interop | [`FutureExt`](blocking::FutureExt) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | +//! | Area | API | Feature | Use | +//! |----------------------|------------------------------------------------|----------------|-----------------------------------------------------------------------------------------------------------------| +//! | Locks and conditions | [`Mutex`](mutex::Mutex) | `mutex` | Protect shared data with asynchronous mutual exclusion. | +//! | | [`RwLock`](rwlock::RwLock) | `rwlock` | Allow multiple readers or one writer. | +//! | | [`UpgradableRwLock`](rwlock::UpgradableRwLock) | `rwlock` | Allow multiple readers or one writer, with one upgradable reader. | +//! | | [`Condvar`](condvar::Condvar) | `condvar` | Wait for notifications while releasing a mutex. | +//! | Coordination | [`Semaphore`](semaphore::Semaphore) | `semaphore` | Limit concurrent work by acquiring permits. | +//! | | [`Barrier`](barrier::Barrier) | `barrier` | Synchronize a fixed number of participants at a reusable rendezvous. | +//! | | [`ManualResetEvent`](event::ManualResetEvent) | `event` | Signal current and future waits until explicitly reset. | +//! | | [`AutoResetEvent`](event::AutoResetEvent) | `event` | Retain one signal and release one waiter per consumed signal. | +//! | | [`Latch`](latch::Latch) | `latch` | Wait until a fixed one-way countdown reaches zero. | +//! | | [`Phaser`](phaser::Phaser) | `phaser` | Coordinate repeated phases with a dynamic participant set. | +//! | | [`WaitGroup`](waitgroup::WaitGroup) | `waitgroup` | Dynamically register participants and wait until all have completed. | +//! | | [`Shutdown`](shutdown::Shutdown) | `shutdown` | Request shutdown and wait until all completion guards are dropped. | +//! | Work coalescing | [`Once`](once::Once) | `once` | Complete one asynchronous initialization; cancelled or panicked attempts may be retried. | +//! | | [`OnceCell`](once::OnceCell) | `once-cell` | Store one value from an access-time initializer; failed, cancelled, or panicked attempts may be retried. | +//! | | [`LazyCell`](once::LazyCell) | `lazy-cell` | Initialize one value with a stored function and resume the same in-flight future after caller cancellation. | +//! | | [`OnceMap`](once::OnceMap) | `once-map` | Coalesce work per key and retain each successful value until explicitly removed. | +//! | | [`Group`](singleflight::Group) | `singleflight` | Coalesce overlapping work per key without retaining completed values. | +//! | Communication | [`Completion`](completion::Completion) | `completion` | Publish one shared result to any number of current and future observers. | +//! | | [`oneshot`] | `oneshot` | Send one value from one sender to one receiver. | +//! | | [`mpmc`] | `mpmc` | Distribute each value to exactly one of multiple competing receivers. | +//! | | [`mpsc`] | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. | +//! | | [`spmc`] | `spmc` | Distribute work from one exclusive sender to multiple competing receivers, with bounded or unbounded storage. | +//! | | [`broadcast`] | `broadcast` | Deliver every value to active receivers with bounded backpressure or unbounded retention. | +//! | | [`watch`] | `watch` | Publish cloneable latest state from one or more senders; receivers independently coalesce intermediate updates. | +//! | Object reuse | [`pool`] | `pool` | Reuse objects through bounded or unbounded pool variants. | +//! | Sync interop | [`FutureExt`](blocking::FutureExt) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | //! //! # Scope and runtime model //! diff --git a/asyncband/src/rwlock/upgradable.rs b/asyncband/src/rwlock/upgradable.rs index 85f60712..48a8c256 100644 --- a/asyncband/src/rwlock/upgradable.rs +++ b/asyncband/src/rwlock/upgradable.rs @@ -71,6 +71,7 @@ use crate::rwlock::write_guard; /// # Example /// /// ``` +/// use asyncband::rwlock::RwLockUpgradableReadGuard; /// use asyncband::rwlock::UpgradableRwLock; /// /// # #[tokio::main] @@ -78,7 +79,7 @@ use crate::rwlock::write_guard; /// let routes = UpgradableRwLock::new(vec!["/health"]); /// let snapshot = routes.upgradable_read().await; /// let snapshot = if !snapshot.contains(&"/metrics") { -/// let mut edit = snapshot.upgrade().await; +/// let mut edit = RwLockUpgradableReadGuard::upgrade(snapshot).await; /// edit.push("/metrics"); /// edit.downgrade() /// } else { @@ -271,12 +272,12 @@ impl<'a, T: ?Sized> RwLockUpgradableReadGuard<'a, T> { /// 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(); + pub async fn upgrade(this: Self) -> RwLockWriteGuard<'a, T> { + let lock = *this.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() + this.into_write() } /// Converts immediately to exclusive access, or returns the unchanged guard on failure. @@ -285,22 +286,23 @@ impl<'a, T: ?Sized> RwLockUpgradableReadGuard<'a, T> { /// limit of one, this always succeeds. /// /// ``` + /// use asyncband::rwlock::RwLockUpgradableReadGuard; /// 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(); + /// let upgradable = RwLockUpgradableReadGuard::try_upgrade(upgradable).unwrap_err(); /// drop(reader); - /// let mut writer = upgradable.try_upgrade().unwrap(); + /// let mut writer = RwLockUpgradableReadGuard::try_upgrade(upgradable).unwrap(); /// *writer += 1; /// ``` - pub fn try_upgrade(self) -> Result, Self> { - let lock = *self.access.owner(); + pub fn try_upgrade(this: Self) -> Result, Self> { + let lock = *this.access.owner(); if lock.s.try_acquire(lock.max_readers - 1) { - Ok(self.into_write()) + Ok(this.into_write()) } else { - Err(self) + Err(this) } } @@ -371,7 +373,7 @@ mod compile_fail_tests { /// let lock = UpgradableRwLock::new(0); /// let guard = lock.try_upgradable_read().unwrap(); /// let borrowed = &*guard; - /// let mut writer = guard.try_upgrade().unwrap(); + /// let mut writer = RwLockUpgradableReadGuard::try_upgrade(guard).unwrap(); /// *writer = 1; /// assert_eq!(*borrowed, 0); /// ``` diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 71a6124b..98b721f4 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -170,7 +170,9 @@ fn movable_public_types_are_send() { 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()); + assert_send_value(RwLockUpgradableReadGuard::upgrade( + lock.try_upgradable_read().unwrap(), + )); let (_tx, mut rx) = watch::channel(0); assert_send_value(rx.changed()); diff --git a/tests-integration/tests/upgradable_rwlock_test.rs b/tests-integration/tests/upgradable_rwlock_test.rs index 85f380df..a665e15b 100644 --- a/tests-integration/tests/upgradable_rwlock_test.rs +++ b/tests-integration/tests/upgradable_rwlock_test.rs @@ -24,6 +24,7 @@ use std::sync::atomic::Ordering; use std::thread; use asyncband::blocking::FutureExt; +use asyncband::rwlock::RwLockUpgradableReadGuard; use asyncband::rwlock::UpgradableRwLock; use tests_integration::assert_completes_without_deadlock; use tests_integration::poll_once; @@ -49,7 +50,7 @@ fn upgrade_mutate_and_downgrade_without_unlocking() { assert!(lock.try_write().is_none()); drop((first, second)); - let mut write = FutureExt::block_on(upgradable.upgrade()); + let mut write = FutureExt::block_on(RwLockUpgradableReadGuard::upgrade(upgradable)); *write = 42; assert!(lock.try_read().is_none()); assert!(lock.try_upgradable_read().is_none()); @@ -99,7 +100,10 @@ fn writer_queued_before_upgradable_acquisition_goes_first() { assert!(upgradable.is_woken()); let upgradable = assert_ready!(upgradable.poll()); assert_eq!(*upgradable, 1); - assert_eq!(*FutureExt::block_on(upgradable.upgrade()), 1); + assert_eq!( + *FutureExt::block_on(RwLockUpgradableReadGuard::upgrade(upgradable)), + 1 + ); } #[test] @@ -111,7 +115,7 @@ fn upgrade_passes_a_writer_queued_after_its_reservation() { 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()); + let mut upgrade = spawn(RwLockUpgradableReadGuard::upgrade(upgradable)); assert_pending!(upgrade.poll()); let mut later_reader = spawn(lock.read()); assert_pending!(later_reader.poll()); @@ -140,7 +144,7 @@ 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()); + let mut upgrade = spawn(RwLockUpgradableReadGuard::upgrade(upgradable)); assert_pending!(upgrade.poll()); let mut writer = spawn(lock.write()); assert_pending!(writer.poll()); @@ -159,7 +163,7 @@ fn readers_queued_before_upgrade_finish_first() { let held = lock.try_read().unwrap(); let mut earlier = spawn(lock.read()); assert_pending!(earlier.poll()); - let mut upgrade = spawn(upgradable.upgrade()); + let mut upgrade = spawn(RwLockUpgradableReadGuard::upgrade(upgradable)); assert_pending!(upgrade.poll()); drop(held); assert!(earlier.is_woken()); @@ -179,7 +183,7 @@ fn cancel_upgrade_before_poll_while_partial_and_after_grant() { let mut reader = Some(lock.try_read().unwrap()); let mut writer = spawn(lock.write()); assert_pending!(writer.poll()); - let mut upgrade = spawn(upgradable.upgrade()); + let mut upgrade = spawn(RwLockUpgradableReadGuard::upgrade(upgradable)); if stage > 0 { // Two of the three additional permits are assigned before returning Pending. assert_pending!(upgrade.poll()); @@ -208,7 +212,7 @@ 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()); + let mut upgrade = spawn(RwLockUpgradableReadGuard::upgrade(upgradable)); assert_pending!(upgrade.poll()); let mut reader = spawn(lock.read()); let mut next = spawn(lock.upgradable_read()); @@ -293,13 +297,13 @@ fn try_upgrade_failure_preserves_guard_and_reservation() { let read = lock.try_read().unwrap(); let mut writer = spawn(lock.write()); assert_pending!(writer.poll()); - let upgradable = upgradable.try_upgrade().unwrap_err(); + let upgradable = RwLockUpgradableReadGuard::try_upgrade(upgradable).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(); + let write = RwLockUpgradableReadGuard::try_upgrade(upgradable).unwrap(); assert_pending!(writer.poll()); drop(write); drop(assert_ready!(writer.poll())); @@ -312,9 +316,9 @@ fn reader_limits_and_repeated_cycles_include_one_and_usize_max() { for value in 1..=20 { let upgradable = FutureExt::block_on(lock.upgradable_read()); let mut write = if value % 2 == 0 { - upgradable.try_upgrade().unwrap() + RwLockUpgradableReadGuard::try_upgrade(upgradable).unwrap() } else { - FutureExt::block_on(upgradable.upgrade()) + FutureExt::block_on(RwLockUpgradableReadGuard::upgrade(upgradable)) }; assert_eq!(*write, value - 1); *write = value; @@ -345,9 +349,9 @@ fn one_reader_upgrade_does_not_wait_behind_queued_readers() { assert_pending!(reader.poll()); assert_pending!(writer.poll()); let mut write = if try_upgrade { - upgradable.try_upgrade().unwrap() + RwLockUpgradableReadGuard::try_upgrade(upgradable).unwrap() } else { - FutureExt::block_on(upgradable.upgrade()) + FutureExt::block_on(RwLockUpgradableReadGuard::upgrade(upgradable)) }; *write = 7; assert_pending!(reader.poll()); @@ -422,7 +426,7 @@ fn cross_thread_guard_transfer() { thread::scope(|scope| { scope .spawn(move || { - let mut write = FutureExt::block_on(upgradable.upgrade()); + let mut write = FutureExt::block_on(RwLockUpgradableReadGuard::upgrade(upgradable)); *write = (1, 1); }) .join() @@ -467,7 +471,8 @@ fn concurrent_readers_writers_and_upgrades_preserve_values() { 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()); + let mut write = + FutureExt::block_on(RwLockUpgradableReadGuard::upgrade(read)); write.0 += 1; thread::yield_now(); write.1 += 1; @@ -490,7 +495,8 @@ fn constructors_unsized_values_and_exclusive_borrows() { 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(); + let mut write = + RwLockUpgradableReadGuard::try_upgrade(lock.try_upgradable_read().unwrap()).unwrap(); write[0] = 2; assert_eq!(&*write.downgrade(), &[2, 1]); } @@ -518,7 +524,7 @@ fn cancelling_an_upgrade_races_with_the_last_readers_release() { 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 upgrade = spawn(RwLockUpgradableReadGuard::upgrade(upgradable)); let mut writer = spawn(lock.write()); assert_pending!(upgrade.poll()); assert_pending!(writer.poll()); @@ -551,7 +557,7 @@ fn upgrade_observes_reader_writes_without_a_thread_join() { 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()); + let mut upgrade = Box::pin(RwLockUpgradableReadGuard::upgrade(upgradable)); if queued { assert_pending!(poll_once(upgrade.as_mut())); } @@ -580,7 +586,9 @@ 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(); + let mut write = + RwLockUpgradableReadGuard::try_upgrade(lock.try_upgradable_read().unwrap()) + .unwrap(); thread::scope(|scope| { let holder = scope.spawn(move || { *write = 42; @@ -627,7 +635,7 @@ fn upgrade_waker_replacement_and_grant_allow_reentry() { }; let upgradable = lock.try_upgradable_read().unwrap(); let reader = lock.try_read().unwrap(); - let mut upgrade = Box::pin(upgradable.upgrade()); + let mut upgrade = Box::pin(RwLockUpgradableReadGuard::upgrade(upgradable)); let old_waker = waker_on_drop(reenter()); assert_pending!(poll_with(upgrade.as_mut(), &old_waker)); drop(old_waker);