diff --git a/CHANGELOG.md b/CHANGELOG.md index bd5e84c9..0930a418 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,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. * Fix undefined behavior when sending or dropping a oneshot sender races with receiver completion or cancellation. +### Improvements + +* 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 * 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/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 13b26424..f26282d1 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; @@ -43,13 +44,18 @@ use crate::internal::waker_batch::WakerBatch; #[derive(Debug)] pub struct Semaphore { /// The current number of available permits in the semaphore. - permits: AtomicUsize, + 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, @@ -58,33 +64,36 @@ struct WaitNode { impl Semaphore { pub const fn new(permits: usize) -> Self { Self { - permits: AtomicUsize::new(permits), + 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, } } } @@ -97,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) { @@ -122,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, @@ -133,26 +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 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 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.distribute_permits(n, self.waiters.lock()); + } + + /// Releases `n` permits by acquiring the queue lock directly. + /// + /// 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); } } @@ -162,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() { @@ -175,20 +215,43 @@ impl Semaphore { wakers.wake_all(); } - fn insert_permits_with_lock<'a>( + /// 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 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 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.available_permits + .compare_exchange(expected, next, Ordering::Release, Ordering::Relaxed) + .map(|_| ()) + } + + /// 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; }; @@ -199,20 +262,30 @@ 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); - rem = 0; + 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_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.contention_detected.store(false, Ordering::Relaxed); + } + remaining = 0; } drop(waiters); batch.wake_all(); - if rem == 0 { + if remaining == 0 { return; } waiters = self.waiters.lock(); @@ -222,7 +295,7 @@ impl Semaphore { #[derive(Debug)] pub struct Acquire<'a> { - permits: usize, + requested_permits: usize, index: Option, semaphore: &'a Semaphore, done: bool, @@ -234,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); } @@ -252,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, @@ -268,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 { @@ -284,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(()); } @@ -310,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, @@ -320,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 { @@ -384,6 +459,8 @@ fn acquired_or_enqueue( #[cfg(test)] mod tests { + use std::panic; + use std::panic::AssertUnwindSafe; use std::sync::Arc; use std::task::Wake; @@ -410,6 +487,89 @@ mod tests { } } + #[test] + fn release_uses_available_permits_without_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_takes_queue_lock_and_clears_detected_contention() { + let semaphore = Semaphore::new(1); + semaphore.contention_detected.store(true, Ordering::Relaxed); + + // 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.contention_detected.load(Ordering::Relaxed)); + } + + #[test] + 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()); + 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()); + + // The queue is empty and permits are available again: the lock is not needed. + let queue = semaphore.waiters.lock(); + semaphore.release(1); + drop(queue); + assert_eq!(semaphore.available_permits(), 2); + } + + #[test] + fn distribute_permits_adds_to_available_permits() { + let semaphore = Semaphore::new(2); + + // 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 distribute_permits_overflow_preserves_available_permits() { + let semaphore = Semaphore::new(usize::MAX); + let result = panic::catch_unwind(AssertUnwindSafe(|| { + semaphore.distribute_permits(1, semaphore.waiters.lock()); + })); + assert!(result.is_err()); + assert_eq!(semaphore.available_permits(), usize::MAX); + assert!(semaphore.waiters.lock().is_empty()); + } + + #[test] + 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()); + assert!(semaphore.try_acquire(1)); + let mut acquire = semaphore.poll_acquire(1); + assert!(acquire.poll_once(&waker).is_pending()); + + 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_with_lock(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 5469d492..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(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(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(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(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/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..043f79e1 --- /dev/null +++ b/tests-integration/tests/semaphore_ordering_test.rs @@ -0,0 +1,118 @@ +// 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 +//! 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 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; +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() { + // 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)); + + // Writes, then releases while a permit is still available. + 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 available permit. + 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 when no permits are available, 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 db1903d9..8f6937dc 100644 --- a/tests-integration/tests/semaphore_test.rs +++ b/tests-integration/tests/semaphore_test.rs @@ -19,11 +19,16 @@ use std::panic::AssertUnwindSafe; use std::panic::catch_unwind; 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; @@ -223,3 +228,107 @@ fn reduce_permits_takes_priority_over_pending_acquires() { drop(permit); assert_eq!(s.available_permits(), 1); } + +/// 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() { + 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 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)); + 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..08f0f058 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -141,6 +141,10 @@ impl CommandMiri { "tests-integration", &["--test", "phaser_test"], )); + run_command(make_miri_cmd( + "tests-integration", + &["--test", "semaphore_ordering_test"], + )); } }