From 54c69ec7678b238037ea17b17ecddb71a8323699 Mon Sep 17 00:00:00 2001 From: tison Date: Sat, 1 Aug 2026 17:05:21 +0800 Subject: [PATCH 1/3] fix(condvar): use non-buffered notifications --- CHANGELOG.md | 2 + README.md | 2 +- mea/src/condvar/mod.rs | 271 ++++++++++++++++++++++++++++++++++----- mea/src/condvar/tests.rs | 223 +++++++++++++++++++++++++++++--- 4 files changed, 446 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 28376f8..78feebf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ 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) diff --git a/README.md b/README.md index f068480..cca8cbe 100644 --- a/README.md +++ b/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. diff --git a/mea/src/condvar/mod.rs b/mea/src/condvar/mod.rs index ff273c8..c4a7f03 100644 --- a/mea/src/condvar/mod.rs +++ b/mea/src/condvar/mod.rs @@ -14,6 +14,17 @@ //! A condition variable that allows tasks to wait for a notification. //! +//! A condition variable is normally paired with a predicate protected by a +//! [`Mutex`](crate::mutex::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. +//! +//! 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 +57,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 as InternalMutex; +use crate::internal::WaitList; use crate::mutex; use crate::mutex::MutexGuard; use crate::mutex::OwnedMutexGuard; @@ -60,7 +77,19 @@ mod tests; /// /// See the [module level documentation](self) for more. pub struct Condvar { - s: internal::Semaphore, + waiters: InternalMutex>, +} + +#[derive(Debug)] +struct WaitNode { + state: WaitState, +} + +#[derive(Debug)] +enum WaitState { + Waiting(Waker), + NotifiedOne, + NotifiedAll, } impl fmt::Debug for Condvar { @@ -87,52 +116,114 @@ impl Condvar { /// ``` pub const fn new() -> Condvar { Condvar { - s: internal::Semaphore::new(0), + waiters: InternalMutex::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(); + Self::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 + .remove_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. + fn notify_one_locked(waiters: &mut WaitList) -> Option { + let mut waker = None; + waiters.remove_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 + } + + /// 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. + /// + /// # Cancellation /// - /// 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. + /// 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 notification = Wait { + condvar: self, + guard: Some(guard), + index: None, + } + .await; + let guard = mutex.lock().await; + notification.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(&self, guard: OwnedMutexGuard) -> OwnedMutexGuard { 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 notification = Wait { + condvar: self, + guard: Some(guard), + index: None, + } + .await; + let guard = mutex.lock_owned().await; + notification.complete(); + guard } /// Yields the current task until this condition variable receives a notification and the @@ -227,3 +318,121 @@ impl Condvar { guard } } + +struct Wait<'a, G> { + condvar: &'a Condvar, + guard: Option, + index: Option, +} + +impl<'a, G> Future for Wait<'a, G> +where + G: Unpin, +{ + type Output = Notification<'a>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + 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 mut notification = None; + waiters.with_mut(index, |node| match &mut node.state { + WaitState::Waiting(waker) => { + if !waker.will_wake(cx.waker()) { + waker.clone_from(cx.waker()); + } + false + } + WaitState::NotifiedOne => { + notification = Some(Notification::one(this.condvar)); + true + } + WaitState::NotifiedAll => { + notification = Some(Notification::all()); + true + } + }); + + if let Some(notification) = notification { + this.index = None; + Poll::Ready(notification) + } else { + Poll::Pending + } + } +} + +impl 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.remove_waiter(index, |node| match &node.state { + WaitState::Waiting(_) => true, + WaitState::NotifiedOne => { + pass_notification = true; + false + } + WaitState::NotifiedAll => false, + }); + waiters.with_mut(index, |_| true); + + if pass_notification { + Condvar::notify_one_locked(&mut waiters) + } else { + None + } + }; + + if let Some(waker) = waker { + waker.wake(); + } + } +} + +struct Notification<'a> { + condvar: Option<&'a Condvar>, +} + +impl<'a> Notification<'a> { + fn one(condvar: &'a Condvar) -> Self { + Self { + condvar: Some(condvar), + } + } + + fn all() -> Self { + Self { condvar: None } + } + + fn complete(mut self) { + self.condvar = None; + } +} + +impl Drop for Notification<'_> { + fn drop(&mut self) { + if let Some(condvar) = self.condvar { + condvar.notify_one(); + } + } +} diff --git a/mea/src/condvar/tests.rs b/mea/src/condvar/tests.rs index 7e1be3e..6b3eeb6 100644 --- a/mea/src/condvar/tests.rs +++ b/mea/src/condvar/tests.rs @@ -12,8 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; -use std::time::Duration; +use std::task::Context; +use std::task::Poll; +use std::task::Waker; use tokio::task::JoinHandle; @@ -21,38 +25,217 @@ use crate::condvar::Condvar; use crate::mutex::Mutex; use crate::test_runtime; +fn poll_once(future: Pin<&mut F>) -> Poll +where + F: Future, +{ + future.poll(&mut Context::from_waker(Waker::noop())) +} + +fn expect_ready(poll: Poll) -> T { + match poll { + Poll::Ready(value) => value, + Poll::Pending => panic!("future should be ready"), + } +} + +#[test] +fn predicate_preserves_state_when_notification_precedes_wait() { + test_runtime().block_on(async { + let mutex = Mutex::new(false); + let condvar = Condvar::new(); + + { + let mut ready = mutex.lock().await; + *ready = true; + condvar.notify_one(); + } + + // The notification itself was not buffered. The predicate is the durable state, so a task + // that arrives later observes it and does not wait. + let ready = condvar + .wait_while(mutex.lock().await, |ready| !*ready) + .await; + assert!(*ready); + }); +} + +#[test] +fn notify_one_is_not_buffered() { + test_runtime().block_on(async { + let mutex = Mutex::new(()); + let condvar = Condvar::new(); + + for _ in 0..3 { + condvar.notify_one(); + } + + let mut wait = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(wait.as_mut()).is_pending()); + }); +} + +#[test] +fn notify_all_is_not_buffered() { + test_runtime().block_on(async { + let mutex = Mutex::new(()); + let condvar = Condvar::new(); + + condvar.notify_all(); + + let mut wait = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(wait.as_mut()).is_pending()); + }); +} + +#[test] +fn notify_one_wakes_one_waiter_at_a_time() { + test_runtime().block_on(async { + let mutex = Mutex::new(()); + let condvar = Condvar::new(); + + let mut first = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(first.as_mut()).is_pending()); + + let mut second = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(second.as_mut()).is_pending()); + + condvar.notify_one(); + drop(expect_ready(poll_once(first.as_mut()))); + assert!(poll_once(second.as_mut()).is_pending()); + + condvar.notify_one(); + drop(expect_ready(poll_once(second.as_mut()))); + }); +} + #[test] -fn notify_all() { +fn notify_all_wakes_current_waiters_using_a_predicate_loop() { test_runtime().block_on(async { + const WAITERS: usize = 10; + + #[derive(Default)] + struct State { + ready: bool, + waiting: usize, + } + + let pair = Arc::new((Mutex::new(State::default()), Condvar::new())); let mut tasks: Vec> = Vec::new(); - let pair = Arc::new((Mutex::new(0u32), Condvar::new())); - for _ in 0..10 { + for _ in 0..WAITERS { let pair = pair.clone(); tasks.push(tokio::spawn(async move { - let (m, c) = &*pair; - let mut count = m.lock().await; - while *count == 0 { - count = c.wait(count).await; - } - *count += 1; + let (mutex, condvar) = &*pair; + let mut state = mutex.lock().await; + state.waiting += 1; + let state = condvar.wait_while(state, |state| !state.ready).await; + assert!(state.ready); })); } - // Give some time for tasks to start up - tokio::time::sleep(Duration::from_millis(50)).await; + let (mutex, condvar) = &*pair; + loop { + let state = mutex.lock().await; + if state.waiting == WAITERS { + break; + } + drop(state); + tokio::task::yield_now().await; + } - let (m, c) = &*pair; + // Seeing every task's `waiting` update while holding this mutex also means every task has + // registered with the condition variable before releasing the same mutex. { - let mut count = m.lock().await; - *count += 1; - c.notify_all(); + let mut state = mutex.lock().await; + state.ready = true; + condvar.notify_all(); } - for t in tasks { - t.await.unwrap(); + for task in tasks { + task.await.unwrap(); } - let count = m.lock().await; - assert_eq!(11, *count); + }); +} + +#[test] +fn cancelling_notified_waiter_passes_notify_one_to_next_waiter() { + test_runtime().block_on(async { + let mutex = Mutex::new(()); + let condvar = Condvar::new(); + + let mut first = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(first.as_mut()).is_pending()); + + let mut second = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(second.as_mut()).is_pending()); + + let held = mutex.lock().await; + condvar.notify_one(); + + // The first waiter consumes the notification, then blocks while reacquiring the mutex. + assert!(poll_once(first.as_mut()).is_pending()); + drop(first); + drop(held); + + // Cancelling the selected waiter passes the notification to an existing waiter. + drop(expect_ready(poll_once(second.as_mut()))); + }); +} + +#[test] +fn cancelling_only_notified_waiter_does_not_buffer_notify_one() { + test_runtime().block_on(async { + let mutex = Mutex::new(()); + let condvar = Condvar::new(); + + let mut first = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(first.as_mut()).is_pending()); + + let held = mutex.lock().await; + condvar.notify_one(); + assert!(poll_once(first.as_mut()).is_pending()); + drop(first); + drop(held); + + let mut late = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(late.as_mut()).is_pending()); + }); +} + +#[test] +fn cancelled_notify_all_waiter_does_not_wake_late_waiter() { + test_runtime().block_on(async { + let mutex = Mutex::new(()); + let condvar = Condvar::new(); + + let mut current = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(current.as_mut()).is_pending()); + condvar.notify_all(); + + let mut late = Box::pin(condvar.wait(mutex.lock().await)); + assert!(poll_once(late.as_mut()).is_pending()); + drop(current); + + assert!(poll_once(late.as_mut()).is_pending()); + condvar.notify_one(); + drop(expect_ready(poll_once(late.as_mut()))); + }); +} + +#[test] +fn wait_owned_reacquires_the_mutex() { + test_runtime().block_on(async { + let mutex = Arc::new(Mutex::new(0)); + let condvar = Condvar::new(); + + let mut wait = Box::pin(condvar.wait_owned(mutex.clone().lock_owned().await)); + assert!(poll_once(wait.as_mut()).is_pending()); + condvar.notify_one(); + + let mut guard = expect_ready(poll_once(wait.as_mut())); + *guard = 1; + drop(guard); + assert_eq!(*mutex.lock().await, 1); }); } From 1503ef674adf48456cded6d969f114673c1308c5 Mon Sep 17 00:00:00 2001 From: tison Date: Sat, 1 Aug 2026 17:24:13 +0800 Subject: [PATCH 2/3] fixup Signed-off-by: tison --- CHANGELOG.md | 3 +-- mea/src/condvar/mod.rs | 17 +++++++++-------- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 78feebf..1421c34 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,8 +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. +* 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) diff --git a/mea/src/condvar/mod.rs b/mea/src/condvar/mod.rs index c4a7f03..cf7e5d2 100644 --- a/mea/src/condvar/mod.rs +++ b/mea/src/condvar/mod.rs @@ -14,11 +14,12 @@ //! A condition variable that allows tasks to wait for a notification. //! -//! A condition variable is normally paired with a predicate protected by a -//! [`Mutex`](crate::mutex::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. +//! 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 @@ -64,7 +65,7 @@ use std::task::Context; use std::task::Poll; use std::task::Waker; -use crate::internal::Mutex as InternalMutex; +use crate::internal::Mutex; use crate::internal::WaitList; use crate::mutex; use crate::mutex::MutexGuard; @@ -77,7 +78,7 @@ mod tests; /// /// See the [module level documentation](self) for more. pub struct Condvar { - waiters: InternalMutex>, + waiters: Mutex>, } #[derive(Debug)] @@ -116,7 +117,7 @@ impl Condvar { /// ``` pub const fn new() -> Condvar { Condvar { - waiters: InternalMutex::new(WaitList::new()), + waiters: Mutex::new(WaitList::new()), } } From 3f91806935479f8de2979bffc7a0993347907fd9 Mon Sep 17 00:00:00 2001 From: tison Date: Sat, 1 Aug 2026 18:41:01 +0800 Subject: [PATCH 3/3] fixup style and comments Signed-off-by: tison --- mea/src/condvar/mod.rs | 91 +++++++++++------------ mea/src/internal/semaphore.rs | 17 +++-- mea/src/internal/waitlist.rs | 131 +++++++++++++++++++--------------- 3 files changed, 124 insertions(+), 115 deletions(-) diff --git a/mea/src/condvar/mod.rs b/mea/src/condvar/mod.rs index cf7e5d2..81b82d5 100644 --- a/mea/src/condvar/mod.rs +++ b/mea/src/condvar/mod.rs @@ -93,6 +93,19 @@ enum WaitState { NotifiedAll, } +fn notify_one_locked(waiters: &mut WaitList) -> Option { + 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 { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Condvar").finish_non_exhaustive() @@ -131,7 +144,7 @@ impl Condvar { pub fn notify_one(&self) { let waker = { let mut waiters = self.waiters.lock(); - Self::notify_one_locked(&mut waiters) + notify_one_locked(&mut waiters) }; if let Some(waker) = waker { @@ -149,7 +162,7 @@ impl Condvar { let mut wakers = Vec::new(); while waiters - .remove_first_waiter(|node| { + .unlink_first_waiter(|node| { let WaitState::Waiting(waker) = mem::replace(&mut node.state, WaitState::NotifiedAll) else { @@ -169,19 +182,6 @@ impl Condvar { } } - fn notify_one_locked(waiters: &mut WaitList) -> Option { - let mut waker = None; - waiters.remove_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 - } - /// 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 @@ -191,7 +191,7 @@ impl Condvar { /// Unlike the standard library equivalent, this function does not check at runtime that the /// same mutex is always used with this condition variable. /// - /// # Cancellation + /// # 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 @@ -199,14 +199,16 @@ impl Condvar { /// for a future waiter. pub async fn wait<'a, T>(&self, guard: MutexGuard<'a, T>) -> MutexGuard<'a, T> { let mutex = mutex::guard_lock(&guard); - let notification = Wait { + let notify_one_baton = Wait { condvar: self, guard: Some(guard), index: None, } .await; let guard = mutex.lock().await; - notification.complete(); + if let Some(baton) = notify_one_baton { + baton.complete(); + } guard } @@ -216,14 +218,16 @@ impl Condvar { /// accepts and returns an owned guard. pub async fn wait_owned(&self, guard: OwnedMutexGuard) -> OwnedMutexGuard { let mutex = mutex::owned_guard_lock(&guard); - let notification = Wait { + let notify_one_baton = Wait { condvar: self, guard: Some(guard), index: None, } .await; let guard = mutex.lock_owned().await; - notification.complete(); + if let Some(baton) = notify_one_baton { + baton.complete(); + } guard } @@ -330,7 +334,7 @@ impl<'a, G> Future for Wait<'a, G> where G: Unpin, { - type Output = Notification<'a>; + type Output = Option>; fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let this = self.get_mut(); @@ -351,30 +355,20 @@ where } let index = this.index.expect("wait future polled after completion"); - let mut notification = None; - waiters.with_mut(index, |node| match &mut node.state { + 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()); } - false + return Poll::Pending; } - WaitState::NotifiedOne => { - notification = Some(Notification::one(this.condvar)); - true - } - WaitState::NotifiedAll => { - notification = Some(Notification::all()); - true - } - }); + WaitState::NotifiedOne => Some(NotifyOneBaton::new(this.condvar)), + WaitState::NotifiedAll => None, + }; - if let Some(notification) = notification { - this.index = None; - Poll::Ready(notification) - } else { - Poll::Pending - } + waiters.remove_unlinked_waiter(index); + this.index = None; + Poll::Ready(notify_one_baton) } } @@ -387,7 +381,7 @@ impl Drop for Wait<'_, G> { let waker = { let mut waiters = self.condvar.waiters.lock(); let mut pass_notification = false; - waiters.remove_waiter(index, |node| match &node.state { + waiters.unlink_waiter(index, |node| match &node.state { WaitState::Waiting(_) => true, WaitState::NotifiedOne => { pass_notification = true; @@ -395,10 +389,10 @@ impl Drop for Wait<'_, G> { } WaitState::NotifiedAll => false, }); - waiters.with_mut(index, |_| true); + waiters.remove_unlinked_waiter(index); if pass_notification { - Condvar::notify_one_locked(&mut waiters) + notify_one_locked(&mut waiters) } else { None } @@ -410,27 +404,24 @@ impl Drop for Wait<'_, G> { } } -struct Notification<'a> { +/// Passes a selected notification onward if the wait is cancelled while reacquiring its mutex. +struct NotifyOneBaton<'a> { condvar: Option<&'a Condvar>, } -impl<'a> Notification<'a> { - fn one(condvar: &'a Condvar) -> Self { +impl<'a> NotifyOneBaton<'a> { + fn new(condvar: &'a Condvar) -> Self { Self { condvar: Some(condvar), } } - fn all() -> Self { - Self { condvar: None } - } - fn complete(mut self) { self.condvar = None; } } -impl Drop for Notification<'_> { +impl Drop for NotifyOneBaton<'_> { fn drop(&mut self) { if let Some(condvar) = self.condvar { condvar.notify_one(); diff --git a/mea/src/internal/semaphore.rs b/mea/src/internal/semaphore.rs index 94ec1e6..eee3581 100644 --- a/mea/src/internal/semaphore.rs +++ b/mea/src/internal/semaphore.rs @@ -144,7 +144,7 @@ impl Semaphore { let mut waiters = self.waiters.lock(); let mut wakers = Vec::new(); loop { - match waiters.remove_first_waiter(|node| { + match waiters.unlink_first_waiter(|node| { node.permits = 0; true }) { @@ -174,7 +174,7 @@ impl Semaphore { while rem > 0 { let mut waiters = lock.take().unwrap_or_else(|| self.waiters.lock()); while wakers.len() < NUM_WAKER { - match waiters.remove_first_waiter(|node| { + match waiters.unlink_first_waiter(|node| { if node.permits <= rem { rem -= node.permits; node.permits = 0; @@ -225,12 +225,12 @@ impl Drop for Acquire<'_> { if let Some(index) = self.index { let mut waiters = self.semaphore.waiters.lock(); let mut acquired = 0; - waiters.remove_waiter(index, |node| { + waiters.unlink_waiter(index, |node| { acquired = self.permits - node.permits; node.permits = 0; true }); - waiters.with_mut(index, |_| true); // drop + waiters.remove_unlinked_waiter(index); if acquired > 0 { self.semaphore.insert_permits_with_lock(acquired, waiters); } @@ -254,8 +254,8 @@ impl Acquire<'_> { match index { Some(idx) => { let mut waiters = semaphore.waiters.lock(); - let mut ready = false; - waiters.with_mut(*idx, |node| { + let ready = { + let node = waiters.waiter_mut(*idx); if node.permits > 0 { let update_waker = node.waker.as_ref().is_none_or(|w| !w.will_wake(waker)); if update_waker { @@ -263,12 +263,11 @@ impl Acquire<'_> { } false } else { - ready = true; true } - }); - + }; if ready { + waiters.remove_unlinked_waiter(*idx); *index = None; *done = true; return Poll::Ready(()); diff --git a/mea/src/internal/waitlist.rs b/mea/src/internal/waitlist.rs index ce53a73..9b2cc69 100644 --- a/mea/src/internal/waitlist.rs +++ b/mea/src/internal/waitlist.rs @@ -14,14 +14,15 @@ use slab::Slab; -/// A guarded linked list. +/// A sentinel-based linked list with stable slab indices. /// -/// * `guard`'s `next` points to the first node (regular head). -/// * `guard`'s `prev` points to the last node (regular tail). +/// * `sentinel`'s `next` points to the first node (regular head). +/// * `sentinel`'s `prev` points to the last node (regular tail). +/// * Unlinked nodes remain addressable by index until they are explicitly removed. #[derive(Debug)] pub(crate) struct WaitList { - // if None, the list is uninitialized and empty - guard: Option, + // If `None`, the list is uninitialized and empty. + sentinel: Option, nodes: Slab>, } @@ -29,30 +30,30 @@ pub(crate) struct WaitList { struct Node { prev: usize, next: usize, - stat: Option, + value: Option, } impl WaitList { - /// Ensures the wait list is initialized, returning the guard index. + /// Ensures the wait list is initialized, returning the sentinel index. fn ensure_init(&mut self) -> usize { - if let Some(guard) = self.guard { - return guard; + if let Some(sentinel) = self.sentinel { + return sentinel; } let first = self.nodes.vacant_entry(); - let guard = first.key(); + let sentinel = first.key(); first.insert(Node { - prev: guard, - next: guard, - stat: None, + prev: sentinel, + next: sentinel, + value: None, }); - self.guard = Some(guard); - guard + self.sentinel = Some(sentinel); + sentinel } pub(crate) const fn new() -> Self { Self { - guard: None, + sentinel: None, nodes: Slab::new(), } } @@ -69,16 +70,16 @@ impl WaitList { ) { assert!(idx.is_none()); - let guard = self.ensure_init(); - let stat = f(); - let prev_head = self.nodes[guard].next; + let sentinel = self.ensure_init(); + let value = f(); + let prev_head = self.nodes[sentinel].next; let new_node = Node { - prev: guard, + prev: sentinel, next: prev_head, - stat, + value, }; let new_key = self.nodes.insert(new_node); - self.nodes[guard].next = new_key; + self.nodes[sentinel].next = new_key; self.nodes[prev_head].prev = new_key; *idx = Some(new_key); } @@ -95,56 +96,67 @@ impl WaitList { ) { assert!(idx.is_none()); - let guard = self.ensure_init(); - let stat = f(); - let prev_tail = self.nodes[guard].prev; + let sentinel = self.ensure_init(); + let value = f(); + let prev_tail = self.nodes[sentinel].prev; let new_node = Node { prev: prev_tail, - next: guard, - stat, + next: sentinel, + value, }; let new_key = self.nodes.insert(new_node); - self.nodes[guard].prev = new_key; + self.nodes[sentinel].prev = new_key; self.nodes[prev_tail].next = new_key; *idx = Some(new_key); } - /// Removes a previously registered waker from the wait list, if the predicate `f` returns + /// Unlinks a previously registered waiter from the wait list if the predicate returns /// `true`. - pub(crate) fn remove_waiter( + /// + /// The slab entry remains available until + /// [`remove_unlinked_waiter`](Self::remove_unlinked_waiter) is called. + /// If the waiter is already unlinked, the predicate still runs but no links are changed. + pub(crate) fn unlink_waiter( &mut self, idx: usize, - f: impl FnOnce(&mut T) -> bool, + should_unlink: impl FnOnce(&mut T) -> bool, ) -> Option<&mut T> { - // SAFETY: the wait list must be initialized before any waiter can be registered - let guard = self.guard.expect("wait list must be uninitialized"); + let sentinel = self.sentinel.expect("wait list must be initialized"); - assert_ne!(idx, guard); + assert_ne!(idx, sentinel); - fn retrieve_stat(node: &mut Node) -> &mut T { - // SAFETY: `idx` is a valid key + non-guard node always has `Some(stat)` - node.stat.as_mut().unwrap() + fn value_mut(node: &mut Node) -> &mut T { + node.value + .as_mut() + .expect("waiter node must contain a value") } - if f(retrieve_stat(&mut self.nodes[idx])) { + if should_unlink(value_mut(&mut self.nodes[idx])) { let prev = self.nodes[idx].prev; let next = self.nodes[idx].next; - self.nodes[prev].next = next; - self.nodes[next].prev = prev; - self.nodes[idx].prev = idx; - self.nodes[idx].next = idx; - Some(retrieve_stat(&mut self.nodes[idx])) + let is_unlinked = prev == idx; + assert_eq!(is_unlinked, next == idx, "waiter links must be consistent"); + if !is_unlinked { + self.nodes[prev].next = next; + self.nodes[next].prev = prev; + self.nodes[idx].prev = idx; + self.nodes[idx].next = idx; + } + Some(value_mut(&mut self.nodes[idx])) } else { None } } - /// Removes the first waiter from the wait list, if the predicate `f` returns `true`. - pub(crate) fn remove_first_waiter(&mut self, f: impl FnOnce(&mut T) -> bool) -> Option<&mut T> { - let guard = self.guard?; - let first = self.nodes[guard].next; - if first != guard { - self.remove_waiter(first, f) + /// Unlinks the first waiter from the wait list if the predicate returns `true`. + pub(crate) fn unlink_first_waiter( + &mut self, + should_unlink: impl FnOnce(&mut T) -> bool, + ) -> Option<&mut T> { + let sentinel = self.sentinel?; + let first = self.nodes[sentinel].next; + if first != sentinel { + self.unlink_waiter(first, should_unlink) } else { None } @@ -152,14 +164,21 @@ impl WaitList { /// Returns `true` if the wait list is empty. pub(crate) fn is_empty(&self) -> bool { - self.guard - .is_none_or(|guard| self.nodes[guard].next == guard) + self.sentinel + .is_none_or(|sentinel| self.nodes[sentinel].next == sentinel) } - pub(crate) fn with_mut(&mut self, idx: usize, drop: impl FnOnce(&mut T) -> bool) { - let node = &mut self.nodes[idx]; - if drop(node.stat.as_mut().unwrap()) { - self.nodes.remove(idx); - } + pub(crate) fn waiter_mut(&mut self, idx: usize) -> &mut T { + self.nodes[idx] + .value + .as_mut() + .expect("waiter node must contain a value") + } + + pub(crate) fn remove_unlinked_waiter(&mut self, idx: usize) { + let node = &self.nodes[idx]; + assert_eq!(node.prev, idx, "waiter must be unlinked before removal"); + assert_eq!(node.next, idx, "waiter must be unlinked before removal"); + self.nodes.remove(idx); } }