From 0a6c1f64bdfbf2f388750c1d8f40b64b799c1249 Mon Sep 17 00:00:00 2001 From: Orthur Date: Fri, 25 Sep 2026 04:28:45 -0400 Subject: [PATCH 1/5] perf(semaphore): avoid the queue lock for positive-balance releases --- CHANGELOG.md | 4 + asyncband/src/internal/semaphore.rs | 134 +++++++++++++++++- asyncband/src/mutex/mod.rs | 8 +- benchmarks/benches/primitives/pool/bounded.rs | 13 ++ benchmarks/benches/primitives/rwlock/read.rs | 11 ++ .../benches/primitives/semaphore/acquire.rs | 9 ++ .../tests/semaphore_ordering_test.rs | 110 ++++++++++++++ tests-integration/tests/semaphore_test.rs | 109 ++++++++++++++ xtask/src/main.rs | 12 ++ 9 files changed, 399 insertions(+), 11 deletions(-) create mode 100644 tests-integration/tests/semaphore_ordering_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f226004..d2697dec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,10 @@ All notable changes to this project will be documented in this file. * Prevent deadlocks in `singleflight::Group::work` and `try_work` when a duplicate key's destructor calls back into the same group. +### Improvements + +* Release `Semaphore` permits without taking the waiter lock while permits remain available, which speeds up dropping `RwLock` read guards, releasing multi-permit `Semaphore` permits, and returning objects to a bounded pool that is not at capacity. `Mutex` releases are unchanged. + ### Notable changes * Executor waker operations, including cloning, waking, and dropping, are expected not to panic; recovery from panicking waker callbacks is no longer supported. diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs index 1a539a2b..ee91633d 100644 --- a/asyncband/src/internal/semaphore.rs +++ b/asyncband/src/internal/semaphore.rs @@ -142,6 +142,30 @@ impl Semaphore { /// Adds `n` permits to the semaphore. pub fn release(&self, n: usize) { + if n == 0 { + return; + } + + // A waiter is linked only after the balance has been drained to zero, and permits reach + // the balance again only once the queue is empty, so a positive balance means no waiter + // is linked and the permits can be added without the queue lock. + // + // Try the exchange once before falling back to the locked path. Other threads can + // still use the fast paths, so a locked balance update may also need to retry. + let current = self.permits.load(Ordering::Relaxed); + if current != 0 && self.try_add_to_balance(current, n).is_ok() { + return; + } + + self.insert_permits_with_lock(n, self.waiters.lock()); + } + + /// Adds `n` permits to a semaphore whose every permit the caller holds. + /// + /// The balance is zero while one caller holds every permit, so the positive-balance path of + /// [`release`](Self::release) cannot succeed and mutex guards skip its probe. The locked path + /// is correct for any balance, so the precondition only affects speed. + pub fn release_all_held(&self, n: usize) { if n != 0 { self.insert_permits_with_lock(n, self.waiters.lock()); } @@ -175,6 +199,24 @@ impl Semaphore { wakers.wake_all(); } + /// Adds `n` permits to a balance expected to hold `current`, or returns the balance observed + /// instead. + /// + /// The overflow check and the addition are one exchange because the positive-balance path of + /// [`release`](Self::release) can grow the balance even while the queue lock is held. The + /// exchange is the strong form because `release` tries it only once. + /// + /// ORDERING: Release publishes the work protected by the released permits to the Acquire + /// load or exchange that next observes this balance. + fn try_add_to_balance(&self, current: usize, n: usize) -> Result<(), usize> { + let next = current.checked_add(n).unwrap_or_else(|| { + panic!("number of added permits ({n}) would overflow usize::MAX (prev: {current})") + }); + self.permits + .compare_exchange(current, next, Ordering::Release, Ordering::Relaxed) + .map(|_| ()) + } + fn insert_permits_with_lock<'a>( &'a self, mut rem: usize, @@ -200,13 +242,17 @@ impl Semaphore { } if rem > 0 && waiters.is_empty() { - // The lock serializes additions; concurrent operations can only remove permits. - let current = self.permits.load(Ordering::Relaxed); - assert!( - current.checked_add(rem).is_some(), - "number of added permits ({rem}) would overflow usize::MAX (prev: {current})" - ); - self.permits.fetch_add(rem, Ordering::Release); + // Fast releases can grow the balance under this lock, so the overflow check and + // addition must be one exchange. A zero balance cannot change under the lock, but + // still needs an RMW to extend the previous release sequence. + let mut current = self.permits.load(Ordering::Relaxed); + if current == 0 { + self.permits.fetch_add(rem, Ordering::Release); + } else { + while let Err(actual) = self.try_add_to_balance(current, rem) { + current = actual; + } + } rem = 0; } @@ -384,6 +430,8 @@ fn acquired_or_enqueue( #[cfg(test)] mod tests { + use std::panic; + use std::panic::AssertUnwindSafe; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; @@ -412,6 +460,78 @@ mod tests { } } + #[test] + fn release_with_positive_balance_does_not_take_the_queue_lock() { + let semaphore = Semaphore::new(1); + + // Holding the queue lock deadlocks a release that needs it. + let queue = semaphore.waiters.lock(); + semaphore.release(2); + assert_eq!(semaphore.available_permits(), 3); + assert!(semaphore.try_acquire(2)); + semaphore.release(2); + assert_eq!(semaphore.available_permits(), 3); + drop(queue); + } + + #[test] + fn release_with_zero_balance_hands_permits_to_the_queue() { + let semaphore = Semaphore::new(0); + let counter = Arc::new(WakeCounter(AtomicUsize::new(0))); + let waker = Waker::from(counter.clone()); + let mut acquire = semaphore.poll_acquire(2); + assert!(acquire.poll_once(&waker).is_pending()); + + semaphore.release(3); + assert_eq!(counter.0.load(Ordering::Relaxed), 1); + assert_eq!(semaphore.available_permits(), 1); + assert!(acquire.poll_once(&waker).is_ready()); + + // Queue empty and balance positive again: the lock is not needed. + let queue = semaphore.waiters.lock(); + semaphore.release(1); + drop(queue); + assert_eq!(semaphore.available_permits(), 2); + } + + #[test] + fn locked_path_adds_to_a_positive_balance() { + let semaphore = Semaphore::new(2); + + // A release that observed zero can find a positive balance once it holds the lock. + semaphore.insert_permits_with_lock(3, semaphore.waiters.lock()); + assert_eq!(semaphore.available_permits(), 5); + } + + #[test] + fn locked_overflow_panics_without_adding_permits() { + let semaphore = Semaphore::new(usize::MAX); + let result = panic::catch_unwind(AssertUnwindSafe(|| { + semaphore.insert_permits_with_lock(1, semaphore.waiters.lock()); + })); + assert!(result.is_err()); + assert_eq!(semaphore.available_permits(), usize::MAX); + assert!(semaphore.waiters.lock().is_empty()); + } + + #[test] + fn release_all_held_hands_permits_to_the_queue() { + let semaphore = Semaphore::new(1); + let counter = Arc::new(WakeCounter(AtomicUsize::new(0))); + let waker = Waker::from(counter.clone()); + assert!(semaphore.try_acquire(1)); + let mut acquire = semaphore.poll_acquire(1); + assert!(acquire.poll_once(&waker).is_pending()); + + semaphore.release_all_held(1); + assert_eq!(counter.0.load(Ordering::Relaxed), 1); + assert_eq!(semaphore.available_permits(), 0); + assert!(acquire.poll_once(&waker).is_ready()); + + semaphore.release_all_held(1); + assert_eq!(semaphore.available_permits(), 1); + } + #[test] fn release_distributes_permits_across_multiple_batches() { const WAITER_COUNT: usize = WakerBatch::STACK_SIZE * 2 + 1; diff --git a/asyncband/src/mutex/mod.rs b/asyncband/src/mutex/mod.rs index ffed4559..a3211793 100644 --- a/asyncband/src/mutex/mod.rs +++ b/asyncband/src/mutex/mod.rs @@ -277,7 +277,7 @@ unsafe impl Sync for MutexGuard<'_, T> {} impl Drop for MutexGuard<'_, T> { fn drop(&mut self) { - self.lock.s.release(1); + self.lock.s.release_all_held(1); } } @@ -442,7 +442,7 @@ unsafe impl Sync for OwnedMutexGuard {} impl Drop for OwnedMutexGuard { fn drop(&mut self) { - self.lock.s.release(1); + self.lock.s.release_all_held(1); } } @@ -638,7 +638,7 @@ unsafe impl Sync for MappedMutexGuard<'_, T> {} impl Drop for MappedMutexGuard<'_, T> { fn drop(&mut self) { - self.s.release(1); + self.s.release_all_held(1); } } @@ -837,7 +837,7 @@ unsafe impl Sync for OwnedMapp impl Drop for OwnedMappedMutexGuard { fn drop(&mut self) { - self.lock.s.release(1); + self.lock.s.release_all_held(1); } } diff --git a/benchmarks/benches/primitives/pool/bounded.rs b/benchmarks/benches/primitives/pool/bounded.rs index 3524e371..79089906 100644 --- a/benchmarks/benches/primitives/pool/bounded.rs +++ b/benchmarks/benches/primitives/pool/bounded.rs @@ -60,6 +60,19 @@ fn bounded_warm_get_and_return(bencher: Bencher) { }); } +#[divan::bench] +fn bounded_warm_get_and_return_with_spare_capacity(bencher: Bencher) { + let pool = bounded::Pool::new(bounded::PoolConfig::new(8), Manager); + let mut context = bench_context(); + drop(poll_ready(pool.get(), &mut context).unwrap()); + + bencher.bench_local(|| { + let object = poll_ready(pool.get(), &mut context).unwrap(); + black_box(*object); + drop(object); + }); +} + #[divan::bench] fn bounded_contended_handoff(bencher: Bencher) { let pool = bounded::Pool::new(bounded::PoolConfig::new(1), Manager); diff --git a/benchmarks/benches/primitives/rwlock/read.rs b/benchmarks/benches/primitives/rwlock/read.rs index 5aff8a54..b498c0e5 100644 --- a/benchmarks/benches/primitives/rwlock/read.rs +++ b/benchmarks/benches/primitives/rwlock/read.rs @@ -38,3 +38,14 @@ fn eight_reads_then_one_write(bencher: Bencher) { black_box(*guard) }); } + +#[divan::bench] +fn read_reuse(bencher: Bencher) { + let lock = RwLock::new(0usize); + let mut context = bench_context(); + + bencher.bench_local(|| { + let guard = poll_ready(lock.read(), &mut context); + black_box(*guard) + }); +} diff --git a/benchmarks/benches/primitives/semaphore/acquire.rs b/benchmarks/benches/primitives/semaphore/acquire.rs index 969e14e4..a61392a8 100644 --- a/benchmarks/benches/primitives/semaphore/acquire.rs +++ b/benchmarks/benches/primitives/semaphore/acquire.rs @@ -54,6 +54,15 @@ fn try_acquire_release(bencher: Bencher) { }); } +#[divan::bench] +fn try_acquire_release_with_spare_permits(bencher: Bencher) { + let semaphore = Semaphore::new(8); + + bencher.bench_local(|| { + drop(black_box(semaphore.try_acquire(black_box(2)).unwrap())); + }); +} + #[divan::bench] fn owned_try_acquire_release(bencher: Bencher) { let semaphore = Arc::new(Semaphore::new(8)); diff --git a/tests-integration/tests/semaphore_ordering_test.rs b/tests-integration/tests/semaphore_ordering_test.rs new file mode 100644 index 00000000..70d7a2fc --- /dev/null +++ b/tests-integration/tests/semaphore_ordering_test.rs @@ -0,0 +1,110 @@ +// 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. + +//! An acquire synchronizes with the release whose release sequence it reads from, and every +//! write to the balance is a read-modify-write, so that sequence extends through later releases +//! and acquisitions. A plain store would end it, and Miri would report a data race here. +//! +//! The race shows only on schedules where the final acquire reads the latest balance rather than +//! a stale one, so the test runs under several seeds. + +use std::cell::UnsafeCell; +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::thread; + +use asyncband::semaphore::Semaphore; + +struct Data(UnsafeCell); + +// SAFETY: accesses are ordered only through the semaphore, which is the property under test. +unsafe impl Sync for Data {} + +/// Sequences the steps in time without adding a happens-before edge. +fn wait_for(step: &AtomicUsize, value: usize) { + while step.load(Ordering::Relaxed) < value { + thread::yield_now(); + } +} + +#[test] +fn acquire_synchronizes_with_every_earlier_release() { + let semaphore = Arc::new(Semaphore::new(1)); + let data = Arc::new(Data(UnsafeCell::new(0))); + let step = Arc::new(AtomicUsize::new(0)); + + // Writes, then releases onto a positive balance. + let writer = { + let (semaphore, data, step) = (semaphore.clone(), data.clone(), step.clone()); + thread::spawn(move || { + // SAFETY: the reader is ordered after this write by the semaphore. + unsafe { *data.0.get() = 42 }; + semaphore.release(1); + step.store(1, Ordering::Relaxed); + }) + }; + + // Takes every permit, so the balance is zero. + let drainer = { + let (semaphore, step) = (semaphore.clone(), step.clone()); + thread::spawn(move || { + wait_for(&step, 1); + loop { + if let Some(permit) = semaphore.try_acquire(2) { + permit.forget(); + break; + } + thread::yield_now(); + } + step.store(2, Ordering::Relaxed); + }) + }; + + // Releases onto the zero balance without any synchronization with the writer. + let releaser = { + let (semaphore, step) = (semaphore.clone(), step.clone()); + thread::spawn(move || { + wait_for(&step, 2); + semaphore.release(1); + step.store(3, Ordering::Relaxed); + }) + }; + + // Acquires the releaser's permit and reads the data. + let reader = { + let (semaphore, data, step) = (semaphore.clone(), data.clone(), step.clone()); + thread::spawn(move || { + wait_for(&step, 3); + let permit = loop { + if let Some(permit) = semaphore.try_acquire(1) { + break permit; + } + thread::yield_now(); + }; + // SAFETY: every release before this acquire happens-before it. + let value = unsafe { *data.0.get() }; + drop(permit); + value + }) + }; + + writer.join().unwrap(); + drainer.join().unwrap(); + releaser.join().unwrap(); + assert_eq!(reader.join().unwrap(), 42); +} diff --git a/tests-integration/tests/semaphore_test.rs b/tests-integration/tests/semaphore_test.rs index 02431f3c..e61a93a0 100644 --- a/tests-integration/tests/semaphore_test.rs +++ b/tests-integration/tests/semaphore_test.rs @@ -17,11 +17,16 @@ use std::pin::pin; use std::sync::Arc; +use std::sync::Barrier; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; use std::task::Context; use std::task::Poll; use std::task::Wake; use std::task::Waker; +use std::thread; +use asyncband::blocking::FutureExt; use asyncband::semaphore::Semaphore; use tests_integration::WakeCounter; use tests_integration::expect_ready; @@ -221,3 +226,107 @@ fn reduce_permits_takes_priority_over_pending_acquires() { drop(permit); assert_eq!(s.available_permits(), 1); } + +/// Releases race acquisitions that drain the balance to zero and link waiters. The permit count +/// must be conserved, and no more permits than the capacity may ever be held at once. +#[test] +fn concurrent_releases_conserve_permits() { + const PERMITS: usize = 3; + const THREADS: usize = 8; + const ITERATIONS: usize = 4_000; + + fn next(state: &mut u64) -> usize { + *state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + (*state >> 33) as usize + } + + let semaphore = Arc::new(Semaphore::new(PERMITS)); + let held = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(Barrier::new(THREADS)); + let workers = (0..THREADS) + .map(|seed| { + let semaphore = semaphore.clone(); + let held = held.clone(); + let start = start.clone(); + thread::spawn(move || { + let mut state = seed as u64 + 1; + start.wait(); + let hold = |n: usize, state: &mut u64| { + let now = held.fetch_add(n, Ordering::AcqRel) + n; + assert!(now <= PERMITS, "{now} permits held at once"); + for _ in 0..next(state) % 8 { + std::hint::spin_loop(); + } + held.fetch_sub(n, Ordering::AcqRel); + }; + for _ in 0..ITERATIONS { + let n = next(&mut state) % PERMITS + 1; + match next(&mut state) % 5 { + 0 => { + if let Some(permit) = semaphore.try_acquire(n) { + hold(n, &mut state); + drop(permit); + } + } + 1 => { + let permit = FutureExt::block_on(semaphore.acquire(n)); + hold(n, &mut state); + permit.forget(); + semaphore.release(n); + } + 2 => { + semaphore.reduce_permits(n); + semaphore.release(n); + } + 3 => { + let mut acquire = pin!(semaphore.acquire(n)); + let poll = poll_with(acquire.as_mut(), Waker::noop()); + if let Poll::Ready(permit) = poll { + hold(n, &mut state); + drop(permit); + } + } + _ => { + let permit = FutureExt::block_on(semaphore.acquire(n)); + hold(n, &mut state); + drop(permit); + } + } + } + }) + }) + .collect::>(); + + for worker in workers { + worker.join().unwrap(); + } + assert_eq!(semaphore.available_permits(), PERMITS); + assert!(semaphore.try_acquire(PERMITS).is_some()); +} + +/// Two releases race on a balance one below `usize::MAX`: exactly one may succeed, and the other +/// must panic before adding anything, whichever path each of them takes. +#[test] +fn concurrent_releases_at_the_limit_panic_exactly_once() { + let semaphore = Arc::new(Semaphore::new(usize::MAX - 1)); + let start = Arc::new(Barrier::new(2)); + let workers = (0..2) + .map(|_| { + let semaphore = semaphore.clone(); + let start = start.clone(); + thread::spawn(move || { + start.wait(); + std::panic::catch_unwind(|| semaphore.release(1)).is_ok() + }) + }) + .collect::>(); + let succeeded = workers + .into_iter() + .map(|worker| worker.join().unwrap()) + .filter(|succeeded| *succeeded) + .count(); + assert_eq!(succeeded, 1); + assert_eq!(semaphore.available_permits(), usize::MAX); +} diff --git a/xtask/src/main.rs b/xtask/src/main.rs index a047a4f4..d9be33e1 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -141,6 +141,11 @@ impl CommandMiri { "tests-integration", &["--test", "phaser_test"], )); + run_command(make_miri_cmd_with_seeds( + "tests-integration", + &["--test", "semaphore_ordering_test"], + "0..8", + )); } } @@ -450,6 +455,13 @@ fn make_miri_cmd(package: &str, target: &[&str]) -> StdCommand { cmd } +/// Runs a Miri target under a range of seeds, for tests whose failure depends on the schedule. +fn make_miri_cmd_with_seeds(package: &str, target: &[&str], seeds: &str) -> StdCommand { + let mut cmd = make_miri_cmd(package, target); + cmd.env("MIRIFLAGS", format!("-Zmiri-many-seeds={seeds}")); + cmd +} + fn make_format_cmd(fix: bool) -> StdCommand { let mut cmd = find_command("cargo"); cmd.args(["+nightly", "fmt", "--all"]); From ad5b6c26d4eab7179b902cf5f9ee82c841e97a7c Mon Sep 17 00:00:00 2001 From: Orthur Date: Fri, 25 Sep 2026 04:36:23 -0400 Subject: [PATCH 2/5] perf(semaphore): route contended releases through the queue lock --- asyncband/src/internal/semaphore.rs | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs index ee91633d..401b4bcb 100644 --- a/asyncband/src/internal/semaphore.rs +++ b/asyncband/src/internal/semaphore.rs @@ -27,6 +27,7 @@ use std::future::Future; use std::pin::Pin; use std::sync::MutexGuard; +use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::task::Context; @@ -44,6 +45,10 @@ use crate::internal::waker_batch::WakerBatch; pub struct Semaphore { /// The current number of available permits in the semaphore. permits: AtomicUsize, + /// Set while a locked release is losing races on the balance, so that other releases join + /// the lock instead of adding to the traffic. A hint only: a stale read in either direction + /// is still correct. + contended: AtomicBool, waiters: Mutex>, } @@ -59,6 +64,7 @@ impl Semaphore { pub const fn new(permits: usize) -> Self { Self { permits: AtomicUsize::new(permits), + contended: AtomicBool::new(false), waiters: Mutex::new(WaitList::new()), } } @@ -153,7 +159,10 @@ impl Semaphore { // Try the exchange once before falling back to the locked path. Other threads can // still use the fast paths, so a locked balance update may also need to retry. let current = self.permits.load(Ordering::Relaxed); - if current != 0 && self.try_add_to_balance(current, n).is_ok() { + if current != 0 + && !self.contended.load(Ordering::Relaxed) + && self.try_add_to_balance(current, n).is_ok() + { return; } @@ -250,8 +259,12 @@ impl Semaphore { self.permits.fetch_add(rem, Ordering::Release); } else { while let Err(actual) = self.try_add_to_balance(current, rem) { + // Losing under the lock means another thread is on the balance now: + // route its release through the lock as well. + self.contended.store(true, Ordering::Relaxed); current = actual; } + self.contended.store(false, Ordering::Relaxed); } rem = 0; } @@ -474,6 +487,17 @@ mod tests { drop(queue); } + #[test] + fn contended_release_takes_the_locked_path_and_clears_the_flag() { + let semaphore = Semaphore::new(1); + semaphore.contended.store(true, Ordering::Relaxed); + + // Only the locked path clears the flag, so a cleared flag proves the fast path was skipped. + semaphore.release(1); + assert_eq!(semaphore.available_permits(), 2); + assert!(!semaphore.contended.load(Ordering::Relaxed)); + } + #[test] fn release_with_zero_balance_hands_permits_to_the_queue() { let semaphore = Semaphore::new(0); From 158bc4966dacc08b4a768f56b81675943ee8dd4a Mon Sep 17 00:00:00 2001 From: Orthur Date: Sat, 10 Oct 2026 05:42:41 -0400 Subject: [PATCH 3/5] test(semaphore): repeat the ordering scenario instead of overriding MIRIFLAGS --- tests-integration/tests/semaphore_ordering_test.rs | 9 ++++++++- xtask/src/main.rs | 10 +--------- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/tests-integration/tests/semaphore_ordering_test.rs b/tests-integration/tests/semaphore_ordering_test.rs index 70d7a2fc..dfdae0de 100644 --- a/tests-integration/tests/semaphore_ordering_test.rs +++ b/tests-integration/tests/semaphore_ordering_test.rs @@ -20,7 +20,7 @@ //! and acquisitions. A plain store would end it, and Miri would report a data race here. //! //! The race shows only on schedules where the final acquire reads the latest balance rather than -//! a stale one, so the test runs under several seeds. +//! a stale one, so the test repeats the scenario to cover several schedules in one run. use std::cell::UnsafeCell; use std::sync::Arc; @@ -44,6 +44,13 @@ fn wait_for(step: &AtomicUsize, value: usize) { #[test] fn acquire_synchronizes_with_every_earlier_release() { + // A plain store shows the race in about one round in four; 16 rounds catch it under one seed. + for _ in 0..16 { + run_once(); + } +} + +fn run_once() { let semaphore = Arc::new(Semaphore::new(1)); let data = Arc::new(Data(UnsafeCell::new(0))); let step = Arc::new(AtomicUsize::new(0)); diff --git a/xtask/src/main.rs b/xtask/src/main.rs index d9be33e1..08f0f058 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -141,10 +141,9 @@ impl CommandMiri { "tests-integration", &["--test", "phaser_test"], )); - run_command(make_miri_cmd_with_seeds( + run_command(make_miri_cmd( "tests-integration", &["--test", "semaphore_ordering_test"], - "0..8", )); } } @@ -455,13 +454,6 @@ fn make_miri_cmd(package: &str, target: &[&str]) -> StdCommand { cmd } -/// Runs a Miri target under a range of seeds, for tests whose failure depends on the schedule. -fn make_miri_cmd_with_seeds(package: &str, target: &[&str], seeds: &str) -> StdCommand { - let mut cmd = make_miri_cmd(package, target); - cmd.env("MIRIFLAGS", format!("-Zmiri-many-seeds={seeds}")); - cmd -} - fn make_format_cmd(fix: bool) -> StdCommand { let mut cmd = find_command("cargo"); cmd.args(["+nightly", "fmt", "--all"]); From b1010edc78b89e39030fe013af778190061d1a7c Mon Sep 17 00:00:00 2001 From: Orthur Date: Sat, 10 Oct 2026 05:42:41 -0400 Subject: [PATCH 4/5] docs: describe the semaphore release change as a fast path --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d2697dec..18a017ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,7 +33,7 @@ All notable changes to this project will be documented in this file. ### Improvements -* Release `Semaphore` permits without taking the waiter lock while permits remain available, which speeds up dropping `RwLock` read guards, releasing multi-permit `Semaphore` permits, and returning objects to a bounded pool that is not at capacity. `Mutex` releases are unchanged. +* Add a lock-free fast path for releasing `Semaphore` permits while permits remain available, which speeds up dropping `RwLock` read guards, releasing multi-permit `Semaphore` permits, and returning objects to a bounded pool that is not at capacity. `Mutex` releases are unchanged. ### Notable changes From 13cfb97bf20ca32cc28af923389080dc31d2d1c5 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 12 Oct 2026 15:46:17 +0100 Subject: [PATCH 5/5] refactor(semaphore): clarify release and permit naming --- asyncband/src/broadcast/mpmc/bounded/mod.rs | 4 +- asyncband/src/internal/semaphore.rs | 256 ++++++++++-------- asyncband/src/mutex/mod.rs | 8 +- asyncband/src/semaphore/mod.rs | 2 +- .../tests/semaphore_ordering_test.rs | 15 +- tests-integration/tests/semaphore_test.rs | 6 +- 6 files changed, 154 insertions(+), 137 deletions(-) diff --git a/asyncband/src/broadcast/mpmc/bounded/mod.rs b/asyncband/src/broadcast/mpmc/bounded/mod.rs index 43735ed1..5514f34f 100644 --- a/asyncband/src/broadcast/mpmc/bounded/mod.rs +++ b/asyncband/src/broadcast/mpmc/bounded/mod.rs @@ -195,7 +195,7 @@ impl Shared { // subscription. let waiting = self.waiting_senders(); if freed > 0 && waiting > 0 { - self.tx_permits.release_if_nonempty(freed.min(waiting)); + self.tx_permits.release_if_waiters(freed.min(waiting)); } } @@ -323,7 +323,7 @@ impl BoundedSender { }; loop { - // Enqueue before rechecking. `release_if_nonempty` adds nothing when no + // Enqueue before rechecking. `release_if_waiters` adds nothing when no // producer is queued, so a reclaim landing between the recheck and the // registration would otherwise drop its wake-up and park this producer for // good. Registering first orders this producer's semaphore acquisition ahead diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs index bfbc43cc..f26282d1 100644 --- a/asyncband/src/internal/semaphore.rs +++ b/asyncband/src/internal/semaphore.rs @@ -44,17 +44,18 @@ use crate::internal::waker_batch::WakerBatch; #[derive(Debug)] pub struct Semaphore { /// The current number of available permits in the semaphore. - permits: AtomicUsize, - /// Set while a locked release is losing races on the balance, so that other releases join - /// the lock instead of adding to the traffic. A hint only: a stale read in either direction - /// is still correct. - contended: AtomicBool, + available_permits: AtomicUsize, + /// Set when a release holding the queue lock loses a CAS race, and cleared after it + /// successfully updates the available count. Releases that observe this flag skip the + /// lock-free attempt. This is a performance hint; stale reads do not affect correctness. + contention_detected: AtomicBool, waiters: Mutex>, } #[derive(Debug)] struct WaitNode { - permits: usize, + /// Permits still needed to satisfy the acquisition or repay the queued permit debt. + needed_permits: usize, /// A linked node without a waker is permit debt owned by the queue. An acquire node only loses /// its waker while being detached, after which its future still owns the node. waker: Option, @@ -63,34 +64,36 @@ struct WaitNode { impl Semaphore { pub const fn new(permits: usize) -> Self { Self { - permits: AtomicUsize::new(permits), - contended: AtomicBool::new(false), + available_permits: AtomicUsize::new(permits), + contention_detected: AtomicBool::new(false), waiters: Mutex::new(WaitList::new()), } } /// Returns the current number of available permits. pub fn available_permits(&self) -> usize { - self.permits.load(Ordering::Acquire) + self.available_permits.load(Ordering::Acquire) } /// Tries to acquire `n` permits from the semaphore. /// /// Returns `true` if the permits were acquired, `false` otherwise. pub fn try_acquire(&self, n: usize) -> bool { - let mut current = self.permits.load(Ordering::Acquire); + let mut available = self.available_permits.load(Ordering::Acquire); loop { - if current < n { + if available < n { return false; } - let next = current - n; - match self - .permits - .compare_exchange(current, next, Ordering::AcqRel, Ordering::Acquire) - { + let next = available - n; + match self.available_permits.compare_exchange( + available, + next, + Ordering::AcqRel, + Ordering::Acquire, + ) { Ok(_) => return true, - Err(actual) => current = actual, + Err(actual) => available = actual, } } } @@ -103,22 +106,22 @@ impl Semaphore { return 0; } - let mut current = self.permits.load(Ordering::Acquire); + let mut available = self.available_permits.load(Ordering::Acquire); loop { - let new = current.saturating_sub(up_to); - match self.permits.compare_exchange_weak( - current, - new, + let next = available.saturating_sub(up_to); + match self.available_permits.compare_exchange_weak( + available, + next, Ordering::AcqRel, Ordering::Acquire, ) { - Ok(_) => return up_to.min(current), - Err(actual) => current = actual, + Ok(_) => return up_to.min(available), + Err(actual) => available = actual, } } } - /// Reduces the semaphore's logical permit balance by exactly `n`. + /// Removes exactly `n` permits, consuming future releases if fewer are available. /// /// If fewer than `n` permits are available, a queue-head debt node consumes future releases. pub fn reduce_permits(&self, n: usize) { @@ -128,7 +131,7 @@ impl Semaphore { /// Acquires `n` permits from the semaphore. pub async fn acquire(&self, n: usize) { let fut = Acquire { - permits: n, + requested_permits: n, index: None, semaphore: self, done: false, @@ -139,53 +142,57 @@ impl Semaphore { /// Returns a future that is resolved when acquired `n` permits from the semaphore. pub fn poll_acquire(&self, n: usize) -> Acquire<'_> { Acquire { - permits: n, + requested_permits: n, index: None, semaphore: self, done: false, } } - /// Adds `n` permits to the semaphore. + /// Releases `n` permits, attempting to update the available count without the queue lock. + /// + /// Acquires the queue lock if no permits are available, contention has been detected, or the + /// compare-exchange fails. pub fn release(&self, n: usize) { if n == 0 { return; } - // A waiter is linked only after the balance has been drained to zero, and permits reach - // the balance again only once the queue is empty, so a positive balance means no waiter - // is linked and the permits can be added without the queue lock. + // A waiter is linked only after all available permits have been taken. Releases make + // permits available again only once the queue is empty, so available permits mean no + // waiter is linked and the count can be increased without the queue lock. // // Try the exchange once before falling back to the locked path. Other threads can - // still use the fast paths, so a locked balance update may also need to retry. - let current = self.permits.load(Ordering::Relaxed); - if current != 0 - && !self.contended.load(Ordering::Relaxed) - && self.try_add_to_balance(current, n).is_ok() + // still acquire or release permits without the lock, so an update under the lock may + // also need to retry. + let available = self.available_permits.load(Ordering::Relaxed); + if available != 0 + && !self.contention_detected.load(Ordering::Relaxed) + && self.try_update_permits(available, n).is_ok() { return; } - self.insert_permits_with_lock(n, self.waiters.lock()); + self.distribute_permits(n, self.waiters.lock()); } - /// Adds `n` permits to a semaphore whose every permit the caller holds. + /// Releases `n` permits by acquiring the queue lock directly. /// - /// The balance is zero while one caller holds every permit, so the positive-balance path of - /// [`release`](Self::release) cannot succeed and mutex guards skip its probe. The locked path - /// is correct for any balance, so the precondition only affects speed. - pub fn release_all_held(&self, n: usize) { + /// Use this when no permits can be available, as when a mutex guard holds its only permit, + /// to skip the speculative attempt in [`release`](Self::release). + /// This method also works when permits are available. + pub fn release_with_lock(&self, n: usize) { if n != 0 { - self.insert_permits_with_lock(n, self.waiters.lock()); + self.distribute_permits(n, self.waiters.lock()); } } - /// Adds `n` permits to the semaphore if there is any waiter. + /// Releases `n` permits only if the queue contains waiters when locked. #[cfg(feature = "broadcast")] - pub fn release_if_nonempty(&self, n: usize) { + pub fn release_if_waiters(&self, n: usize) { let waiters = self.waiters.lock(); if !waiters.is_empty() { - self.insert_permits_with_lock(n, waiters); + self.distribute_permits(n, waiters); } } @@ -195,7 +202,7 @@ impl Semaphore { let mut waiters = self.waiters.lock(); let mut wakers = WakerBatch::new(); while let Some((id, waiter)) = waiters.unlink_first_waiter(|node| { - node.permits = 0; + node.needed_permits = 0; true }) { if let Some(waker) = waiter.waker.take() { @@ -208,38 +215,43 @@ impl Semaphore { wakers.wake_all(); } - /// Adds `n` permits to a balance expected to hold `current`, or returns the balance observed - /// instead. + /// Tries to update the available permit count from `expected` to `expected + released`. + /// + /// Returns the observed count if it does not match `expected`. Panics if the sum would + /// overflow. /// - /// The overflow check and the addition are one exchange because the positive-balance path of - /// [`release`](Self::release) can grow the balance even while the queue lock is held. The - /// exchange is the strong form because `release` tries it only once. + /// The overflow check and addition must use the same count because other releases can add + /// permits even while the queue lock is held. The exchange is the strong form because + /// [`release`](Self::release) tries it only once. /// /// ORDERING: Release publishes the work protected by the released permits to the Acquire - /// load or exchange that next observes this balance. - fn try_add_to_balance(&self, current: usize, n: usize) -> Result<(), usize> { - let next = current.checked_add(n).unwrap_or_else(|| { - panic!("number of added permits ({n}) would overflow usize::MAX (prev: {current})") + /// load or exchange that next observes this permit count. + fn try_update_permits(&self, expected: usize, released: usize) -> Result<(), usize> { + let next = expected.checked_add(released).unwrap_or_else(|| { + panic!( + "number of added permits ({released}) would overflow usize::MAX (prev: {expected})" + ) }); - self.permits - .compare_exchange(current, next, Ordering::Release, Ordering::Relaxed) + self.available_permits + .compare_exchange(expected, next, Ordering::Release, Ordering::Relaxed) .map(|_| ()) } - fn insert_permits_with_lock<'a>( + /// Assigns permits to queued waiters, then makes any surplus available. + fn distribute_permits<'a>( &'a self, - mut rem: usize, + mut remaining: usize, mut waiters: MutexGuard<'a, WaitList>, ) { let mut batch = WakerBatch::new(); loop { - while rem > 0 && !batch.will_spill() { + while remaining > 0 && !batch.will_spill() { let Some((id, waiter)) = waiters.unlink_first_waiter(|node| { - let acquired = node.permits.min(rem); - node.permits -= acquired; - rem -= acquired; - node.permits == 0 + let assigned = node.needed_permits.min(remaining); + node.needed_permits -= assigned; + remaining -= assigned; + node.needed_permits == 0 }) else { break; }; @@ -250,28 +262,30 @@ impl Semaphore { } } - if rem > 0 && waiters.is_empty() { - // Fast releases can grow the balance under this lock, so the overflow check and - // addition must be one exchange. A zero balance cannot change under the lock, but - // still needs an RMW to extend the previous release sequence. - let mut current = self.permits.load(Ordering::Relaxed); - if current == 0 { - self.permits.fetch_add(rem, Ordering::Release); + if remaining > 0 && waiters.is_empty() { + // Releases can add available permits without taking this lock, so the overflow + // check and addition must use the same count. If none are available, other threads + // cannot add any under this lock, but an RMW is still needed to extend the previous + // release sequence. + let mut available = self.available_permits.load(Ordering::Relaxed); + if available == 0 { + self.available_permits + .fetch_add(remaining, Ordering::Release); } else { - while let Err(actual) = self.try_add_to_balance(current, rem) { - // Losing under the lock means another thread is on the balance now: - // route its release through the lock as well. - self.contended.store(true, Ordering::Relaxed); - current = actual; + while let Err(actual) = self.try_update_permits(available, remaining) { + // Another thread changed the available count despite our queue lock; + // ask subsequent releases to acquire the lock as well. + self.contention_detected.store(true, Ordering::Relaxed); + available = actual; } - self.contended.store(false, Ordering::Relaxed); + self.contention_detected.store(false, Ordering::Relaxed); } - rem = 0; + remaining = 0; } drop(waiters); batch.wake_all(); - if rem == 0 { + if remaining == 0 { return; } waiters = self.waiters.lock(); @@ -281,7 +295,7 @@ impl Semaphore { #[derive(Debug)] pub struct Acquire<'a> { - permits: usize, + requested_permits: usize, index: Option, semaphore: &'a Semaphore, done: bool, @@ -293,13 +307,13 @@ impl Drop for Acquire<'_> { let mut waiters = self.semaphore.waiters.lock(); let mut acquired = 0; waiters.unlink_waiter(index, |node| { - acquired = self.permits - node.permits; - node.permits = 0; + acquired = self.requested_permits - node.needed_permits; + node.needed_permits = 0; true }); let waiter = waiters.remove_unlinked_waiter(index); if acquired > 0 { - self.semaphore.insert_permits_with_lock(acquired, waiters); + self.semaphore.distribute_permits(acquired, waiters); } else { drop(waiters); } @@ -311,7 +325,7 @@ impl Drop for Acquire<'_> { impl Acquire<'_> { pub fn poll_once(&mut self, waker: &Waker) -> Poll<()> { let Self { - permits, + requested_permits, index, semaphore, done, @@ -327,7 +341,7 @@ impl Acquire<'_> { let mut waiters = semaphore.waiters.lock(); let ready = { let node = waiters.waiter_mut(*idx); - if node.permits > 0 { + if node.needed_permits > 0 { old_waker = register_waker(&mut node.waker, waker); false } else { @@ -343,9 +357,9 @@ impl Acquire<'_> { } None => { // not yet enqueued - let needed = *permits; + let requested = *requested_permits; - if acquired_or_enqueue(semaphore, needed, Some(index), Some(waker), true) { + if acquired_or_enqueue(semaphore, requested, Some(index), Some(waker), true) { *done = true; return Poll::Ready(()); } @@ -369,7 +383,7 @@ impl Future for Acquire<'_> { /// Returns `true` if successfully acquired the semaphore; `false` otherwise. fn acquired_or_enqueue( sem: &Semaphore, - needed: usize, + requested: usize, index: Option<&mut Option>, waker: Option<&Waker>, enqueue_last: bool, @@ -379,50 +393,52 @@ fn acquired_or_enqueue( waker.is_some(), "only acquire waiters have a future owner" ); - let mut current = sem.permits.load(Ordering::Acquire); + let mut available = sem.available_permits.load(Ordering::Acquire); let mut lock = None; loop { - let (remaining, next) = if current >= needed { - (0, current - needed) + let (needed, next) = if available >= requested { + (0, available - requested) } else { - (needed - current, 0) + (requested - available, 0) }; - if remaining > 0 && lock.is_none() { + if needed > 0 && lock.is_none() { // No permits were immediately available, so this permit will // (probably) need to wait. We'll need to acquire a lock on the // wait queue before continuing. We need to do this _before_ the - // CAS that sets the new value of the semaphore's `permits` + // CAS that sets the new value of the semaphore's `available_permits` // counter. Otherwise, if we subtract the permits and then // acquire the lock, we might miss additional permits being // added while waiting for the lock. lock = Some(sem.waiters.lock()); } - if let Err(actual) = - sem.permits - .compare_exchange(current, next, Ordering::AcqRel, Ordering::Acquire) - { + if let Err(actual) = sem.available_permits.compare_exchange( + available, + next, + Ordering::AcqRel, + Ordering::Acquire, + ) { // other thread changed the permits; retry - current = actual; + available = actual; continue; } - // all needed permits were acquired - if remaining == 0 { + // all requested permits were acquired + if needed == 0 { return true; } // all available permits were acquired, but more are needed; - // enqueue a waiter with the remaining needed permits + // enqueue a waiter for the permits still needed let mut waiters = lock.take().unwrap_or_else(|| { - unreachable!("lock must be acquired when remaining {remaining} > 0"); + unreachable!("lock must be acquired when needed {needed} > 0"); }); let node = WaitNode { - permits: remaining, + needed_permits: needed, waker: waker.cloned(), }; let id = if enqueue_last { @@ -472,7 +488,7 @@ mod tests { } #[test] - fn release_with_positive_balance_does_not_take_the_queue_lock() { + fn release_uses_available_permits_without_queue_lock() { let semaphore = Semaphore::new(1); // Holding the queue lock deadlocks a release that needs it. @@ -486,18 +502,18 @@ mod tests { } #[test] - fn contended_release_takes_the_locked_path_and_clears_the_flag() { + fn release_takes_queue_lock_and_clears_detected_contention() { let semaphore = Semaphore::new(1); - semaphore.contended.store(true, Ordering::Relaxed); + semaphore.contention_detected.store(true, Ordering::Relaxed); - // Only the locked path clears the flag, so a cleared flag proves the fast path was skipped. + // Only the locked path clears this flag, so clearing it proves that the lock was taken. semaphore.release(1); assert_eq!(semaphore.available_permits(), 2); - assert!(!semaphore.contended.load(Ordering::Relaxed)); + assert!(!semaphore.contention_detected.load(Ordering::Relaxed)); } #[test] - fn release_with_zero_balance_hands_permits_to_the_queue() { + fn release_hands_permits_to_waiters_when_none_are_available() { let semaphore = Semaphore::new(0); let counter = Arc::new(WakeCounter(AtomicUsize::new(0))); let waker = Waker::from(counter.clone()); @@ -509,7 +525,7 @@ mod tests { assert_eq!(semaphore.available_permits(), 1); assert!(acquire.poll_once(&waker).is_ready()); - // Queue empty and balance positive again: the lock is not needed. + // The queue is empty and permits are available again: the lock is not needed. let queue = semaphore.waiters.lock(); semaphore.release(1); drop(queue); @@ -517,19 +533,19 @@ mod tests { } #[test] - fn locked_path_adds_to_a_positive_balance() { + fn distribute_permits_adds_to_available_permits() { let semaphore = Semaphore::new(2); - // A release that observed zero can find a positive balance once it holds the lock. - semaphore.insert_permits_with_lock(3, semaphore.waiters.lock()); + // Another release can make permits available before this one acquires the lock. + semaphore.distribute_permits(3, semaphore.waiters.lock()); assert_eq!(semaphore.available_permits(), 5); } #[test] - fn locked_overflow_panics_without_adding_permits() { + fn distribute_permits_overflow_preserves_available_permits() { let semaphore = Semaphore::new(usize::MAX); let result = panic::catch_unwind(AssertUnwindSafe(|| { - semaphore.insert_permits_with_lock(1, semaphore.waiters.lock()); + semaphore.distribute_permits(1, semaphore.waiters.lock()); })); assert!(result.is_err()); assert_eq!(semaphore.available_permits(), usize::MAX); @@ -537,7 +553,7 @@ mod tests { } #[test] - fn release_all_held_hands_permits_to_the_queue() { + fn release_with_lock_hands_permits_to_waiters() { let semaphore = Semaphore::new(1); let counter = Arc::new(WakeCounter(AtomicUsize::new(0))); let waker = Waker::from(counter.clone()); @@ -545,12 +561,12 @@ mod tests { let mut acquire = semaphore.poll_acquire(1); assert!(acquire.poll_once(&waker).is_pending()); - semaphore.release_all_held(1); + semaphore.release_with_lock(1); assert_eq!(counter.0.load(Ordering::Relaxed), 1); assert_eq!(semaphore.available_permits(), 0); assert!(acquire.poll_once(&waker).is_ready()); - semaphore.release_all_held(1); + semaphore.release_with_lock(1); assert_eq!(semaphore.available_permits(), 1); } diff --git a/asyncband/src/mutex/mod.rs b/asyncband/src/mutex/mod.rs index 5a83558e..da8e81c2 100644 --- a/asyncband/src/mutex/mod.rs +++ b/asyncband/src/mutex/mod.rs @@ -278,7 +278,7 @@ unsafe impl Sync for MutexGuard<'_, T> {} impl Drop for MutexGuard<'_, T> { fn drop(&mut self) { - self.lock.s.release_all_held(1); + self.lock.s.release_with_lock(1); } } @@ -445,7 +445,7 @@ unsafe impl Sync for OwnedMutexGuard {} impl Drop for OwnedMutexGuard { fn drop(&mut self) { - self.lock.s.release_all_held(1); + self.lock.s.release_with_lock(1); } } @@ -641,7 +641,7 @@ unsafe impl Sync for MappedMutexGuard<'_, T> {} impl Drop for MappedMutexGuard<'_, T> { fn drop(&mut self) { - self.s.release_all_held(1); + self.s.release_with_lock(1); } } @@ -840,7 +840,7 @@ unsafe impl Sync for OwnedMapp impl Drop for OwnedMappedMutexGuard { fn drop(&mut self) { - self.lock.s.release_all_held(1); + self.lock.s.release_with_lock(1); } } diff --git a/asyncband/src/semaphore/mod.rs b/asyncband/src/semaphore/mod.rs index 93340e31..2d7aeb61 100644 --- a/asyncband/src/semaphore/mod.rs +++ b/asyncband/src/semaphore/mod.rs @@ -124,7 +124,7 @@ impl Semaphore { self.s.drain_permits(up_to) } - /// Reduces the semaphore's logical permit balance by exactly `n`. + /// Removes exactly `n` permits, consuming future releases if fewer are available. /// /// This method returns immediately. If fewer than `n` permits are currently available, future /// releases repay the resulting deficit before ordinary queued acquisitions may proceed. diff --git a/tests-integration/tests/semaphore_ordering_test.rs b/tests-integration/tests/semaphore_ordering_test.rs index dfdae0de..043f79e1 100644 --- a/tests-integration/tests/semaphore_ordering_test.rs +++ b/tests-integration/tests/semaphore_ordering_test.rs @@ -16,11 +16,12 @@ // under the License. //! An acquire synchronizes with the release whose release sequence it reads from, and every -//! write to the balance is a read-modify-write, so that sequence extends through later releases -//! and acquisitions. A plain store would end it, and Miri would report a data race here. +//! update to the available permit count is a read-modify-write, so that sequence extends through +//! later releases and acquisitions. A plain store would end it, and Miri would report a data race +//! here. //! -//! The race shows only on schedules where the final acquire reads the latest balance rather than -//! a stale one, so the test repeats the scenario to cover several schedules in one run. +//! The race shows only on schedules where the final acquire reads the latest permit count rather +//! than a stale one, so the test repeats the scenario to cover several schedules in one run. use std::cell::UnsafeCell; use std::sync::Arc; @@ -55,7 +56,7 @@ fn run_once() { let data = Arc::new(Data(UnsafeCell::new(0))); let step = Arc::new(AtomicUsize::new(0)); - // Writes, then releases onto a positive balance. + // Writes, then releases while a permit is still available. let writer = { let (semaphore, data, step) = (semaphore.clone(), data.clone(), step.clone()); thread::spawn(move || { @@ -66,7 +67,7 @@ fn run_once() { }) }; - // Takes every permit, so the balance is zero. + // Takes every available permit. let drainer = { let (semaphore, step) = (semaphore.clone(), step.clone()); thread::spawn(move || { @@ -82,7 +83,7 @@ fn run_once() { }) }; - // Releases onto the zero balance without any synchronization with the writer. + // Releases when no permits are available, without any synchronization with the writer. let releaser = { let (semaphore, step) = (semaphore.clone(), step.clone()); thread::spawn(move || { diff --git a/tests-integration/tests/semaphore_test.rs b/tests-integration/tests/semaphore_test.rs index 51d9a77f..8f6937dc 100644 --- a/tests-integration/tests/semaphore_test.rs +++ b/tests-integration/tests/semaphore_test.rs @@ -229,7 +229,7 @@ fn reduce_permits_takes_priority_over_pending_acquires() { assert_eq!(s.available_permits(), 1); } -/// Releases race acquisitions that drain the balance to zero and link waiters. The permit count +/// Releases race acquisitions that take all available permits and link waiters. The permit count /// must be conserved, and no more permits than the capacity may ever be held at once. #[test] fn concurrent_releases_conserve_permits() { @@ -308,8 +308,8 @@ fn concurrent_releases_conserve_permits() { assert!(semaphore.try_acquire(PERMITS).is_some()); } -/// Two releases race on a balance one below `usize::MAX`: exactly one may succeed, and the other -/// must panic before adding anything, whichever path each of them takes. +/// Two releases race with `usize::MAX - 1` permits available: exactly one may succeed, and the +/// other must panic before adding anything, whichever path each of them takes. #[test] fn concurrent_releases_at_the_limit_panic_exactly_once() { let semaphore = Arc::new(Semaphore::new(usize::MAX - 1));