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 <utility>
26 :
27 : /* async_waker implementation notes
28 : ===================================
29 :
30 : wake() must be callable from foreign threads (that is the whole
31 : point: the user's thread provides the timing). A waiter-side
32 : claimed_ flag is not enough there -- the
33 : waker has to dereference the waiter, and nothing would pin the
34 : waiter's frame between reading the pointer and claiming it.
35 :
36 : So the three-state st_ atomic is the single arbiter:
37 :
38 : empty --arm--> armed --wake/cancel CAS--> empty
39 : empty --wake--> token --wait consumes--> empty
40 :
41 : Whoever wins the armed->empty CAS owns the resume and may
42 : dereference waiter_: the frame cannot die underneath the
43 : winner because the coroutine only resumes when the winner
44 : posts it. The loser never touches the waiter. When the stop
45 : callback wins, a concurrent wake retries, finds empty, and
46 : latches a token -- a racing wakeup is deferred, never lost.
47 :
48 : Serialized resumption is required: await_suspend keeps
49 : writing after the publishing armed-CAS (the stop_cb
50 : placement-new and active_ = true), so a wake/cancel winner
51 : can post the continuation while that tail is still running.
52 : The posted resume must be ordered after await_suspend's
53 : return, which holds on a single-threaded executor (the one
54 : thread is still inside await_suspend) and on a strand (the
55 : resume is a later turn, synchronized with the current one).
56 : A raw multi-threaded executor lets another worker run
57 : await_resume against those in-flight writes. async_event and
58 : async_mutex make the same assumption; it is stated explicitly
59 : here because wake() invites foreign threads into the picture.
60 : */
61 :
62 : namespace boost {
63 : namespace capy {
64 :
65 : /** A single-slot waker that hands one wakeup to a waiting coroutine.
66 :
67 : This is the escape hatch for timing and other external events:
68 : the user provides the thread and the clock, capy provides the
69 : suspension point. One coroutine suspends in `wait()`; any
70 : thread wakes it with `wake()`.
71 :
72 : A wakeup with no waiter present is latched as a single pending
73 : token, and the next `wait()` consumes it immediately. This
74 : makes the wake-before-wait race benign without any lock
75 : protocol. Multiple wakes collapse into one token.
76 :
77 : @par Cancellation
78 :
79 : If the environment's stop token is triggered while suspended,
80 : the wait completes with `error::canceled`. A wake that loses
81 : the race against cancellation is latched for the next `wait()`
82 : rather than dropped.
83 :
84 : @par Zero Allocation
85 :
86 : No heap allocation occurs for wait or wake operations.
87 :
88 : @par Thread Safety
89 :
90 : Distinct objects: Safe.@n
91 : Shared objects: `wake()` may be called from any thread.
92 : `wait()` must only be awaited by one coroutine at a time. The
93 : executor must never run the coroutine's continuations
94 : concurrently: use a single-threaded executor, or a strand over
95 : a multi-threaded one. That is the same threading model as
96 : `async_event` and `async_mutex`. Awaiting `wait()` directly
97 : on a multi-threaded executor is undefined.
98 :
99 : This type is non-copyable and non-movable because a suspended
100 : waiter holds a pointer into the object.
101 :
102 : @par Example
103 : @par !example example
104 :
105 : */
106 : class async_waker
107 : {
108 : public:
109 : class wait_awaiter;
110 :
111 : private:
112 : static constexpr int state_empty = 0; // no token, no waiter
113 : static constexpr int state_token = 1; // latched wakeup
114 : static constexpr int state_armed = 2; // waiter suspended
115 :
116 : std::atomic<int> st_{state_empty};
117 : wait_awaiter* waiter_ = nullptr;
118 :
119 : public:
120 : /** Suspends the caller until `wake()` runs, or resumes it with `error::canceled` on a stop request.
121 : */
122 : class wait_awaiter
123 : {
124 : friend class async_waker;
125 :
126 : async_waker* waker_;
127 : continuation cont_;
128 : executor_ref ex_;
129 :
130 : // Declared before stop_cb_buf_: the callback accesses
131 : // these members, so they must still be alive if the
132 : // stop_cb_ destructor blocks.
133 : bool canceled_ = false;
134 : bool active_ = false;
135 : bool published_ = false;
136 :
137 : struct cancel_fn
138 : {
139 : wait_awaiter* self_;
140 :
141 HIT 9 : void operator()() const noexcept
142 : {
143 9 : int expected = state_armed;
144 18 : if(self_->waker_->st_.compare_exchange_strong(
145 : expected, state_empty,
146 : std::memory_order_acq_rel,
147 : std::memory_order_acquire))
148 : {
149 8 : self_->canceled_ = true;
150 8 : self_->ex_.post(self_->cont_);
151 : }
152 9 : }
153 : };
154 :
155 : using stop_cb_t = std::stop_callback<cancel_fn>;
156 :
157 : // Aligned storage for stop_cb_t. Declared last: its
158 : // destructor may block while the callback accesses the
159 : // members above.
160 : BOOST_CAPY_MSVC_WARNING_PUSH
161 : BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
162 : alignas(stop_cb_t)
163 : unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
164 : BOOST_CAPY_MSVC_WARNING_POP
165 :
166 19 : stop_cb_t& stop_cb_() noexcept
167 : {
168 19 : return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_);
169 : }
170 :
171 : public:
172 : /** Destroy the awaiter, leaving the waker unable to reach it.
173 :
174 : Destroys the stop callback if one is registered. If the awaiter
175 : is still armed, it also returns the waker's slot to the empty
176 : state, so a later `wake()` cannot dereference a destroyed
177 : awaiter. That case means the frame is being torn down without
178 : ever being resumed; a wake arriving afterward latches a token
179 : instead.
180 : */
181 294 : ~wait_awaiter()
182 : {
183 294 : if(active_)
184 1 : stop_cb_().~stop_cb_t();
185 294 : if(published_)
186 : {
187 : // Destroyed while still armed (frame torn down
188 : // without resuming): deregister so a later
189 : // wake cannot touch the dead frame.
190 1 : int expected = state_armed;
191 1 : waker_->st_.compare_exchange_strong(
192 : expected, state_empty,
193 : std::memory_order_acq_rel,
194 : std::memory_order_acquire);
195 : }
196 294 : }
197 :
198 : /** Construct an awaiter for the given waker.
199 :
200 : @param waker The waker to wait on. It must outlive the awaiter.
201 : */
202 147 : explicit wait_awaiter(async_waker* waker) noexcept
203 147 : : waker_(waker)
204 : {
205 147 : }
206 :
207 : /** Construct by moving.
208 :
209 : The moved-from awaiter is left inert: its destructor no longer
210 : destroys the stop callback and no longer deregisters from the
211 : waker.
212 :
213 : @param o The awaiter to move from.
214 : */
215 147 : wait_awaiter(wait_awaiter&& o) noexcept
216 147 : : waker_(o.waker_)
217 147 : , cont_(o.cont_)
218 147 : , ex_(o.ex_)
219 147 : , canceled_(o.canceled_)
220 147 : , active_(std::exchange(o.active_, false))
221 147 : , published_(std::exchange(o.published_, false))
222 : {
223 147 : }
224 :
225 : /** Copy construction is disabled; an armed waiter is registered
226 : with the waker by address.
227 :
228 : @param other The awaiter that would be copied.
229 : */
230 : wait_awaiter(wait_awaiter const& other) = delete;
231 :
232 : /** Copy assignment is disabled; an armed waiter is registered
233 : with the waker by address.
234 :
235 : @param other The awaiter that would be assigned from.
236 :
237 : @return A reference to `*this`.
238 : */
239 : wait_awaiter& operator=(wait_awaiter const& other) = delete;
240 :
241 : /** Move assignment is disabled; an armed waiter is registered
242 : with the waker by address.
243 :
244 : @param other The awaiter that would be moved from.
245 :
246 : @return A reference to `*this`.
247 : */
248 : wait_awaiter& operator=(wait_awaiter&& other) = delete;
249 :
250 : /** Consume a latched token, completing synchronously.
251 :
252 : This is not a pure query: the check is a compare-exchange that
253 : takes the token. Calling it twice is not idempotent: the second
254 : call reports `false`, because the first already consumed the
255 : wakeup.
256 :
257 : @return `true` if a pending wakeup token was latched and has now
258 : been consumed, in which case the awaiting coroutine does not
259 : suspend; otherwise `false`.
260 : */
261 147 : bool await_ready() noexcept
262 : {
263 147 : int expected = state_token;
264 147 : return waker_->st_.compare_exchange_strong(
265 : expected, state_empty,
266 : std::memory_order_acq_rel,
267 147 : std::memory_order_acquire);
268 : }
269 :
270 : /** Arm the waker with the awaiting coroutine.
271 :
272 : This is the @ref IoAwaitable overload of `await_suspend`.
273 : Unlike `async_event` and `async_mutex`, it has three outcomes,
274 : because a `wake()` from another thread can land in the window
275 : between `await_ready` and this call.
276 :
277 : @li A stop request is already pending on `env->stop_token`: the
278 : awaiter records the cancellation and does not arm.
279 :
280 : @li The waker's slot is no longer empty. Under the single-waiter
281 : precondition that means a wakeup was latched after
282 : `await_ready` looked, so the token is consumed here instead
283 : and the wait succeeds.
284 :
285 : @li Otherwise the slot moves to the armed state, publishing this
286 : awaiter to the waker, and a stop callback is registered on
287 : `env->stop_token`. Whichever of `wake()` and that callback
288 : wins the armed-to-empty transition posts `h` through
289 : `env->executor`. The loser does nothing, and a losing
290 : `wake()` re-latches its token for the next `wait()`.
291 :
292 : @param h The awaiting coroutine, resumed when the waker fires
293 : or the wait is canceled.
294 :
295 : @param env The execution environment. Its executor posts the
296 : resumption and its stop token is watched for the duration of
297 : the wait. It must outlive the wait.
298 :
299 : @return `h` in the first two cases, which resumes the awaiting
300 : coroutine immediately; otherwise `std::noop_coroutine()`, which
301 : leaves the coroutine suspended and returns control to the
302 : resumer.
303 : */
304 : std::coroutine_handle<>
305 51 : await_suspend(
306 : std::coroutine_handle<> h,
307 : io_env const* env) noexcept
308 : {
309 51 : if(env->stop_token.stop_requested())
310 : {
311 32 : canceled_ = true;
312 32 : return h;
313 : }
314 19 : cont_.h = h;
315 19 : ex_ = env->executor;
316 19 : waker_->waiter_ = this;
317 :
318 19 : int expected = state_empty;
319 38 : if(!waker_->st_.compare_exchange_strong(
320 : expected, state_armed,
321 : std::memory_order_acq_rel,
322 : std::memory_order_acquire))
323 : {
324 : // Single-waiter precondition: a second concurrent
325 : // wait would find the slot armed.
326 MIS 0 : BOOST_CAPY_ASSERT(expected == state_token);
327 :
328 : // A wake latched between await_ready and here;
329 : // consume it and resume inline.
330 0 : waker_->st_.store(
331 : state_empty, std::memory_order_release);
332 0 : return h;
333 : }
334 HIT 19 : published_ = true;
335 :
336 57 : ::new(stop_cb_buf_) stop_cb_t(
337 19 : env->stop_token, cancel_fn{this});
338 19 : active_ = true;
339 19 : return std::noop_coroutine();
340 : }
341 :
342 : /** Complete the wait and report the outcome.
343 :
344 : Destroys the stop callback if one is registered and clears the
345 : armed bookkeeping, so the destructor does not deregister a slot
346 : the resumption already consumed.
347 :
348 : @return An empty `io_result<>` if the wait was woken, whether by
349 : `wake()` or by a token consumed inline. Otherwise one holding
350 : `error::canceled`, which means the stop token won the race.
351 : */
352 146 : [[nodiscard]] io_result<> await_resume() noexcept
353 : {
354 146 : if(active_)
355 : {
356 18 : stop_cb_().~stop_cb_t();
357 18 : active_ = false;
358 : }
359 146 : published_ = false;
360 146 : if(canceled_)
361 39 : return {make_error_code(error::canceled)};
362 107 : return {{}};
363 : }
364 : };
365 :
366 : /// Construct with no token latched.
367 1 : async_waker() = default;
368 :
369 : /** Copy construction is disabled; an armed waiter points into the
370 : waker.
371 :
372 : @param other The waker that would be copied.
373 : */
374 : async_waker(async_waker const& other) = delete;
375 :
376 : /** Copy assignment is disabled; an armed waiter points into the waker.
377 :
378 : @param other The waker that would be assigned from.
379 :
380 : @return A reference to `*this`.
381 : */
382 : async_waker& operator=(async_waker const& other) = delete;
383 :
384 : /** Move construction is disabled; an armed waiter points into the
385 : waker.
386 :
387 : @param other The waker that would be moved from.
388 : */
389 : async_waker(async_waker&& other) = delete;
390 :
391 : /** Move assignment is disabled; an armed waiter points into the waker.
392 :
393 : @param other The waker that would be moved from.
394 :
395 : @return A reference to `*this`.
396 : */
397 : async_waker& operator=(async_waker&& other) = delete;
398 :
399 : /** Asynchronously wait until woken.
400 :
401 : If a token is latched, completes immediately and consumes
402 : it. Otherwise suspends until `wake()` or the stop token
403 : fires.
404 :
405 : @par Preconditions
406 : No other coroutine is currently waiting on this object.
407 :
408 : @return An awaitable that await-returns `io_result<>`;
409 : empty on wakeup, `error::canceled` if the stop
410 : token wins.
411 : */
412 147 : wait_awaiter wait() noexcept
413 : {
414 147 : return wait_awaiter{this};
415 : }
416 :
417 : /** Wake the waiter, or latch the wakeup if none waits.
418 :
419 : Callable from any thread. The waiter's resumption is
420 : posted through its executor; this call never resumes a
421 : coroutine inline. Multiple calls without an intervening
422 : `wait()` collapse into a single token.
423 : */
424 109 : void wake() noexcept
425 : {
426 : for(;;)
427 : {
428 109 : int s = st_.load(std::memory_order_acquire);
429 109 : if(s == state_token)
430 109 : return;
431 107 : if(s == state_empty)
432 : {
433 192 : if(st_.compare_exchange_weak(
434 : s, state_token,
435 : std::memory_order_acq_rel,
436 : std::memory_order_acquire))
437 96 : return;
438 MIS 0 : continue;
439 : }
440 : // armed: winning this CAS claims the waiter, whose
441 : // frame is pinned until we post its resumption.
442 HIT 22 : if(st_.compare_exchange_weak(
443 : s, state_empty,
444 : std::memory_order_acq_rel,
445 : std::memory_order_acquire))
446 : {
447 11 : auto* w = waiter_;
448 11 : w->ex_.post(w->cont_);
449 11 : return;
450 : }
451 MIS 0 : }
452 : }
453 : };
454 :
455 : } // namespace capy
456 : } // namespace boost
457 :
458 : #endif
|
