Index: Cargo.toml ================================================================== --- Cargo.toml +++ Cargo.toml @@ -1,8 +1,8 @@ [package] name = "wakerizer" -version = "0.1.0" +version = "0.2.0" edition = "2021" license = "0BSD" categories = [ "concurrency", "asynchronous" ] keywords = [ "async", "future" ] repository = "https://repos.qrnch.tech/pub/wakerizer" @@ -16,12 +16,12 @@ "bacon.toml", "rustfmt.toml" ] [dependencies] -indexmap = { version = "2.3.0" } parking_lot = { version = "0.12.3" } +rustc-hash = { version = "2.1.1" } [dev-dependencies] tokio = { version = "1.44.1", features = [ "macros", "rt", "rt-multi-thread", "time" ] } Index: src/lib.rs ================================================================== --- src/lib.rs +++ src/lib.rs @@ -1,11 +1,10 @@ //! _wakerizer_ is intended to be used to keep track of multiple `Future`s //! waiting for a single (shared) resource. //! -//! I can assist in developing behaviors vaguely similar to `Condvar`'s -//! `notify_all()` and `notify_one()` (though there's a massive caveat -//! regarding the latter -- see warning below). +//! It can assist in developing behaviors vaguely similar to `Condvar`'s +//! `notify_all()`. //! //! # Usage //! A resource that may be waited on creates and stores a [`Wakers`] object. //! //! Each time a `Future` is created that will be waiting for the resource, its @@ -13,31 +12,25 @@ //! //! If the Future's `poll()` function returns `Poll::Pending`, it calls its //! `Waiter::prime()` to indicate that it is a Future that is actively waiting. //! //! Whenever the resource is ready, it can signal waiting futures using -//! [`Wakers::wake_one()`] or [`Wakers::wake_all()`]. -//! -//! # Warning -//! The `Wakers::wake_one()` function can be unsafe to use in certain -//! circumstances, and should generally be avoided. It's better to wake all -//! and let the `Future`s be polled again and the starved `Future`s return -//! pending. +//! [`Wakers::wake_all()`]. use std::{ sync::Arc, task::{Context, Waker} }; use parking_lot::Mutex; -use indexmap::IndexMap; +use rustc_hash::FxHashMap; struct Inner { wake_on_drop: bool, - wakers: IndexMap, - idgen: usize + wakers: FxHashMap, + idgen: u32 } /// A set of wakers that can be used to wake up pending futures. #[repr(transparent)] @@ -54,37 +47,17 @@ pub fn wake_on_drop(&self) { let mut inner = self.0.lock(); inner.wake_on_drop = true; } - /// Wake one waiting task. - /// - /// # Warning - /// If the chosen task is in the process of being cancelled (it may be part - /// of a `select` that just had another arm wake it up), then this wake - /// will be lost. For this reason, `wake_one()` should be avoided -- it's - /// better to call [`Wakers::wake_all()`] and let the `Future`s return - /// `Pending` as appopriate. - /// - /// # Safety - /// This function is unsafe to use in multithreaded runtimes where the - /// `Future` being woken up is in a cancelable context (such as in a - /// `select!` arm). - pub unsafe fn wake_one(&self) { - let mut g = self.0.lock(); - if let Some((_, waker)) = g.wakers.pop() { - waker.wake(); - } - } - /// Wake all waiting tasks. pub fn wake_all(&self) { self .0 .lock() .wakers - .drain(..) + .drain() .for_each(|(_, waker)| waker.wake()); } /// Allocate a new, unprimed, [`Waiter`]. /// @@ -114,11 +87,11 @@ impl Default for Wakers { fn default() -> Self { let inner = Inner { wake_on_drop: false, - wakers: IndexMap::new(), + wakers: FxHashMap::default(), idgen: 0 }; Self(Arc::new(Mutex::new(inner))) } } @@ -125,11 +98,11 @@ impl Drop for Wakers { fn drop(&mut self) { let mut inner = self.0.lock(); if inner.wake_on_drop { - inner.wakers.drain(..).for_each(|(_, waker)| waker.wake()); + inner.wakers.drain().for_each(|(_, waker)| waker.wake()); } } } @@ -141,11 +114,11 @@ /// /// This ensures that the waker is automatically removed from the collection of /// wakers when it is dropped. pub struct Waiter { sh: Arc>, - id: Option + id: Option } impl Waiter { /// Prime this waiter for waiting for a Waker. /// @@ -165,11 +138,11 @@ impl Drop for Waiter { fn drop(&mut self) { let mut g = self.sh.lock(); if let Some(id) = self.id { - g.wakers.swap_remove(&id); + g.wakers.remove(&id); } } } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: tests/basic.rs ================================================================== --- tests/basic.rs +++ tests/basic.rs @@ -1,11 +1,11 @@ use std::{ future::Future, pin::Pin, sync::{ - atomic::{AtomicBool, Ordering}, - Arc + Arc, + atomic::{AtomicBool, Ordering} }, task::{Context, Poll}, time::Duration }; @@ -18,16 +18,11 @@ state: Arc, wakers: Wakers } impl Trigger { - fn trigger_one(&self) { - self.state.store(true, Ordering::SeqCst); - unsafe { self.wakers.wake_one() }; - } - - fn trigger_all(&self) { + fn trigger(&self) { self.state.store(true, Ordering::SeqCst); self.wakers.wake_all(); } fn waiter(&self) -> TriggerWaiter { @@ -77,11 +72,11 @@ let waiter = button.waiter(); let jh3 = task::spawn(async { waiter.await; }); - button.trigger_all(); + button.trigger(); jh1.await.unwrap(); jh2.await.unwrap(); jh3.await.unwrap(); } @@ -105,72 +100,13 @@ waiter.await; }); time::sleep(Duration::from_millis(100)).await; - button.trigger_all(); - - jh1.await.unwrap(); - jh2.await.unwrap(); - jh3.await.unwrap(); -} - - -#[tokio::test] -async fn one_nowait() { - let button = Trigger::default(); - - let waiter = button.waiter(); - let jh1 = task::spawn(async { - waiter.await; - }); - - let waiter = button.waiter(); - let jh2 = task::spawn(async { - waiter.await; - }); - - let waiter = button.waiter(); - let jh3 = task::spawn(async { - waiter.await; - }); - - button.trigger_one(); - button.trigger_one(); - button.trigger_one(); - - jh1.await.unwrap(); - jh2.await.unwrap(); - jh3.await.unwrap(); -} - -#[tokio::test] -async fn one_wait() { - let button = Trigger::default(); - - let waiter = button.waiter(); - let jh1 = task::spawn(async { - waiter.await; - }); - - let waiter = button.waiter(); - let jh2 = task::spawn(async { - waiter.await; - }); - - let waiter = button.waiter(); - let jh3 = task::spawn(async { - waiter.await; - }); - - time::sleep(Duration::from_millis(100)).await; - - button.trigger_one(); - button.trigger_one(); - button.trigger_one(); + button.trigger(); jh1.await.unwrap(); jh2.await.unwrap(); jh3.await.unwrap(); } // vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 : Index: www/changelog.md ================================================================== --- www/changelog.md +++ www/changelog.md @@ -2,18 +2,32 @@ ⚠️ indicates a breaking change. ## [Unreleased] -[Details](/vdiff?from=wakerizer-0.1.0&to=trunk) +[Details](/vdiff?from=wakerizer-0.2.0&to=trunk) ### Added ### Changed ### Removed +--- + +## [0.2.0] - 2025-04-05 + +[Details](/vdiff?from=wakerizer-0.1.0&to=wakerizer-0.2.0) + +### Changed + +- Internals: Switch from `IndexMap` to `rustc-hash::FxHasMap`. + +### Removed + +- ⚠️ Removed `wake_one()`, because it had too sharp edges. + --- ## [0.1.0] - 2025-04-01 Initial release.