wakerizer

Check-in Differences
Login

Check-in Differences

Invert Patch Path Unified Diff

Many hyperlinks are disabled.
Use anonymous login to enable hyperlinks.

Difference From wakerizer-0.1.0 To wakerizer-0.2.0

2025-04-05
14:41
Release maintenance. Leaf check-in: ef7a5da54f user: jan tags: trunk, wakerizer-0.2.0
14:28
Merge. check-in: e57772d3a5 user: jan tags: trunk
01:30
Remove wake_one(). Replace IndexMap with rustc-hash::FxHashMap. Closed-Leaf check-in: 390d406790 user: jan tags: 0.2.0-wip
2025-04-01
01:15
Move from old repo. check-in: 579045bbba user: jan tags: trunk, wakerizer-0.1.0
01:02
initial empty check-in check-in: a8735ea926 user: jan tags: trunk

Changes to Cargo.toml. 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

23
24
25
26
27
28
29

[package]
name = "wakerizer"
version = "0.1.0"
edition = "2021"
license = "0BSD"
categories = [ "concurrency", "asynchronous" ]
keywords = [ "async", "future" ]
repository = "https://repos.qrnch.tech/pub/wakerizer"
description = "Helpers for resources that may have multiple concurrent wakers."
rust-version = "1.56"
exclude = [
  ".fossil-settings",
  ".efiles",
  ".fslckout",
  "www",
  "bacon.toml",
  "rustfmt.toml"
]

[dependencies]
indexmap = { version = "2.3.0" }
parking_lot = { version = "0.12.3" }


[dev-dependencies]
tokio = { version = "1.44.1", features = [
  "macros", "rt", "rt-multi-thread", "time"
] }

[package.metadata.docs.rs]



|

















<

>








1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

21
22
23
24
25
26
27
28
29

[package]
name = "wakerizer"
version = "0.2.0"
edition = "2021"
license = "0BSD"
categories = [ "concurrency", "asynchronous" ]
keywords = [ "async", "future" ]
repository = "https://repos.qrnch.tech/pub/wakerizer"
description = "Helpers for resources that may have multiple concurrent wakers."
rust-version = "1.56"
exclude = [
  ".fossil-settings",
  ".efiles",
  ".fslckout",
  "www",
  "bacon.toml",
  "rustfmt.toml"
]

[dependencies]

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"
] }

[package.metadata.docs.rs]
︙︙
Changes to src/lib.rs. 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

//! _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).
//!
//! # 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
//! `Wakers` spawns a [`Waiter`], which is stored with the `Future`.
//!
//! 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.

use std::{
  sync::Arc,
  task::{Context, Waker}
};

use parking_lot::Mutex;

use indexmap::IndexMap;

struct Inner {
  wake_on_drop: bool,
  wakers: IndexMap<usize, Waker>,
  idgen: usize
}


/// A set of wakers that can be used to wake up pending futures.
#[repr(transparent)]
#[derive(Clone)]
pub struct Wakers(Arc<Mutex<Inner>>);

impl Wakers {
  #[must_use]
  pub fn new() -> Self {
    Self::default()
  }

  /// Make `Wakers` wake all registered [`Waiter`]s when it's dropped.
  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(..)
      .for_each(|(_, waker)| waker.wake());
  }

  /// Allocate a new, unprimed, [`Waiter`].
  ///
  /// Call `Waiter::prime()` to "activate" the `Waiter`.
  ///




|
|
<











|
<
<
<
<
<
<








|



|
|




















<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<






|








1
2
3
4
5

6
7
8
9
10
11
12
13
14
15
16
17






18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51




















52
53
54
55
56
57
58
59
60
61
62
63
64
65

//! _wakerizer_ is intended to be used to keep track of multiple `Future`s
//! waiting for a single (shared) resource.
//!
//! 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
//! `Wakers` spawns a [`Waiter`], which is stored with the `Future`.
//!
//! 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_all()`].







use std::{
  sync::Arc,
  task::{Context, Waker}
};

use parking_lot::Mutex;

use rustc_hash::FxHashMap;

struct Inner {
  wake_on_drop: bool,
  wakers: FxHashMap<u32, Waker>,
  idgen: u32
}


/// A set of wakers that can be used to wake up pending futures.
#[repr(transparent)]
#[derive(Clone)]
pub struct Wakers(Arc<Mutex<Inner>>);

impl Wakers {
  #[must_use]
  pub fn new() -> Self {
    Self::default()
  }

  /// Make `Wakers` wake all registered [`Waiter`]s when it's dropped.
  pub fn wake_on_drop(&self) {
    let mut inner = self.0.lock();
    inner.wake_on_drop = true;
  }





















  /// Wake all waiting tasks.
  pub fn wake_all(&self) {
    self
      .0
      .lock()
      .wakers
      .drain()
      .for_each(|(_, waker)| waker.wake());
  }

  /// Allocate a new, unprimed, [`Waiter`].
  ///
  /// Call `Waiter::prime()` to "activate" the `Waiter`.
  ///
︙︙
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

  }
}

impl Default for Wakers {
  fn default() -> Self {
    let inner = Inner {
      wake_on_drop: false,
      wakers: IndexMap::new(),
      idgen: 0
    };
    Self(Arc::new(Mutex::new(inner)))
  }
}

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());
    }
  }
}


