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/await_context.hpp>
19 : #include <boost/capy/results.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 1049 : bool claim_armed() noexcept
125 : {
126 1049 : int expected = state_armed;
127 : for(;;)
128 : {
129 2098 : if(st_.compare_exchange_weak(
130 : expected, state_empty,
131 : std::memory_order_acq_rel,
132 : std::memory_order_acquire))
133 206 : return true;
134 843 : if(expected == state_armed)
135 : continue; // LCOV_EXCL_LINE spurious weak CAS failure
136 843 : if(expected != state_arming)
137 843 : 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 631 : void operator()() const noexcept
172 : {
173 631 : if(self_->waker_->claim_armed())
174 : {
175 204 : self_->canceled_ = true;
176 204 : self_->claimed_.store(
177 : true, std::memory_order_release);
178 204 : self_->ex_.post(self_->cont_);
179 : }
180 631 : }
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 946 : stop_cb_t& stop_cb_() noexcept
195 : {
196 946 : 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 4906 : ~wait_awaiter()
210 : {
211 4906 : if(active_)
212 3 : stop_cb_().~stop_cb_t();
213 4911 : 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 4906 : }
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 2453 : explicit wait_awaiter(async_waker* waker) noexcept
232 2453 : : waker_(waker)
233 : {
234 2453 : }
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 2453 : wait_awaiter(wait_awaiter&& o) noexcept
245 2453 : : waker_(o.waker_)
246 2453 : , cont_(o.cont_)
247 2453 : , ex_(o.ex_)
248 2453 : , canceled_(o.canceled_)
249 2453 : , active_(std::exchange(o.active_, false))
250 2453 : , published_(std::exchange(o.published_, false))
251 2453 : , claimed_(o.claimed_.load(std::memory_order_relaxed))
252 : {
253 2453 : }
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 2453 : bool await_ready() noexcept
292 : {
293 2453 : int expected = state_token;
294 2453 : return waker_->st_.compare_exchange_strong(
295 : expected, state_empty,
296 : std::memory_order_acq_rel,
297 2453 : std::memory_order_acquire);
298 : }
299 :
300 : /** Arm the waker with the awaiting coroutine.
301 :
302 : This is the @ref ContextAwaitable 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 2354 : await_suspend(
339 : std::coroutine_handle<> h,
340 : await_context const* env) noexcept
341 : {
342 2354 : if(env->stop_token.stop_requested())
343 : {
344 1408 : canceled_ = true;
345 1408 : return h;
346 : }
347 946 : cont_.h = h;
348 946 : cont_.owner = env->owner;
349 946 : ex_ = env->executor;
350 946 : waker_->waiter_ = this;
351 :
352 : // Registered before publishing, so the wake() that claims
353 : // this waiter can deregister it before posting.
354 2838 : ::new(stop_cb_buf_) stop_cb_t(
355 946 : env->stop_token, cancel_fn{this});
356 946 : active_ = true;
357 946 : published_ = true;
358 :
359 946 : int expected = state_empty;
360 1892 : if(!waker_->st_.compare_exchange_strong(
361 : expected, state_arming,
362 : std::memory_order_acq_rel,
363 : std::memory_order_acquire))
364 : {
365 : // LCOV_EXCL_START race: a wake latched between await_ready and here
366 : // Single-waiter precondition: a second concurrent
367 : // wait would find the slot armed.
368 : BOOST_CAPY_ASSERT(expected == state_token);
369 :
370 : // A wake latched between await_ready and here;
371 : // consume it and resume inline.
372 : published_ = false;
373 : stop_cb_().~stop_cb_t();
374 : active_ = false;
375 : waker_->st_.store(
376 : state_empty, std::memory_order_release);
377 : return h;
378 : // LCOV_EXCL_STOP
379 : }
380 :
381 : // While arming nothing else can claim this waiter, so a stop
382 : // requested as the callback registered is settled here, and
383 : // the waker is not touched again once the waiter is armed.
384 945 : if(env->stop_token.stop_requested())
385 : {
386 427 : published_ = false;
387 427 : waker_->st_.store(
388 : state_empty, std::memory_order_release);
389 427 : canceled_ = true;
390 427 : return h;
391 : }
392 518 : waker_->st_.store(state_armed, std::memory_order_release);
393 518 : return std::noop_coroutine();
394 : }
395 :
396 : /** Complete the wait and report the outcome.
397 :
398 : Destroys the stop callback if one is registered and clears the
399 : armed bookkeeping, so the destructor does not deregister a slot
400 : the resumption already consumed.
401 :
402 : @return An empty `results<>` if the wait was woken, whether by
403 : `wake()` or by a token consumed inline. Otherwise one holding
404 : `error::canceled`, which means the stop token won the race or
405 : the waker was destroyed.
406 : */
407 2448 : [[nodiscard]] results<> await_resume() noexcept
408 : {
409 2448 : if(active_)
410 : {
411 629 : stop_cb_().~stop_cb_t();
412 629 : active_ = false;
413 : }
414 2448 : published_ = false;
415 2448 : if(canceled_)
416 2039 : return {make_error_code(error::canceled)};
417 409 : return {{}};
418 : }
419 : };
420 :
421 : /// Construct with no token latched.
422 307 : async_waker() = default;
423 :
424 : /** Destroy the waker, canceling a coroutine still waiting.
425 :
426 : The waiter is posted to its executor to resume with
427 : `error::canceled`, as if its stop token had been triggered. An
428 : executor that can no longer run it abandons it instead, possibly
429 : before this destructor returns. Either way the waiter no longer
430 : refers to the waker.
431 :
432 : @pre No `wake()` call is in progress or made afterward.
433 : */
434 418 : ~async_waker()
435 : {
436 418 : if(claim_armed())
437 : {
438 2 : auto* w = waiter_;
439 : // The stop callback reaches into the waker; deregistering
440 : // it waits out one already running.
441 2 : if(w->active_)
442 : {
443 : using cb_t = wait_awaiter::stop_cb_t;
444 2 : w->stop_cb_().~cb_t();
445 2 : w->active_ = false;
446 : }
447 2 : w->canceled_ = true;
448 2 : w->claimed_.store(true, std::memory_order_release);
449 2 : w->ex_.post(w->cont_);
450 : }
451 418 : }
452 :
453 : /** Copy construction is disabled; an armed waiter points into the
454 : waker.
455 :
456 : @param other The waker that would be copied.
457 : */
458 : async_waker(async_waker const& other) = delete;
459 :
460 : /** Copy assignment is disabled; an armed waiter points into the waker.
461 :
462 : @param other The waker that would be assigned from.
463 :
464 : @return A reference to `*this`.
465 : */
466 : async_waker& operator=(async_waker const& other) = delete;
467 :
468 : /** Move construction is disabled; an armed waiter points into the
469 : waker.
470 :
471 : @param other The waker that would be moved from.
472 : */
473 : async_waker(async_waker&& other) = delete;
474 :
475 : /** Move assignment is disabled; an armed waiter points into the waker.
476 :
477 : @param other The waker that would be moved from.
478 :
479 : @return A reference to `*this`.
480 : */
481 : async_waker& operator=(async_waker&& other) = delete;
482 :
483 : /** Asynchronously wait until woken.
484 :
485 : If a token is latched, completes immediately and consumes
486 : it. Otherwise suspends until `wake()` or the stop token
487 : fires.
488 :
489 : @par Preconditions
490 : No other coroutine is currently waiting on this object; a
491 : coroutine woken or canceled but not yet resumed still is.
492 :
493 : @return An awaitable that await-returns `results<>`;
494 : empty on wakeup, `error::canceled` if the stop
495 : token wins.
496 : */
497 2453 : wait_awaiter wait() noexcept
498 : {
499 2453 : return wait_awaiter{this};
500 : }
501 :
502 : /** Wake the waiter, or latch the wakeup if none waits.
503 :
504 : Callable from any thread. The waiter's resumption is
505 : posted through its executor; this call never resumes a
506 : coroutine inline. Multiple calls without an intervening
507 : `wait()` collapse into a single token. It may wait briefly
508 : for a waiter that is still suspending, or for a stop callback
509 : of the waiter that is running on another thread.
510 : */
511 413 : void wake() noexcept
512 : {
513 : for(;;)
514 : {
515 413 : int s = st_.load(std::memory_order_acquire);
516 413 : if(s == state_token)
517 413 : return;
518 411 : if(s == state_empty)
519 : {
520 200 : if(st_.compare_exchange_weak(
521 : s, state_token,
522 : std::memory_order_acq_rel,
523 : std::memory_order_acquire))
524 100 : return;
525 : continue; // LCOV_EXCL_LINE spurious weak CAS failure
526 : }
527 : // The waiter is still publishing itself; it settles in a
528 : // few instructions.
529 311 : if(s == state_arming)
530 : {
531 : std::this_thread::yield(); // LCOV_EXCL_LINE race with a waiter still publishing itself
532 : continue; // LCOV_EXCL_LINE race with a waiter still publishing itself
533 : }
534 : // armed: winning this CAS claims the waiter, whose
535 : // frame is pinned until we post its resumption.
536 622 : if(st_.compare_exchange_weak(
537 : s, state_empty,
538 : std::memory_order_acq_rel,
539 : std::memory_order_acquire))
540 : {
541 311 : auto* w = waiter_;
542 : // Nothing may reach this waker once the waiter is
543 : // posted; this waits out a stop callback already
544 : // running, which then finds the slot empty.
545 : using cb_t = wait_awaiter::stop_cb_t;
546 311 : w->stop_cb_().~cb_t();
547 311 : w->active_ = false;
548 311 : w->claimed_.store(true, std::memory_order_release);
549 311 : w->ex_.post(w->cont_);
550 311 : return;
551 : }
552 : } // LCOV_EXCL_LINE armed CAS lost to a stop callback or failed spuriously
553 : }
554 : };
555 :
556 : } // namespace capy
557 : } // namespace boost
558 :
559 : #endif
|
