TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Steve Gerbino
3 : // Copyright (c) 2026 Michael Vandeberg
4 : //
5 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 : //
8 : // Official repository: https://github.com/cppalliance/capy
9 : //
10 :
11 : #ifndef BOOST_CAPY_EX_ASYNC_WAKER_HPP
12 : #define BOOST_CAPY_EX_ASYNC_WAKER_HPP
13 :
14 : #include <boost/capy/detail/config.hpp>
15 : #include <boost/capy/continuation.hpp>
16 : #include <boost/capy/error.hpp>
17 : #include <boost/capy/ex/executor_ref.hpp>
18 : #include <boost/capy/ex/io_env.hpp>
19 : #include <boost/capy/io_result.hpp>
20 :
21 : #include <atomic>
22 : #include <coroutine>
23 : #include <new>
24 : #include <stop_token>
25 : #include <thread>
26 : #include <utility>
27 :
28 : /* async_waker implementation notes
29 : ===================================
30 :
31 : wake() must be callable from foreign threads (that is the whole
32 : point: the user's thread provides the timing). A waiter-side
33 : claimed_ flag is not enough there -- the
34 : waker has to dereference the waiter, and nothing would pin the
35 : waiter's frame between reading the pointer and claiming it.
36 :
37 : So the st_ atomic is the single arbiter:
38 :
39 : empty --arm--> arming --publish--> armed --wake/cancel CAS--> empty
40 : arming --stop already requested--> empty
41 : empty --wake--> token --wait consumes--> empty
42 :
43 : Whoever wins the armed->empty CAS owns the resume and may
44 : dereference waiter_: the frame cannot die underneath the
45 : winner because the coroutine only resumes when the winner
46 : posts it. The loser never touches the waiter. When the stop
47 : callback wins, a concurrent wake retries, finds empty, and
48 : latches a token -- a racing wakeup is deferred, never lost.
49 :
50 : The stop callback is registered before the waiter enters the
51 : arming state, and a wake() winner destroys it before posting.
52 : While arming, nothing can claim the waiter: wake(), the stop
53 : callback and the destructor wait the few instructions it takes.
54 : So the waiter settles a stop requested during registration
55 : itself, its last touch of the waker is the publishing store,
56 : and the waker can die right after the wake() that claimed it.
57 :
58 : Serialized resumption is still required: the executor must not
59 : run the waiter's continuations concurrently, since a winner can
60 : post the waiter before its await_suspend has returned. async_event
61 : and async_mutex make the same assumption; it is stated explicitly
62 : here because wake() invites foreign threads into the picture.
63 : */
64 :
65 : namespace boost {
66 : namespace capy {
67 :
68 : /** A single-slot waker that hands one wakeup to a waiting coroutine.
69 :
70 : This is the escape hatch for timing and other external events:
71 : the user provides the thread and the clock, capy provides the
72 : suspension point. One coroutine suspends in `wait()`; any
73 : thread wakes it with `wake()`.
74 :
75 : A wakeup with no waiter present is latched as a single pending
76 : token, and the next `wait()` consumes it immediately. This
77 : makes the wake-before-wait race benign without any lock
78 : protocol. Multiple wakes collapse into one token.
79 :
80 : @par Cancellation
81 :
82 : If the environment's stop token is triggered while suspended,
83 : the wait completes with `error::canceled`. A wake that loses
84 : the race against cancellation is latched for the next `wait()`
85 : rather than dropped.
86 :
87 : @par Zero Allocation
88 :
89 : No heap allocation occurs for wait or wake operations.
90 :
91 : @par Thread Safety
92 :
93 : Distinct objects: Safe.@n
94 : Shared objects: `wake()` may be called from any thread.
95 : `wait()` must only be awaited by one coroutine at a time. The
96 : executor must never run the coroutine's continuations
97 : concurrently: use a single-threaded executor, or a strand over
98 : a multi-threaded one. That is the same threading model as
99 : `async_event` and `async_mutex`. Awaiting `wait()` directly
100 : on a multi-threaded executor is undefined.
101 :
102 : This type is non-copyable and non-movable because a suspended
103 : waiter holds a pointer into the object.
104 :
105 : @par Example
106 : @par !example example
107 :
108 : */
109 : class async_waker
110 : {
111 : public:
112 : class wait_awaiter;
113 :
114 : private:
115 : static constexpr int state_empty = 0; // no token, no waiter
116 : static constexpr int state_token = 1; // latched wakeup
117 : static constexpr int state_armed = 2; // waiter suspended
118 : static constexpr int state_arming = 3; // waiter publishing itself
119 :
120 : std::atomic<int> st_{state_empty};
121 : wait_awaiter* waiter_ = nullptr;
122 :
123 : // Take the armed waiter, waiting out one still publishing itself.
124 HIT 2427 : bool claim_armed() noexcept
125 : {
126 2427 : int expected = state_armed;
127 : for(;;)
128 : {
129 4854 : if(st_.compare_exchange_weak(
130 : expected, state_empty,
131 : std::memory_order_acq_rel,
132 : std::memory_order_acquire))
133 2011 : return true;
134 416 : if(expected == state_armed)
135 : continue; // LCOV_EXCL_LINE spurious weak CAS failure
136 416 : if(expected != state_arming)
137 416 : return false;
138 : std::this_thread::yield(); // LCOV_EXCL_LINE race with a waiter still publishing itself
139 : expected = state_armed; // LCOV_EXCL_LINE race with a waiter still publishing itself
140 : }
141 : }
142 :
143 : public:
144 : /** Suspends the caller until `wake()` runs, or resumes it with `error::canceled` on a stop request.
145 : */
146 : class wait_awaiter
147 : {
148 : friend class async_waker;
149 :
150 : async_waker* waker_;
151 : continuation cont_;
152 : executor_ref ex_;
153 :
154 : // Declared before stop_cb_buf_: the callback accesses
155 : // these members, so they must still be alive if the
156 : // stop_cb_ destructor blocks.
157 : bool canceled_ = false;
158 : bool active_ = false;
159 : bool published_ = false;
160 :
161 : // Set by whoever wins the armed->empty transition, before it
162 : // posts. A claimed waiter belongs to the executor's queue and
163 : // may be discarded after the waker is gone, so its destructor
164 : // must not touch the waker.
165 : std::atomic<bool> claimed_{false};
166 :
167 : struct cancel_fn
168 : {
169 : wait_awaiter* self_;
170 :
171 2009 : void operator()() const noexcept
172 : {
173 2009 : if(self_->waker_->claim_armed())
174 : {
175 2009 : self_->canceled_ = true;
176 2009 : self_->claimed_.store(
177 : true, std::memory_order_release);
178 2009 : self_->ex_.post(self_->cont_);
179 : }
180 2009 : }
181 : };
182 :
183 : using stop_cb_t = std::stop_callback<cancel_fn>;
184 :
185 : // Aligned storage for stop_cb_t. Declared last: its
186 : // destructor may block while the callback accesses the
187 : // members above.
188 : BOOST_CAPY_MSVC_WARNING_PUSH
189 : BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
190 : alignas(stop_cb_t)
191 : unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
192 : BOOST_CAPY_MSVC_WARNING_POP
193 :
194 2325 : stop_cb_t& stop_cb_() noexcept
195 : {
196 2325 : return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_);
197 : }
198 :
199 : public:
200 : /** Destroy the awaiter, leaving the waker unable to reach it.
201 :
202 : Destroys the stop callback if one is registered. If the awaiter
203 : is still armed, it also returns the waker's slot to the empty
204 : state, so a later `wake()` cannot dereference a destroyed
205 : awaiter. That case means the frame is being torn down without
206 : ever being resumed; a wake arriving afterward latches a token
207 : instead.
208 : */
209 4918 : ~wait_awaiter()
210 : {
211 4918 : if(active_)
212 3 : stop_cb_().~stop_cb_t();
213 4923 : if(published_ &&
214 5 : !claimed_.load(std::memory_order_acquire))
215 : {
216 : // Destroyed while still armed (frame torn down
217 : // without resuming): deregister so a later
218 : // wake cannot touch the dead frame.
219 1 : int expected = state_armed;
220 1 : waker_->st_.compare_exchange_strong(
221 : expected, state_empty,
222 : std::memory_order_acq_rel,
223 : std::memory_order_acquire);
224 : }
225 4918 : }
226 :
227 : /** Construct an awaiter for the given waker.
228 :
229 : @param waker The waker to wait on. It must outlive the awaiter.
230 : */
231 2459 : explicit wait_awaiter(async_waker* waker) noexcept
232 2459 : : waker_(waker)
233 : {
234 2459 : }
235 :
236 : /** Construct by moving.
237 :
238 : The moved-from awaiter is left inert: its destructor no longer
239 : destroys the stop callback and no longer deregisters from the
240 : waker.
241 :
242 : @param o The awaiter to move from.
243 : */
244 2459 : wait_awaiter(wait_awaiter&& o) noexcept
245 2459 : : waker_(o.waker_)
246 2459 : , cont_(o.cont_)
247 2459 : , ex_(o.ex_)
248 2459 : , canceled_(o.canceled_)
249 2459 : , active_(std::exchange(o.active_, false))
250 2459 : , published_(std::exchange(o.published_, false))
251 2459 : , claimed_(o.claimed_.load(std::memory_order_relaxed))
252 : {
253 2459 : }
254 :
255 : /** Copy construction is disabled; an armed waiter is registered
256 : with the waker by address.
257 :
258 : @param other The awaiter that would be copied.
259 : */
260 : wait_awaiter(wait_awaiter const& other) = delete;
261 :
262 : /** Copy assignment is disabled; an armed waiter is registered
263 : with the waker by address.
264 :
265 : @param other The awaiter that would be assigned from.
266 :
267 : @return A reference to `*this`.
268 : */
269 : wait_awaiter& operator=(wait_awaiter const& other) = delete;
270 :
271 : /** Move assignment is disabled; an armed waiter is registered
272 : with the waker by address.
273 :
274 : @param other The awaiter that would be moved from.
275 :
276 : @return A reference to `*this`.
277 : */
278 : wait_awaiter& operator=(wait_awaiter&& other) = delete;
279 :
280 : /** Consume a latched token, completing synchronously.
281 :
282 : This is not a pure query: the check is a compare-exchange that
283 : takes the token. Calling it twice is not idempotent: the second
284 : call reports `false`, because the first already consumed the
285 : wakeup.
286 :
287 : @return `true` if a pending wakeup token was latched and has now
288 : been consumed, in which case the awaiting coroutine does not
289 : suspend; otherwise `false`.
290 : */
291 2459 : bool await_ready() noexcept
292 : {
293 2459 : int expected = state_token;
294 2459 : return waker_->st_.compare_exchange_strong(
295 : expected, state_empty,
296 : std::memory_order_acq_rel,
297 2459 : std::memory_order_acquire);
298 : }
299 :
300 : /** Arm the waker with the awaiting coroutine.
301 :
302 : This is the @ref IoAwaitable overload of `await_suspend`.
303 : Unlike `async_event` and `async_mutex`, it has three outcomes,
304 : because a `wake()` from another thread can land in the window
305 : between `await_ready` and this call.
306 :
307 : @li A stop request is already pending on `env->stop_token`: the
308 : awaiter records the cancellation and does not arm.
309 :
310 : @li The waker's slot is no longer empty. Under the single-waiter
311 : precondition that means a wakeup was latched after
312 : `await_ready` looked, so the token is consumed here instead
313 : and the wait succeeds.
314 :
315 : @li Otherwise a stop callback is registered on
316 : `env->stop_token` and the slot moves to the armed state,
317 : publishing this awaiter to the waker. A stop requested
318 : while the callback registered completes the wait canceled
319 : instead. Whichever of `wake()` and that callback wins the
320 : armed-to-empty transition posts `h` through
321 : `env->executor`; a winning `wake()` first deregisters the
322 : callback. The loser does nothing, and a losing `wake()`
323 : re-latches its token for the next `wait()`.
324 :
325 : @param h The awaiting coroutine, resumed when the waker fires
326 : or the wait is canceled.
327 :
328 : @param env The execution environment. Its executor posts the
329 : resumption and its stop token is watched for the duration of
330 : the wait. It must outlive the wait.
331 :
332 : @return `h` in the first two cases, which resumes the awaiting
333 : coroutine immediately; otherwise `std::noop_coroutine()`, which
334 : leaves the coroutine suspended and returns control to the
335 : resumer.
336 : */
337 : std::coroutine_handle<>
338 2361 : await_suspend(
339 : std::coroutine_handle<> h,
340 : io_env const* env) noexcept
341 : {
342 2361 : if(env->stop_token.stop_requested())
343 : {
344 36 : canceled_ = true;
345 36 : return h;
346 : }
347 2325 : cont_.h = h;
348 2325 : cont_.owner = env->owner;
349 2325 : ex_ = env->executor;
350 2325 : waker_->waiter_ = this;
351 :
352 : // Registered before publishing, so the wake() that claims
353 : // this waiter can deregister it before posting.
354 6975 : ::new(stop_cb_buf_) stop_cb_t(
355 2325 : env->stop_token, cancel_fn{this});
356 2325 : active_ = true;
357 2325 : published_ = true;
358 :
359 2325 : int expected = state_empty;
360 4650 : if(!waker_->st_.compare_exchange_strong(
361 : expected, state_arming,
362 : std::memory_order_acq_rel,
363 : std::memory_order_acquire))
364 : {
365 : // Single-waiter precondition: a second concurrent
366 : // wait would find the slot armed.
367 MIS 0 : BOOST_CAPY_ASSERT(expected == state_token);
368 :
369 : // A wake latched between await_ready and here;
370 : // consume it and resume inline.
371 0 : published_ = false;
372 0 : stop_cb_().~stop_cb_t();
373 0 : active_ = false;
374 0 : waker_->st_.store(
375 : state_empty, std::memory_order_release);
376 0 : return h;
377 : }
378 :
379 : // While arming nothing else can claim this waiter, so a stop
380 : // requested as the callback registered is settled here, and
381 : // the waker is not touched again once the waiter is armed.
382 HIT 2325 : if(env->stop_token.stop_requested())
383 : {
384 MIS 0 : published_ = false;
385 0 : waker_->st_.store(
386 : state_empty, std::memory_order_release);
387 0 : canceled_ = true;
388 0 : return h;
389 : }
390 HIT 2325 : waker_->st_.store(state_armed, std::memory_order_release);
391 2325 : return std::noop_coroutine();
392 : }
393 :
394 : /** Complete the wait and report the outcome.
395 :
396 : Destroys the stop callback if one is registered and clears the
397 : armed bookkeeping, so the destructor does not deregister a slot
398 : the resumption already consumed.
399 :
400 : @return An empty `io_result<>` if the wait was woken, whether by
401 : `wake()` or by a token consumed inline. Otherwise one holding
402 : `error::canceled`, which means the stop token won the race or
403 : the waker was destroyed.
404 : */
405 2454 : [[nodiscard]] io_result<> await_resume() noexcept
406 : {
407 2454 : if(active_)
408 : {
409 2007 : stop_cb_().~stop_cb_t();
410 2007 : active_ = false;
411 : }
412 2454 : published_ = false;
413 2454 : if(canceled_)
414 2045 : return {make_error_code(error::canceled)};
415 409 : return {{}};
416 : }
417 : };
418 :
419 : /// Construct with no token latched.
420 307 : async_waker() = default;
421 :
422 : /** Destroy the waker, canceling a coroutine still waiting.
423 :
424 : The waiter is posted to its executor to resume with
425 : `error::canceled`, as if its stop token had been triggered. An
426 : executor that can no longer run it abandons it instead, possibly
427 : before this destructor returns. Either way the waiter no longer
428 : refers to the waker.
429 :
430 : @pre No `wake()` call is in progress or made afterward.
431 : */
432 418 : ~async_waker()
433 : {
434 418 : if(claim_armed())
435 : {
436 2 : auto* w = waiter_;
437 : // The stop callback reaches into the waker; deregistering
438 : // it waits out one already running.
439 2 : if(w->active_)
440 : {
441 : using cb_t = wait_awaiter::stop_cb_t;
442 2 : w->stop_cb_().~cb_t();
443 2 : w->active_ = false;
444 : }
445 2 : w->canceled_ = true;
446 2 : w->claimed_.store(true, std::memory_order_release);
447 2 : w->ex_.post(w->cont_);
448 : }
449 418 : }
450 :
451 : /** Copy construction is disabled; an armed waiter points into the
452 : waker.
453 :
454 : @param other The waker that would be copied.
455 : */
456 : async_waker(async_waker const& other) = delete;
457 :
458 : /** Copy assignment is disabled; an armed waiter points into the waker.
459 :
460 : @param other The waker that would be assigned from.
461 :
462 : @return A reference to `*this`.
463 : */
464 : async_waker& operator=(async_waker const& other) = delete;
465 :
466 : /** Move construction is disabled; an armed waiter points into the
467 : waker.
468 :
469 : @param other The waker that would be moved from.
470 : */
471 : async_waker(async_waker&& other) = delete;
472 :
473 : /** Move assignment is disabled; an armed waiter points into the waker.
474 :
475 : @param other The waker that would be moved from.
476 :
477 : @return A reference to `*this`.
478 : */
479 : async_waker& operator=(async_waker&& other) = delete;
480 :
481 : /** Asynchronously wait until woken.
482 :
483 : If a token is latched, completes immediately and consumes
484 : it. Otherwise suspends until `wake()` or the stop token
485 : fires.
486 :
487 : @par Preconditions
488 : No other coroutine is currently waiting on this object; a
489 : coroutine woken or canceled but not yet resumed still is.
490 :
491 : @return An awaitable that await-returns `io_result<>`;
492 : empty on wakeup, `error::canceled` if the stop
493 : token wins.
494 : */
495 2459 : wait_awaiter wait() noexcept
496 : {
497 2459 : return wait_awaiter{this};
498 : }
499 :
500 : /** Wake the waiter, or latch the wakeup if none waits.
501 :
502 : Callable from any thread. The waiter's resumption is
503 : posted through its executor; this call never resumes a
504 : coroutine inline. Multiple calls without an intervening
505 : `wait()` collapse into a single token. It may wait briefly
506 : for a waiter that is still suspending, or for a stop callback
507 : of the waiter that is running on another thread.
508 : */
509 413 : void wake() noexcept
510 : {
511 : for(;;)
512 : {
513 413 : int s = st_.load(std::memory_order_acquire);
514 413 : if(s == state_token)
515 413 : return;
516 411 : if(s == state_empty)
517 : {
518 196 : if(st_.compare_exchange_weak(
519 : s, state_token,
520 : std::memory_order_acq_rel,
521 : std::memory_order_acquire))
522 98 : return;
523 MIS 0 : continue;
524 : }
525 : // The waiter is still publishing itself; it settles in a
526 : // few instructions.
527 HIT 313 : if(s == state_arming)
528 : {
529 MIS 0 : std::this_thread::yield();
530 0 : continue;
531 : }
532 : // armed: winning this CAS claims the waiter, whose
533 : // frame is pinned until we post its resumption.
534 HIT 626 : if(st_.compare_exchange_weak(
535 : s, state_empty,
536 : std::memory_order_acq_rel,
537 : std::memory_order_acquire))
538 : {
539 313 : auto* w = waiter_;
540 : // Nothing may reach this waker once the waiter is
541 : // posted; this waits out a stop callback already
542 : // running, which then finds the slot empty.
543 : using cb_t = wait_awaiter::stop_cb_t;
544 313 : w->stop_cb_().~cb_t();
545 313 : w->active_ = false;
546 313 : w->claimed_.store(true, std::memory_order_release);
547 313 : w->ex_.post(w->cont_);
548 313 : return;
549 : }
550 MIS 0 : }
551 : }
552 : };
553 :
554 : } // namespace capy
555 : } // namespace boost
556 :
557 : #endif
|
