refactor: harden library ergonomics and test contracts

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.

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 {

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<bool>` 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<bool>` 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<T, F>(