fix(condvar): align notifications with standard semantics

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file.
 * Wake waiting tasks only after releasing internal locks. ([#125](https://github.com/fast/mea/pull/125))
 * Retry spurious atomic failures when completing `Once` initialization. ([#126](https://github.com/fast/mea/pull/126))
 * Make cloning a `WaitGroup` panic on counter overflow instead of silently losing track of a handle.
+* Align `Condvar` with standard condition-variable semantics by notifying only current waiters and passing a cancelled `notify_one` wakeup to another current waiter instead of storing a permit.
 
 ## v0.6.5 (2026-07-30)
 

README.md

@@ -73,7 +73,7 @@ This crate collects runtime-agnostic synchronization primitives from spare parts
 
 * **admission::FairShare** is written from scratch to bound global concurrency while balancing held permits across contending keys.
 * **Barrier** is inspired by `std::sync::Barrier` and `tokio::sync::Barrier`, with a different implementation based on the internal `WaitSet` primitive.
-* **Condvar** is inspired by `std::sync::Condvar` and `async_std::sync::Condvar`, with a different implementation based on the internal `Semaphore` primitive. Different from the async_std implementation, this condvar is fair.
+* **Condvar** is inspired by `std::sync::Condvar` and `async_std::sync::Condvar`, with a fair FIFO waiter queue and standard non-buffered notification semantics.
 * **Latch** is inspired by [`latches`](https://github.com/mirromutth/latches), with a different implementation based on the internal `CountdownState` primitive. No `wait` or `watch` method is provided, since it can be easily implemented by [composing delay futures](https://docs.rs/fastimer/*/fastimer/fn.timeout.html). No sync variant is provided, since it can be easily implemented with block_on of any runtime.
 * **Mutex** is derived from `tokio::sync::Mutex`. No blocking method is provided, since it can be easily implemented with block_on of any runtime.
 * **OnceCell** is derived from `tokio::sync::OnceCell`, but using our own semaphore implementation.

mea/src/condvar/mod.rs

@@ -14,6 +14,18 @@
 
 //! A condition variable that allows tasks to wait for a notification.
 //!
+//! A condition variable is normally paired with a predicate protected by a [`Mutex`]. The predicate
+//! records the state of the application; notifications only wake tasks that may need to check that
+//! state again. Notifications are not buffered, so calling [`Condvar::notify_one`] or
+//! [`Condvar::notify_all`] when no task is waiting has no effect.
+//!
+//! [`Mutex`]: mutex::Mutex
+//!
+//! Always check the predicate while holding the mutex and wait in a loop. [`Condvar::wait`]
+//! registers the task before releasing the mutex, so a notifier that updates the predicate under
+//! the same mutex cannot race with the transition into the wait state. [`Condvar::wait_while`]
+//! expresses this pattern directly.
+//!
 //! # Examples
 //!
 //! ```
@@ -46,9 +58,15 @@
 //! ```
 
 use std::fmt;
+use std::future::Future;
+use std::mem;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
 use std::task::Waker;
 
-use crate::internal;
+use crate::internal::Mutex;
+use crate::internal::WaitList;
 use crate::mutex;
 use crate::mutex::MutexGuard;
 use crate::mutex::OwnedMutexGuard;
@@ -60,7 +78,32 @@ mod tests;
 ///
 /// See the [module level documentation](self) for more.
 pub struct Condvar {
-    s: internal::Semaphore,
+    waiters: Mutex<WaitList<WaitNode>>,
+}
+
+#[derive(Debug)]
+struct WaitNode {
+    state: WaitState,
+}
+
+#[derive(Debug)]
+enum WaitState {
+    Waiting(Waker),
+    NotifiedOne,
+    NotifiedAll,
+}
+
+fn notify_one_locked(waiters: &mut WaitList<WaitNode>) -> Option<Waker> {
+    let mut waker = None;
+    waiters.unlink_first_waiter(|node| {
+        let WaitState::Waiting(waiting) = mem::replace(&mut node.state, WaitState::NotifiedOne)
+        else {
+            unreachable!("only waiting tasks remain linked")
+        };
+        waker = Some(waiting);
+        true
+    });
+    waker
 }
 
 impl fmt::Debug for Condvar {
@@ -87,52 +130,105 @@ impl Condvar {
     /// ```
     pub const fn new() -> Condvar {
         Condvar {
-            s: internal::Semaphore::new(0),
+            waiters: Mutex::new(WaitList::new()),
         }
     }
 
-    /// Wakes up one blocked task on this condvar.
+    /// Wakes up one task currently blocked on this condition variable.
+    ///
+    /// If no task is currently waiting, this call has no effect. Notifications are not buffered for
+    /// future calls to [`wait`](Self::wait) or [`wait_owned`](Self::wait_owned).
+    ///
+    /// If the selected task is cancelled before its wait completes, the notification is passed to
+    /// another task that is waiting at that point, if one exists.
     pub fn notify_one(&self) {
-        self.s.release(1);
+        let waker = {
+            let mut waiters = self.waiters.lock();
+            notify_one_locked(&mut waiters)
+        };
+
+        if let Some(waker) = waker {
+            waker.wake();
+        }
     }
 
-    /// Wakes up all blocked tasks on this condvar.
+    /// Wakes up all tasks currently blocked on this condition variable.
+    ///
+    /// If no task is currently waiting, this call has no effect. Notifications are not buffered for
+    /// future calls to [`wait`](Self::wait) or [`wait_owned`](Self::wait_owned).
     pub fn notify_all(&self) {
-        self.s.notify_all();
+        let wakers = {
+            let mut waiters = self.waiters.lock();
+            let mut wakers = Vec::new();
+
+            while waiters
+                .unlink_first_waiter(|node| {
+                    let WaitState::Waiting(waker) =
+                        mem::replace(&mut node.state, WaitState::NotifiedAll)
+                    else {
+                        unreachable!("only waiting tasks remain linked")
+                    };
+                    wakers.push(waker);
+                    true
+                })
+                .is_some()
+            {}
+
+            wakers
+        };
+
+        for waker in wakers {
+            waker.wake();
+        }
     }
 
-    /// Yields the current task until this condition variable receives a notification.
+    /// Waits for a notification, atomically releasing and then reacquiring the mutex.
+    ///
+    /// The task is registered with this condition variable before the mutex is released. When this
+    /// function returns, the mutex has been reacquired. The associated predicate must be checked
+    /// again after every return; prefer [`wait_while`](Self::wait_while) when possible.
+    ///
+    /// Unlike the standard library equivalent, this function does not check at runtime that the
+    /// same mutex is always used with this condition variable.
     ///
-    /// Unlike the std equivalent, this does not check that a single mutex is used at runtime.
-    /// However, as a best practice avoid using with multiple mutexes.
+    /// # Cancel safety
+    ///
+    /// Cancelling this wait removes the task from the wait queue. If the task was selected by
+    /// [`notify_one`](Self::notify_one) but has not yet reacquired the mutex, the notification is
+    /// passed to another task that is waiting at that point, if one exists. It is never buffered
+    /// for a future waiter.
     pub async fn wait<'a, T>(&self, guard: MutexGuard<'a, T>) -> MutexGuard<'a, T> {
         let mutex = mutex::guard_lock(&guard);
-
-        // register waiter while holding lock
-        let mut acquire = self.s.poll_acquire(1);
-        let _ = acquire.poll_once(Waker::noop());
-        drop(guard);
-
-        // await for notification, and then reacquire the lock
-        acquire.await;
-        mutex.lock().await
+        let notify_one_baton = Wait {
+            condvar: self,
+            guard: Some(guard),
+            index: None,
+        }
+        .await;
+        let guard = mutex.lock().await;
+        if let Some(baton) = notify_one_baton {
+            baton.complete();
+        }
+        guard
     }
 
-    /// Yields the current task until this condition variable receives a notification.
+    /// Waits for a notification, atomically releasing and then reacquiring the owned mutex.
     ///
-    /// Unlike the std equivalent, this does not check that a single mutex is used at runtime.
-    /// However, as a best practice avoid using with multiple mutexes.
+    /// This has the same notification and cancellation semantics as [`wait`](Self::wait), but
+    /// accepts and returns an owned guard.
     pub async fn wait_owned<T>(&self, guard: OwnedMutexGuard<T>) -> OwnedMutexGuard<T> {
         let mutex = mutex::owned_guard_lock(&guard);
-
-        // register waiter while holding lock
-        let mut acquire = self.s.poll_acquire(1);
-        let _ = acquire.poll_once(Waker::noop());
-        drop(guard);
-
-        // await for notification, and then reacquire the lock
-        acquire.await;
-        mutex.lock_owned().await
+        let notify_one_baton = Wait {
+            condvar: self,
+            guard: Some(guard),
+            index: None,
+        }
+        .await;
+        let guard = mutex.lock_owned().await;
+        if let Some(baton) = notify_one_baton {
+            baton.complete();
+        }
+        guard
     }
 
     /// Yields the current task until this condition variable receives a notification and the
@@ -227,3 +323,108 @@ impl Condvar {
         guard
     }
 }
+
+struct Wait<'a, G> {
+    condvar: &'a Condvar,
+    guard: Option<G>,
+    index: Option<usize>,
+}
+
+impl<'a, G> Future for Wait<'a, G>
+where
+    G: Unpin,
+{
+    type Output = Option<NotifyOneBaton<'a>>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+        let mut waiters = this.condvar.waiters.lock();
+
+        if let Some(guard) = this.guard.take() {
+            waiters.register_waiter_to_tail(&mut this.index, || {
+                Some(WaitNode {
+                    state: WaitState::Waiting(cx.waker().clone()),
+                })
+            });
+
+            // Registration must happen before unlocking the associated mutex. A notifier that
+            // acquires the mutex after this point will therefore observe this waiter.
+            drop(waiters);
+            drop(guard);
+            return Poll::Pending;
+        }
+
+        let index = this.index.expect("wait future polled after completion");
+        let notify_one_baton = match &mut waiters.waiter_mut(index).state {
+            WaitState::Waiting(waker) => {
+                if !waker.will_wake(cx.waker()) {
+                    waker.clone_from(cx.waker());
+                }
+                return Poll::Pending;
+            }
+            WaitState::NotifiedOne => Some(NotifyOneBaton::new(this.condvar)),
+            WaitState::NotifiedAll => None,
+        };
+
+        waiters.remove_unlinked_waiter(index);
+        this.index = None;
+        Poll::Ready(notify_one_baton)
+    }
+}
+
+impl<G> Drop for Wait<'_, G> {
+    fn drop(&mut self) {
+        let Some(index) = self.index.take() else {
+            return;
+        };
+
+        let waker = {
+            let mut waiters = self.condvar.waiters.lock();
+            let mut pass_notification = false;
+            waiters.unlink_waiter(index, |node| match &node.state {
+                WaitState::Waiting(_) => true,
+                WaitState::NotifiedOne => {
+                    pass_notification = true;
+                    false
+                }
+                WaitState::NotifiedAll => false,
+            });
+            waiters.remove_unlinked_waiter(index);
+
+            if pass_notification {
+                notify_one_locked(&mut waiters)
+            } else {
+                None
+            }
+        };
+
+        if let Some(waker) = waker {
+            waker.wake();
+        }
+    }
+}
+
+/// Passes a selected notification onward if the wait is cancelled while reacquiring its mutex.
+struct NotifyOneBaton<'a> {
+    condvar: Option<&'a Condvar>,
+}
+
+impl<'a> NotifyOneBaton<'a> {
+    fn new(condvar: &'a Condvar) -> Self {
+        Self {
+            condvar: Some(condvar),
+        }
+    }
+
+    fn complete(mut self) {
+        self.condvar = None;
+    }
+}
+
+impl Drop for NotifyOneBaton<'_> {
+    fn drop(&mut self) {
+        if let Some(condvar) = self.condvar {
+            condvar.notify_one();
+        }
+    }
+}