From e8bdb6419b1f4755533b1562c0147d26260a4001 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 21:58:30 +0800 Subject: [PATCH 01/32] refactor(latch): hide internal wait futures Latch::wait and Latch::wait_owned are async functions, so callers receive compiler-generated futures rather than these concrete structs. Keeping types that safe callers can neither construct nor obtain adds semver surface without adding capability; make them implementation details while preserving the method APIs. --- CHANGELOG.md | 1 + asyncband/src/latch/mod.rs | 23 ++--------------------- 2 files changed, 3 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d6d6f5e..9543c02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ All notable changes to this project will be documented in this file. * Remove the `asyncband::atomicbox` module and its `AtomicBox` and `AtomicOptionBox` types from the public API. * Remove the lossy `broadcast::overflow` channel; use the new lossless `broadcast::mpmc::unbounded` channel instead. * Remove `OnceMap::with_capacity` and `OnceMap::with_capacity_and_hasher`; use `OnceMap::new` or `OnceMap::with_hasher`, which allocate the backing table lazily. +* Remove the unconstructible `LatchWait` and `OwnedLatchWait` types from the public API; `Latch::wait` and `Latch::wait_owned` continue to return anonymous futures through their `async fn` signatures. * Rename `oneshot::Sender::is_closed` and `oneshot::Receiver::is_closed` to `is_disconnected`. * Remove `Semaphore::try_acquire_and_forget`, `Semaphore::acquire_and_forget`, `Semaphore::try_acquire_owned_and_forget`, and `Semaphore::acquire_owned_and_forget`; acquire a permit and call its `forget` method instead. * Replace `Semaphore::forget` with `Semaphore::drain_permits` and `Semaphore::forget_exact` with `Semaphore::reduce_permits`; permit-level `forget` methods are unchanged. diff --git a/asyncband/src/latch/mod.rs b/asyncband/src/latch/mod.rs index cfb4e1f..8763c26 100644 --- a/asyncband/src/latch/mod.rs +++ b/asyncband/src/latch/mod.rs @@ -55,7 +55,6 @@ //! [`count_down()`]: Latch::count_down //! [`arrive()`]: Latch::arrive -use std::fmt; use std::future::Future; use std::pin::Pin; use std::sync::Arc; @@ -258,21 +257,12 @@ impl Latch { } } -/// A wait future returned by [`Latch::wait()`]. -/// -/// This future will complete when the latch count reaches zero. #[must_use = "futures do nothing unless you `.await` or poll them"] -pub struct LatchWait<'a> { +struct LatchWait<'a> { token: Option, latch: &'a Latch, } -impl fmt::Debug for LatchWait<'_> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("LatchWait").finish_non_exhaustive() - } -} - impl Future for LatchWait<'_> { type Output = (); @@ -288,21 +278,12 @@ impl Drop for LatchWait<'_> { } } -/// An owned wait future returned by [`Latch::wait()`]. -/// -/// This future will complete when the latch count reaches zero. #[must_use = "futures do nothing unless you `.await` or poll them"] -pub struct OwnedLatchWait { +struct OwnedLatchWait { token: Option, latch: Arc, } -impl fmt::Debug for OwnedLatchWait { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("OwnedLatchWait").finish_non_exhaustive() - } -} - impl Future for OwnedLatchWait { type Output = (); From 4cfd4c754338cbe6b70d5b1d251b5e13c4353970 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 21:59:56 +0800 Subject: [PATCH 02/32] test: stop asserting debug text Debug output is diagnostic and intentionally non-contractual. Assertions on exact fields or rendered text freeze formatting without protecting user-visible behavior, so retain trait coverage and behavioral state tests instead. --- tests-integration/tests/once_map_test.rs | 3 -- tests-integration/tests/once_test.rs | 14 --------- tests-integration/tests/rwlock_test.rs | 37 ------------------------ 3 files changed, 54 deletions(-) diff --git a/tests-integration/tests/once_map_test.rs b/tests-integration/tests/once_map_test.rs index aa150d3..ef0cf3d 100644 --- a/tests-integration/tests/once_map_test.rs +++ b/tests-integration/tests/once_map_test.rs @@ -31,9 +31,6 @@ fn constructors_and_default() { let _: OnceMap = OnceMap::default(); let _: OnceMap = OnceMap::new(); let _: OnceMap = OnceMap::with_hasher(RandomState::new()); - - let map: OnceMap = OnceMap::new(); - assert!(format!("{map:?}").contains("OnceMap")); } #[tokio::test] diff --git a/tests-integration/tests/once_test.rs b/tests-integration/tests/once_test.rs index 71faaae..c341b2e 100644 --- a/tests-integration/tests/once_test.rs +++ b/tests-integration/tests/once_test.rs @@ -106,20 +106,6 @@ async fn test_once_cancelled() { assert!(ONCE.is_completed()); } -#[tokio::test] -async fn test_once_debug() { - let once = Once::new(); - let debug_str = format!("{:?}", once); - assert!(debug_str.contains("Once")); - assert!(debug_str.contains("done")); - assert!(debug_str.contains("false")); - - once.call_once(async || {}).await; - - let debug_str = format!("{:?}", once); - assert!(debug_str.contains("true")); -} - #[tokio::test] async fn test_once_default() { let once = Once::default(); diff --git a/tests-integration/tests/rwlock_test.rs b/tests-integration/tests/rwlock_test.rs index a368f44..12d2f35 100644 --- a/tests-integration/tests/rwlock_test.rs +++ b/tests-integration/tests/rwlock_test.rs @@ -349,43 +349,6 @@ async fn test_memory_ordering_correctness() { assert_eq!(*guard, vec![100, 2, 3, 4]); } -#[tokio::test] -async fn test_rwlock_debug_when_locked() { - let rwlock = Arc::new(RwLock::new(78)); - - let rwlock_debug_unlocked = format!("{rwlock:?}"); - assert!( - rwlock_debug_unlocked.contains("78"), - "RwLock Debug should show value when unlocked, got: {rwlock_debug_unlocked}" - ); - - let write_guard = rwlock.write().await; - let rwlock_debug_write = format!("{rwlock:?}"); - assert!( - rwlock_debug_write.contains(""), - "RwLock Debug should show when write lock is held, got: {rwlock_debug_write}" - ); - drop(write_guard); - - let read_guard = rwlock.read().await; - let rwlock_debug_read = format!("{rwlock:?}"); - - let shows_value = rwlock_debug_read.contains("78"); - let shows_locked = rwlock_debug_read.contains(""); - assert!( - shows_value || shows_locked, - "RwLock Debug with read lock should show either value or , got: {rwlock_debug_read}" - ); - - drop(read_guard); - - let rwlock_debug_final = format!("{rwlock:?}"); - assert!( - rwlock_debug_final.contains("78"), - "RwLock Debug should show value when all locks are released, got: {rwlock_debug_final}" - ); -} - #[tokio::test] async fn test_rwlock_zst() { // Test that RwLock works correctly with Zero-Sized Types From 7417ded9a4618867c9d5cf8e95d69404afdf57a5 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:01:16 +0800 Subject: [PATCH 03/32] test(latch): replace timing checks with state transitions Wall-clock delays test scheduler luck rather than latch semantics and make the suite slower. Poll wait futures directly to verify pending, wake, cancellation, and ready transitions while retaining a real concurrent-arrival case. --- tests-integration/tests/latch_test.rs | 204 +++++++------------------- 1 file changed, 54 insertions(+), 150 deletions(-) diff --git a/tests-integration/tests/latch_test.rs b/tests-integration/tests/latch_test.rs index 6e5d7a7..b527c0b 100644 --- a/tests-integration/tests/latch_test.rs +++ b/tests-integration/tests/latch_test.rs @@ -16,28 +16,14 @@ // under the License. use std::future::Future; -use std::prelude::rust_2015::Vec; use std::sync::Arc; -use std::sync::atomic::AtomicU32; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::task::Context; use std::task::Wake; use std::task::Waker; -use std::time::Duration; -use std::time::Instant; - -use asyncband::latch::*; - -macro_rules! assert_time { - ($time:expr, $mills:literal $(,)?) => { - assert!( - (Duration::from_millis($mills - 1) - ..Duration::from_millis($mills + std::cmp::max($mills >> 1, 50))) - .contains(&$time) - ) - }; -} + +use asyncband::latch::Latch; struct TrackWake(AtomicUsize); @@ -48,24 +34,22 @@ impl Wake for TrackWake { } #[test] -fn test_count_down() { - let latch = Latch::new(3); - latch.count_down(); - latch.count_down(); - latch.count_down(); - assert_eq!(latch.count(), 0); -} +fn countdown_operations_saturate_at_zero() { + let latch = Latch::new(5); -#[test] -fn test_try_wait() { - let latch = Latch::new(0); + latch.arrive(0); + assert_eq!(latch.try_wait(), Err(5)); + + latch.arrive(3); + assert_eq!(latch.try_wait(), Err(2)); + + latch.count_down(); + latch.arrive(2); assert_eq!(latch.try_wait(), Ok(())); -} -#[test] -fn test_try_wait_err() { - let latch = Latch::new(3); - assert_eq!(latch.try_wait(), Err(3)); + latch.count_down(); + latch.arrive(u32::MAX); + assert_eq!(latch.count(), 0); } #[test] @@ -85,142 +69,62 @@ fn cancelled_wait_releases_its_waker() { } #[test] -fn test_arrive_zero() { +fn final_arrival_wakes_every_waiter() { let latch = Latch::new(2); - latch.arrive(0); - assert_eq!(latch.count(), 2); -} - -#[test] -fn test_more_arrive() { - let latch = Latch::new(10); - for _ in 0..4 { - latch.arrive(3); - } - assert_eq!(latch.count(), 0); -} - -#[tokio::test] -async fn test_arrive() { - let latch = Latch::new(3); - latch.arrive(3); - latch.wait().await; - assert_eq!(latch.count(), 0); -} - -#[tokio::test] -async fn test_last_one_signal() { - let latch = Arc::new(Latch::new(3)); - let l1 = latch.clone(); + let first_tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let second_tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let first_waker = Waker::from(first_tracker.clone()); + let second_waker = Waker::from(second_tracker.clone()); + let mut first_context = Context::from_waker(&first_waker); + let mut second_context = Context::from_waker(&second_waker); + let mut first = Box::pin(latch.wait()); + let mut second = Box::pin(latch.wait()); + + assert!(first.as_mut().poll(&mut first_context).is_pending()); + assert!(second.as_mut().poll(&mut second_context).is_pending()); latch.count_down(); - latch.count_down(); + assert_eq!(first_tracker.0.load(Ordering::Relaxed), 0); + assert_eq!(second_tracker.0.load(Ordering::Relaxed), 0); - let start = Instant::now(); - - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(32)).await; - l1.count_down() - }); - - latch.wait().await; - assert_time!(start.elapsed(), 32); - assert_eq!(latch.count(), 0); + latch.count_down(); + assert_eq!(first_tracker.0.load(Ordering::Relaxed), 1); + assert_eq!(second_tracker.0.load(Ordering::Relaxed), 1); + assert!(first.as_mut().poll(&mut first_context).is_ready()); + assert!(second.as_mut().poll(&mut second_context).is_ready()); } #[tokio::test] -async fn test_gate_wait() { +async fn owned_wait_can_move_to_another_task() { let latch = Arc::new(Latch::new(1)); - let tasks: Vec<_> = (0..4) - .map(|_| { - let latch = latch.clone(); - let start = Instant::now(); - - tokio::spawn(async move { - latch.wait().await; - start.elapsed() - }) - }) - .collect(); - - tokio::time::sleep(Duration::from_millis(20)).await; - latch.count_down(); + let waiter = tokio::spawn(latch.clone().wait_owned()); - for t in tasks { - assert_time!(t.await.unwrap(), 20); - } + latch.count_down(); + waiter.await.unwrap(); } -#[tokio::test] -async fn test_multi_tasks() { - const SIZE: u32 = 16; - let latch = Arc::new(Latch::new(SIZE)); - let counter = Arc::new(AtomicU32::new(0)); +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_arrivals_complete_the_latch() { + const TASKS: usize = 32; - for _ in 0..SIZE { - let latch = latch.clone(); - let counter = counter.clone(); + let latch = Arc::new(Latch::new(TASKS as u32)); + let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let mut tasks = Vec::with_capacity(TASKS); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(20)).await; - counter.fetch_add(1, Ordering::Relaxed); - latch.count_down() - }); + for _ in 0..TASKS { + let latch = latch.clone(); + let start = start.clone(); + tasks.push(tokio::spawn(async move { + start.wait().await; + latch.count_down(); + })); } - let start = Instant::now(); - + start.wait().await; latch.wait().await; - assert_time!(start.elapsed(), 20); - assert_eq!(counter.load(Ordering::Relaxed), SIZE); - assert_eq!(latch.count(), 0); -} - -#[tokio::test] -async fn test_more_count_down() { - const SIZE: u32 = 16; - let latch = Arc::new(Latch::new(SIZE)); - let counter = Arc::new(AtomicU32::new(0)); - - for _ in 0..(SIZE + (SIZE >> 1)) { - let latch = latch.clone(); - let counter = counter.clone(); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(10)).await; - counter.fetch_add(1, Ordering::Relaxed); - latch.count_down() - }); + for task in tasks { + task.await.unwrap(); } - - latch.wait().await; - assert!(counter.load(Ordering::Relaxed) >= SIZE); - assert_eq!(latch.count(), 0); - latch.count_down(); assert_eq!(latch.count(), 0); } - -#[tokio::test] -async fn test_select_two_wait() { - let latch1 = Arc::new(Latch::new(1)); - let latch2 = Arc::new(Latch::new(1)); - let l1 = latch1.clone(); - let l2 = latch2.clone(); - - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(50)).await; - l1.count_down(); - }); - - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(10)).await; - l2.count_down(); - }); - - assert!(tokio::select! { - _ = latch1.wait() => false, - _ = latch2.wait() => true, - }); - - latch1.wait().await; -} From 9c92b440ec54f08baa33f692a635be1c5331cd42 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:01:54 +0800 Subject: [PATCH 04/32] test(waitgroup): make lifecycle tests deterministic Sleep-based completion checks can pass or fail according to task scheduling. Drive the IntoFuture state and worker-handle drops explicitly so the tests document ownership, cancellation, and the last-handle transition. --- tests-integration/tests/waitgroup_test.rs | 81 +++++++++-------------- 1 file changed, 31 insertions(+), 50 deletions(-) diff --git a/tests-integration/tests/waitgroup_test.rs b/tests-integration/tests/waitgroup_test.rs index 281ee49..970db9d 100644 --- a/tests-integration/tests/waitgroup_test.rs +++ b/tests-integration/tests/waitgroup_test.rs @@ -16,69 +16,50 @@ // under the License. use std::future::IntoFuture; -use std::time::Duration; use asyncband::waitgroup::WaitGroup; -use tests_integration::test_runtime; +use tests_integration::poll_once; -#[test] -fn test_wait_group_drop() { +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn waits_for_all_worker_handles() { let wg = WaitGroup::new(); - for _i in 0..100 { - let w = wg.clone(); - test_runtime().spawn(async move { - drop(w); - }); + let mut tasks = Vec::new(); + for _ in 0..100 { + let worker = wg.clone(); + tasks.push(tokio::spawn(async move { + drop(worker); + })); } - pollster::block_on(wg.into_future()); -} -#[test] -fn test_wait_group_await() { - let wg = WaitGroup::new(); - for _i in 0..100 { - let w = wg.clone(); - test_runtime().spawn(async move { - w.await; - }); + wg.await; + for task in tasks { + task.await.unwrap(); } - pollster::block_on(wg.into_future()); } #[test] -fn test_wait_group_timeout() { +fn wait_is_pending_until_the_last_handle_drops() { let wg = WaitGroup::new(); - let _wg_clone = wg.clone(); - let timeout = test_runtime().block_on(async move { - tokio::select! { - _ = tokio::time::sleep(Duration::from_millis(50)) => true , - _ = wg => false, - } - }); - assert!(timeout); + let worker = wg.clone(); + let mut wait = Box::pin(wg.into_future()); + + assert!(poll_once(wait.as_mut()).is_pending()); + drop(worker); + assert!(poll_once(wait.as_mut()).is_ready()); } #[test] -fn test_wait_group_cancel() { +fn cancelling_one_wait_does_not_cancel_another() { let wg = WaitGroup::new(); - let wg_clone = wg.clone().into_future(); - let wg_clone_2 = wg.clone().into_future(); - test_runtime().block_on(async move { - tokio::select! { - _ = tokio::time::sleep(Duration::ZERO) => {}, - _ = wg_clone => {} - } - }); - let fut = test_runtime().spawn(async move { - wg_clone_2.await; - }); - std::thread::sleep(Duration::from_millis(50)); - drop(wg); - let timeout = test_runtime().block_on(async move { - tokio::select! { - _ = tokio::time::sleep(Duration::from_secs(60)) => true , - _ = fut => false, - } - }); - assert!(!timeout); + let worker = wg.clone(); + let first = wg.into_future(); + let mut second = Box::pin(first.clone()); + let mut first = Box::pin(first); + + assert!(poll_once(first.as_mut()).is_pending()); + assert!(poll_once(second.as_mut()).is_pending()); + drop(first); + + drop(worker); + assert!(poll_once(second.as_mut()).is_ready()); } From a5e32940a417fff71a6db041431493c07dc90af3 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:02:50 +0800 Subject: [PATCH 05/32] test(once): coordinate retries without timers Artificial delays were only keeping an initializer in flight long enough for another caller to join. Poll the initializer to Pending and coordinate cancellation, panic, and retry directly so the same contracts are covered deterministically. --- tests-integration/tests/once_test.rs | 179 +++++++++++---------------- 1 file changed, 75 insertions(+), 104 deletions(-) diff --git a/tests-integration/tests/once_test.rs b/tests-integration/tests/once_test.rs index c341b2e..622f746 100644 --- a/tests-integration/tests/once_test.rs +++ b/tests-integration/tests/once_test.rs @@ -15,146 +15,117 @@ // specific language governing permissions and limitations // under the License. +use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; -use std::time::Duration; use asyncband::once::Once; -use tests_integration::test_runtime; -use tokio_test::assert_ready; +use tests_integration::poll_once; #[tokio::test] -async fn test_call_once_runs_only_once() { - static ONCE: Once = Once::new(); - static COUNTER: AtomicUsize = AtomicUsize::new(0); +async fn call_once_runs_only_one_initializer() { + let once = Once::new(); + let counter = AtomicUsize::new(0); - assert!(!ONCE.is_completed()); + assert!(!once.is_completed()); - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); }) .await; - assert!(ONCE.is_completed()); - assert_eq!(COUNTER.load(Ordering::SeqCst), 1); + assert!(once.is_completed()); + assert_eq!(counter.load(Ordering::SeqCst), 1); - // Second call should not run the closure - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); }) .await; - assert_eq!(COUNTER.load(Ordering::SeqCst), 1); + assert_eq!(counter.load(Ordering::SeqCst), 1); } -#[test] -fn test_once_multi_task() { - static ONCE: Once = Once::new(); - static COUNTER: AtomicUsize = AtomicUsize::new(0); - - test_runtime().block_on(async { - const N: usize = 100; - - let mut handles = Vec::with_capacity(N); - - for _ in 0..N { - handles.push(tokio::spawn(async move { - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); - }) - .await; - })); - } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_callers_run_one_initializer() { + const TASKS: usize = 100; + + let once = Arc::new(Once::new()); + let counter = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let mut tasks = Vec::with_capacity(TASKS); + + for _ in 0..TASKS { + let once = once.clone(); + let counter = counter.clone(); + let start = start.clone(); + tasks.push(tokio::spawn(async move { + start.wait().await; + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); + }) + .await; + })); + } - for handle in handles { - handle.await.unwrap(); - } + start.wait().await; + for task in tasks { + task.await.unwrap(); + } - // Only one task should have incremented the counter - assert_eq!(COUNTER.load(Ordering::SeqCst), 1); - assert!(ONCE.is_completed()); - }); + assert_eq!(counter.load(Ordering::SeqCst), 1); + assert!(once.is_completed()); } #[tokio::test] -async fn test_once_cancelled() { - static ONCE: Once = Once::new(); - static COUNTER: AtomicUsize = AtomicUsize::new(0); - - let handle1 = tokio::spawn(async { - let fut = ONCE.call_once(async || { - tokio::time::sleep(Duration::from_millis(1000)).await; - COUNTER.fetch_add(1, Ordering::SeqCst); - }); - let timeout = tokio::time::timeout(Duration::from_millis(1), fut).await; - assert!(timeout.is_err()); - }); +async fn cancelled_initializer_can_be_retried() { + let once = Once::new(); + let mut first = Box::pin(once.call_once(async || std::future::pending::<()>().await)); - let handle2 = tokio::spawn(async { - tokio::time::sleep(Duration::from_millis(100)).await; - ONCE.call_once(async || { - COUNTER.fetch_add(10, Ordering::SeqCst); - }) - .await; - }); - - handle1.await.unwrap(); - handle2.await.unwrap(); + assert!(poll_once(first.as_mut()).is_pending()); + drop(first); - // The second task should have run since the first was cancelled - assert_eq!(COUNTER.load(Ordering::SeqCst), 10); - assert!(ONCE.is_completed()); + once.call_once(async || {}).await; + assert!(once.is_completed()); } #[tokio::test] -async fn test_once_default() { - let once = Once::default(); - assert!(!once.is_completed()); -} - -#[tokio::test] -async fn test_once_retry_after_panic() { - static ONCE: Once = Once::new(); - static COUNTER: AtomicUsize = AtomicUsize::new(0); - - let handle = tokio::spawn(async { - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); - panic!("boom"); - }) - .await; +async fn panicked_initializer_can_be_retried() { + let once = Arc::new(Once::new()); + let counter = Arc::new(AtomicUsize::new(0)); + + let handle = tokio::spawn({ + let once = once.clone(); + let counter = counter.clone(); + async move { + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); + panic!("boom"); + }) + .await; + } }); - let err = handle.await.expect_err("once init should panic"); - assert!(err.is_panic()); + let error = handle.await.expect_err("initializer should panic"); + assert!(error.is_panic()); - ONCE.call_once(async || { - COUNTER.fetch_add(1, Ordering::SeqCst); + once.call_once(async || { + counter.fetch_add(1, Ordering::SeqCst); }) .await; - assert_eq!(COUNTER.load(Ordering::SeqCst), 2); - assert!(ONCE.is_completed()); + assert_eq!(counter.load(Ordering::SeqCst), 2); + assert!(once.is_completed()); } #[tokio::test] -async fn test_once_wait() { - // wait after call_once completed - { - let once = Once::new(); - once.call_once(async || {}).await; - assert_ready!(tokio_test::task::spawn(once.wait()).poll()); - } +async fn wait_observes_completion() { + let once = Once::new(); + let mut waiting = Box::pin(once.wait()); - // wait before call_once completed - { - static ONCE: Once = Once::new(); - let handle = tokio::spawn(async { - ONCE.wait().await; - }); + assert!(poll_once(waiting.as_mut()).is_pending()); + once.call_once(async || {}).await; + assert!(poll_once(waiting.as_mut()).is_ready()); - tokio::time::sleep(Duration::from_millis(100)).await; - ONCE.call_once(async || {}).await; - handle.await.unwrap(); - } + let mut completed = Box::pin(once.wait()); + assert!(poll_once(completed.as_mut()).is_ready()); } From b6e16a743ed08e8e949e41d0e46bc180e394cfa3 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:03:53 +0800 Subject: [PATCH 06/32] test(once-cell): drive initialization state explicitly Timer sleeps made initialization races dependent on executor scheduling. Explicitly drive the winning initializer and waiting callers through their states so retry, error, cancellation, and publication behavior remain deterministic. --- tests-integration/tests/once_cell_test.rs | 223 ++++++++-------------- 1 file changed, 79 insertions(+), 144 deletions(-) diff --git a/tests-integration/tests/once_cell_test.rs b/tests-integration/tests/once_cell_test.rs index 5370f02..f930339 100644 --- a/tests-integration/tests/once_cell_test.rs +++ b/tests-integration/tests/once_cell_test.rs @@ -17,187 +17,122 @@ use std::sync::Arc; use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; -use std::time::Duration; use asyncband::once::OnceCell; -use tokio::sync::Mutex; +use tests_integration::poll_once; -struct Foo { - value: Arc, -} - -impl Foo { - async fn new(value: Arc) -> Self { - // simulate some async initialization work - tokio::time::sleep(Duration::from_millis(100)).await; - Foo { value } - } +struct DropFlag(Arc); - fn value(&self) -> bool { - self.value.load(Ordering::Acquire) - } -} - -impl Drop for Foo { +impl Drop for DropFlag { fn drop(&mut self) { - self.value.store(true, Ordering::Release); + self.0.store(true, Ordering::Release); } } #[tokio::test] -async fn drop_cell() { +async fn dropping_the_cell_drops_its_value() { let dropped = Arc::new(AtomicBool::new(false)); { let cell = OnceCell::new(); - assert!(cell.get().is_none()); - - let state = dropped.clone(); - cell.get_or_init(|| async { Foo::new(state).await }).await; - - let foo = cell.get().unwrap(); - assert!(!foo.value()); + cell.get_or_init(async || DropFlag(dropped.clone())).await; assert!(!dropped.load(Ordering::Acquire)); } assert!(dropped.load(Ordering::Acquire)); } -#[test] -fn multi_init() { - let rt = tokio::runtime::Builder::new_multi_thread() - .worker_threads(4) - .build() - .unwrap(); - - static CELL: OnceCell = OnceCell::new(); - - rt.block_on(async { - const N: usize = 100; - - let values = Arc::new(Mutex::new(vec![0; N])); - let mut handles = Vec::with_capacity(N); - - for i in 0..N { - let values = values.clone(); - handles.push(rt.spawn(async move { - let result = CELL.get_or_init(move || async move { i + 1000 }).await; - let mut values = values.lock().await; - values[i] = *result; - })); - } - - for handle in handles { - handle.await.unwrap(); - } - let cell_value = CELL.get().unwrap(); - for (index, value) in values.lock().await.iter().enumerate() { - assert_eq!(*value, *cell_value, "mismatch at index {index}"); - } - }); -} +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_callers_publish_one_value() { + const TASKS: usize = 100; + + let cell = Arc::new(OnceCell::new()); + let attempts = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let mut tasks = Vec::with_capacity(TASKS); + + for value in 0..TASKS { + let cell = cell.clone(); + let attempts = attempts.clone(); + let start = start.clone(); + tasks.push(tokio::spawn(async move { + start.wait().await; + *cell + .get_or_init(async || { + attempts.fetch_add(1, Ordering::SeqCst); + value + }) + .await + })); + } -#[tokio::test] -async fn init_cancelled() { - static CELL: OnceCell = OnceCell::new(); - - let handle1 = tokio::spawn(async { - let fut = CELL.get_or_init(|| async { - tokio::time::sleep(Duration::from_millis(1000)).await; - 1 - }); - let timeout = tokio::time::timeout(Duration::from_millis(1), fut).await; - assert!(timeout.is_err()); - }); - - let handle2 = tokio::spawn(async { - tokio::time::sleep(Duration::from_millis(100)).await; - let value = CELL.get_or_init(|| async { 2 }).await; - assert_eq!(*value, 2); - }); - - handle1.await.unwrap(); - handle2.await.unwrap(); + start.wait().await; + let expected = tasks.remove(0).await.unwrap(); + for task in tasks { + assert_eq!(task.await.unwrap(), expected); + } + assert_eq!(attempts.load(Ordering::SeqCst), 1); } #[tokio::test] -async fn init_error() { - { - static CELL: OnceCell = OnceCell::new(); - - let handle1 = tokio::spawn(async { - let result = CELL.get_or_try_init(|| async { Err(()) }).await; - assert!(result.is_err()); - }); - - let handle2 = tokio::spawn(async { - tokio::time::sleep(Duration::from_millis(100)).await; - let value = CELL.get_or_try_init(|| async { Ok::<_, ()>(2) }).await; - assert_eq!(*value.unwrap(), 2); - }); - - handle1.await.unwrap(); - handle2.await.unwrap(); - } - - { - static CELL: OnceCell = OnceCell::new(); +async fn cancelled_initialization_can_be_retried() { + let cell = OnceCell::new(); + let mut first = Box::pin(cell.get_or_init(async || std::future::pending::().await)); - let handle1 = tokio::spawn(async { - let value = CELL.get_or_try_init(|| async { Ok::<_, ()>(2) }).await; - assert_eq!(*value.unwrap(), 2); - }); + assert!(poll_once(first.as_mut()).is_pending()); + drop(first); - let handle2 = tokio::spawn(async { - tokio::time::sleep(Duration::from_millis(100)).await; - let value = CELL.get_or_try_init(|| async { Err(()) }).await; - assert_eq!(*value.unwrap(), 2); - }); - - handle1.await.unwrap(); - handle2.await.unwrap(); - } + assert_eq!(*cell.get_or_init(async || 2).await, 2); } #[tokio::test] -async fn get_mut_or_init() { - let mut cell: OnceCell = OnceCell::new(); - let v = cell - .get_mut_or_init(async || { - tokio::time::sleep(Duration::from_millis(1)).await; - 41 +async fn failed_initialization_can_be_retried_and_success_is_cached() { + let cell = OnceCell::new(); + let attempts = AtomicUsize::new(0); + + let error = cell + .get_or_try_init(async || { + attempts.fetch_add(1, Ordering::SeqCst); + Err::("not ready") }) .await; - *v += 1; + assert_eq!(error, Err("not ready")); - let v = tokio::spawn(async move { *cell.get_or_init(async || 0).await }) - .await - .unwrap(); - assert_eq!(v, 42); + let value = cell + .get_or_try_init(async || { + attempts.fetch_add(1, Ordering::SeqCst); + Ok::<_, &str>(2) + }) + .await; + assert_eq!(value, Ok(&2)); + + let cached = cell + .get_or_try_init(async || { + attempts.fetch_add(1, Ordering::SeqCst); + Ok::<_, &str>(3) + }) + .await; + assert_eq!(cached, Ok(&2)); + assert_eq!(attempts.load(Ordering::SeqCst), 2); } #[tokio::test] -async fn get_mut_or_try_init() { +async fn exclusive_initialization_returns_mutable_access() { let mut cell: OnceCell = OnceCell::new(); - let r = cell - .get_mut_or_try_init(async || { - tokio::time::sleep(Duration::from_millis(1)).await; - Err(()) - }) + + let failed = cell + .get_mut_or_try_init(async || Err::("not ready")) .await; - assert!(r.is_err()); + assert_eq!(failed, Err("not ready")); assert_eq!(cell.get_mut(), None); - let v = tokio::spawn(async move { - let v = cell - .get_mut_or_try_init(async || Ok::<_, ()>(10)) - .await - .unwrap(); - *v += 5; - *v - }) - .await - .unwrap(); - assert_eq!(v, 15); + let value = cell.get_mut_or_init(async || 41).await; + *value += 1; + + let value = tokio::spawn(async move { *cell.get_or_init(async || 0).await }) + .await + .unwrap(); + assert_eq!(value, 42); } From d8814c0b158a6e95f16fe914fdea86c529c13575 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:04:57 +0800 Subject: [PATCH 07/32] test(rwlock): remove redundant unwind scenarios These scenarios panic only after a guard has been created, so unwinding exercises the same ordinary Drop path as non-panicking scope exit. Removing the variants reduces noise while the mapping and lock-release invariants remain covered by tests that reach distinct code paths. --- tests-integration/tests/rwlock_test.rs | 164 ------------------------- 1 file changed, 164 deletions(-) diff --git a/tests-integration/tests/rwlock_test.rs b/tests-integration/tests/rwlock_test.rs index 12d2f35..3e95cb4 100644 --- a/tests-integration/tests/rwlock_test.rs +++ b/tests-integration/tests/rwlock_test.rs @@ -202,132 +202,6 @@ async fn test_guard_prevents_concurrent_access() { assert_eq!(*final_guard, 123); } -#[test] -fn test_lock_panic_safety() { - // Test panic safety with synchronous locks - use std::panic::AssertUnwindSafe; - - let rwlock = Arc::new(RwLock::new(0)); - let rwlock_clone = rwlock.clone(); - - let result = std::panic::catch_unwind(AssertUnwindSafe(move || { - let _guard = rwlock_clone.try_read().unwrap(); - panic!("test panic"); - })); - - assert!(result.is_err()); - // Lock should be released after panic - assert!(rwlock.try_read().is_some()); -} - -#[tokio::test] -async fn test_async_lock_panic_safety() { - // Test panic safety with async locks - let rwlock = Arc::new(RwLock::new(0)); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let _guard = rwlock_clone.read().await; - panic!("async test panic"); - }); - - // panic - assert!(handle.await.is_err()); - - let guard = rwlock.try_read(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_owned_guard_panic_safety() { - // Test panic safety with owned guards - let rwlock = Arc::new(RwLock::new(0)); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let _guard = rwlock_clone.clone().read_owned().await; - panic!("owned guard panic"); - }); - - assert!(handle.await.is_err()); - - // Lock should be available after the panicked task - let guard = rwlock.try_read(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_mapped_guard_panic_safety() { - // Test panic safety with mapped guards - let rwlock = Arc::new(RwLock::new((66, vec![1, 2, 3]))); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let guard = rwlock_clone.read().await; - let _mapped = RwLockReadGuard::map(guard, |data| &data.0); - panic!("mapped guard panic"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_read(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_mapped_write_guard_panic_safety() { - // Test panic safety with mapped write guards - let rwlock = Arc::new(RwLock::new((42, String::from("test")))); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let guard = rwlock_clone.write().await; - let _mapped = RwLockWriteGuard::map(guard, |data| &mut data.0); - panic!("mapped write guard panic"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_write(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_owned_mapped_read_guard_panic_safety() { - // Test panic safety with owned mapped read guards - let rwlock = Arc::new(RwLock::new((100, vec![4, 5, 6]))); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let guard = rwlock_clone.read_owned().await; - let _mapped = OwnedRwLockReadGuard::map(guard, |data| &data.1); - panic!("owned mapped read guard panic"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_read(); - assert!(guard.is_some()); -} - -#[tokio::test] -async fn test_owned_mapped_write_guard_panic_safety() { - // Test panic safety with owned mapped write guards - let rwlock = Arc::new(RwLock::new((200, String::from("owned")))); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let guard = rwlock_clone.write_owned().await; - let _mapped = OwnedRwLockWriteGuard::map(guard, |data| &mut data.1); - panic!("owned mapped write guard panic"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_write(); - assert!(guard.is_some()); -} - #[tokio::test] async fn test_memory_ordering_correctness() { // Test that rwlock provides proper memory ordering guarantees @@ -801,44 +675,6 @@ async fn test_downgrade_with_max_readers() { assert_eq!(*write_guard2, 100); } -#[tokio::test] -async fn test_downgrade_panic_safety() { - let rwlock = Arc::new(RwLock::new((0, "test".to_string()))); - - { - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let mut write_guard = rwlock_clone.write().await; - write_guard.0 = 42; - let read_guard = write_guard.downgrade(); - assert_eq!(read_guard.0, 42); - panic!("test panic after downgrade"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_read().unwrap(); - assert_eq!(guard.0, 42); - drop(guard); - } - - { - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let owned_write = rwlock_clone.write_owned().await; - let mut mapped_write = OwnedRwLockWriteGuard::map(owned_write, |data| &mut data.1); - *mapped_write = "panic_test".to_string(); - let _mapped_read = mapped_write.downgrade(); - panic!("test panic with owned mapped downgrade"); - }); - - assert!(handle.await.is_err()); - - let guard = rwlock.try_read().unwrap(); - assert_eq!(guard.1, "panic_test"); - } -} - #[tokio::test] async fn test_downgrade_prevents_deadlock() { // Test the classic deadlock prevention scenario From 1109f2530bd7d5573012b20eb480f60d66631536 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:06:02 +0800 Subject: [PATCH 08/32] test(rwlock): poll fairness transitions directly Yielding tasks cannot prove FIFO ordering because the executor chooses their registration order. Register a writer and a later reader by polling them directly while the lock is held, then verify that the writer becomes ready first. --- tests-integration/tests/rwlock_test.rs | 252 ++----------------------- 1 file changed, 20 insertions(+), 232 deletions(-) diff --git a/tests-integration/tests/rwlock_test.rs b/tests-integration/tests/rwlock_test.rs index 3e95cb4..81c1dea 100644 --- a/tests-integration/tests/rwlock_test.rs +++ b/tests-integration/tests/rwlock_test.rs @@ -20,6 +20,9 @@ use std::sync::Arc; use std::sync::Weak; use asyncband::rwlock::*; +use tests_integration::poll_once; +use tokio_test::assert_pending; +use tokio_test::assert_ready; #[test] fn test_try_read_write_never_blocks() { @@ -159,49 +162,6 @@ async fn test_stress_concurrent_readers_writers() { } } -#[tokio::test] -async fn test_guard_prevents_concurrent_access() { - let rwlock = Arc::new(RwLock::new(0)); - let rwlock_clone = rwlock.clone(); - let writer_queued = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let writer_queued_clone = writer_queued.clone(); - let writer_completed = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let writer_completed_clone = writer_completed.clone(); - - let read_guard = rwlock.read().await; - - assert!(rwlock.try_write().is_none()); - - let handle = tokio::spawn(async move { - writer_queued_clone.store(true, std::sync::atomic::Ordering::SeqCst); - let mut write_guard = rwlock_clone.write().await; - *write_guard = 123; - writer_completed_clone.store(true, std::sync::atomic::Ordering::SeqCst); - *write_guard - }); - - while !writer_queued.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - for _ in 0..10 { - tokio::task::yield_now().await; - } - - assert!(!writer_completed.load(std::sync::atomic::Ordering::SeqCst)); - assert!(rwlock.try_write().is_none()); - - drop(read_guard); - - let result = handle.await.unwrap(); - assert_eq!(result, 123); - assert!(writer_completed.load(std::sync::atomic::Ordering::SeqCst)); - - // Verify the write took effect and lock is available - let final_guard = rwlock.try_read().unwrap(); - assert_eq!(*final_guard, 123); -} - #[tokio::test] async fn test_memory_ordering_correctness() { // Test that rwlock provides proper memory ordering guarantees @@ -587,63 +547,6 @@ async fn test_downgrade_atomicity() { } } -#[tokio::test] -async fn test_downgrade_allows_concurrent_readers() { - // Test that downgrading a write lock allows other readers to acquire the lock. - let rwlock = Arc::new(RwLock::new(0i32)); - let writer_started = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let downgrade_completed = Arc::new(std::sync::atomic::AtomicBool::new(false)); - - let writer_rwlock = rwlock.clone(); - let writer_started_clone = writer_started.clone(); - let downgrade_completed_clone = downgrade_completed.clone(); - - let writer_handle = tokio::spawn(async move { - let mut write_guard = writer_rwlock.write().await; - *write_guard = 42; - writer_started_clone.store(true, std::sync::atomic::Ordering::SeqCst); - - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; - - let read_guard = write_guard.downgrade(); - downgrade_completed_clone.store(true, std::sync::atomic::Ordering::SeqCst); - - assert_eq!(*read_guard, 42); - - tokio::time::sleep(tokio::time::Duration::from_millis(20)).await; - *read_guard - }); - - while !writer_started.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - let mut reader_handles = vec![]; - for i in 0..5 { - let reader_rwlock = rwlock.clone(); - let downgrade_completed_clone = downgrade_completed.clone(); - - let handle = tokio::spawn(async move { - while !downgrade_completed_clone.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - let read_guard = reader_rwlock.read().await; - assert_eq!(*read_guard, 42); - (i, *read_guard) - }); - reader_handles.push(handle); - } - - // All tasks should complete successfully - let writer_result = writer_handle.await.unwrap(); - assert_eq!(writer_result, 42); - - for handle in reader_handles { - let (reader_id, value) = handle.await.unwrap(); - assert_eq!(value, 42, "Reader {reader_id} should see the written value"); - } -} #[tokio::test] async fn test_downgrade_with_max_readers() { let rwlock = Arc::new(RwLock::with_max_readers(0, NonZeroUsize::new(3).unwrap())); @@ -676,138 +579,23 @@ async fn test_downgrade_with_max_readers() { } #[tokio::test] -async fn test_downgrade_prevents_deadlock() { - // Test the classic deadlock prevention scenario - // Demonstrates how downgrade enables safe lock ordering patterns - - let rwlock = Arc::new(RwLock::new(vec![1, 2, 3])); - - let rwlock1 = rwlock.clone(); - let task1 = tokio::spawn(async move { - let mut write_guard = rwlock1.write().await; - write_guard.push(4); - let len_after_write = write_guard.len(); - - let read_guard = write_guard.downgrade(); - - assert_eq!(read_guard.len(), len_after_write); - - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; - - let final_len = read_guard.len(); - drop(read_guard); - final_len - }); - - let rwlock2 = rwlock.clone(); - let task2 = tokio::spawn(async move { - tokio::time::sleep(tokio::time::Duration::from_millis(5)).await; - - let mut write_guard = rwlock2.write().await; - write_guard.push(5); - - let read_guard = write_guard.downgrade(); - let final_len = read_guard.len(); - - drop(read_guard); - final_len - }); - - let rwlock3 = rwlock.clone(); - let task3 = tokio::spawn(async move { - tokio::time::sleep(tokio::time::Duration::from_millis(15)).await; - - let read_guard = rwlock3.read().await; - read_guard.len() - }); - - let result1 = task1.await.unwrap(); - let result2 = task2.await.unwrap(); - let result3 = task3.await.unwrap(); - - assert_eq!(result1, 4); // After first push - assert_eq!(result2, 5); // After second push - assert_eq!(result3, 5); // Final state - - let final_read = rwlock.read().await; - assert_eq!(final_read.len(), 5); - assert_eq!(*final_read, vec![1, 2, 3, 4, 5]); -} - -#[tokio::test] -async fn test_downgrade_with_waiting_writers() { - let rwlock = Arc::new(RwLock::new(0i32)); - let writer_queued = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let downgrade_done = Arc::new(std::sync::atomic::AtomicBool::new(false)); - - let rwlock1 = rwlock.clone(); - let downgrade_done_clone = downgrade_done.clone(); - let downgrade_task = tokio::spawn(async move { - let mut write_guard = rwlock1.write().await; - *write_guard = 42; - - let read_guard = write_guard.downgrade(); - downgrade_done_clone.store(true, std::sync::atomic::Ordering::SeqCst); - - tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; - assert_eq!(*read_guard, 42); - - drop(read_guard); - 42 - }); - - while !downgrade_done.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - let rwlock2 = rwlock.clone(); - let writer_queued_clone = writer_queued.clone(); - let writer_task = tokio::spawn(async move { - writer_queued_clone.store(true, std::sync::atomic::Ordering::SeqCst); - - let mut write_guard = rwlock2.write().await; - - assert_eq!(*write_guard, 42); - *write_guard = 100; - - drop(write_guard); - 100 - }); - - while !writer_queued.load(std::sync::atomic::Ordering::SeqCst) { - tokio::task::yield_now().await; - } - - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; - - let mut reader_tasks = vec![]; - for i in 0..3 { - let rwlock_clone = rwlock.clone(); - - let reader_task = tokio::spawn(async move { - let read_guard = rwlock_clone.read().await; - let value = *read_guard; - drop(read_guard); - (i, value) - }); - reader_tasks.push(reader_task); - } - - let downgrade_result = downgrade_task.await.unwrap(); - assert_eq!(downgrade_result, 42); - - let writer_result = writer_task.await.unwrap(); - assert_eq!(writer_result, 100); - - for reader_task in reader_tasks { - let (reader_id, value) = reader_task.await.unwrap(); - // The readers might see either value depending on exact timing, - // so let's not make strict assertions here - println!("Reader {reader_id} saw value: {value}"); - } - - let final_read = rwlock.read().await; - assert_eq!(*final_read, 100); +async fn queued_writer_precedes_a_later_reader() { + let rwlock = RwLock::new(0); + let first_reader = rwlock.read().await; + let mut writer = Box::pin(rwlock.write()); + let mut later_reader = Box::pin(rwlock.read()); + + assert_pending!(poll_once(writer.as_mut())); + assert_pending!(poll_once(later_reader.as_mut())); + + drop(first_reader); + let mut writer_guard = assert_ready!(poll_once(writer.as_mut())); + assert_pending!(poll_once(later_reader.as_mut())); + + *writer_guard = 100; + drop(writer_guard); + let reader_guard = assert_ready!(poll_once(later_reader.as_mut())); + assert_eq!(*reader_guard, 100); } #[tokio::test] From 0a7df8a512098c0480bc2408b920eebc63b31f0c Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:07:42 +0800 Subject: [PATCH 09/32] test(rwlock): consolidate owned guard mapping coverage Ten nearly identical tests repeated Weak counts and individual map or filter_map operations. A pair of mapping chains covers ownership retention, successful and failed projections, and final unlock behavior with substantially less scaffolding. --- tests-integration/tests/rwlock_test.rs | 347 ++++--------------------- 1 file changed, 49 insertions(+), 298 deletions(-) diff --git a/tests-integration/tests/rwlock_test.rs b/tests-integration/tests/rwlock_test.rs index 81c1dea..157afac 100644 --- a/tests-integration/tests/rwlock_test.rs +++ b/tests-integration/tests/rwlock_test.rs @@ -17,7 +17,6 @@ use std::num::NonZeroUsize; use std::sync::Arc; -use std::sync::Weak; use asyncband::rwlock::*; use tests_integration::poll_once; @@ -219,270 +218,66 @@ async fn test_rwlock_zst() { } #[tokio::test] -async fn test_owned_write_guard_map_memory_leak() { - // Test OwnedRwLockWriteGuard::map memory management - let rwlock = Arc::new(RwLock::new(29u32)); - let weak_ref: Weak> = Arc::downgrade(&rwlock); +async fn owned_write_mappings_preserve_lock_ownership() { + let rwlock = Arc::new(RwLock::new(Some(vec![1, 2, 3]))); + let weak = Arc::downgrade(&rwlock); - { - let write_guard = rwlock.clone().write_owned().await; - let mut mapped_guard = OwnedRwLockWriteGuard::map(write_guard, |data| data); - *mapped_guard = 100; - assert_eq!(*mapped_guard, 100); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped (no strong refs)" - ); -} - -#[tokio::test] -async fn test_owned_write_guard_filter_map_memory_leak() { - // Test OwnedRwLockWriteGuard::filter_map memory management + let identity = OwnedRwLockWriteGuard::map(rwlock.clone().write_owned().await, |value| value); + drop(identity); - // Test success case - { - let rwlock = Arc::new(RwLock::new(Some(29u32))); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mut mapped_guard = - OwnedRwLockWriteGuard::filter_map(write_guard, |data| data.as_mut()) - .expect("Should succeed"); - *mapped_guard = 100; - assert_eq!(*mapped_guard, 100); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map success" - ); - } - - // Test failure case - { - let rwlock = Arc::new(RwLock::new(None::)); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let result = OwnedRwLockWriteGuard::filter_map(write_guard, |data| data.as_mut()); - assert!(result.is_err(), "filter_map should have failed for None"); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map failure" - ); - } -} - -#[tokio::test] -async fn test_owned_mapped_write_guard_map_memory_leak() { - // Test OwnedMappedRwLockWriteGuard::map memory management - let rwlock = Arc::new(RwLock::new("test".to_string())); - let weak_ref: Weak> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mapped_guard1 = OwnedRwLockWriteGuard::map(write_guard, |s| s); - let mut mapped_guard2 = OwnedMappedRwLockWriteGuard::map(mapped_guard1, |s| s.as_mut_str()); - mapped_guard2.make_ascii_uppercase(); - assert_eq!(&*mapped_guard2, "TEST"); - } + let failed = OwnedRwLockWriteGuard::filter_map(rwlock.clone().write_owned().await, |_| { + None::<&mut Vec> + }); + let original = match failed { + Ok(_) => panic!("mapping should fail"), + Err(original) => original, + }; + drop(original); + + let value = + OwnedRwLockWriteGuard::filter_map(rwlock.clone().write_owned().await, Option::as_mut) + .unwrap(); + let value = OwnedMappedRwLockWriteGuard::map(value, Vec::as_mut_slice); + let mut value = + OwnedMappedRwLockWriteGuard::filter_map(value, |value| value.first_mut()).unwrap(); + *value = 100; + let value = value.downgrade(); drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped (no strong refs)" - ); -} - -#[tokio::test] -async fn test_owned_mapped_write_guard_filter_map_memory_leak() { - // Test OwnedMappedRwLockWriteGuard::filter_map memory management - - // Test success case - { - let rwlock = Arc::new(RwLock::new(vec![1, 2, 3])); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mapped_guard1 = OwnedRwLockWriteGuard::map(write_guard, |v| v); - let mut mapped_guard2 = OwnedMappedRwLockWriteGuard::filter_map(mapped_guard1, |v| { - if !v.is_empty() { Some(&mut v[0]) } else { None } - }) - .expect("Should succeed"); - *mapped_guard2 = 100; - assert_eq!(*mapped_guard2, 100); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "Memory leak detected on filter_map success" - ); - } + assert!(weak.upgrade().is_some()); + assert_eq!(*value, 100); - // Test failure case - { - let rwlock = Arc::new(RwLock::new(Vec::::new())); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mapped_guard1 = OwnedRwLockWriteGuard::map(write_guard, |v| v); - let result = OwnedMappedRwLockWriteGuard::filter_map(mapped_guard1, |v| { - if !v.is_empty() { Some(&mut v[0]) } else { None } - }); - - assert!( - result.is_err(), - "filter_map should have failed for empty vector" - ); - if let Err(original_guard) = result { - assert!(original_guard.is_empty()); - } - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "Memory leak detected on filter_map failure" - ); - } + drop(value); + assert!(weak.upgrade().is_none()); } #[tokio::test] -async fn test_owned_read_guard_map_memory_leak() { - // Test OwnedRwLockReadGuard::map memory management - let rwlock = Arc::new(RwLock::new(29u32)); - let weak_ref: Weak> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard = OwnedRwLockReadGuard::map(read_guard, |data| data); - assert_eq!(*mapped_guard, 29); - } +async fn owned_read_mappings_preserve_lock_ownership() { + let rwlock = Arc::new(RwLock::new(Some(vec![1, 2, 3]))); + let weak = Arc::downgrade(&rwlock); + + let identity = OwnedRwLockReadGuard::map(rwlock.clone().read_owned().await, |value| value); + drop(identity); + + let failed = + OwnedRwLockReadGuard::filter_map(rwlock.clone().read_owned().await, |_| None::<&Vec>); + let original = match failed { + Ok(_) => panic!("mapping should fail"), + Err(original) => original, + }; + drop(original); + + let value = OwnedRwLockReadGuard::filter_map(rwlock.clone().read_owned().await, Option::as_ref) + .unwrap(); + let value = OwnedMappedRwLockReadGuard::map(value, Vec::as_slice); + let value = OwnedMappedRwLockReadGuard::filter_map(value, |value| value.first()).unwrap(); drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped (no strong ref; potential memory leaks)" - ); -} - -#[tokio::test] -async fn test_owned_read_guard_filter_map_memory_leak() { - // Test OwnedRwLockReadGuard::filter_map memory management - - // Test success case - { - let rwlock = Arc::new(RwLock::new(Some(29u32))); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); + assert!(weak.upgrade().is_some()); + assert_eq!(*value, 1); - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard = OwnedRwLockReadGuard::filter_map(read_guard, |data| data.as_ref()) - .expect("filter_map should succeed for Some(_) value"); - assert_eq!(*mapped_guard, 29); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map success" - ); - } - - // Test failure case - { - let rwlock = Arc::new(RwLock::new(None::)); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let result = OwnedRwLockReadGuard::filter_map(read_guard, |data| data.as_ref()); - assert!(result.is_err(), "filter_map should have failed for None"); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map failure" - ); - } -} - -#[tokio::test] -async fn test_owned_mapped_read_guard_map_memory_leak() { - // Test OwnedMappedRwLockReadGuard::map memory management - let rwlock = Arc::new(RwLock::new("test".to_string())); - let weak_ref: Weak> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard1 = OwnedRwLockReadGuard::map(read_guard, |s| s); - let mapped_guard2 = OwnedMappedRwLockReadGuard::map(mapped_guard1, |s| s.as_str()); - assert_eq!(&*mapped_guard2, "test"); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "Memory leak detected: Arc was not deallocated" - ); -} - -#[tokio::test] -async fn test_owned_mapped_read_guard_filter_map_memory_leak() { - // Test OwnedMappedRwLockReadGuard::filter_map memory management - - // Test success case - { - let rwlock = Arc::new(RwLock::new(Some(29u32))); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard1 = OwnedRwLockReadGuard::map(read_guard, |data| data); - let mapped_guard2 = - OwnedMappedRwLockReadGuard::filter_map(mapped_guard1, |opt| opt.as_ref()) - .expect("filter_map should succeed for Some(_) value"); - assert_eq!(*mapped_guard2, 29); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "expected Arc to be dropped after filter_map success" - ); - } - - // Test failure case - { - let rwlock = Arc::new(RwLock::new(None::)); - let weak_ref: Weak>> = Arc::downgrade(&rwlock); - - { - let read_guard = rwlock.clone().read_owned().await; - let mapped_guard1 = OwnedRwLockReadGuard::map(read_guard, |data| data); - let result = OwnedMappedRwLockReadGuard::filter_map(mapped_guard1, |opt| opt.as_ref()); - assert!(result.is_err(), "filter_map should have failed for None"); - } - - drop(rwlock); - assert!( - weak_ref.upgrade().is_none(), - "Memory leak detected on filter_map failure" - ); - } + drop(value); + assert!(weak.upgrade().is_none()); } #[tokio::test] @@ -597,47 +392,3 @@ async fn queued_writer_precedes_a_later_reader() { let reader_guard = assert_ready!(poll_once(later_reader.as_mut())); assert_eq!(*reader_guard, 100); } - -#[tokio::test] -async fn test_owned_write_guard_downgrade_no_memory_leak() { - let rwlock = Arc::new(RwLock::new(42i32)); - let weak_ref = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let _read_guard = write_guard.downgrade(); - } - - drop(rwlock); - - assert_eq!( - weak_ref.strong_count(), - 0, - "Memory leak detected in OwnedRwLockWriteGuard downgrade!" - ); -} - -#[tokio::test] -async fn test_owned_mapped_write_guard_downgrade_no_memory_leak() { - #[derive(Debug)] - struct Data { - value: i32, - } - - let rwlock = Arc::new(RwLock::new(Data { value: 42 })); - let weak_ref = Arc::downgrade(&rwlock); - - { - let write_guard = rwlock.clone().write_owned().await; - let mapped_write_guard = OwnedRwLockWriteGuard::map(write_guard, |data| &mut data.value); - let _mapped_read_guard = mapped_write_guard.downgrade(); - } - - drop(rwlock); - - assert_eq!( - weak_ref.strong_count(), - 0, - "Memory leak detected in OwnedMappedRwLockWriteGuard downgrade!" - ); -} From 4bd82bbd399031858ece05a538bc94cc3499db67 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:08:52 +0800 Subject: [PATCH 10/32] test(rwlock): focus concurrency tests on observable guarantees The pseudo memory-ordering and oversized stress cases duplicated basic locking behavior without defining deterministic contracts. Keep focused coverage for reader limits, concurrent readers and writers, writer priority, every downgrade form, mapped ownership, and zero-sized values. --- tests-integration/tests/rwlock_test.rs | 214 +++++++------------------ 1 file changed, 59 insertions(+), 155 deletions(-) diff --git a/tests-integration/tests/rwlock_test.rs b/tests-integration/tests/rwlock_test.rs index 157afac..1f95bd6 100644 --- a/tests-integration/tests/rwlock_test.rs +++ b/tests-integration/tests/rwlock_test.rs @@ -24,197 +24,107 @@ use tokio_test::assert_pending; use tokio_test::assert_ready; #[test] -fn test_try_read_write_never_blocks() { - // Test that try_read and try_write never block +fn try_methods_respect_held_guards() { let rwlock = Arc::new(RwLock::new(42)); - let r1 = rwlock.try_read().unwrap(); - let _r2 = rwlock.try_read().unwrap(); - assert_eq!(*r1, 42); - assert_eq!(*_r2, 42); + let first_reader = rwlock.try_read().unwrap(); + let second_reader = rwlock.try_read().unwrap(); + assert_eq!(*first_reader, 42); + assert_eq!(*second_reader, 42); assert!(rwlock.try_write().is_none()); - drop(r1); - drop(_r2); + drop(first_reader); + drop(second_reader); - let w = rwlock.try_write().unwrap(); - assert_eq!(*w, 42); + let writer = rwlock.try_write().unwrap(); + assert_eq!(*writer, 42); assert!(rwlock.try_read().is_none()); assert!(rwlock.clone().try_read_owned().is_none()); } #[test] -fn test_get_mut_provides_exclusive_access() { - // Test that get_mut provides exclusive access to the data +fn get_mut_and_into_inner_use_exclusive_access() { let mut rwlock = RwLock::new(100); - let data = rwlock.get_mut(); - *data = 200; + *rwlock.get_mut() = 200; assert_eq!(*rwlock.get_mut(), 200); - - let inner = rwlock.into_inner(); - assert_eq!(inner, 200); + assert_eq!(rwlock.into_inner(), 200); } #[test] -fn test_with_max_readers() { +fn max_readers_limits_concurrent_readers() { let rwlock = RwLock::with_max_readers(10, NonZeroUsize::new(2).unwrap()); - let r1 = rwlock.try_read().unwrap(); - let r2 = rwlock.try_read().unwrap(); - assert_eq!(*r1, 10); - assert_eq!(*r2, 10); - + let first = rwlock.try_read().unwrap(); + let second = rwlock.try_read().unwrap(); assert!(rwlock.try_read().is_none()); - assert!(rwlock.try_write().is_none()); - drop(r1); - - let r3 = rwlock.try_read().unwrap(); - assert_eq!(*r3, 10); - + drop(first); + let replacement = rwlock.try_read().unwrap(); assert!(rwlock.try_read().is_none()); - drop(r2); - drop(r3); - - let mut w = rwlock.try_write().unwrap(); - *w = 20; - drop(w); - - let r = rwlock.try_read().unwrap(); - assert_eq!(*r, 20); + drop(second); + drop(replacement); + assert!(rwlock.try_write().is_some()); } -#[tokio::test] -async fn test_stress_concurrent_readers_writers() { - // Test concurrent readers and writers with RwLock - let rwlock = Arc::new(RwLock::new(0i32)); - let mut reader_results = Vec::new(); - let mut writer_results = Vec::new(); - - // Spawn reader tasks - for i in 0..50 { - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let guard = rwlock_clone.read().await; - let value = *guard; +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_readers_and_writers_preserve_values() { + const READERS: usize = 50; + const WRITERS: usize = 10; + + let rwlock = Arc::new(RwLock::new(0)); + let start = Arc::new(tokio::sync::Barrier::new(READERS + WRITERS + 1)); + let mut readers = Vec::with_capacity(READERS); + let mut writers = Vec::with_capacity(WRITERS); + + for _ in 0..READERS { + let rwlock = rwlock.clone(); + let start = start.clone(); + readers.push(tokio::spawn(async move { + start.wait().await; + let value = *rwlock.read().await; tokio::task::yield_now().await; - (i, value) - }); - reader_results.push(handle); + value + })); } - // Spawn writer tasks - for i in 0..10 { - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let mut guard = rwlock_clone.write().await; - let old_value = *guard; + for _ in 0..WRITERS { + let rwlock = rwlock.clone(); + let start = start.clone(); + writers.push(tokio::spawn(async move { + start.wait().await; + let mut guard = rwlock.write().await; *guard += 1; tokio::task::yield_now().await; - (i + 100, old_value, *guard) - }); - writer_results.push(handle); + })); } - for handle in reader_results { - let (reader_id, value) = handle.await.unwrap(); - assert!( - (0..=10).contains(&value), - "Reader {reader_id} saw invalid value: {value}" - ); + start.wait().await; + for reader in readers { + assert!((0..=WRITERS as i32).contains(&reader.await.unwrap())); } - - let mut writer_values = Vec::new(); - for handle in writer_results { - let (writer_id, old_value, new_value) = handle.await.unwrap(); - assert_eq!( - new_value, - old_value + 1, - "Writer {writer_id} increment failed: {old_value} -> {new_value}" - ); - writer_values.push((old_value, new_value)); - } - - let final_guard = rwlock.read().await; - assert_eq!( - *final_guard, 10, - "Final value should be 10 after 10 increments" - ); - - writer_values.sort_by_key(|(old, _)| *old); - for (i, (old_value, new_value)) in writer_values.iter().enumerate() { - assert_eq!( - *old_value, i as i32, - "Writer operations should be sequential" - ); - assert_eq!( - *new_value, - (i + 1) as i32, - "Each increment should be atomic" - ); + for writer in writers { + writer.await.unwrap(); } + assert_eq!(*rwlock.read().await, WRITERS as i32); } #[tokio::test] -async fn test_memory_ordering_correctness() { - // Test that rwlock provides proper memory ordering guarantees - // When one task modifies data under rwlock protection, - // another task should see the modification after acquiring the lock - let rwlock = Arc::new(RwLock::new(vec![1, 2, 3])); - let rwlock_clone = rwlock.clone(); - - let handle = tokio::spawn(async move { - let mut guard = rwlock_clone.write().await; - guard.push(4); - guard[0] = 100; - // Lock is released when guard is dropped - }); - - handle.await.unwrap(); - - let guard = rwlock.read().await; - assert_eq!(*guard, vec![100, 2, 3, 4]); -} - -#[tokio::test] -async fn test_rwlock_zst() { - // Test that RwLock works correctly with Zero-Sized Types +async fn zero_sized_values_follow_locking_rules() { let rwlock = Arc::new(RwLock::new(())); - let rwlock_clone = rwlock.clone(); - let handle = tokio::spawn(async move { - let guard = rwlock_clone.read().await; - *guard; - }); - - handle.await.unwrap(); - - let guard1 = rwlock.read().await; - let guard2 = rwlock.clone().read_owned().await; - *guard1; - *guard2; - + let borrowed = rwlock.read().await; + let owned = rwlock.clone().read_owned().await; assert!(rwlock.try_write().is_none()); - drop(guard1); - drop(guard2); - - let mut write_guard = rwlock.write().await; - *write_guard = (); - drop(write_guard); - - let try_write_guard = rwlock.try_write().unwrap(); - *try_write_guard; - drop(try_write_guard); - - let guard = rwlock.try_read().unwrap(); - *guard; + drop(borrowed); + drop(owned); + assert!(rwlock.try_write().is_some()); } #[tokio::test] @@ -281,11 +191,9 @@ async fn owned_read_mappings_preserve_lock_ownership() { } #[tokio::test] -async fn test_downgrade_atomicity() { - // Test atomic downgrade behavior for all guard types +async fn every_write_guard_type_can_downgrade() { let rwlock = Arc::new(RwLock::new((42, "test".to_string()))); - // Test basic write guard downgrade { let mut write_guard = rwlock.write().await; write_guard.0 = 100; @@ -293,7 +201,6 @@ async fn test_downgrade_atomicity() { let read_guard = write_guard.downgrade(); assert_eq!(read_guard.0, 100); - // Writers blocked, readers allowed assert!(rwlock.try_write().is_none()); let concurrent_read = rwlock.try_read().unwrap(); assert_eq!(concurrent_read.0, 100); @@ -301,7 +208,6 @@ async fn test_downgrade_atomicity() { drop(read_guard); } - // Test owned write guard downgrade { let mut owned_write = rwlock.clone().write_owned().await; owned_write.1 = "updated".to_string(); @@ -313,7 +219,6 @@ async fn test_downgrade_atomicity() { drop(owned_read); } - // Test mapped write guard downgrade { let write_guard = rwlock.write().await; let mut mapped_write = RwLockWriteGuard::map(write_guard, |data| &mut data.0); @@ -326,7 +231,6 @@ async fn test_downgrade_atomicity() { drop(mapped_read); } - // Test owned mapped write guard downgrade { let owned_write = rwlock.clone().write_owned().await; let mut owned_mapped = OwnedRwLockWriteGuard::map(owned_write, |data| &mut data.1); @@ -343,7 +247,7 @@ async fn test_downgrade_atomicity() { } #[tokio::test] -async fn test_downgrade_with_max_readers() { +async fn downgraded_guard_counts_against_max_readers() { let rwlock = Arc::new(RwLock::with_max_readers(0, NonZeroUsize::new(3).unwrap())); let mut write_guard = rwlock.write().await; From 384f2df76ee30b166a4dacac61897520906ab675 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:12:24 +0800 Subject: [PATCH 11/32] fix(pool): reject zero-sized bounded pools A maximum size of zero creates a semaphore with no capacity, so every checkout can remain pending forever. Asyncband has no resize or closed-pool control that could make that state useful; reject it at Pool::new so configuration mistakes fail immediately and document the boundary. --- asyncband/src/pool/bounded.rs | 15 +++++++++++++-- tests-integration/tests/pool_behavior_test.rs | 12 ++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/asyncband/src/pool/bounded.rs b/asyncband/src/pool/bounded.rs index d3584e8..50214c8 100644 --- a/asyncband/src/pool/bounded.rs +++ b/asyncband/src/pool/bounded.rs @@ -91,7 +91,7 @@ use crate::semaphore::Semaphore; #[derive(Clone, Copy, Debug)] #[non_exhaustive] pub struct PoolConfig { - /// Maximum size of the [`Pool`]. + /// Maximum size of the [`Pool`]. Must be greater than zero. pub max_size: usize, /// Queue strategy of the [`Pool`]. @@ -104,7 +104,9 @@ pub struct PoolConfig { } impl PoolConfig { - /// Creates a new [`PoolConfig`]. + /// Creates a new [`PoolConfig`] for a pool with the given maximum size. + /// + /// [`Pool::new`] panics if `max_size` is zero. pub fn new(max_size: usize) -> Self { Self { max_size, @@ -174,7 +176,16 @@ where impl Pool { /// Creates a new [`Pool`]. + /// + /// # Panics + /// + /// Panics if `config.max_size` is zero. pub fn new(config: PoolConfig, manager: M) -> Arc { + assert!( + config.max_size > 0, + "bounded pool max_size must be greater than zero" + ); + let permits = Arc::new(Semaphore::new(config.max_size)); let slots = Mutex::new(PoolState::new()); diff --git a/tests-integration/tests/pool_behavior_test.rs b/tests-integration/tests/pool_behavior_test.rs index 864ce28..3ea95a8 100644 --- a/tests-integration/tests/pool_behavior_test.rs +++ b/tests-integration/tests/pool_behavior_test.rs @@ -56,6 +56,18 @@ impl ManageObject for CountingManager { } } +#[test] +#[should_panic(expected = "bounded pool max_size must be greater than zero")] +fn bounded_pool_rejects_zero_capacity() { + bounded::Pool::new( + bounded::PoolConfig::new(0), + CountingManager { + next: Arc::new(AtomicUsize::new(0)), + detached: Arc::new(AtomicUsize::new(0)), + }, + ); +} + #[test] fn bounded_construction_allocates_idle_storage_lazily() { let pool = bounded::Pool::new( From c6dd38c41bb8a7571bf8e59889c3aee30547381a Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:13:22 +0800 Subject: [PATCH 12/32] docs(waitgroup): explain handle-based completion WaitGroup completion is driven by handle drops rather than task completion itself, and awaiting consumes the coordinator handle. The previous prose obscured that ownership model, making it easy to await the wrong clone and wait forever; describe the lifecycle directly. --- asyncband/src/waitgroup/mod.rs | 57 +++++++++++++++------------------- 1 file changed, 25 insertions(+), 32 deletions(-) diff --git a/asyncband/src/waitgroup/mod.rs b/asyncband/src/waitgroup/mod.rs index 93567f3..1e08255 100644 --- a/asyncband/src/waitgroup/mod.rs +++ b/asyncband/src/waitgroup/mod.rs @@ -15,43 +15,38 @@ // specific language governing permissions and limitations // under the License. -//! A synchronization primitive for waiting on multiple tasks to complete. +//! Wait for a set of worker handles to be dropped. //! -//! Similar to Go's WaitGroup, this type allows a task to wait for multiple other -//! tasks to finish. Each task holds a handle to the WaitGroup, and the main task -//! can wait for all handles to be dropped before proceeding. -//! -//! A WaitGroup waits for a collection of tasks to finish. The main task calls -//! [`clone()`] to create a new worker handle for each task, and can then wait -//! for all tasks to complete by calling `.await` on the WaitGroup. +//! A [`WaitGroup`] starts with one coordinator handle. Clone that handle once for each unit of work +//! and move the clones into their workers. Dropping a worker handle marks that worker as complete. +//! Awaiting the coordinator consumes it and waits until every remaining handle has been dropped. //! //! # Examples //! //! ``` //! # #[tokio::main] //! # async fn main() { -//! use std::time::Duration; -//! //! use asyncband::waitgroup::WaitGroup; -//! let wg = WaitGroup::new(); //! -//! for i in 0..3 { -//! let wg = wg.clone(); -//! tokio::spawn(async move { -//! println!("Task {} starting", i); -//! tokio::time::sleep(Duration::from_millis(100)).await; -//! // wg is automatically decremented when dropped -//! drop(wg); -//! }); +//! async fn do_work() {} +//! +//! let group = WaitGroup::new(); +//! let mut tasks = Vec::new(); +//! +//! for _ in 0..3 { +//! let worker = group.clone(); +//! tasks.push(tokio::spawn(async move { +//! do_work().await; +//! drop(worker); // Signals completion. This would also happen at the end of the task. +//! })); //! } //! -//! // Wait for all tasks to complete -//! wg.await; -//! println!("All tasks completed"); +//! group.await; +//! for task in tasks { +//! task.await.unwrap(); +//! } //! # } //! ``` -//! -//! [`clone()`]: WaitGroup::clone use std::fmt; use std::future::Future; @@ -67,7 +62,7 @@ use crate::internal::waitset::WakerToken; #[cfg(test)] mod tests; -/// A synchronization primitive for waiting on multiple tasks to complete. +/// A group of handles whose collective completion can be awaited. /// /// See the [module level documentation](self) for more. pub struct WaitGroup { @@ -104,7 +99,7 @@ impl WaitGroup { } impl Clone for WaitGroup { - /// Creates a new worker handle for the WaitGroup. + /// Creates a new worker handle for the wait group. /// /// This increments the WaitGroup counter. The counter will be decremented /// when the new handle is dropped. @@ -133,8 +128,7 @@ impl IntoFuture for WaitGroup { type Output = (); type IntoFuture = Wait; - /// Converts the WaitGroup into a future that completes when all tasks finish. This decreases - /// the WaitGroup counter. + /// Consumes this handle and waits for all other handles to be dropped. fn into_future(self) -> Self::IntoFuture { let state = self.state.clone(); drop(self); @@ -142,11 +136,10 @@ impl IntoFuture for WaitGroup { } } -/// A future that completes when all tasks in a WaitGroup have finished. +/// A future that completes when every [`WaitGroup`] handle has been dropped. /// -/// This type is created by either: (1) calling `.await` on a `WaitGroup`, or (2) cloning -/// itself, which does not increase the WaitGroup counter, but creates a new future that -/// will complete when the WaitGroup counter reaches zero. +/// Awaiting a [`WaitGroup`] creates this future. Cloning a `Wait` creates another observer without +/// adding a worker to the group. #[must_use = "futures do nothing unless you `.await` or poll them"] pub struct Wait { token: Option, From 872633040583327fd6f3a88329384230c03b619c Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:14:04 +0800 Subject: [PATCH 13/32] test: cover documentation without default features The crate promises an empty default feature set, but the crate-level mutex example failed to compile in that configuration. All-feature tests masked the problem, so gate the example correctly and make the repository test command exercise no-default-feature documentation explicitly. --- asyncband/src/lib.rs | 3 +++ xtask/src/main.rs | 13 +++++++++++++ 2 files changed, 16 insertions(+) diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 349f3cc..6beb257 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -36,6 +36,7 @@ //! Then use the selected APIs directly: //! //! ``` +//! # #[cfg(feature = "mutex")] //! # #[tokio::main] //! # async fn main() { //! use asyncband::mutex::Mutex; @@ -47,6 +48,8 @@ //! } //! assert_eq!(*counter.lock().await, 1); //! # } +//! # #[cfg(not(feature = "mutex"))] +//! # fn main() {} //! ``` //! //! # API map diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 5e0ba84..c6413c6 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -132,6 +132,7 @@ struct CommandTest { impl CommandTest { fn run(self) { + run_command(make_default_feature_doc_test_cmd()); run_command(make_test_cmd(self.no_capture, &asyncband_features())); } } @@ -394,6 +395,18 @@ fn make_test_cmd(no_capture: bool, features: &[String]) -> StdCommand { cmd } +fn make_default_feature_doc_test_cmd() -> StdCommand { + let mut cmd = find_command("cargo"); + cmd.args([ + "test", + "--package", + PACKAGE_NAME, + "--doc", + "--no-default-features", + ]); + cmd +} + fn make_check_cmd(features: &[String]) -> StdCommand { let mut cmd = find_command("cargo"); cmd.env("RUSTFLAGS", "-Dwarnings"); From 758411d5b46af2bce72a94b2b29162d85d502570 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:14:53 +0800 Subject: [PATCH 14/32] docs(barrier): describe generations and leader results The documentation still described threads and boolean return values even though this is an async task primitive returning BarrierWaitResult. Explain reusable generations and the single leader as observable API concepts, and replace print-only examples with assertions. --- asyncband/src/barrier/mod.rs | 96 +++++++++--------------------------- 1 file changed, 22 insertions(+), 74 deletions(-) diff --git a/asyncband/src/barrier/mod.rs b/asyncband/src/barrier/mod.rs index 69cef81..9f43625 100644 --- a/asyncband/src/barrier/mod.rs +++ b/asyncband/src/barrier/mod.rs @@ -15,17 +15,11 @@ // specific language governing permissions and limitations // under the License. -//! A synchronization primitive that enables multiple tasks to wait for each other. +//! Synchronize a fixed number of tasks at a reusable rendezvous point. //! -//! The barrier ensures that no task proceeds past a certain point until all tasks have reached it. -//! This is useful for scenarios where multiple tasks need to proceed together after reaching a -//! certain point in their execution. -//! -//! A barrier enables multiple tasks to synchronize the beginning of some computation. -//! When a barrier is created, it is initialized with a count of the number of tasks -//! that will synchronize on the barrier. Each task can then call [`wait()`] on the -//! barrier to indicate it is ready to proceed. The barrier ensures that no task -//! proceeds past the barrier point until all tasks have made the call. +//! A [`Barrier`] releases one generation after the configured number of participants have awaited +//! [`Barrier::wait`]. It can then be reused for the next generation. Exactly one participant in +//! each generation receives a [`BarrierWaitResult`] marked as the leader. //! //! # Examples //! @@ -37,38 +31,21 @@ //! use asyncband::barrier::Barrier; //! //! let barrier = Arc::new(Barrier::new(3)); -//! let mut handles = Vec::new(); +//! let mut tasks = Vec::new(); //! -//! for i in 0..3 { +//! for _ in 0..3 { //! let barrier = barrier.clone(); -//! handles.push(tokio::spawn(async move { -//! println!("Task {} before barrier", i); -//! let result = barrier.wait().await; -//! println!("Task {} after barrier (leader: {})", i, result.is_leader()); -//! })); -//! } -//! -//! for handle in std::mem::take(&mut handles) { -//! handle.await.unwrap(); +//! let task = tokio::spawn(async move { barrier.wait().await.is_leader() }); +//! tasks.push(task); //! } //! -//! // The barrier can be reused (generation is increased). -//! for i in 0..3 { -//! let barrier = barrier.clone(); -//! handles.push(tokio::spawn(async move { -//! println!("Task {} before barrier", i); -//! let result = barrier.wait().await; -//! println!("Task {} after barrier (leader: {})", i, result.is_leader()); -//! })); -//! } -//! -//! for handle in handles { -//! handle.await.unwrap(); +//! let mut leaders = 0; +//! for task in tasks { +//! leaders += usize::from(task.await.unwrap()); //! } +//! assert_eq!(leaders, 1); //! # } //! ``` -//! -//! [`wait()`]: Barrier::wait use std::fmt; use std::future::Future; @@ -105,20 +82,9 @@ impl fmt::Debug for BarrierState { } } -/// A `BarrierWaitResult` is returned by [`Barrier::wait()`] when all threads -/// in the [`Barrier`] have rendezvoused. +/// The result of participating in one [`Barrier`] generation. /// -/// # Examples -/// -/// ``` -/// # #[tokio::main] -/// # async fn main() { -/// use asyncband::barrier::Barrier; -/// -/// let barrier = Barrier::new(1); -/// let barrier_wait_result = barrier.wait().await; -/// # } -/// ``` +/// Exactly one participant in each completed generation is designated as the leader. pub struct BarrierWaitResult(bool); impl fmt::Debug for BarrierWaitResult { @@ -130,10 +96,9 @@ impl fmt::Debug for BarrierWaitResult { } impl BarrierWaitResult { - /// Returns `true` if this worker is the "leader" for the call to [`Barrier::wait()`]. + /// Returns `true` if this participant is the leader for its barrier generation. /// - /// Only one worker will have `true` returned from their result, all other - /// workers will have `false` returned. + /// Exactly one participant per generation returns `true`. /// /// # Examples /// @@ -143,8 +108,7 @@ impl BarrierWaitResult { /// use asyncband::barrier::Barrier; /// /// let barrier = Barrier::new(1); - /// let barrier_wait_result = barrier.wait().await; - /// println!("{:?}", barrier_wait_result.is_leader()); + /// assert!(barrier.wait().await.is_leader()); /// # } /// ``` #[must_use] @@ -188,9 +152,8 @@ impl Barrier { /// Waits for all tasks to reach this point. /// - /// The barrier will block the current task until all `n` tasks have called `wait()`. - /// The last task to call `wait()` will be designated as the leader and receive `true` - /// as the return value. All other tasks will receive `false`. + /// The barrier holds the current task until all `n` participants have arrived. The final + /// participant is designated as the leader for this generation. /// /// # Cancel safety /// @@ -206,32 +169,17 @@ impl Barrier { /// wait futures. Callers that need to keep waiting after another operation completes first /// should retain and continue polling the same `wait` future instead of creating a new one. /// - /// # Returns - /// - /// Returns a `Future` that resolves to: - /// * `true` if this task is the last (leader) task to arrive at the barrier - /// * `false` for all other tasks + /// Returns a [`BarrierWaitResult`] that identifies whether this participant is the leader. /// /// # Examples /// /// ``` /// # #[tokio::main] /// # async fn main() { - /// use std::sync::Arc; - /// /// use asyncband::barrier::Barrier; /// - /// let barrier = Arc::new(Barrier::new(2)); - /// let barrier2 = barrier.clone(); - /// - /// let handle = tokio::spawn(async move { - /// let result = barrier2.wait().await; - /// println!("Task 1: leader = {}", result.is_leader()); - /// }); - /// - /// let result = barrier.wait().await; - /// println!("Task 2: leader = {}", result.is_leader()); - /// handle.await.unwrap(); + /// let barrier = Barrier::new(1); + /// assert!(barrier.wait().await.is_leader()); /// # } /// ``` pub async fn wait(&self) -> BarrierWaitResult { From 57cfb8e9f9352ffa41f1e8379a93bb3c3c76dcc6 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:15:50 +0800 Subject: [PATCH 15/32] test(semaphore): remove timer-based pseudo stress Launching one hundred tasks with staggered sleeps did not establish a meaningful order or concurrency invariant; it only rechecked that dropped guards return permits. Deterministic permit lifecycle tests already cover that contract, so remove the slow pseudo stress case. --- tests-integration/tests/semaphore_test.rs | 24 ----------------------- 1 file changed, 24 deletions(-) diff --git a/tests-integration/tests/semaphore_test.rs b/tests-integration/tests/semaphore_test.rs index 4adbef7..fead9a3 100644 --- a/tests-integration/tests/semaphore_test.rs +++ b/tests-integration/tests/semaphore_test.rs @@ -22,7 +22,6 @@ use std::task::Context; use std::task::Poll; use std::task::Wake; use std::task::Waker; -use std::vec::Vec; use asyncband::semaphore::Semaphore; @@ -81,29 +80,6 @@ fn forget() { assert!(sem.try_acquire(1).is_none()); } -#[tokio::test] -async fn stress_test() { - let sem = Arc::new(Semaphore::new(5)); - let mut join_handles = Vec::new(); - for i in 0..100 { - let sem_clone = sem.clone(); - join_handles.push(tokio::spawn(async move { - let _p = sem_clone.acquire(1).await; - tokio::time::sleep(std::time::Duration::from_millis(100 - i)).await; - })); - } - for j in join_handles { - j.await.unwrap(); - } - // there should be exactly 5 semaphores available now - let _p1 = sem.try_acquire(1).unwrap(); - let _p2 = sem.try_acquire(1).unwrap(); - let _p3 = sem.try_acquire(1).unwrap(); - let _p4 = sem.try_acquire(1).unwrap(); - let _p5 = sem.try_acquire(1).unwrap(); - assert!(sem.try_acquire(1).is_none()); -} - #[test] fn add_max_amount_permits() { let s = Semaphore::new(0); From 6242da2a998acb5f442e8dea1e1f611f50b409f8 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:16:22 +0800 Subject: [PATCH 16/32] test(broadcast): remove scheduler timing assumptions Sleeping before send tried to guess that a receiver had parked and duplicated the adjacent direct-waker test. Remove that race, and use a channel timeout solely as a deadlock watchdog instead of repeatedly polling an atomic flag with sleeps. --- .../tests/broadcast_mpmc_unbounded_test.rs | 30 ++++--------------- 1 file changed, 5 insertions(+), 25 deletions(-) diff --git a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs b/tests-integration/tests/broadcast_mpmc_unbounded_test.rs index 5347b34..8469ab9 100644 --- a/tests-integration/tests/broadcast_mpmc_unbounded_test.rs +++ b/tests-integration/tests/broadcast_mpmc_unbounded_test.rs @@ -24,7 +24,6 @@ use std::task::Wake; use std::task::Waker; use std::thread; use std::time::Duration; -use std::time::Instant; use asyncband::broadcast::mpmc::*; @@ -330,8 +329,7 @@ fn panicking_clone_leaves_the_channel_consistent() { #[test] fn message_destructors_run_outside_the_channel_lock() { - let finished = Arc::new(AtomicUsize::new(0)); - let flag = finished.clone(); + let (finished_tx, finished_rx) = std::sync::mpsc::channel(); let worker = thread::spawn(move || { let (tx, mut rx1) = unbounded(); @@ -357,33 +355,15 @@ fn message_destructors_run_outside_the_channel_lock() { }); drop(tx); - flag.store(1, Ordering::SeqCst); + finished_tx.send(()).unwrap(); }); - let deadline = Instant::now() + Duration::from_secs(10); - while Instant::now() < deadline && finished.load(Ordering::SeqCst) == 0 { - thread::sleep(Duration::from_millis(10)); - } - assert_eq!( - finished.load(Ordering::SeqCst), - 1, - "a message destructor deadlocked against the channel lock" - ); + finished_rx + .recv_timeout(Duration::from_secs(10)) + .expect("a message destructor deadlocked against the channel lock"); worker.join().unwrap(); } -#[tokio::test] -async fn test_wait_mechanism() { - let (tx, mut rx) = unbounded(); - - let handle = tokio::spawn(async move { rx.recv().await }); - - tokio::time::sleep(Duration::from_millis(100)).await; - tx.send(42); - - assert_eq!(handle.await.unwrap(), Ok(42)); -} - #[test] fn send_wakes_a_parked_receiver_exactly_once() { let (tx, mut rx) = unbounded(); From f9eaedf4cb4e7c82186bb0b4d38ad6f7e40b7743 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:18:17 +0800 Subject: [PATCH 17/32] refactor(pool): distinguish state invariants from unsafe code The Option unwraps rely on ordinary moved-value state invariants and do not justify unsafe operations. Calling those comments SAFETY obscures actual unsafe review obligations, so label them as invariants without changing behavior. --- asyncband/src/pool/bounded.rs | 12 ++++++------ asyncband/src/pool/unbounded.rs | 12 ++++++------ 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/asyncband/src/pool/bounded.rs b/asyncband/src/pool/bounded.rs index 50214c8..0d4dbec 100644 --- a/asyncband/src/pool/bounded.rs +++ b/asyncband/src/pool/bounded.rs @@ -482,14 +482,14 @@ impl Drop for Object { impl Deref for Object { type Target = M::Object; fn deref(&self) -> &M::Object { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. &self.state.as_ref().unwrap().o } } impl DerefMut for Object { fn deref_mut(&mut self) -> &mut Self::Target { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. &mut self.state.as_mut().unwrap().o } } @@ -511,7 +511,7 @@ impl Object { /// /// This reduces the size of the pool by one. pub fn detach(mut self) -> M::Object { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. let mut o = self.state.take().unwrap().o; if let Some(pool) = self.pool.upgrade() { pool.detach_object(&mut o); @@ -521,7 +521,7 @@ impl Object { /// Returns the status of the object. pub fn status(&self) -> ObjectStatus { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. self.state.as_ref().unwrap().status } } @@ -557,7 +557,7 @@ impl Drop for UnreadyObject { impl UnreadyObject { fn ready(mut self, permit: OwnedSemaphorePermit) -> Object { - // SAFETY: `state` is always `Some` when `UnreadyObject` is owned. + // INVARIANT: `state` is `Some` until this object becomes ready, detaches, or is dropped. let state = Some(self.state.take().unwrap()); let pool = self.pool.clone(); Object { @@ -576,7 +576,7 @@ impl UnreadyObject { } fn state(&mut self) -> &mut ObjectState { - // SAFETY: `state` is always `Some` when `UnreadyObject` is owned. + // INVARIANT: `state` is `Some` until this object becomes ready, detaches, or is dropped. self.state.as_mut().unwrap() } } diff --git a/asyncband/src/pool/unbounded.rs b/asyncband/src/pool/unbounded.rs index 8483224..24358b7 100644 --- a/asyncband/src/pool/unbounded.rs +++ b/asyncband/src/pool/unbounded.rs @@ -499,14 +499,14 @@ impl> Drop for Object { impl> Deref for Object { type Target = T; fn deref(&self) -> &T { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. &self.state.as_ref().unwrap().o } } impl> DerefMut for Object { fn deref_mut(&mut self) -> &mut Self::Target { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. &mut self.state.as_mut().unwrap().o } } @@ -528,7 +528,7 @@ impl> Object { /// /// This reduces the size of the pool by one. pub fn detach(mut self) -> M::Object { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. let mut o = self.state.take().unwrap().o; if let Some(pool) = self.pool.upgrade() { pool.detach_object(&mut o); @@ -538,7 +538,7 @@ impl> Object { /// Returns the status of the object. pub fn status(&self) -> ObjectStatus { - // SAFETY: `state` is always `Some` when `Object` is owned. + // INVARIANT: `state` is `Some` until this object is detached or dropped. self.state.as_ref().unwrap().status } } @@ -574,7 +574,7 @@ impl> Drop for UnreadyObject { impl> UnreadyObject { fn ready(mut self) -> Object { - // SAFETY: `state` is always `Some` when `UnreadyObject` is owned. + // INVARIANT: `state` is `Some` until this object becomes ready, detaches, or is dropped. let state = Some(self.state.take().unwrap()); let pool = self.pool.clone(); Object { state, pool } @@ -589,7 +589,7 @@ impl> UnreadyObject { } fn state(&mut self) -> &mut ObjectState { - // SAFETY: `state` is always `Some` when `UnreadyObject` is owned. + // INVARIANT: `state` is `Some` until this object becomes ready, detaches, or is dropped. self.state.as_mut().unwrap() } } From 16b8747f253a5987da1e01056d37cbbb57f5c59e Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:19:15 +0800 Subject: [PATCH 18/32] docs(latch): state countdown and cancellation semantics The previous documentation repeated generic coordination prose, used print-only examples, and omitted important behavior. State that the countdown saturates at zero, cannot be reused, and has cancel-safe waits, then demonstrate those contracts with assertions. --- asyncband/src/latch/mod.rs | 107 ++++++++++++------------------------- 1 file changed, 33 insertions(+), 74 deletions(-) diff --git a/asyncband/src/latch/mod.rs b/asyncband/src/latch/mod.rs index 8763c26..048c261 100644 --- a/asyncband/src/latch/mod.rs +++ b/asyncband/src/latch/mod.rs @@ -15,15 +15,12 @@ // specific language governing permissions and limitations // under the License. -//! A countdown latch that allows one or more tasks to wait until a set of operations completes. +//! Wait for a one-way countdown to reach zero. //! -//! Unlike a barrier, a latch's count can only decrease and cannot be reused once it reaches zero. -//! This makes it ideal for scenarios where you need to wait for a specific number of events or -//! operations to complete. -//! -//! A latch starts with an initial count and tasks can wait for this count to reach zero. -//! The count can be decremented by calling [`count_down()`] or [`arrive()`]. Once the count -//! reaches zero, all waiting tasks are unblocked. +//! A [`Latch`] starts with a fixed count. [`Latch::count_down`] decrements it by one and +//! [`Latch::arrive`] decrements it by an arbitrary amount. Both operations saturate at zero. Once +//! zero is reached, all current and future waits complete immediately; a latch cannot be reset or +//! reused for another countdown. //! //! # Examples //! @@ -35,25 +32,21 @@ //! use asyncband::latch::Latch; //! //! let latch = Arc::new(Latch::new(3)); -//! let mut handles = Vec::new(); +//! let mut tasks = Vec::new(); //! -//! for i in 0..3 { +//! for _ in 0..3 { //! let latch = latch.clone(); -//! handles.push(tokio::spawn(async move { -//! println!("Task {} starting", i); -//! // Simulate some work -//! latch.count_down(); // Signal completion -//! })); +//! let task = tokio::spawn(async move { latch.count_down() }); +//! tasks.push(task); //! } //! -//! // Wait for all tasks to complete //! latch.wait().await; -//! println!("All tasks completed"); +//! for task in tasks { +//! task.await.unwrap(); +//! } +//! assert_eq!(latch.count(), 0); //! # } //! ``` -//! -//! [`count_down()`]: Latch::count_down -//! [`arrive()`]: Latch::arrive use std::future::Future; use std::pin::Pin; @@ -64,7 +57,7 @@ use std::task::Poll; use crate::internal::countdown::CountdownState; use crate::internal::waitset::WakerToken; -/// A synchronization primitive that can be used to coordinate multiple tasks. +/// A one-shot countdown that can wake any number of waiting tasks. /// /// See the [module level documentation](self) for more. #[derive(Debug)] @@ -75,16 +68,12 @@ pub struct Latch { impl Latch { /// Creates a new latch initialized with the given count. /// - /// # Arguments - /// - /// * `count` - The initial count value. Tasks will wait until this count reaches zero. - /// /// # Examples /// /// ``` /// use asyncband::latch::Latch; /// - /// let latch = Latch::new(3); // Creates a latch with count of 3 + /// let latch = Latch::new(3); /// ``` pub fn new(count: u32) -> Self { Self { @@ -94,8 +83,6 @@ impl Latch { /// Returns the current count. /// - /// This method is typically used for debugging and testing purposes. - /// /// # Examples /// /// ``` @@ -118,8 +105,8 @@ impl Latch { /// use asyncband::latch::Latch; /// /// let latch = Latch::new(2); - /// latch.count_down(); // Count is now 1 - /// latch.count_down(); // Count is now 0, all waiting tasks are woken + /// latch.count_down(); + /// assert_eq!(latch.count(), 1); /// ``` pub fn count_down(&self) { if self.state.decrement(1) { @@ -129,19 +116,8 @@ impl Latch { /// Decrements the latch count by `n`, waking up all waiting tasks if the counter reaches zero. /// - /// This method provides a way to decrement the counter by more than one at a time. - /// It will not cause an overflow when decrementing the counter. - /// - /// # Arguments - /// - /// * `n` - The amount to decrement the counter by - /// - /// # Behavior - /// - /// * If `n` is zero or the counter has already reached zero, nothing happens - /// * If the current count is greater than `n`, it is decremented by `n` - /// * If the current count is greater than 0 but less than or equal to `n`, the count becomes - /// zero and all waiting tasks are woken + /// The count saturates at zero. Passing zero or calling this after the latch has completed has + /// no effect. /// /// # Examples /// @@ -149,8 +125,10 @@ impl Latch { /// use asyncband::latch::Latch; /// /// let latch = Latch::new(5); - /// latch.arrive(3); // Count is now 2 - /// latch.arrive(2); // Count is now 0, all waiting tasks are woken + /// latch.arrive(3); + /// assert_eq!(latch.count(), 2); + /// latch.arrive(10); + /// assert_eq!(latch.count(), 0); /// ``` pub fn arrive(&self, n: u32) { if n != 0 && self.state.decrement(n) { @@ -160,10 +138,7 @@ impl Latch { /// Attempts to wait for the latch count to reach zero without blocking. /// - /// # Returns - /// - /// * `Ok(())` if the count is zero - /// * `Err(count)` if the count is not zero, where `count` is the current count + /// Returns `Ok(())` if the latch is complete, or `Err(count)` with the current nonzero count. /// /// # Examples /// @@ -183,27 +158,19 @@ impl Latch { /// Returns a future that will complete when the latch count reaches zero. /// + /// This method is cancel safe. Dropping a pending wait does not change the countdown or affect + /// other waiters. + /// /// # Examples /// /// ``` /// # #[tokio::main] /// # async fn main() { - /// use std::sync::Arc; - /// /// use asyncband::latch::Latch; /// - /// let latch = Arc::new(Latch::new(1)); - /// let latch2 = latch.clone(); - /// - /// // Spawn a task that will wait for the latch - /// let handle = tokio::spawn(async move { - /// latch2.wait().await; - /// println!("Latch reached zero!"); - /// }); - /// - /// // Count down the latch + /// let latch = Latch::new(1); /// latch.count_down(); - /// handle.await.unwrap(); + /// latch.wait().await; /// # } /// ``` pub async fn wait(&self) { @@ -216,8 +183,8 @@ impl Latch { /// Returns a future that will complete when the latch count reaches zero. /// - /// The latch must be wrapped in an [`Arc`] to call this method. Thus, the returned future has - /// no lifetime constraints. + /// The latch must be wrapped in an [`Arc`] to call this method. The future owns that `Arc`, so + /// it can be moved into a spawned task. Like [`wait`](Self::wait), this method is cancel safe. /// /// # Examples /// @@ -229,17 +196,9 @@ impl Latch { /// use asyncband::latch::Latch; /// /// let latch = Arc::new(Latch::new(1)); - /// let latch2 = latch.clone(); - /// - /// // Spawn a task that will wait for the latch - /// let handle = tokio::spawn(async move { - /// latch2.wait_owned().await; - /// println!("Latch reached zero!"); - /// }); - /// - /// // Count down the latch + /// let waiter = tokio::spawn(latch.clone().wait_owned()); /// latch.count_down(); - /// handle.await.unwrap(); + /// waiter.await.unwrap(); /// # } /// ``` pub async fn wait_owned(self: Arc) { From bd6057240a786ceaf1b99523ad2716b45481a2e2 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:19:33 +0800 Subject: [PATCH 19/32] refactor(mpsc): label endpoint state invariants accurately Endpoint Options are cleared only by their destructors, and safe borrows cannot overlap those destructors. These unwraps therefore rely on a normal ownership invariant rather than an unsafe proof; naming that distinction keeps safety review focused. --- asyncband/src/mpsc/bounded.rs | 4 ++-- asyncband/src/mpsc/unbounded.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/asyncband/src/mpsc/bounded.rs b/asyncband/src/mpsc/bounded.rs index c4d4bcf..cef6316 100644 --- a/asyncband/src/mpsc/bounded.rs +++ b/asyncband/src/mpsc/bounded.rs @@ -193,7 +193,7 @@ impl BoundedSender { /// # } /// ``` pub fn try_send(&self, value: T) -> Result<(), TrySendError> { - // SAFETY: The sender is guaranteed to be non-null before dropped. + // INVARIANT: A shared borrow of the endpoint cannot overlap its destructor. let sender = self.sender.as_ref().unwrap(); match sender.try_send(value) { Ok(()) => { @@ -270,7 +270,7 @@ impl BoundedReceiver { /// # } /// ``` pub fn try_recv(&mut self) -> Result { - // SAFETY: The receiver is guaranteed to be non-null before dropped. + // INVARIANT: A mutable borrow of the endpoint cannot overlap its destructor. let receiver = self.receiver.as_ref().unwrap(); match receiver.try_recv() { Ok(v) => { diff --git a/asyncband/src/mpsc/unbounded.rs b/asyncband/src/mpsc/unbounded.rs index 57552d2..3993b68 100644 --- a/asyncband/src/mpsc/unbounded.rs +++ b/asyncband/src/mpsc/unbounded.rs @@ -113,7 +113,7 @@ impl UnboundedSender { /// If the receiver has been dropped, this function returns an error. The error includes /// the value passed to `send`. pub fn send(&self, value: T) -> Result<(), SendError> { - // SAFETY: The sender is guaranteed to be non-null before dropped. + // INVARIANT: A shared borrow of the endpoint cannot overlap its destructor. let sender = self.sender.as_ref().unwrap(); sender.send(value).map_err(|err| SendError::new(err.0))?; From 37dabd588b049712a075d7f808938feb1cd822da Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:20:11 +0800 Subject: [PATCH 20/32] test(mpsc): replace pressure loops with boundary cases Million-message loops with Instant and println are benchmarks without thresholds, while testing every capacity from 1 through 100 repeats the same control flow. Replace them with representative capacity boundaries and a focused multi-producer delivery case. --- tests-integration/tests/mpsc_test.rs | 89 ++++++---------------------- 1 file changed, 18 insertions(+), 71 deletions(-) diff --git a/tests-integration/tests/mpsc_test.rs b/tests-integration/tests/mpsc_test.rs index ed132d3..50d5a10 100644 --- a/tests-integration/tests/mpsc_test.rs +++ b/tests-integration/tests/mpsc_test.rs @@ -16,7 +16,6 @@ // under the License. use std::task::Poll; -use std::time::Instant; use asyncband::mpsc; use asyncband::mpsc::RecvError; @@ -34,31 +33,11 @@ fn expect_ready(poll: Poll) -> T { } #[test] -fn test_unbounded_pressure() { - let n = 1024 * 1024; +fn unbounded_collects_from_multiple_producers() { let (tx, mut rx) = mpsc::unbounded(); test_runtime().block_on(async move { - let start = Instant::now(); - tokio::spawn(async move { - for i in 0..n { - tx.send(i).unwrap(); - } - }); - - for i in 0..n { - assert_eq!(rx.recv().await, Ok(i)); - } - println!("Elapsed: {:?}", start.elapsed()); - }); -} - -#[test] -fn test_unbounded_sum() { - let (tx, mut rx) = mpsc::unbounded(); - - test_runtime().block_on(async move { - for i in 0..100 { + for i in 0..8 { let tx = tx.clone(); tokio::spawn(async move { tx.send(i).unwrap(); @@ -70,7 +49,7 @@ fn test_unbounded_sum() { while let Ok(i) = rx.recv().await { sum += i; } - assert_eq!(sum, 4950); + assert_eq!(sum, 28); }); } @@ -182,31 +161,19 @@ async fn async_send_recv_unbounded() { } #[test] -fn try_recv_unbounded() { - for num in 0..100 { - let (tx, mut rx) = mpsc::unbounded(); - - for i in 0..num { - tx.send(i).unwrap(); - } - - for i in 0..num { - assert_eq!(rx.try_recv(), Ok(i)); - } +fn unbounded_try_recv_preserves_order_and_reports_state() { + let (tx, mut rx) = mpsc::unbounded(); - assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); - drop(tx); - assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); + for i in 0..4 { + tx.send(i).unwrap(); } -} -#[test] -fn try_recv_reports_disconnection_while_empty_unbounded() { - let (tx, mut rx) = mpsc::unbounded::<()>(); - - assert_eq!(Err(TryRecvError::Empty), rx.try_recv()); + for i in 0..4 { + assert_eq!(rx.try_recv(), Ok(i)); + } + assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); drop(tx); - assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv()); + assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); } #[tokio::test] @@ -236,17 +203,17 @@ async fn async_send_recv_bounded() { } #[test] -fn try_send_recv_bounded() { - for num in 1..101 { - let (tx, mut rx) = mpsc::bounded(num); +fn bounded_try_send_respects_capacity_and_order() { + for capacity in [1, 4, 16] { + let (tx, mut rx) = mpsc::bounded(capacity); - for i in 0..num { + for i in 0..capacity { tx.try_send(i).unwrap(); } - assert_eq!(tx.try_send(num), Err(TrySendError::Full(num))); + assert_eq!(tx.try_send(capacity), Err(TrySendError::Full(capacity))); - for i in 0..num { + for i in 0..capacity { assert_eq!(rx.try_recv(), Ok(i)); } @@ -340,23 +307,3 @@ fn bounded_receiver_drop_returns_values_to_all_blocked_senders() { assert_eq!(first_error.into_inner(), 1); assert_eq!(second_error.into_inner(), 2); } - -#[test] -fn test_bounded_pressure() { - let n = 1024 * 1024; - let (tx, mut rx) = mpsc::bounded(1024); - - test_runtime().block_on(async move { - let start = Instant::now(); - tokio::spawn(async move { - for i in 0..n { - tx.send(i).await.unwrap(); - } - }); - - for i in 0..n { - assert_eq!(rx.recv().await, Ok(i)); - } - println!("Elapsed: {:?}", start.elapsed()); - }); -} From 0431604ede07c7e04e975bc8910959ee9ad1dbf3 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:20:35 +0800 Subject: [PATCH 21/32] docs(mpsc): document capacity and waiting behavior Callers need to know that bounded channels apply backpressure, unbounded channels can grow with a slow receiver, and a zero bounded capacity is rejected. Correct the send and recv wording as well so asynchronous waiting is not described as thread sleeping. --- asyncband/src/mpsc/bounded.rs | 8 ++++++-- asyncband/src/mpsc/mod.rs | 6 +++++- asyncband/src/mpsc/unbounded.rs | 4 ++-- 3 files changed, 13 insertions(+), 5 deletions(-) diff --git a/asyncband/src/mpsc/bounded.rs b/asyncband/src/mpsc/bounded.rs index cef6316..4f888f3 100644 --- a/asyncband/src/mpsc/bounded.rs +++ b/asyncband/src/mpsc/bounded.rs @@ -42,6 +42,10 @@ use crate::internal::semaphore::Semaphore; /// A `send` on this channel will wait if the buffer of the channel is full until a /// `recv` is called on the receiver, which will consume the message and /// free up space in the buffer. +/// +/// # Panics +/// +/// Panics if `buffer` is zero. #[track_caller] pub fn bounded(buffer: usize) -> (BoundedSender, BoundedReceiver) { assert!(buffer > 0, "mpsc bounded channel requires buffer > 0"); @@ -110,7 +114,7 @@ impl Drop for BoundedSender { } impl BoundedSender { - /// Attempts to send a message to the associated receiver. + /// Sends a message to the associated receiver. /// /// This method will wait if the buffer of the channel is full until a `recv` is called on the /// receiver, which will consume the message and free up space in the buffer. @@ -288,7 +292,7 @@ impl BoundedReceiver { /// no buffered messages remain. At that point, this `Receiver` can never receive another /// value. /// - /// If the buffer is empty while a sender remains, this method sleeps until a message is sent or + /// If the buffer is empty while a sender remains, this method waits until a message is sent or /// the final sender is dropped. /// /// # Cancel safety diff --git a/asyncband/src/mpsc/mod.rs b/asyncband/src/mpsc/mod.rs index 45bc740..040c98b 100644 --- a/asyncband/src/mpsc/mod.rs +++ b/asyncband/src/mpsc/mod.rs @@ -15,7 +15,11 @@ // specific language governing permissions and limitations // under the License. -//! A multi-producer, single-consumer queue for sending values between asynchronous tasks. +//! Multi-producer, single-consumer channels for asynchronous tasks. +//! +//! [`bounded`] applies backpressure once its fixed-capacity buffer fills. [`unbounded`] never waits +//! to send while the receiver is alive, but a slow receiver can cause memory use to grow without a +//! configured limit. mod bounded; mod error; diff --git a/asyncband/src/mpsc/unbounded.rs b/asyncband/src/mpsc/unbounded.rs index 3993b68..ed6eb56 100644 --- a/asyncband/src/mpsc/unbounded.rs +++ b/asyncband/src/mpsc/unbounded.rs @@ -104,7 +104,7 @@ impl Drop for UnboundedSender { } impl UnboundedSender { - /// Attempts to send a message without blocking. + /// Sends a message without blocking. /// /// This method is not marked async because sending a message to an unbounded channel /// never requires any form of waiting. Because of this, the `send` method can be @@ -189,7 +189,7 @@ impl UnboundedReceiver { /// no buffered messages remain. At that point, this `Receiver` can never receive another /// value. /// - /// If the buffer is empty while a sender remains, this method sleeps until a message is sent or + /// If the buffer is empty while a sender remains, this method waits until a message is sent or /// the final sender is dropped. /// /// # Cancel safety From db9f067027bd28a13851eb3adb055d248c9681dd Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:21:22 +0800 Subject: [PATCH 22/32] docs(semaphore): remove blocking I/O from async example The file example held an async permit while calling std::fs, teaching users to block an executor thread and distracting from the semaphore contract. Replace it with a direct permit lifecycle example that is runtime-agnostic and safe to copy. --- asyncband/src/semaphore/mod.rs | 53 +++++----------------------------- 1 file changed, 7 insertions(+), 46 deletions(-) diff --git a/asyncband/src/semaphore/mod.rs b/asyncband/src/semaphore/mod.rs index c812f51..3f67bcf 100644 --- a/asyncband/src/semaphore/mod.rs +++ b/asyncband/src/semaphore/mod.rs @@ -15,21 +15,14 @@ // specific language governing permissions and limitations // under the License. -//! An async counting semaphore for controlling access to a set of resources. +//! Limit concurrent access with a set of permits. //! -//! A semaphore maintains a set of permits. Permits are used to synchronize access -//! to a pool of resources. Each [`acquire`] call blocks until a permit is available, -//! and then takes one permit. Each [`release`] call adds a new permit, potentially -//! releasing a blocked acquirer. -//! -//! Semaphores are often used to restrict the number of tasks that can access some -//! (physical or logical) resource. For example, here is a class that uses a -//! semaphore to control access to a pool of connections: +//! [`Semaphore::acquire`] waits for the requested number of permits and returns a guard that puts +//! them back when dropped. [`Semaphore::try_acquire`] performs the same operation without waiting, +//! while [`Semaphore::release`] adds permits that were not represented by a guard. //! //! # Examples //! -//! ## Basic usage -//! //! ``` //! # #[tokio::main] //! # async fn main() { @@ -43,43 +36,11 @@ //! //! let permit_attempt = semaphore.try_acquire(1); //! assert!(permit_attempt.is_none()); -//! # } -//! ``` -//! -//! ## Limit the number of simultaneously opened files in your program -//! -//! Most operating systems have limits on the number of open file -//! handles. Even in systems without explicit limits, resource constraints -//! implicitly set an upper bound on the number of open files. If your -//! program attempts to open a large number of files and exceeds this -//! limit, it will result in an error. -//! -//! This example uses a Semaphore with 100 permits. By acquiring a permit from -//! the Semaphore before accessing a file, you ensure that your program opens -//! no more than 100 files at a time. When trying to open the 101st -//! file, the program will wait until a permit becomes available before -//! proceeding to open another file. //! +//! drop(a_permit); +//! assert_eq!(semaphore.available_permits(), 1); +//! # } //! ``` -//! use std::fs::File; -//! use std::io::Result; -//! use std::io::Write; -//! use std::sync::LazyLock; -//! -//! use asyncband::semaphore::Semaphore; -//! -//! static PERMITS: LazyLock = LazyLock::new(|| Semaphore::new(100)); -//! -//! async fn write_to_file(message: &[u8]) -> Result<()> { -//! let _permit = PERMITS.acquire(1).await; -//! let mut buffer = File::create("example.txt")?; -//! buffer.write_all(message)?; -//! Ok(()) // Permit goes out of scope here, and is available again for acquisition -//! } -//! ``` -//! -//! [`acquire`]: Semaphore::acquire -//! [`release`]: Semaphore::release use std::sync::Arc; From f1f19f8a870e7637c451460cff3df4aaad65d4e6 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:21:52 +0800 Subject: [PATCH 23/32] docs(semaphore): correct owned permit examples The OwnedSemaphorePermit documentation was copied from the borrowed permit and still constructed SemaphorePermit values, so its doctests compiled without exercising the documented API. Use Arc-based owned acquisition and name the correct permit type. --- asyncband/src/semaphore/mod.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/asyncband/src/semaphore/mod.rs b/asyncband/src/semaphore/mod.rs index 3f67bcf..764561b 100644 --- a/asyncband/src/semaphore/mod.rs +++ b/asyncband/src/semaphore/mod.rs @@ -497,7 +497,7 @@ impl OwnedSemaphorePermit { self.permits = 0; } - /// Merge two [`SemaphorePermit`] instances together, consuming `other` + /// Merge two [`OwnedSemaphorePermit`] instances together, consuming `other` /// without releasing the permits it holds. /// /// Permits held by both `self` and `other` are released when `self` drops. @@ -515,10 +515,10 @@ impl OwnedSemaphorePermit { /// use asyncband::semaphore::Semaphore; /// /// let sem = Arc::new(Semaphore::new(10)); - /// let mut permit = sem.try_acquire(1).unwrap(); + /// let mut permit = sem.clone().try_acquire_owned(1).unwrap(); /// /// for _ in 0..9 { - /// let new_permit = sem.try_acquire(1).unwrap(); + /// let new_permit = sem.clone().try_acquire_owned(1).unwrap(); /// // Merge individual permits into a single one. /// permit.merge(new_permit) /// } @@ -582,10 +582,12 @@ impl OwnedSemaphorePermit { /// # Examples /// /// ``` + /// use std::sync::Arc; + /// /// use asyncband::semaphore::Semaphore; /// - /// let sem = Semaphore::new(5); - /// let permit = sem.try_acquire(3).unwrap(); + /// let sem = Arc::new(Semaphore::new(5)); + /// let permit = sem.try_acquire_owned(3).unwrap(); /// assert_eq!(permit.permits(), 3); /// ``` pub fn permits(&self) -> usize { From d27ff619c5759b68b665d294325f9040629453f9 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:22:36 +0800 Subject: [PATCH 24/32] docs(once-cell): assert initialization behavior directly Printed output is not a checked contract, and the word nominally left the initialization guarantee ambiguous. Assert that competing initializers publish one shared value and consistently describe asynchronous callers as tasks. --- asyncband/src/once/once_cell/mod.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/asyncband/src/once/once_cell/mod.rs b/asyncband/src/once/once_cell/mod.rs index ddb2323..7c4e47f 100644 --- a/asyncband/src/once/once_cell/mod.rs +++ b/asyncband/src/once/once_cell/mod.rs @@ -22,7 +22,7 @@ use crate::internal::value_cell::ValueCell; use crate::semaphore::Semaphore; use crate::semaphore::SemaphorePermit; -/// A thread-safe cell which can nominally be written to only once. +/// A thread-safe cell whose value is asynchronously initialized at most once. /// /// Callers provide an initializer when accessing an empty cell. An initializer that returns an /// error, panics, or is cancelled leaves the cell empty so a later caller can retry. Use @@ -44,12 +44,10 @@ use crate::semaphore::SemaphorePermit; /// let handle2 = tokio::spawn(async { CELL.get_or_init(move || async { 2 }).await }); /// let result1 = handle1.await.unwrap(); /// let result2 = handle2.await.unwrap(); -/// println!("Results: {}, {}", result1, result2); +/// assert_eq!(result1, result2); +/// assert!(*result1 == 1 || *result1 == 2); /// # } /// ``` -/// -/// The outputs must be either `Results: 1, 1` or `Results: 2, 2`, i.e. once the value is set via -/// an asynchronous function, the value inside the `OnceCell` will be immutable. pub struct OnceCell { value: ValueCell, semaphore: Semaphore, @@ -253,7 +251,7 @@ impl OnceCell { /// Initializes the contents of the cell to `value` if the cell was uninitialized, /// then returns a reference to it. /// - /// May wait if another thread is currently attempting to initialize the cell. The cell is + /// May wait if another task is currently attempting to initialize the cell. The cell is /// guaranteed to contain a value when `try_insert` returns, though not necessarily the /// one provided. /// From 8a9bdad3fbcff5d7090e02deabff599dc6eb0c18 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:23:03 +0800 Subject: [PATCH 25/32] docs(condvar): demonstrate predicate-based task waiting The example called a Tokio task a thread and hand-wrote the predicate loop despite exposing wait_while for that purpose. Show the durable predicate pattern directly and join the notifier so the example does not leave detached work. --- asyncband/src/condvar/mod.rs | 35 +++++++++++++++-------------------- 1 file changed, 15 insertions(+), 20 deletions(-) diff --git a/asyncband/src/condvar/mod.rs b/asyncband/src/condvar/mod.rs index f1a7205..ef3fb2e 100644 --- a/asyncband/src/condvar/mod.rs +++ b/asyncband/src/condvar/mod.rs @@ -40,23 +40,20 @@ //! use asyncband::mutex::Mutex; //! //! let pair = Arc::new((Mutex::new(false), Condvar::new())); -//! let pair_clone = pair.clone(); +//! let notifier_pair = pair.clone(); //! -//! // Inside our lock, spawn a new thread, and then wait for it to start. -//! tokio::spawn(async move { -//! let (lock, cvar) = &*pair_clone; -//! let mut started = lock.lock().await; -//! *started = true; -//! // We notify the condvar that the value has changed. +//! let notifier = tokio::spawn(async move { +//! let (lock, cvar) = &*notifier_pair; +//! let mut ready = lock.lock().await; +//! *ready = true; //! cvar.notify_one(); //! }); //! -//! // Wait for the thread to start up. //! let (lock, cvar) = &*pair; -//! let mut started = lock.lock().await; -//! while !*started { -//! started = cvar.wait(started).await; -//! } +//! let ready = cvar.wait_while(lock.lock().await, |ready| !*ready).await; +//! assert!(*ready); +//! drop(ready); +//! notifier.await.unwrap(); //! # } //! ``` @@ -247,21 +244,20 @@ impl Condvar { /// let pair = Arc::new((Mutex::new(false), Condvar::new())); /// let pair_clone = pair.clone(); /// - /// tokio::spawn(async move { + /// let notifier = tokio::spawn(async move { /// let (lock, cvar) = &*pair_clone; /// let mut started = lock.lock().await; /// *started = true; - /// // We notify the condvar that the value has changed. /// cvar.notify_one(); /// }); /// - /// // Wait for the thread to start up. /// let (lock, cvar) = &*pair; - /// // As long as the value inside the `Mutex` is `false`, we wait. /// let guard = cvar /// .wait_while(lock.lock().await, |started| !*started) /// .await; /// assert!(*guard); + /// drop(guard); + /// notifier.await.unwrap(); /// # } /// ``` pub async fn wait_while<'a, T, F>( @@ -293,21 +289,20 @@ impl Condvar { /// let pair = (Arc::new(Mutex::new(false)), Arc::new(Condvar::new())); /// let pair_clone = pair.clone(); /// - /// tokio::spawn(async move { + /// let notifier = tokio::spawn(async move { /// let (lock, cvar) = pair_clone; /// let mut started = lock.lock_owned().await; /// *started = true; - /// // We notify the condvar that the value has changed. /// cvar.notify_one(); /// }); /// - /// // Wait for the thread to start up. /// let (lock, cvar) = pair; - /// // As long as the value inside the `Mutex` is `false`, we wait. /// let guard = cvar /// .wait_while_owned(lock.lock_owned().await, |started| !*started) /// .await; /// assert!(*guard); + /// drop(guard); + /// notifier.await.unwrap(); /// # } /// ``` pub async fn wait_while_owned( From b05146326dd9560d637fb8a6e39a27928fae1bf0 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 22:23:24 +0800 Subject: [PATCH 26/32] docs(shutdown): focus on observable guard semantics Public documentation exposed the internal latch and wait-group composition and printed from unjoined tasks. Focus on request and guard ownership semantics, join the example tasks, and clarify that an owned watch future does not itself keep completion pending while its source guard still does. --- asyncband/src/shutdown/mod.rs | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs index 424bd52..713188e 100644 --- a/asyncband/src/shutdown/mod.rs +++ b/asyncband/src/shutdown/mod.rs @@ -24,10 +24,9 @@ //! shutdown request. //! * [`ShutdownWatch`] can observe the shutdown request without delaying completion. //! -//! Internally, the shutdown signal is implemented using a countdown latch, and the task completion -//! is tracked using a wait group. [`Shutdown`] is cloneable, allowing multiple control handles to -//! request shutdown or wait for completion. [`ShutdownGuard`] is also cloneable; each clone keeps -//! completion pending independently until it is dropped. +//! [`Shutdown`] is cloneable, allowing multiple control handles to request shutdown or wait for +//! completion. [`ShutdownGuard`] is also cloneable; each clone keeps completion pending +//! independently until it is dropped. //! //! Awaiting [`Shutdown`] requests shutdown and then waits until all [`ShutdownGuard`] handles have //! been dropped. The request is made when the future is first polled, not when the value is created @@ -40,18 +39,24 @@ //! # #[tokio::main] //! # async fn main() { //! let (shutdown, guard) = asyncband::shutdown::new(); +//! let mut tasks = Vec::new(); //! -//! for i in 0..3 { +//! for _ in 0..3 { //! let guard = guard.clone(); -//! tokio::spawn(async move { -//! println!("Task {} starting", i); +//! let task = tokio::spawn(async move { //! guard.shutdown_requested().await; -//! println!("Task {} done", i); +//! 1 //! }); +//! tasks.push(task); //! } //! drop(guard); //! //! shutdown.await; +//! let mut completed = 0; +//! for task in tasks { +//! completed += task.await.unwrap(); +//! } +//! assert_eq!(completed, 3); //! # } //! ``` @@ -182,8 +187,8 @@ impl ShutdownGuard { /// Returns an owned future that resolves when shutdown is requested. /// - /// The returned future has no lifetime constraints and does not keep shutdown completion - /// pending. + /// The returned future has no lifetime constraints and does not itself keep shutdown + /// completion pending. This guard continues to do so until it is dropped. pub fn shutdown_requested_owned(&self) -> impl Future + 'static { self.latch.clone().wait_owned() } From f3f0870d3a9b6eb88d5d63cff39a3ef0005716b1 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 23:03:15 +0800 Subject: [PATCH 27/32] docs(shutdown): clarify owned request future semantics The previous documentation used an ambiguous 'do so' reference, making it unclear whether the returned future or the source guard delays completion. State directly that the future only observes the request while the guard continues to participate in completion. --- asyncband/src/shutdown/mod.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs index 713188e..0a6e251 100644 --- a/asyncband/src/shutdown/mod.rs +++ b/asyncband/src/shutdown/mod.rs @@ -185,10 +185,10 @@ impl ShutdownGuard { self.latch.wait().await; } - /// Returns an owned future that resolves when shutdown is requested. + /// Returns a future that can outlive this guard and resolves when shutdown is requested. /// - /// The returned future has no lifetime constraints and does not itself keep shutdown - /// completion pending. This guard continues to do so until it is dropped. + /// The future only observes the request and does not delay shutdown completion. This guard + /// delays completion until it is dropped. pub fn shutdown_requested_owned(&self) -> impl Future + 'static { self.latch.clone().wait_owned() } From 990f66f8ae9f7491fd3ea7dff2ec1d7b6858c7e8 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 23:05:46 +0800 Subject: [PATCH 28/32] docs(shutdown): keep owned wait documentation concise This method differs from shutdown_requested only by returning a future that does not borrow the guard. Describe that distinction directly instead of repeating the broader ShutdownGuard lifecycle semantics. --- asyncband/src/shutdown/mod.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs index 0a6e251..928b280 100644 --- a/asyncband/src/shutdown/mod.rs +++ b/asyncband/src/shutdown/mod.rs @@ -185,10 +185,7 @@ impl ShutdownGuard { self.latch.wait().await; } - /// Returns a future that can outlive this guard and resolves when shutdown is requested. - /// - /// The future only observes the request and does not delay shutdown completion. This guard - /// delays completion until it is dropped. + /// Returns a future that resolves when shutdown is requested without borrowing this guard. pub fn shutdown_requested_owned(&self) -> impl Future + 'static { self.latch.clone().wait_owned() } From 7ee64ff013ec580c121973c9c358e723d00b9278 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 23:08:12 +0800 Subject: [PATCH 29/32] docs(shutdown): describe owned wait by its use Callers care that shutdown_requested_owned can be moved into a spawned task, not how it relates to the receiver borrow. Match Latch::wait_owned by documenting that practical distinction and cancel safety. --- asyncband/src/shutdown/mod.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs index 928b280..bc2705c 100644 --- a/asyncband/src/shutdown/mod.rs +++ b/asyncband/src/shutdown/mod.rs @@ -185,7 +185,10 @@ impl ShutdownGuard { self.latch.wait().await; } - /// Returns a future that resolves when shutdown is requested without borrowing this guard. + /// Returns a future that resolves when shutdown is requested. + /// + /// The future can be moved into a spawned task. Like + /// [`shutdown_requested`](Self::shutdown_requested), this method is cancel safe. pub fn shutdown_requested_owned(&self) -> impl Future + 'static { self.latch.clone().wait_owned() } From cae2e9fc25dcd2b53fdb002df88334c0317a2554 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 23:12:17 +0800 Subject: [PATCH 30/32] docs(shutdown): summarize owned wait in one sentence The practical distinction is that the returned future can be spawned. State that directly and omit lifecycle and cancellation details that obscure the method purpose. --- asyncband/src/shutdown/mod.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs index bc2705c..44ed97a 100644 --- a/asyncband/src/shutdown/mod.rs +++ b/asyncband/src/shutdown/mod.rs @@ -185,10 +185,7 @@ impl ShutdownGuard { self.latch.wait().await; } - /// Returns a future that resolves when shutdown is requested. - /// - /// The future can be moved into a spawned task. Like - /// [`shutdown_requested`](Self::shutdown_requested), this method is cancel safe. + /// Returns a future that can be spawned and resolves when shutdown is requested. pub fn shutdown_requested_owned(&self) -> impl Future + 'static { self.latch.clone().wait_owned() } From 7cc1949ce5e8afd8d0a5e1eb2e1e31f7f76f7d71 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 23:14:23 +0800 Subject: [PATCH 31/32] test: remove special-case no-feature doctest pass The extra command reran doctests only for asyncband with every feature disabled, adding a second test invocation for a configuration with no distinct documented contract. Keep cargo x test focused on the workspace feature run and remove the now-unused command builder. --- xtask/src/main.rs | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/xtask/src/main.rs b/xtask/src/main.rs index c6413c6..5e0ba84 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -132,7 +132,6 @@ struct CommandTest { impl CommandTest { fn run(self) { - run_command(make_default_feature_doc_test_cmd()); run_command(make_test_cmd(self.no_capture, &asyncband_features())); } } @@ -395,18 +394,6 @@ fn make_test_cmd(no_capture: bool, features: &[String]) -> StdCommand { cmd } -fn make_default_feature_doc_test_cmd() -> StdCommand { - let mut cmd = find_command("cargo"); - cmd.args([ - "test", - "--package", - PACKAGE_NAME, - "--doc", - "--no-default-features", - ]); - cmd -} - fn make_check_cmd(features: &[String]) -> StdCommand { let mut cmd = find_command("cargo"); cmd.env("RUSTFLAGS", "-Dwarnings"); From 99487947f8c031dc67e254d92b6f344bc40038bd Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 23:14:57 +0800 Subject: [PATCH 32/32] Apply suggestion from @tisonkun --- asyncband/src/shutdown/mod.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs index 44ed97a..faae3f5 100644 --- a/asyncband/src/shutdown/mod.rs +++ b/asyncband/src/shutdown/mod.rs @@ -185,7 +185,9 @@ impl ShutdownGuard { self.latch.wait().await; } - /// Returns a future that can be spawned and resolves when shutdown is requested. + /// Returns a future that resolves when shutdown is requested. + /// + /// The returned future can be moved into a spawned task. pub fn shutdown_requested_owned(&self) -> impl Future + 'static { self.latch.clone().wait_owned() }