/// Representation of a waker in waiting state.
///
/// Instance of this should be created in `Future`'s `poll()` method when
/// returning `Poll::Pending`.  The object should be stored in the same object
/// that implements `Future`.
///
/// This ensures that the waker is automatically removed from the collection of
/// wakers when it is dropped.
pub struct Waiter {
  sh: Arc<Mutex<Inner>>,
  id: Option<usize>
}

impl Waiter {
  /// Prime this waiter for waiting for a Waker.
  ///
  /// This function is typically called just before returning `Poll::Pending`.
  pub fn prime(&mut self, ctx: &mut Context<'_>) {








|










|















|








85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

  }
}

impl Default for Wakers {
  fn default() -> Self {
    let inner = Inner {
      wake_on_drop: false,
      wakers: FxHashMap::default(),
      idgen: 0
    };
    Self(Arc::new(Mutex::new(inner)))
  }
}

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());
    }
  }
}


/// Representation of a waker in waiting state.
///
/// Instance of this should be created in `Future`'s `poll()` method when
/// returning `Poll::Pending`.  The object should be stored in the same object
/// that implements `Future`.
///
/// This ensures that the waker is automatically removed from the collection of
/// wakers when it is dropped.
pub struct Waiter {
  sh: Arc<Mutex<Inner>>,
  id: Option<u32>
}

impl Waiter {
  /// Prime this waiter for waiting for a Waker.
  ///
  /// This function is typically called just before returning `Poll::Pending`.
  pub fn prime(&mut self, ctx: &mut Context<'_>) {
︙︙
163
164
165
166
167
168
169
170
171
172
173
174
175

  }
}

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);
    }
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :








|






136
137
138
139
140
141
142
143
144
145
146
147
148

  }
}

impl Drop for Waiter {
  fn drop(&mut self) {
    let mut g = self.sh.lock();
    if let Some(id) = self.id {
      g.wakers.remove(&id);
    }
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :
Changes to tests/basic.rs. 
1
2
3
4
5
6

7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

use std::{
  future::Future,
  pin::Pin,
  sync::{
    atomic::{AtomicBool, Ordering},
    Arc

  },
  task::{Context, Poll},
  time::Duration
};

use tokio::{task, time};

use wakerizer::{Waiter, Wakers};

#[derive(Default)]
struct Trigger {
  state: Arc<AtomicBool>,
  wakers: Wakers
}

impl Trigger {
  fn trigger_one(&self) {
    self.state.store(true, Ordering::SeqCst);
    unsafe { self.wakers.wake_one() };
  }

  fn trigger_all(&self) {
    self.state.store(true, Ordering::SeqCst);
    self.wakers.wake_all();
  }

  fn waiter(&self) -> TriggerWaiter {
    TriggerWaiter {
      state: Arc::clone(&self.state),





<
|
>
















|
<
<
<
<
<








1
2
3
4

5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23





24
25
26
27
28
29
30

use std::{
  future::Future,
  pin::Pin,
  sync::{

    Arc,
    atomic::{AtomicBool, Ordering}
  },
  task::{Context, Poll},
  time::Duration
};

use tokio::{task, time};

use wakerizer::{Waiter, Wakers};

#[derive(Default)]
struct Trigger {
  state: Arc<AtomicBool>,
  wakers: Wakers
}

impl Trigger {
  fn trigger(&self) {





    self.state.store(true, Ordering::SeqCst);
    self.wakers.wake_all();
  }

  fn waiter(&self) -> TriggerWaiter {
    TriggerWaiter {
      state: Arc::clone(&self.state),
︙︙
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

  });

  let waiter = button.waiter();
  let jh3 = task::spawn(async {
    waiter.await;
  });

  button.trigger_all();

  jh1.await.unwrap();
  jh2.await.unwrap();
  jh3.await.unwrap();
}

#[tokio::test]








|








70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

  });

  let waiter = button.waiter();
  let jh3 = task::spawn(async {
    waiter.await;
  });

  button.trigger();

  jh1.await.unwrap();
  jh2.await.unwrap();
  jh3.await.unwrap();
}

#[tokio::test]
︙︙
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176

  let waiter = button.waiter();
  let jh3 = task::spawn(async {
    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();

  jh1.await.unwrap();
  jh2.await.unwrap();
  jh3.await.unwrap();
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :








|






<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<


98
99
100
101
102
103
104
105
106
107
108
109
110
111



























































112

  let waiter = button.waiter();
  let jh3 = task::spawn(async {
    waiter.await;
  });

  time::sleep(Duration::from_millis(100)).await;

  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 :
Changes to www/changelog.md. 
1
2
3
4
5
6
7
8
9
10
11
12
13
14














15
16
17
18
19
20

# Change Log

⚠️  indicates a breaking change.

## [Unreleased]

[Details](/vdiff?from=wakerizer-0.1.0&to=trunk)

### Added

### Changed

### Removed















---

## [0.1.0] - 2025-04-01

Initial release.








|







>
>
>
>
>
>
>
>
>
>
>
>
>
>







1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# Change Log

⚠️  indicates a breaking change.

## [Unreleased]

[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.