TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Steve Gerbino
3 : //
4 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
5 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6 : //
7 : // Official repository: https://github.com/cppalliance/capy
8 : //
9 :
10 : #ifndef BOOST_CAPY_EX_ASYNC_WAKER_HPP
11 : #define BOOST_CAPY_EX_ASYNC_WAKER_HPP
12 :
13 : #include <boost/capy/detail/config.hpp>
14 : #include <boost/capy/continuation.hpp>
15 : #include <boost/capy/error.hpp>
16 : #include <boost/capy/ex/executor_ref.hpp>
17 : #include <boost/capy/ex/io_env.hpp>
18 : #include <boost/capy/io_result.hpp>
19 :
20 : #include <atomic>
21 : #include <coroutine>
22 : #include <new>
23 : #include <stop_token>
24 : #include <utility>
25 :
26 : /* async_waker implementation notes
27 : ===================================
28 :
29 : wake() must be callable from foreign threads (that is the whole
30 : point: the user's thread provides the timing). A waiter-side
31 : claimed_ flag is not enough there -- the
32 : waker has to dereference the waiter, and nothing would pin the
33 : waiter's frame between reading the pointer and claiming it.
34 :
35 : So the three-state st_ atomic is the single arbiter:
36 :
37 : empty --arm--> armed --wake/cancel CAS--> empty
38 : empty --wake--> token --wait consumes--> empty
39 :
40 : Whoever wins the armed->empty CAS owns the resume and may
41 : dereference waiter_: the frame cannot die underneath the
42 : winner because the coroutine only resumes when the winner
43 : posts it. The loser never touches the waiter. When the stop
44 : callback wins, a concurrent wake retries, finds empty, and
45 : latches a token -- a racing wakeup is deferred, never lost.
46 :
47 : Serialized resumption is required: await_suspend keeps
48 : writing after the publishing armed-CAS (the stop_cb
49 : placement-new and active_ = true), so a wake/cancel winner
50 : can post the continuation while that tail is still running.
51 : The posted resume must be ordered after await_suspend's
52 : return, which holds on a single-threaded executor (the one
53 : thread is still inside await_suspend) and on a strand (the
54 : resume is a later turn, synchronized with the current one).
55 : A raw multi-threaded executor lets another worker run
56 : await_resume against those in-flight writes. async_event and
57 : async_mutex make the same assumption; it is stated explicitly
58 : here because wake() invites foreign threads into the picture.
59 : */
60 :
61 : namespace boost {
62 : namespace capy {
63 :
64 : /** A single-slot waker that hands one wakeup to a waiting coroutine.
65 :
66 : This is the escape hatch for timing and other external events:
67 : the user provides the thread and the clock, capy provides the
68 : suspension point. One coroutine suspends in `wait()`; any
69 : thread wakes it with `wake()`.
70 :
71 : A wakeup with no waiter present is latched as a single pending
72 : token, and the next `wait()` consumes it immediately. This
73 : makes the wake-before-wait race benign without any lock
74 : protocol. Multiple wakes collapse into one token.
75 :
76 : @par Cancellation
77 :
78 : If the environment's stop token is triggered while suspended,
79 : the wait completes with `error::canceled`. A wake that loses
80 : the race against cancellation is latched for the next `wait()`
81 : rather than dropped.
82 :
83 : @par Zero Allocation
84 :
85 : No heap allocation occurs for wait or wake operations.
86 :
87 : @par Thread Safety
88 :
89 : Distinct objects: Safe.@n
90 : Shared objects: `wake()` may be called from any thread.
91 : `wait()` must only be awaited by one coroutine at a time, and
92 : only on an executor that never runs the coroutine's
93 : continuations concurrently: a single-threaded executor or a
94 : strand over a multi-threaded one (the same threading model as
95 : `async_event` and `async_mutex`). Awaiting `wait()` directly
96 : on a multi-threaded executor is undefined.
97 :
98 : This type is non-copyable and non-movable because a suspended
99 : waiter holds a pointer into the object.
100 :
101 : @par Example
102 : @code
103 : async_waker waker;
104 :
105 : // user-provided timing thread
106 : std::thread th([&waker] {
107 : std::this_thread::sleep_for(100ms);
108 : waker.wake();
109 : });
110 :
111 : task<> waiter() {
112 : auto [ec] = co_await waker.wait();
113 : // resumed on the executor after ~100ms
114 : }
115 : // ... th.join() after the pool drains
116 : @endcode
117 : */
118 : class async_waker
119 : {
120 : public:
121 : class wait_awaiter;
122 :
123 : private:
124 : static constexpr int state_empty = 0; // no token, no waiter
125 : static constexpr int state_token = 1; // latched wakeup
126 : static constexpr int state_armed = 2; // waiter suspended
127 :
128 : std::atomic<int> st_{state_empty};
129 : wait_awaiter* waiter_ = nullptr;
130 :
131 : public:
132 : /** Awaiter returned by wait().
133 : */
134 : class wait_awaiter
135 : {
136 : friend class async_waker;
137 :
138 : async_waker* waker_;
139 : continuation cont_;
140 : executor_ref ex_;
141 :
142 : // Declared before stop_cb_buf_: the callback accesses
143 : // these members, so they must still be alive if the
144 : // stop_cb_ destructor blocks.
145 : bool canceled_ = false;
146 : bool active_ = false;
147 : bool published_ = false;
148 :
149 : struct cancel_fn
150 : {
151 : wait_awaiter* self_;
152 :
153 HIT 8 : void operator()() const noexcept
154 : {
155 8 : int expected = state_armed;
156 16 : if(self_->waker_->st_.compare_exchange_strong(
157 : expected, state_empty,
158 : std::memory_order_acq_rel,
159 : std::memory_order_acquire))
160 : {
161 4 : self_->canceled_ = true;
162 4 : self_->ex_.post(self_->cont_);
163 : }
164 8 : }
165 : };
166 :
167 : using stop_cb_t = std::stop_callback<cancel_fn>;
168 :
169 : // Aligned storage for stop_cb_t. Declared last: its
170 : // destructor may block while the callback accesses the
171 : // members above.
172 : BOOST_CAPY_MSVC_WARNING_PUSH
173 : BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
174 : alignas(stop_cb_t)
175 : unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
176 : BOOST_CAPY_MSVC_WARNING_POP
177 :
178 14 : stop_cb_t& stop_cb_() noexcept
179 : {
180 14 : return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_);
181 : }
182 :
183 : public:
184 264 : ~wait_awaiter()
185 : {
186 264 : if(active_)
187 1 : stop_cb_().~stop_cb_t();
188 264 : if(published_)
189 : {
190 : // Destroyed while still armed (frame torn down
191 : // without resuming): deregister so a later
192 : // wake cannot touch the dead frame.
193 1 : int expected = state_armed;
194 1 : waker_->st_.compare_exchange_strong(
195 : expected, state_empty,
196 : std::memory_order_acq_rel,
197 : std::memory_order_acquire);
198 : }
199 264 : }
200 :
201 132 : explicit wait_awaiter(async_waker* waker) noexcept
202 132 : : waker_(waker)
203 : {
204 132 : }
205 :
206 132 : wait_awaiter(wait_awaiter&& o) noexcept
207 132 : : waker_(o.waker_)
208 132 : , cont_(o.cont_)
209 132 : , ex_(o.ex_)
210 132 : , canceled_(o.canceled_)
211 132 : , active_(std::exchange(o.active_, false))
212 132 : , published_(std::exchange(o.published_, false))
213 : {
214 132 : }
215 :
216 : wait_awaiter(wait_awaiter const&) = delete;
217 : wait_awaiter& operator=(wait_awaiter const&) = delete;
218 : wait_awaiter& operator=(wait_awaiter&&) = delete;
219 :
220 : /// Consume a latched token, completing synchronously.
221 132 : bool await_ready() noexcept
222 : {
223 132 : int expected = state_token;
224 132 : return waker_->st_.compare_exchange_strong(
225 : expected, state_empty,
226 : std::memory_order_acq_rel,
227 132 : std::memory_order_acquire);
228 : }
229 :
230 : /** IoAwaitable protocol overload. */
231 : std::coroutine_handle<>
232 36 : await_suspend(
233 : std::coroutine_handle<> h,
234 : io_env const* env) noexcept
235 : {
236 36 : if(env->stop_token.stop_requested())
237 : {
238 22 : canceled_ = true;
239 22 : return h;
240 : }
241 14 : cont_.h = h;
242 14 : ex_ = env->executor;
243 14 : waker_->waiter_ = this;
244 :
245 14 : int expected = state_empty;
246 28 : if(!waker_->st_.compare_exchange_strong(
247 : expected, state_armed,
248 : std::memory_order_acq_rel,
249 : std::memory_order_acquire))
250 : {
251 : // Single-waiter precondition: a second concurrent
252 : // wait would find the slot armed.
253 MIS 0 : BOOST_CAPY_ASSERT(expected == state_token);
254 :
255 : // A wake latched between await_ready and here;
256 : // consume it and resume inline.
257 0 : waker_->st_.store(
258 : state_empty, std::memory_order_release);
259 0 : return h;
260 : }
261 HIT 14 : published_ = true;
262 :
263 42 : ::new(stop_cb_buf_) stop_cb_t(
264 14 : env->stop_token, cancel_fn{this});
265 14 : active_ = true;
266 14 : return std::noop_coroutine();
267 : }
268 :
269 131 : io_result<> await_resume() noexcept
270 : {
271 131 : if(active_)
272 : {
273 13 : stop_cb_().~stop_cb_t();
274 13 : active_ = false;
275 : }
276 131 : published_ = false;
277 131 : if(canceled_)
278 25 : return {make_error_code(error::canceled)};
279 106 : return {{}};
280 : }
281 : };
282 :
283 : /// Construct with no token latched.
284 : async_waker() = default;
285 :
286 : /// Copy constructor (deleted).
287 : async_waker(async_waker const&) = delete;
288 :
289 : /// Copy assignment (deleted).
290 : async_waker& operator=(async_waker const&) = delete;
291 :
292 : /// Move constructor (deleted).
293 : async_waker(async_waker&&) = delete;
294 :
295 : /// Move assignment (deleted).
296 : async_waker& operator=(async_waker&&) = delete;
297 :
298 : /** Asynchronously wait until woken.
299 :
300 : If a token is latched, completes immediately and consumes
301 : it. Otherwise suspends until `wake()` or the stop token
302 : fires.
303 :
304 : @par Preconditions
305 : No other coroutine is currently waiting on this object.
306 :
307 : @return An awaitable that await-returns `io_result<>`;
308 : empty on wakeup, `error::canceled` if the stop
309 : token wins.
310 : */
311 132 : wait_awaiter wait() noexcept
312 : {
313 132 : return wait_awaiter{this};
314 : }
315 :
316 : /** Wake the waiter, or latch the wakeup if none waits.
317 :
318 : Callable from any thread. The waiter's resumption is
319 : posted through its executor; this call never resumes a
320 : coroutine inline. Multiple calls without an intervening
321 : `wait()` collapse into a single token.
322 : */
323 108 : void wake() noexcept
324 : {
325 : for(;;)
326 : {
327 108 : int s = st_.load(std::memory_order_acquire);
328 108 : if(s == state_token)
329 108 : return;
330 106 : if(s == state_empty)
331 : {
332 192 : if(st_.compare_exchange_weak(
333 : s, state_token,
334 : std::memory_order_acq_rel,
335 : std::memory_order_acquire))
336 96 : return;
337 MIS 0 : continue;
338 : }
339 : // armed: winning this CAS claims the waiter, whose
340 : // frame is pinned until we post its resumption.
341 HIT 20 : if(st_.compare_exchange_weak(
342 : s, state_empty,
343 : std::memory_order_acq_rel,
344 : std::memory_order_acquire))
345 : {
346 10 : auto* w = waiter_;
347 10 : w->ex_.post(w->cont_);
348 10 : return;
349 : }
350 MIS 0 : }
351 : }
352 : };
353 :
354 : } // namespace capy
355 : } // namespace boost
356 :
357 : #endif
|
