From 3b1eebb97e8f8db2cbff6c92e778a0a18059f50d Mon Sep 17 00:00:00 2001 From: Matthew Hambrecht Date: Sun, 23 Aug 2026 23:09:38 -0400 Subject: [PATCH] core swap to internal raw type --- .github/workflows/ci.yml | 22 ++ CHANGELOG.md | 2 + Cargo.lock | 181 ++++++++++ Cargo.toml | 2 + README.md | 14 +- asyncband/Cargo.toml | 3 + asyncband/src/internal/mod.rs | 11 +- asyncband/src/rwlock/loom_rwlock.rs | 326 ++++++++++++++++++ asyncband/src/rwlock/mapped_read_guard.rs | 18 +- asyncband/src/rwlock/mapped_write_guard.rs | 31 +- asyncband/src/rwlock/mod.rs | 59 +++- .../src/rwlock/owned_mapped_read_guard.rs | 2 +- .../src/rwlock/owned_mapped_write_guard.rs | 21 +- asyncband/src/rwlock/owned_read_guard.rs | 6 +- .../src/rwlock/owned_upgradable_read_guard.rs | 227 ++++++++++++ asyncband/src/rwlock/owned_write_guard.rs | 60 ++-- asyncband/src/rwlock/read_guard.rs | 10 +- asyncband/src/rwlock/upgradable_read_guard.rs | 214 ++++++++++++ asyncband/src/rwlock/write_guard.rs | 60 ++-- 19 files changed, 1155 insertions(+), 114 deletions(-) create mode 100644 asyncband/src/rwlock/loom_rwlock.rs create mode 100644 asyncband/src/rwlock/owned_upgradable_read_guard.rs create mode 100644 asyncband/src/rwlock/upgradable_read_guard.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 566de0d..4d1fbea 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -103,6 +103,26 @@ jobs: - uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 - run: cargo x bench + loom: + name: Loom concurrency check + runs-on: ubuntu-24.04 + timeout-minutes: 30 + env: + RUSTUP_TOOLCHAIN: stable + RUSTFLAGS: --cfg loom + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - name: Install stable toolchain + run: >- + rustup toolchain install stable + --profile minimal + --no-self-update + - uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 + - name: Explore modeled interleavings + run: >- + cargo test -p asyncband --lib --all-features loom_ --release + -- --test-threads=1 + required: name: Required runs-on: ubuntu-24.04 @@ -110,6 +130,7 @@ jobs: needs: - benchmark - check + - loom - test steps: - name: Guardian @@ -117,6 +138,7 @@ jobs: if [[ ! ( \ "${{ needs.benchmark.result }}" == "success" \ && "${{ needs.check.result }}" == "success" \ + && "${{ needs.loom.result }}" == "success" \ && "${{ needs.test.result }}" == "success" \ ) ]]; then echo "Required jobs haven't been completed successfully." diff --git a/CHANGELOG.md b/CHANGELOG.md index a0d21b9..dd9eddc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ All notable changes to this project will be documented in this file. * Add an opt-in `asyncband::blocking::FutureExt` bridge with `block_on` and `wait_timeout` methods for waiting on runtime-agnostic futures from synchronous code. * Add opt-in bounded and unbounded runtime-agnostic object pools under `asyncband::pool`. +* Add borrowed and owned upgradable read guards to `asyncband::rwlock::RwLock`, with atomic upgrade and downgrade operations. ### Breaking changes @@ -28,4 +29,5 @@ All notable changes to this project will be documented in this file. ### Improvements +* Replace `RwLock`'s semaphore-based coordination with a mode-aware scheduler that avoids locking the waiter queue on uncontended acquisition and release. * Remove the `slab` dependency in favor of a focused internal waiter arena. diff --git a/Cargo.lock b/Cargo.lock index c72436f..c89e891 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,15 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aho-corasick" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" +dependencies = [ + "memchr", +] + [[package]] name = "anstream" version = "1.0.0" @@ -57,6 +66,7 @@ name = "asyncband" version = "0.6.7" dependencies = [ "hashbrown", + "loom", "tokio", ] @@ -72,6 +82,7 @@ version = "0.0.0" dependencies = [ "asyncband", "divan", + "tokio", ] [[package]] @@ -304,6 +315,21 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" +[[package]] +name = "generator" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3b854b0e584ead1a33f18b2fcad7cf7be18b3875c78816b753639aa501513ae" +dependencies = [ + "cc", + "cfg-if", + "libc", + "log", + "rustversion", + "windows-link", + "windows-result", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -468,6 +494,12 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "libc" version = "0.2.189" @@ -507,6 +539,29 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "loom" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca" +dependencies = [ + "cfg-if", + "generator", + "pin-utils", + "scoped-tls", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "memchr" version = "2.8.3" @@ -524,6 +579,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num-conv" version = "0.2.2" @@ -577,6 +641,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "pollster" version = "1.0.1" @@ -639,12 +709,29 @@ dependencies = [ "bitflags", ] +[[package]] +name = "regex-automata" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + [[package]] name = "regex-lite" version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973" +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + [[package]] name = "ring" version = "0.17.14" @@ -707,6 +794,18 @@ dependencies = [ "untrusted", ] +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + +[[package]] +name = "scoped-tls" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294" + [[package]] name = "scopeguard" version = "1.2.0" @@ -766,6 +865,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "2.0.1" @@ -889,6 +997,15 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "time" version = "0.3.54" @@ -979,6 +1096,55 @@ dependencies = [ "tokio-stream", ] +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", +] + [[package]] name = "unicode-ident" version = "1.0.24" @@ -1052,6 +1218,12 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "version_check" version = "0.9.5" @@ -1088,6 +1260,15 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.52.0" diff --git a/Cargo.toml b/Cargo.toml index e5c3ad8..0420d10 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,6 +38,7 @@ hashbrown = { version = "0.17.1", default-features = false } cargo_metadata = { version = "0.23.1" } clap = { version = "4.6.5" } divan = { version = "0.1.21" } +loom = { version = "0.7.2", features = ["futures"] } pollster = { version = "1.0.1" } semver = { version = "1.0.28" } serde = { version = "1.0.229" } @@ -47,6 +48,7 @@ ureq = { version = "3.3.0", default-features = false } which = { version = "8.0.5" } [workspace.lints.rust] +unexpected_cfgs = { level = "warn", check-cfg = ["cfg(loom)"] } unknown_lints = "deny" [workspace.lints.clippy] diff --git a/README.md b/README.md index 44ecbb6..806cebe 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,7 @@ Public paths stay direct—such as `asyncband::mutex`, `asyncband::pool`, and `a | Area | API | Feature | Use | | ----------------------- | ------------------------------------------------------------------------------------ | -------------- | ----------------------------------------------------------------------- | | Shared state | [`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. | +| | [`RwLock`](https://docs.rs/asyncband/*/asyncband/rwlock/struct.RwLock.html) | `rwlock` | Allow readers, a writer, or one atomically upgradable reader. | | | [`Condvar`](https://docs.rs/asyncband/*/asyncband/condvar/struct.Condvar.html) | `condvar` | Wait for notifications while releasing a mutex. | | Initialization | [`Once`](https://docs.rs/asyncband/*/asyncband/once/struct.Once.html) | `once` | Run asynchronous initialization exactly once. | | | [`OnceCell`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceCell.html) | `once-cell` | Initialize and store one asynchronous value. | @@ -113,6 +113,18 @@ The `blocking` module is a lightweight, thread-parking single-future executor, n Asyncband types implement `Send` and `Sync` only when the protected, transferred, or managed value satisfies the necessary bounds. See each API's documentation for its exact contract. +### Loom model checking + +Asyncband uses bounded [Loom](https://github.com/tokio-rs/loom) models for concurrency-critical internals. During model checking, participating components conditionally replace their synchronization primitives with Loom's instrumented atomics, mutexes, threads, cells, and future executor. This allows focused models to explore relevant execution orderings. + +Run all current and future models separately from the normal test suite: + +```shell +RUSTFLAGS="--cfg loom" cargo test -p asyncband --lib --all-features loom_ --release -- --test-threads=1 +``` + +Models remain small and bounded so they are practical in CI. Set `LOOM_MAX_PREEMPTIONS` or `LOOM_MAX_BRANCHES` to request a deeper local run. Tests intended for the shared CI job use the `loom_` name prefix so models for additional primitives are discovered automatically. + ## Minimum Supported Rust Version (MSRV) This crate is built against the latest stable release, and its minimum supported rustc version is 1.86.0. diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml index af8e642..33819e0 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -69,5 +69,8 @@ hashbrown = { workspace = true, default-features = false, features = [ [dev-dependencies] tokio = { workspace = true, features = ["full"] } +[target.'cfg(loom)'.dependencies] +loom = { workspace = true } + [lints] workspace = true diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index caac4a2..5bb6b4b 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -53,14 +53,9 @@ pub(crate) mod once_table; ))] pub(crate) mod mutex; -#[cfg(any( - feature = "mpsc", - feature = "mutex", - feature = "rwlock", - feature = "semaphore", -))] -// `mpsc` uses `poll_acquire`, `release_if_nonempty`, and `notify_all`; mutexes and rwlocks use -// `acquire`, `try_acquire`, and `release`; the public semaphore also uses the accounting methods. +#[cfg(any(feature = "mpsc", feature = "mutex", feature = "semaphore",))] +// `mpsc` uses `poll_acquire`, `release_if_nonempty`, and `notify_all`; mutexes use `acquire`, +// `try_acquire`, and `release`; the public semaphore also uses the accounting methods. // Each single-primitive build intentionally leaves the other groups unused. #[allow(dead_code)] pub(crate) mod semaphore; diff --git a/asyncband/src/rwlock/loom_rwlock.rs b/asyncband/src/rwlock/loom_rwlock.rs new file mode 100644 index 0000000..9fbb814 --- /dev/null +++ b/asyncband/src/rwlock/loom_rwlock.rs @@ -0,0 +1,326 @@ +// 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. + +//! Loom models for the RwLock scheduler. + +use std::future::Future; +use std::future::poll_fn; +use std::pin::Pin; +use std::sync::atomic::Ordering::AcqRel; +use std::sync::atomic::Ordering::Acquire; +use std::sync::atomic::Ordering::Release; +use std::task::Poll; + +use loom::future::block_on; +use loom::model::Builder; +use loom::sync::Arc; +use loom::sync::atomic::AtomicBool; +use loom::sync::atomic::AtomicUsize; +use loom::sync::mpsc; +use loom::thread; + +use super::raw::RawRwLock; + +const MAX_READERS: usize = 2; + +/// Runs one model with CI-friendly exploration bounds. +fn check(model: impl Fn() + Send + Sync + 'static) { + let mut builder = Builder::new(); + builder.max_threads = 4; + if std::env::var_os("LOOM_MAX_BRANCHES").is_none() { + builder.max_branches = 400; + } + if std::env::var_os("LOOM_MAX_PREEMPTIONS").is_none() { + builder.preemption_bound = Some(3); + } + builder.check(model); +} + +/// Confirms that no ownership or queued waiter remains. +fn assert_fully_unlocked(lock: &RawRwLock) { + assert!(lock.try_write(), "lock ownership or a waiter was leaked"); + lock.unlock_write(); +} + +/// Polls an acquisition once and confirms that it queued. +fn poll_pending_once(future: Pin<&mut F>) +where + F: Future, +{ + let mut future = future; + block_on(poll_fn(|cx| { + assert!(future.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + })); +} + +/// Checks that read and write ownership never overlap. +#[test] +fn loom_reader_writer_exclusion() { + check(|| { + let lock = Arc::new(RawRwLock::new(MAX_READERS)); + let readers = Arc::new(AtomicUsize::new(0)); + let writer = Arc::new(AtomicBool::new(false)); + + let reader_thread = { + let lock = lock.clone(); + let readers = readers.clone(); + let writer = writer.clone(); + thread::spawn(move || { + block_on(lock.read()); + assert!(!writer.load(Acquire)); + readers.fetch_add(1, AcqRel); + thread::yield_now(); + assert!(!writer.load(Acquire)); + readers.fetch_sub(1, AcqRel); + lock.unlock_read(); + }) + }; + + let writer_thread = { + let lock = lock.clone(); + let readers = readers.clone(); + let writer = writer.clone(); + thread::spawn(move || { + block_on(lock.write()); + assert_eq!(readers.load(Acquire), 0); + assert!(!writer.swap(true, AcqRel)); + thread::yield_now(); + assert_eq!(readers.load(Acquire), 0); + writer.store(false, Release); + lock.unlock_write(); + }) + }; + + reader_thread.join().unwrap(); + writer_thread.join().unwrap(); + assert_fully_unlocked(&lock); + }); +} + +/// Checks writer publication racing the final reader release. +#[test] +fn loom_writer_publication_races_last_reader_release() { + check(|| { + let lock = Arc::new(RawRwLock::new(MAX_READERS)); + block_on(lock.read()); + + let writer = { + let lock = lock.clone(); + thread::spawn(move || { + block_on(lock.write()); + lock.unlock_write(); + }) + }; + + lock.unlock_read(); + writer.join().unwrap(); + assert_fully_unlocked(&lock); + }); +} + +/// Checks reader publication racing a writer release. +#[test] +fn loom_reader_publication_races_writer_release() { + check(|| { + let lock = Arc::new(RawRwLock::new(MAX_READERS)); + block_on(lock.write()); + + let reader = { + let lock = lock.clone(); + thread::spawn(move || { + block_on(lock.read()); + lock.unlock_read(); + }) + }; + + lock.unlock_write(); + reader.join().unwrap(); + assert_fully_unlocked(&lock); + }); +} + +/// Checks cancellation immediately before or after a writer grant. +#[test] +fn loom_writer_cancellation_races_grant() { + check(|| { + let lock = Arc::new(RawRwLock::new(MAX_READERS)); + block_on(lock.read()); + let (published, published_rx) = mpsc::channel(); + + let cancellation = { + let lock = lock.clone(); + thread::spawn(move || { + let mut acquire = Box::pin(lock.write()); + poll_pending_once(acquire.as_mut()); + published.send(()).unwrap(); + thread::yield_now(); + drop(acquire); + }) + }; + + published_rx.recv().unwrap(); + lock.unlock_read(); + cancellation.join().unwrap(); + assert_fully_unlocked(&lock); + }); +} + +/// Checks cancellation immediately before or after an upgrade grant. +#[test] +fn loom_upgrade_cancellation_races_grant() { + check(|| { + let lock = Arc::new(RawRwLock::new(MAX_READERS)); + block_on(lock.upgradable_read()); + block_on(lock.read()); + let (published, published_rx) = mpsc::channel(); + + let cancellation = { + let lock = lock.clone(); + thread::spawn(move || { + let mut upgrade = Box::pin(lock.upgrade()); + poll_pending_once(upgrade.as_mut()); + published.send(()).unwrap(); + thread::yield_now(); + drop(upgrade); + }) + }; + + published_rx.recv().unwrap(); + lock.unlock_read(); + cancellation.join().unwrap(); + assert_fully_unlocked(&lock); + }); +} + +/// Checks that cancelling the queue head wakes its follower. +#[test] +fn loom_queue_head_cancellation_wakes_follower() { + check(|| { + let lock = Arc::new(RawRwLock::new(MAX_READERS)); + block_on(lock.write()); + + let (head_published, head_published_rx) = mpsc::channel(); + let (cancel_head, cancel_head_rx) = mpsc::channel(); + let head = { + let lock = lock.clone(); + thread::spawn(move || { + let mut acquire = Box::pin(lock.write()); + poll_pending_once(acquire.as_mut()); + head_published.send(()).unwrap(); + cancel_head_rx.recv().unwrap(); + drop(acquire); + }) + }; + head_published_rx.recv().unwrap(); + + let (follower_published, follower_published_rx) = mpsc::channel(); + let follower = { + let lock = lock.clone(); + thread::spawn(move || { + let mut acquire = Box::pin(lock.read()); + poll_pending_once(acquire.as_mut()); + follower_published.send(()).unwrap(); + block_on(acquire.as_mut()); + lock.unlock_read(); + }) + }; + follower_published_rx.recv().unwrap(); + + cancel_head.send(()).unwrap(); + lock.unlock_write(); + head.join().unwrap(); + follower.join().unwrap(); + assert_fully_unlocked(&lock); + }); +} + +/// Checks that an upgrade precedes a later queued writer. +#[test] +fn loom_upgrade_precedes_queued_writer() { + check(|| { + let lock = Arc::new(RawRwLock::new(MAX_READERS)); + let acquisition_order = Arc::new(AtomicUsize::new(0)); + block_on(lock.upgradable_read()); + block_on(lock.read()); + + let (writer_published, writer_published_rx) = mpsc::channel(); + let writer = { + let lock = lock.clone(); + let acquisition_order = acquisition_order.clone(); + thread::spawn(move || { + let mut acquire = Box::pin(lock.write()); + poll_pending_once(acquire.as_mut()); + writer_published.send(()).unwrap(); + block_on(acquire.as_mut()); + assert_eq!( + acquisition_order.compare_exchange(1, 2, AcqRel, Acquire), + Ok(1) + ); + lock.unlock_write(); + }) + }; + writer_published_rx.recv().unwrap(); + + let (upgrade_published, upgrade_published_rx) = mpsc::channel(); + let upgrade = { + let lock = lock.clone(); + let acquisition_order = acquisition_order.clone(); + thread::spawn(move || { + let mut acquire = Box::pin(lock.upgrade()); + poll_pending_once(acquire.as_mut()); + upgrade_published.send(()).unwrap(); + block_on(acquire.as_mut()); + assert_eq!( + acquisition_order.compare_exchange(0, 1, AcqRel, Acquire), + Ok(0) + ); + lock.unlock_write(); + }) + }; + upgrade_published_rx.recv().unwrap(); + + lock.unlock_read(); + upgrade.join().unwrap(); + writer.join().unwrap(); + assert_eq!(acquisition_order.load(Acquire), 2); + assert_fully_unlocked(&lock); + }); +} + +/// Checks a write-to-read downgrade racing reader publication. +#[test] +fn loom_downgrade_races_reader_publication() { + check(|| { + let lock = Arc::new(RawRwLock::new(MAX_READERS)); + block_on(lock.write()); + + let reader = { + let lock = lock.clone(); + thread::spawn(move || { + block_on(lock.read()); + lock.unlock_read(); + }) + }; + + lock.downgrade_write_to_read(); + assert!(!lock.try_write()); + lock.unlock_read(); + reader.join().unwrap(); + assert_fully_unlocked(&lock); + }); +} diff --git a/asyncband/src/rwlock/mapped_read_guard.rs b/asyncband/src/rwlock/mapped_read_guard.rs index 4d2094d..2b0923e 100644 --- a/asyncband/src/rwlock/mapped_read_guard.rs +++ b/asyncband/src/rwlock/mapped_read_guard.rs @@ -20,7 +20,7 @@ use std::marker::PhantomData; use std::ops::Deref; use std::ptr::NonNull; -use crate::internal::semaphore; +use crate::rwlock::RawRwLock; /// RAII structure used to release the shared read access of a lock when dropped, for a mapped /// component of the locked data. @@ -30,7 +30,7 @@ use crate::internal::semaphore; /// access control while maintaining the same locking semantics. /// /// As long as you have this guard, you have shared read access to the underlying `T`. The guard -/// internally keeps a reference to the original rwlock's semaphore, so the original lock is +/// internally keeps a reference to the original rwlock's scheduler, so the original lock is /// maintained until this guard is dropped. /// /// `MappedRwLockReadGuard` implements [`Send`] and [`Sync`] when `T: Sync`, allowing it to be @@ -82,14 +82,14 @@ use crate::internal::semaphore; #[must_use = "if unused the RwLock will immediately unlock"] pub struct MappedRwLockReadGuard<'a, T: ?Sized> { d: NonNull, - s: &'a semaphore::Semaphore, + raw: &'a RawRwLock, variance: PhantomData T>, } // SAFETY: MappedRwLockReadGuard is Send when T: Sync. We don't require T: Send because // the guard RwLockReadGuard doesn't transfer ownership of T - it only holds a shared reference. // When moved to another thread, the guard maintains the read lock and the new thread -// can safely access &T (which is allowed since T: Sync). The semaphore reference +// can safely access &T (which is allowed since T: Sync). The scheduler reference // and NonNull pointer are both safe to transfer between threads. unsafe impl Send for MappedRwLockReadGuard<'_, T> {} @@ -98,10 +98,10 @@ unsafe impl Send for MappedRwLockReadGuard<'_, T> {} unsafe impl Sync for MappedRwLockReadGuard<'_, T> {} impl<'a, T: ?Sized> MappedRwLockReadGuard<'a, T> { - pub(crate) fn new(d: NonNull, s: &'a semaphore::Semaphore) -> Self { + pub(super) fn new(d: NonNull, raw: &'a RawRwLock) -> Self { Self { d, - s, + raw, variance: PhantomData, } } @@ -109,7 +109,7 @@ impl<'a, T: ?Sized> MappedRwLockReadGuard<'a, T> { impl Drop for MappedRwLockReadGuard<'_, T> { fn drop(&mut self) { - self.s.release(1); + self.raw.unlock_read(); } } @@ -192,7 +192,7 @@ impl<'a, T: ?Sized> MappedRwLockReadGuard<'a, T> { // access to the data through the rwlock, so dereferencing is safe. let d = NonNull::from(f(unsafe { orig.d.as_ref() })); let orig = std::mem::ManuallyDrop::new(orig); - MappedRwLockReadGuard::new(d, orig.s) + MappedRwLockReadGuard::new(d, orig.raw) } /// Attempts to make a new [`MappedRwLockReadGuard`] for a component of the locked data. The @@ -260,7 +260,7 @@ impl<'a, T: ?Sized> MappedRwLockReadGuard<'a, T> { Some(d) => { let d = NonNull::from(d); let orig = std::mem::ManuallyDrop::new(orig); - Ok(MappedRwLockReadGuard::new(d, orig.s)) + Ok(MappedRwLockReadGuard::new(d, orig.raw)) } None => Err(orig), } diff --git a/asyncband/src/rwlock/mapped_write_guard.rs b/asyncband/src/rwlock/mapped_write_guard.rs index 23c05dc..dbf207a 100644 --- a/asyncband/src/rwlock/mapped_write_guard.rs +++ b/asyncband/src/rwlock/mapped_write_guard.rs @@ -22,8 +22,8 @@ use std::ops::Deref; use std::ops::DerefMut; use std::ptr::NonNull; -use crate::internal::semaphore; use crate::rwlock::MappedRwLockReadGuard; +use crate::rwlock::RawRwLock; /// RAII structure used to release the exclusive write access of a lock when dropped, for a mapped /// component of the locked data. @@ -33,8 +33,8 @@ use crate::rwlock::MappedRwLockReadGuard; /// access control while maintaining the same locking semantics. /// /// As long as you have this guard, you have exclusive write access to the underlying `T`. The guard -/// internally keeps a reference to the original rwlock's semaphore and tracks the number of permits -/// acquired, so the original lock is maintained until this guard is dropped. +/// internally keeps a reference to the original rwlock's scheduler, so the original lock is +/// maintained until this guard is dropped. /// /// `MappedRwLockWriteGuard` implements [`Send`] when the underlying data type implements [`Send`], /// and implements [`Sync`] when the underlying data type implements both [`Send`] and [`Sync`], @@ -103,8 +103,7 @@ use crate::rwlock::MappedRwLockReadGuard; #[must_use = "if unused the RwLock will immediately unlock"] pub struct MappedRwLockWriteGuard<'a, T: ?Sized> { d: NonNull, - s: &'a semaphore::Semaphore, - permits_acquired: usize, + raw: &'a RawRwLock, // Mutable access requires invariance over T. variance: PhantomData<&'a mut T>, } @@ -119,11 +118,10 @@ unsafe impl Sync for MappedRwLockWriteGuard<'_, T> {} unsafe impl Send for MappedRwLockWriteGuard<'_, T> {} impl<'a, T: ?Sized> MappedRwLockWriteGuard<'a, T> { - pub(crate) fn new(d: NonNull, s: &'a semaphore::Semaphore, permits_acquired: usize) -> Self { + pub(super) fn new(d: NonNull, raw: &'a RawRwLock) -> Self { Self { d, - s, - permits_acquired, + raw, variance: PhantomData, } } @@ -131,7 +129,7 @@ impl<'a, T: ?Sized> MappedRwLockWriteGuard<'a, T> { impl Drop for MappedRwLockWriteGuard<'_, T> { fn drop(&mut self) { - self.s.release(self.permits_acquired); + self.raw.unlock_write(); } } @@ -222,9 +220,8 @@ impl<'a, T: ?Sized> MappedRwLockWriteGuard<'a, T> { // when the original MappedRwLockWriteGuard was constructed. The guard guarantees exclusive // access to the data through the rwlock, so dereferencing is safe. let d = NonNull::from(f(unsafe { orig.d.as_mut() })); - let permits_acquired = orig.permits_acquired; let orig = ManuallyDrop::new(orig); - MappedRwLockWriteGuard::new(d, orig.s, permits_acquired) + MappedRwLockWriteGuard::new(d, orig.raw) } /// Attempts to make a new [`MappedRwLockWriteGuard`] for a component of the locked data. The @@ -302,9 +299,8 @@ impl<'a, T: ?Sized> MappedRwLockWriteGuard<'a, T> { match f(unsafe { orig.d.as_mut() }) { Some(d) => { let d = NonNull::from(d); - let permits_acquired = orig.permits_acquired; let orig = ManuallyDrop::new(orig); - Ok(MappedRwLockWriteGuard::new(d, orig.s, permits_acquired)) + Ok(MappedRwLockWriteGuard::new(d, orig.raw)) } None => Err(orig), } @@ -354,14 +350,13 @@ impl<'a, T: ?Sized> MappedRwLockWriteGuard<'a, T> { /// ``` pub fn downgrade(self) -> MappedRwLockReadGuard<'a, T> { // Prevent the original write guard from running its Drop implementation, - // which would release all permits. This must be done BEFORE any operation + // which would unlock the rwlock. This must be done BEFORE any operation // that might panic to ensure panic safety. let guard = ManuallyDrop::new(self); - // Release max_readers - 1 permits to convert the write lock to a read lock. - guard.s.release(guard.permits_acquired - 1); + guard.raw.downgrade_write_to_read(); - // Create the mapped read guard with 1 permit (standard for read locks) - MappedRwLockReadGuard::new(guard.d, guard.s) + // Create a mapped read guard retaining the new shared ownership. + MappedRwLockReadGuard::new(guard.d, guard.raw) } } diff --git a/asyncband/src/rwlock/mod.rs b/asyncband/src/rwlock/mod.rs index 9f04847..13c9897 100644 --- a/asyncband/src/rwlock/mod.rs +++ b/asyncband/src/rwlock/mod.rs @@ -15,11 +15,13 @@ // specific language governing permissions and limitations // under the License. -//! A reader-writer lock that allows multiple readers or a single writer at a time. +//! A reader-writer lock with atomically upgradable read access. //! //! This type of lock allows a number of readers or at most one writer at any point in time. The //! write portion of this lock typically allows modification of the underlying data (exclusive //! access) and the read portion of this lock typically allows for read-only access (shared access). +//! One [`upgradable read`](RwLock::upgradable_read) may coexist with ordinary readers and can later +//! be atomically promoted to exclusive write access. //! //! In comparison, a [`Mutex`] does not distinguish between readers or writers that acquire the //! lock, therefore causing any tasks waiting for the lock to become available to yield. An RwLock @@ -32,6 +34,10 @@ //! the lock until that writer has acquired and released it. In contrast, the priority policy of //! the Rust standard library's `std::sync::RwLock` depends on the operating system. //! +//! An upgradable read reserves promotion priority over requests made after it acquires the lock. +//! When promotion is explicitly requested, existing readers finish but later writers and readers +//! cannot overtake the upgrade. This reservation makes promotion atomic and deadlock-free. +//! //! The type parameter `T` represents the data that this lock protects. It is required that `T` //! satisfies [`Send`] to be shared across threads. The RAII guards returned from the locking //! methods implement [`Deref`] (and [`DerefMut`] for the `write` method) to allow access to the @@ -61,6 +67,16 @@ //! assert_eq!(*w, 6); //! } // write lock is dropped here //! +//! // an upgradable reader can inspect before deciding whether to write +//! { +//! let r = lock.upgradable_read().await; +//! if *r == 6 { +//! let mut w = r.upgrade().await; +//! *w += 1; +//! } +//! } +//! assert_eq!(*lock.read().await, 7); +//! //! # } //! ``` //! @@ -73,8 +89,6 @@ use std::cell::UnsafeCell; use std::fmt; use std::num::NonZeroUsize; -use crate::internal::semaphore::Semaphore; - mod mapped_read_guard; pub use mapped_read_guard::MappedRwLockReadGuard; mod mapped_write_guard; @@ -87,21 +101,25 @@ mod owned_read_guard; pub use owned_read_guard::OwnedRwLockReadGuard; mod owned_write_guard; pub use owned_write_guard::OwnedRwLockWriteGuard; +mod owned_upgradable_read_guard; +pub use owned_upgradable_read_guard::OwnedRwLockUpgradableReadGuard; mod read_guard; pub use read_guard::RwLockReadGuard; +mod upgradable_read_guard; +pub use upgradable_read_guard::RwLockUpgradableReadGuard; mod write_guard; pub use write_guard::RwLockWriteGuard; +mod raw; +use raw::RawRwLock; +#[cfg(all(test, loom))] +mod loom_rwlock; -/// A reader-writer lock that allows multiple readers or a single writer at a time. +/// A reader-writer lock allowing readers, one writer, or one upgradable reader at a time. /// /// See the [module level documentation](self) for more. pub struct RwLock { - /// Maximum number of concurrent readers. - /// - /// This is ensured to be non-zero. - max_readers: usize, - /// Semaphore to coordinate read and write access to T - s: Semaphore, + /// Mode-aware scheduler coordinating readers, writers, and upgrades. + raw: RawRwLock, /// The inner data. c: UnsafeCell, } @@ -142,11 +160,18 @@ impl RwLock { /// /// let rwlock = RwLock::new(5); /// ``` + #[cfg(not(loom))] pub const fn new(t: T) -> RwLock { // large enough while not touch the edge RwLock::with_max_readers(t, NonZeroUsize::new(usize::MAX >> 1).unwrap()) } + #[cfg(loom)] + /// Creates a new reader-writer lock in an unlocked state for Loom model checking. + pub fn new(t: T) -> RwLock { + RwLock::with_max_readers(t, NonZeroUsize::new(usize::MAX >> 1).unwrap()) + } + /// Creates a new reader-writer lock in an unlocked state, and allows a maximum of /// `max_readers` concurrent readers. /// @@ -162,11 +187,21 @@ impl RwLock { /// let max_readers = NonZeroUsize::new(1024).expect("max_readers must be non-zero"); /// let rwlock = RwLock::with_max_readers(5, max_readers); /// ``` + #[cfg(not(loom))] pub const fn with_max_readers(t: T, max_readers: NonZeroUsize) -> RwLock { let max_readers = max_readers.get(); - let s = Semaphore::new(max_readers); + let raw = RawRwLock::new(max_readers); + let c = UnsafeCell::new(t); + RwLock { raw, c } + } + + #[cfg(loom)] + /// Creates an unlocked reader-writer lock with a reader limit for Loom model checking. + pub fn with_max_readers(t: T, max_readers: NonZeroUsize) -> RwLock { + let max_readers = max_readers.get(); + let raw = RawRwLock::new(max_readers); let c = UnsafeCell::new(t); - RwLock { max_readers, c, s } + RwLock { raw, c } } /// Consumes the lock, returning the underlying data. diff --git a/asyncband/src/rwlock/owned_mapped_read_guard.rs b/asyncband/src/rwlock/owned_mapped_read_guard.rs index 449bcd4..e5763eb 100644 --- a/asyncband/src/rwlock/owned_mapped_read_guard.rs +++ b/asyncband/src/rwlock/owned_mapped_read_guard.rs @@ -116,7 +116,7 @@ impl OwnedMappedRwLockReadGuard { } impl Drop for OwnedMappedRwLockReadGuard { fn drop(&mut self) { - self.lock.s.release(1); + self.lock.raw.unlock_read(); } } diff --git a/asyncband/src/rwlock/owned_mapped_write_guard.rs b/asyncband/src/rwlock/owned_mapped_write_guard.rs index d34a2b8..62a11f5 100644 --- a/asyncband/src/rwlock/owned_mapped_write_guard.rs +++ b/asyncband/src/rwlock/owned_mapped_write_guard.rs @@ -35,8 +35,8 @@ use crate::rwlock::RwLock; /// the `'static` lifetime. /// /// As long as you have this guard, you have exclusive write access to the underlying `T`. The guard -/// internally keeps an `Arc` reference to the original rwlock and tracks the number of permits -/// acquired, so the original lock is maintained until this guard is dropped. +/// internally keeps an `Arc` reference to the original rwlock, so the original lock is maintained +/// until this guard is dropped. /// /// `OwnedMappedRwLockWriteGuard` implements [`Send`] and [`Sync`] /// when the underlying data type supports these traits, allowing it to be used across task @@ -109,7 +109,6 @@ use crate::rwlock::RwLock; pub struct OwnedMappedRwLockWriteGuard { d: NonNull, lock: Arc>, - permits_acquired: usize, // Mutable access requires invariance over U. variance: PhantomData<*mut U>, } @@ -125,18 +124,17 @@ unsafe impl Sync for OwnedMappedRwLoc unsafe impl Send for OwnedMappedRwLockWriteGuard {} impl OwnedMappedRwLockWriteGuard { - pub(crate) fn new(d: NonNull, lock: Arc>, permits_acquired: usize) -> Self { + pub(crate) fn new(d: NonNull, lock: Arc>) -> Self { Self { d, lock, - permits_acquired, variance: PhantomData, } } } impl Drop for OwnedMappedRwLockWriteGuard { fn drop(&mut self) { - self.lock.s.release(self.permits_acquired); + self.lock.raw.unlock_write(); } } @@ -232,13 +230,12 @@ impl OwnedMappedRwLockWriteGuard { let d = NonNull::from(f(unsafe { orig.d.as_mut() })); let orig = ManuallyDrop::new(orig); - let permits_acquired = orig.permits_acquired; // SAFETY: The original guard is wrapped in `ManuallyDrop` and will not be dropped. // This allows us to safely move the `Arc` out of it and transfer ownership to the new // guard. let lock = unsafe { std::ptr::read(&orig.lock) }; - OwnedMappedRwLockWriteGuard::new(d, lock, permits_acquired) + OwnedMappedRwLockWriteGuard::new(d, lock) } /// Attempts to make a new [`OwnedMappedRwLockWriteGuard`] for a component of the locked data. @@ -324,13 +321,12 @@ impl OwnedMappedRwLockWriteGuard { let d = NonNull::from(d); let orig = ManuallyDrop::new(orig); - let permits_acquired = orig.permits_acquired; // SAFETY: The original guard is wrapped in `ManuallyDrop` and will not be dropped. // This allows us to safely move the `Arc` out of it and transfer ownership to the // new guard. let lock = unsafe { std::ptr::read(&orig.lock) }; - Ok(OwnedMappedRwLockWriteGuard::new(d, lock, permits_acquired)) + Ok(OwnedMappedRwLockWriteGuard::new(d, lock)) } None => Err(orig), } @@ -381,12 +377,11 @@ impl OwnedMappedRwLockWriteGuard { /// ``` pub fn downgrade(self) -> OwnedMappedRwLockReadGuard { // Prevent the original write guard from running its Drop implementation, - // which would release all permits. This must be done BEFORE any operation + // which would unlock the rwlock. This must be done BEFORE any operation // that might panic to ensure panic safety. let guard = ManuallyDrop::new(self); - // Release max_readers - 1 permits to convert the write lock to a read lock. - guard.lock.s.release(guard.permits_acquired - 1); + guard.lock.raw.downgrade_write_to_read(); // SAFETY: The `guard` is wrapped in `ManuallyDrop`, so its destructor will not be run. // We can safely move the `Arc` out of the guard, as the guard is not used after this. diff --git a/asyncband/src/rwlock/owned_read_guard.rs b/asyncband/src/rwlock/owned_read_guard.rs index d902dd0..2d30b4b 100644 --- a/asyncband/src/rwlock/owned_read_guard.rs +++ b/asyncband/src/rwlock/owned_read_guard.rs @@ -71,7 +71,7 @@ impl RwLock { /// # } /// ``` pub async fn read_owned(self: Arc) -> OwnedRwLockReadGuard { - self.s.acquire(1).await; + self.raw.read().await; OwnedRwLockReadGuard { lock: self } } @@ -102,7 +102,7 @@ impl RwLock { /// assert!(lock.clone().try_read_owned().is_none()); /// ``` pub fn try_read_owned(self: Arc) -> Option> { - if self.s.try_acquire(1) { + if self.raw.try_read() { Some(OwnedRwLockReadGuard { lock: self }) } else { None @@ -122,7 +122,7 @@ pub struct OwnedRwLockReadGuard { impl Drop for OwnedRwLockReadGuard { fn drop(&mut self) { - self.lock.s.release(1); + self.lock.raw.unlock_read(); } } diff --git a/asyncband/src/rwlock/owned_upgradable_read_guard.rs b/asyncband/src/rwlock/owned_upgradable_read_guard.rs new file mode 100644 index 0000000..d4a82b3 --- /dev/null +++ b/asyncband/src/rwlock/owned_upgradable_read_guard.rs @@ -0,0 +1,227 @@ +// 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::ManuallyDrop; +use std::ops::Deref; +use std::sync::Arc; + +use crate::rwlock::OwnedRwLockReadGuard; +use crate::rwlock::OwnedRwLockWriteGuard; +use crate::rwlock::RwLock; + +impl RwLock { + /// Acquires an owned upgradable read guard from an [`Arc`]. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use std::sync::Arc; + /// + /// use asyncband::rwlock::RwLock; + /// + /// let lock = Arc::new(RwLock::new(5)); + /// let guard = lock.clone().upgradable_read_owned().await; + /// assert_eq!(*guard, 5); + /// # } + /// ``` + pub async fn upgradable_read_owned(self: Arc) -> OwnedRwLockUpgradableReadGuard { + self.raw.upgradable_read().await; + OwnedRwLockUpgradableReadGuard { lock: self } + } + + /// Attempts to acquire an owned upgradable read guard without waiting. + /// + /// # Examples + /// + /// ``` + /// use std::sync::Arc; + /// + /// use asyncband::rwlock::RwLock; + /// + /// let lock = Arc::new(RwLock::new(1)); + /// let guard = lock + /// .clone() + /// .try_upgradable_read_owned() + /// .expect("lock is available"); + /// assert!(lock.clone().try_upgradable_read_owned().is_none()); + /// + /// drop(guard); + /// assert!(lock.try_upgradable_read_owned().is_some()); + /// ``` + pub fn try_upgradable_read_owned(self: Arc) -> Option> { + self.raw + .try_upgradable_read() + .then(|| OwnedRwLockUpgradableReadGuard { lock: self }) + } +} + +/// An owned shared read guard that can be atomically promoted to exclusive write access. +/// +/// The guard keeps its lock alive and can be moved independently of the borrow used to acquire it. +/// At most one owned or borrowed upgradable guard may exist for a lock. +/// +/// # Examples +/// +/// ``` +/// # #[tokio::main] +/// # async fn main() { +/// use std::sync::Arc; +/// +/// use asyncband::rwlock::OwnedRwLockUpgradableReadGuard; +/// use asyncband::rwlock::RwLock; +/// +/// let lock = Arc::new(RwLock::new(4)); +/// let guard: OwnedRwLockUpgradableReadGuard = lock.clone().upgradable_read_owned().await; +/// +/// let value = tokio::spawn(async move { *guard }).await.unwrap(); +/// assert_eq!(value, 4); +/// # } +/// ``` +#[must_use = "if unused the RwLock will immediately unlock"] +pub struct OwnedRwLockUpgradableReadGuard { + pub(super) lock: Arc>, +} + +unsafe impl Send for OwnedRwLockUpgradableReadGuard {} +unsafe impl Sync for OwnedRwLockUpgradableReadGuard {} + +impl Drop for OwnedRwLockUpgradableReadGuard { + fn drop(&mut self) { + self.lock.raw.unlock_upgradable(); + } +} + +impl Deref for OwnedRwLockUpgradableReadGuard { + type Target = T; + + fn deref(&self) -> &T { + // SAFETY: an upgradable guard owns shared access to the protected value. + unsafe { &*self.lock.c.get() } + } +} + +impl fmt::Debug for OwnedRwLockUpgradableReadGuard { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Debug::fmt(&**self, f) + } +} + +impl fmt::Display for OwnedRwLockUpgradableReadGuard { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&**self, f) + } +} + +impl OwnedRwLockUpgradableReadGuard { + /// Atomically promotes this guard to owned exclusive write access. + /// + /// Cancelling the operation releases the upgradable read guard. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use std::sync::Arc; + /// + /// use asyncband::rwlock::RwLock; + /// + /// let lock = Arc::new(RwLock::new(1)); + /// let guard = lock.clone().upgradable_read_owned().await; + /// let mut writer = guard.upgrade().await; + /// *writer += 1; + /// + /// assert_eq!(*writer, 2); + /// # } + /// ``` + pub async fn upgrade(self) -> OwnedRwLockWriteGuard { + let guard = ManuallyDrop::new(self); + // SAFETY: the source guard will never be dropped; ownership of its Arc moves into this + // local, which remains alive across the raw upgrade future and moves into the result. + let lock = unsafe { std::ptr::read(&guard.lock) }; + lock.raw.upgrade().await; + OwnedRwLockWriteGuard { lock } + } + + /// Attempts to promote immediately, returning the original guard when other readers remain. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use std::sync::Arc; + /// + /// use asyncband::rwlock::RwLock; + /// + /// let lock = Arc::new(RwLock::new(1)); + /// let upgradable = lock.clone().upgradable_read_owned().await; + /// let reader = lock.read().await; + /// + /// let upgradable = upgradable + /// .try_upgrade() + /// .expect_err("reader still holds the lock"); + /// drop(reader); + /// let writer = upgradable.try_upgrade().expect("last reader has left"); + /// assert_eq!(*writer, 1); + /// # } + /// ``` + pub fn try_upgrade(self) -> Result, Self> { + if self.lock.raw.try_upgrade() { + let guard = ManuallyDrop::new(self); + // SAFETY: the source guard will not drop and ownership moves to the write guard. + let lock = unsafe { std::ptr::read(&guard.lock) }; + Ok(OwnedRwLockWriteGuard { lock }) + } else { + Err(self) + } + } + + /// Atomically converts this guard to an owned ordinary shared read guard. + /// + /// Downgrading relinquishes the unique promotion reservation while retaining shared access. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use std::sync::Arc; + /// + /// use asyncband::rwlock::RwLock; + /// + /// let lock = Arc::new(RwLock::new(1)); + /// let upgradable = lock.clone().upgradable_read_owned().await; + /// let reader = upgradable.downgrade(); + /// + /// assert!(lock.clone().try_upgradable_read_owned().is_some()); + /// assert!(lock.clone().try_write_owned().is_none()); + /// drop(reader); + /// assert!(lock.try_write_owned().is_some()); + /// # } + /// ``` + pub fn downgrade(self) -> OwnedRwLockReadGuard { + let guard = ManuallyDrop::new(self); + guard.lock.raw.downgrade_upgradable_to_read(); + // SAFETY: the source guard will not drop and ownership moves to the read guard. + let lock = unsafe { std::ptr::read(&guard.lock) }; + OwnedRwLockReadGuard { lock } + } +} diff --git a/asyncband/src/rwlock/owned_write_guard.rs b/asyncband/src/rwlock/owned_write_guard.rs index 634dacd..4d5efa2 100644 --- a/asyncband/src/rwlock/owned_write_guard.rs +++ b/asyncband/src/rwlock/owned_write_guard.rs @@ -24,6 +24,7 @@ use std::sync::Arc; use crate::rwlock::OwnedMappedRwLockWriteGuard; use crate::rwlock::OwnedRwLockReadGuard; +use crate::rwlock::OwnedRwLockUpgradableReadGuard; use crate::rwlock::RwLock; impl RwLock { @@ -60,11 +61,8 @@ impl RwLock { /// # } /// ``` pub async fn write_owned(self: Arc) -> OwnedRwLockWriteGuard { - self.s.acquire(self.max_readers).await; - OwnedRwLockWriteGuard { - permits_acquired: self.max_readers, - lock: self, - } + self.raw.write().await; + OwnedRwLockWriteGuard { lock: self } } /// Attempts to acquire this `RwLock` with exclusive write access. @@ -94,11 +92,8 @@ impl RwLock { /// *v = 2; /// ``` pub fn try_write_owned(self: Arc) -> Option> { - if self.s.try_acquire(self.max_readers) { - Some(OwnedRwLockWriteGuard { - permits_acquired: self.max_readers, - lock: self, - }) + if self.raw.try_write() { + Some(OwnedRwLockWriteGuard { lock: self }) } else { None } @@ -129,7 +124,6 @@ impl RwLock { /// ``` #[must_use = "if unused the RwLock will immediately unlock"] pub struct OwnedRwLockWriteGuard { - pub(super) permits_acquired: usize, pub(super) lock: Arc>, } @@ -138,7 +132,7 @@ unsafe impl Sync for OwnedRwLockWriteGuard {} impl Drop for OwnedRwLockWriteGuard { fn drop(&mut self) { - self.lock.s.release(self.permits_acquired); + self.lock.raw.unlock_write(); } } @@ -216,13 +210,12 @@ impl OwnedRwLockWriteGuard { let d = NonNull::from(f(unsafe { &mut *orig.lock.c.get() })); let orig = ManuallyDrop::new(orig); - let permits_acquired = orig.permits_acquired; // SAFETY: The original guard is wrapped in `ManuallyDrop` and will not be dropped. // This allows us to safely move the `Arc` out of it and transfer ownership to the new // guard. let lock = unsafe { std::ptr::read(&orig.lock) }; - OwnedMappedRwLockWriteGuard::new(d, lock, permits_acquired) + OwnedMappedRwLockWriteGuard::new(d, lock) } /// Attempts to make a new [`OwnedMappedRwLockWriteGuard`] for a component of the @@ -285,13 +278,12 @@ impl OwnedRwLockWriteGuard { let orig = ManuallyDrop::new(orig); - let permits_acquired = orig.permits_acquired; // SAFETY: The original guard is wrapped in `ManuallyDrop` and will not be dropped. // This allows us to safely move the `Arc` out of it and transfer ownership to the new // guard. let lock = unsafe { std::ptr::read(&orig.lock) }; - Ok(OwnedMappedRwLockWriteGuard::new(d, lock, permits_acquired)) + Ok(OwnedMappedRwLockWriteGuard::new(d, lock)) } /// Atomically downgrades the write lock to a read lock. @@ -327,13 +319,11 @@ impl OwnedRwLockWriteGuard { /// ``` pub fn downgrade(self) -> OwnedRwLockReadGuard { // Prevent the original write guard from running its Drop implementation, - // which would release all permits. This must be done BEFORE any operation + // which would unlock the rwlock. This must be done BEFORE any operation // that might panic to ensure panic safety. let guard = ManuallyDrop::new(self); - // Release max_readers - 1 permits to convert the write lock to a read lock. - // The remaining 1 permit is kept for the read lock. - guard.lock.s.release(guard.permits_acquired - 1); + guard.lock.raw.downgrade_write_to_read(); // SAFETY: The `guard` is wrapped in `ManuallyDrop`, so its destructor will not be run. // We can safely move the `Arc` out of the guard, as the guard is not used after this. @@ -341,4 +331,34 @@ impl OwnedRwLockWriteGuard { let lock = unsafe { std::ptr::read(&guard.lock) }; OwnedRwLockReadGuard { lock } } + + /// Atomically downgrades this guard to an owned upgradable read guard. + /// + /// The returned guard keeps the lock alive and reserves the ability to promote again. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use std::sync::Arc; + /// + /// use asyncband::rwlock::RwLock; + /// + /// let lock = Arc::new(RwLock::new(1)); + /// let writer = lock.clone().write_owned().await; + /// let upgradable = writer.downgrade_to_upgradable(); + /// + /// assert!(lock.try_read().is_some()); + /// assert!(lock.clone().try_upgradable_read_owned().is_none()); + /// assert_eq!(*upgradable, 1); + /// # } + /// ``` + pub fn downgrade_to_upgradable(self) -> OwnedRwLockUpgradableReadGuard { + let guard = ManuallyDrop::new(self); + guard.lock.raw.downgrade_write_to_upgradable(); + // SAFETY: the source guard will not drop and ownership moves to the upgradable guard. + let lock = unsafe { std::ptr::read(&guard.lock) }; + OwnedRwLockUpgradableReadGuard { lock } + } } diff --git a/asyncband/src/rwlock/read_guard.rs b/asyncband/src/rwlock/read_guard.rs index ec70afe..3bacdce 100644 --- a/asyncband/src/rwlock/read_guard.rs +++ b/asyncband/src/rwlock/read_guard.rs @@ -67,7 +67,7 @@ impl RwLock { /// # } /// ``` pub async fn read(&self) -> RwLockReadGuard<'_, T> { - self.s.acquire(1).await; + self.raw.read().await; RwLockReadGuard { lock: self } } @@ -93,7 +93,7 @@ impl RwLock { /// assert!(lock.try_read().is_none()); /// ``` pub fn try_read(&self) -> Option> { - if self.s.try_acquire(1) { + if self.raw.try_read() { Some(RwLockReadGuard { lock: self }) } else { None @@ -116,7 +116,7 @@ unsafe impl Sync for RwLockReadGuard<'_, T> {} impl Drop for RwLockReadGuard<'_, T> { fn drop(&mut self) { - self.lock.s.release(1); + self.lock.raw.unlock_read(); } } @@ -174,7 +174,7 @@ impl<'a, T: ?Sized> RwLockReadGuard<'a, T> { { let d = NonNull::from(f(&*orig)); let orig = ManuallyDrop::new(orig); - MappedRwLockReadGuard::new(d, &orig.lock.s) + MappedRwLockReadGuard::new(d, &orig.lock.raw) } /// Attempts to make a new [`MappedRwLockReadGuard`] for a component of the @@ -216,7 +216,7 @@ impl<'a, T: ?Sized> RwLockReadGuard<'a, T> { Some(d) => { let d = NonNull::from(d); let orig = ManuallyDrop::new(orig); - Ok(MappedRwLockReadGuard::new(d, &orig.lock.s)) + Ok(MappedRwLockReadGuard::new(d, &orig.lock.raw)) } None => Err(orig), } diff --git a/asyncband/src/rwlock/upgradable_read_guard.rs b/asyncband/src/rwlock/upgradable_read_guard.rs new file mode 100644 index 0000000..3c8c64d --- /dev/null +++ b/asyncband/src/rwlock/upgradable_read_guard.rs @@ -0,0 +1,214 @@ +// 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::ManuallyDrop; +use std::ops::Deref; + +use crate::rwlock::RwLock; +use crate::rwlock::RwLockReadGuard; +use crate::rwlock::RwLockWriteGuard; + +impl RwLock { + /// Acquires shared read access that can later be atomically upgraded. + /// + /// At most one upgradable read guard may exist at a time. Ordinary readers can coexist with + /// it. The upgradable guard reserves promotion priority over requests made after it acquired + /// the lock. + /// + /// Cancelling this operation loses its position in the lock's FIFO queue. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use asyncband::rwlock::RwLock; + /// + /// let lock = RwLock::new(7); + /// let upgradable = lock.upgradable_read().await; + /// + /// assert_eq!(*upgradable, 7); + /// assert!(lock.try_read().is_some()); + /// assert!(lock.try_upgradable_read().is_none()); + /// # } + /// ``` + pub async fn upgradable_read(&self) -> RwLockUpgradableReadGuard<'_, T> { + self.raw.upgradable_read().await; + RwLockUpgradableReadGuard { lock: self } + } + + /// Attempts to acquire upgradable shared access without waiting. + /// + /// # Examples + /// + /// ``` + /// use asyncband::rwlock::RwLock; + /// + /// let lock = RwLock::new(1); + /// let guard = lock.try_upgradable_read().expect("lock is available"); + /// assert!(lock.try_upgradable_read().is_none()); + /// + /// drop(guard); + /// assert!(lock.try_upgradable_read().is_some()); + /// ``` + pub fn try_upgradable_read(&self) -> Option> { + self.raw + .try_upgradable_read() + .then(|| RwLockUpgradableReadGuard { lock: self }) + } +} + +/// A shared read guard that can be atomically promoted to exclusive write access. +/// +/// Ordinary readers may coexist with this guard, but only one upgradable guard may exist. Use +/// [`upgrade`](Self::upgrade) to wait for exclusive access without releasing the reservation. +/// +/// # Examples +/// +/// ``` +/// # #[tokio::main] +/// # async fn main() { +/// use asyncband::rwlock::RwLock; +/// use asyncband::rwlock::RwLockUpgradableReadGuard; +/// +/// let lock = RwLock::new(3); +/// let guard: RwLockUpgradableReadGuard<'_, i32> = lock.upgradable_read().await; +/// assert_eq!(*guard, 3); +/// # } +/// ``` +#[must_use = "if unused the RwLock will immediately unlock"] +pub struct RwLockUpgradableReadGuard<'a, T: ?Sized> { + pub(super) lock: &'a RwLock, +} + +unsafe impl Send for RwLockUpgradableReadGuard<'_, T> {} +unsafe impl Sync for RwLockUpgradableReadGuard<'_, T> {} + +impl Drop for RwLockUpgradableReadGuard<'_, T> { + fn drop(&mut self) { + self.lock.raw.unlock_upgradable(); + } +} + +impl Deref for RwLockUpgradableReadGuard<'_, T> { + type Target = T; + + fn deref(&self) -> &T { + // SAFETY: an upgradable guard owns shared access to the protected value. + unsafe { &*self.lock.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> { + /// Atomically promotes this guard to exclusive write access. + /// + /// Existing readers are allowed to finish and new requests wait behind the upgrade. Cancelling + /// the operation releases the upgradable read guard. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use asyncband::rwlock::RwLock; + /// + /// let lock = RwLock::new(1); + /// let guard = lock.upgradable_read().await; + /// let mut writer = guard.upgrade().await; + /// *writer += 1; + /// + /// assert_eq!(*writer, 2); + /// # } + /// ``` + pub async fn upgrade(self) -> RwLockWriteGuard<'a, T> { + // The raw upgrade future assumes responsibility for releasing upgradable ownership if it + // is cancelled, so the source guard must no longer run its destructor after this point. + let guard = ManuallyDrop::new(self); + guard.lock.raw.upgrade().await; + RwLockWriteGuard { lock: guard.lock } + } + + /// Attempts to promote immediately, returning the original guard when other readers remain. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use asyncband::rwlock::RwLock; + /// + /// let lock = RwLock::new(1); + /// let upgradable = lock.upgradable_read().await; + /// let reader = lock.read().await; + /// + /// let upgradable = upgradable + /// .try_upgrade() + /// .expect_err("reader still holds the lock"); + /// drop(reader); + /// let writer = upgradable.try_upgrade().expect("last reader has left"); + /// assert_eq!(*writer, 1); + /// # } + /// ``` + pub fn try_upgrade(self) -> Result, Self> { + if self.lock.raw.try_upgrade() { + let guard = ManuallyDrop::new(self); + Ok(RwLockWriteGuard { lock: guard.lock }) + } else { + Err(self) + } + } + + /// Atomically converts this guard to an ordinary shared read guard. + /// + /// Downgrading relinquishes the unique promotion reservation while retaining shared access. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use asyncband::rwlock::RwLock; + /// + /// let lock = RwLock::new(1); + /// let upgradable = lock.upgradable_read().await; + /// let reader = upgradable.downgrade(); + /// + /// assert!(lock.try_upgradable_read().is_some()); + /// assert!(lock.try_write().is_none()); + /// drop(reader); + /// assert!(lock.try_write().is_some()); + /// # } + /// ``` + pub fn downgrade(self) -> RwLockReadGuard<'a, T> { + let guard = ManuallyDrop::new(self); + guard.lock.raw.downgrade_upgradable_to_read(); + RwLockReadGuard { lock: guard.lock } + } +} diff --git a/asyncband/src/rwlock/write_guard.rs b/asyncband/src/rwlock/write_guard.rs index d2f26b6..086f2d9 100644 --- a/asyncband/src/rwlock/write_guard.rs +++ b/asyncband/src/rwlock/write_guard.rs @@ -24,6 +24,7 @@ use std::ptr::NonNull; use crate::rwlock::MappedRwLockWriteGuard; use crate::rwlock::RwLock; use crate::rwlock::RwLockReadGuard; +use crate::rwlock::RwLockUpgradableReadGuard; impl RwLock { /// Locks this `RwLock` with exclusive write access, causing the current task to yield until the @@ -52,11 +53,8 @@ impl RwLock { /// # } /// ``` pub async fn write(&self) -> RwLockWriteGuard<'_, T> { - self.s.acquire(self.max_readers).await; - RwLockWriteGuard { - permits_acquired: self.max_readers, - lock: self, - } + self.raw.write().await; + RwLockWriteGuard { lock: self } } /// Attempts to acquire this `RwLock` with exclusive write access. @@ -81,11 +79,8 @@ impl RwLock { /// *v = 2; /// ``` pub fn try_write(&self) -> Option> { - if self.s.try_acquire(self.max_readers) { - Some(RwLockWriteGuard { - permits_acquired: self.max_readers, - lock: self, - }) + if self.raw.try_write() { + Some(RwLockWriteGuard { lock: self }) } else { None } @@ -116,7 +111,6 @@ impl RwLock { /// ``` #[must_use = "if unused the RwLock will immediately unlock"] pub struct RwLockWriteGuard<'a, T: ?Sized> { - pub(super) permits_acquired: usize, pub(super) lock: &'a RwLock, } @@ -125,7 +119,7 @@ unsafe impl Sync for RwLockWriteGuard<'_, T> {} impl Drop for RwLockWriteGuard<'_, T> { fn drop(&mut self) { - self.lock.s.release(self.permits_acquired); + self.lock.raw.unlock_write(); } } @@ -195,9 +189,8 @@ impl<'a, T: ?Sized> RwLockWriteGuard<'a, T> { U: ?Sized, { let d = NonNull::from(f(unsafe { &mut *orig.lock.c.get() })); - let permits_acquired = orig.permits_acquired; let orig = ManuallyDrop::new(orig); - MappedRwLockWriteGuard::new(d, &orig.lock.s, permits_acquired) + MappedRwLockWriteGuard::new(d, &orig.lock.raw) } /// Attempts to make a new [`MappedRwLockWriteGuard`] for a component of the @@ -250,13 +243,8 @@ impl<'a, T: ?Sized> RwLockWriteGuard<'a, T> { match f(unsafe { &mut *orig.lock.c.get() }) { Some(d) => { let d = NonNull::from(d); - let permits_acquired = orig.permits_acquired; let orig = ManuallyDrop::new(orig); - Ok(MappedRwLockWriteGuard::new( - d, - &orig.lock.s, - permits_acquired, - )) + Ok(MappedRwLockWriteGuard::new(d, &orig.lock.raw)) } None => Err(orig), } @@ -294,13 +282,37 @@ impl<'a, T: ?Sized> RwLockWriteGuard<'a, T> { /// ``` pub fn downgrade(self) -> RwLockReadGuard<'a, T> { // Prevent the original write guard from running its Drop implementation, - // which would release all permits. This must be done BEFORE any operation + // which would unlock the rwlock. This must be done BEFORE any operation // that might panic to ensure panic safety. let guard = ManuallyDrop::new(self); - // Release max_readers - 1 permits to convert the write lock to a read lock. - // The remaining 1 permit is kept for the read lock. - guard.lock.s.release(guard.permits_acquired - 1); + guard.lock.raw.downgrade_write_to_read(); RwLockReadGuard { lock: guard.lock } } + + /// Atomically downgrades this write guard to an upgradable read guard. + /// + /// The returned guard retains shared access and reserves the ability to promote again. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use asyncband::rwlock::RwLock; + /// + /// let lock = RwLock::new(1); + /// let writer = lock.write().await; + /// let upgradable = writer.downgrade_to_upgradable(); + /// + /// assert!(lock.try_read().is_some()); + /// assert!(lock.try_upgradable_read().is_none()); + /// assert_eq!(*upgradable, 1); + /// # } + /// ``` + pub fn downgrade_to_upgradable(self) -> RwLockUpgradableReadGuard<'a, T> { + let guard = ManuallyDrop::new(self); + guard.lock.raw.downgrade_write_to_upgradable(); + RwLockUpgradableReadGuard { lock: guard.lock } + } }