TLA Line data Source code
1 : //
2 : // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
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_ASYNC_EVENT_HPP
12 : #define BOOST_CAPY_ASYNC_EVENT_HPP
13 :
14 : #include <boost/capy/detail/config.hpp>
15 : #include <boost/capy/detail/intrusive.hpp>
16 : #include <boost/capy/continuation.hpp>
17 : #include <boost/capy/concept/executor.hpp>
18 : #include <boost/capy/error.hpp>
19 : #include <boost/capy/ex/io_env.hpp>
20 : #include <boost/capy/io_result.hpp>
21 :
22 : #include <stop_token>
23 :
24 : #include <atomic>
25 : #include <coroutine>
26 : #include <new>
27 : #include <utility>
28 :
29 : /* async_event implementation notes
30 : =================================
31 :
32 : Same cancellation pattern as async_mutex (see that file for the
33 : full discussion on claimed_, stop_cb lifetime, member ordering,
34 : and threading assumptions).
35 :
36 : Key difference: set() wakes ALL waiters (broadcast), not one.
37 : It pops every waiter from the list and posts the ones it
38 : claims. Waiters already claimed by a stop callback are skipped.
39 :
40 : Because set() pops all waiters, a canceled waiter may have been
41 : removed from the list by set() before its await_resume runs.
42 : This requires a separate in_list_ flag (unlike async_mutex where
43 : active_ served double duty). await_resume only calls remove()
44 : when in_list_ is true.
45 : */
46 :
47 : namespace boost {
48 : namespace capy {
49 :
50 : /** Queues coroutines in `wait()` and resumes all of them when `set()` is called.
51 :
52 : This event provides a way to notify multiple coroutines that some
53 : condition has occurred. When a coroutine awaits an unset event, it
54 : suspends and is added to a wait queue. When the event is set, all
55 : waiting coroutines are resumed.
56 :
57 : @par Cancellation
58 :
59 : When a coroutine is suspended waiting for the event and its stop
60 : token is triggered, the waiter completes with `error::canceled`
61 : instead of waiting for `set()`.
62 :
63 : Cancellation only applies while the coroutine is suspended in the
64 : wait queue. If the event is already set when `wait()` is called,
65 : the wait completes immediately even if the stop token is already
66 : signaled.
67 :
68 : @par Zero Allocation
69 :
70 : No heap allocation occurs for wait operations.
71 :
72 : @par Thread Safety
73 :
74 : Distinct objects: Safe.@n
75 : Shared objects: Unsafe.
76 :
77 : The event operations are designed for single-threaded use on one
78 : executor. The stop callback may fire from any thread.
79 :
80 : This type is non-copyable and non-movable because suspended
81 : waiters hold intrusive pointers into the event's internal list.
82 :
83 : @par Example
84 : @code
85 : async_event event;
86 :
87 : task<> waiter() {
88 : auto [ec] = co_await event.wait();
89 : if(ec)
90 : co_return;
91 : // ... event was set ...
92 : }
93 :
94 : task<> notifier() {
95 : // ... do some work ...
96 : event.set(); // Wake all waiters
97 : }
98 : @endcode
99 : */
100 : class async_event
101 : {
102 : public:
103 : class wait_awaiter;
104 :
105 : private:
106 : bool set_ = false;
107 : detail::intrusive_list<wait_awaiter> waiters_;
108 :
109 : public:
110 : /** Suspends the caller until `set()` runs, or resumes it with `error::canceled` on a stop request.
111 : */
112 : class wait_awaiter
113 : : public detail::intrusive_list<wait_awaiter>::node
114 : {
115 : friend class async_event;
116 :
117 : async_event* e_;
118 : continuation cont_;
119 : executor_ref ex_;
120 :
121 : // Declared before stop_cb_buf_: the callback
122 : // accesses these members, so they must still be
123 : // alive if the stop_cb_ destructor blocks.
124 : std::atomic<bool> claimed_{false};
125 : bool canceled_ = false;
126 : bool active_ = false;
127 : bool in_list_ = false;
128 :
129 : struct cancel_fn
130 : {
131 : wait_awaiter* self_;
132 :
133 HIT 4 : void operator()() const noexcept
134 : {
135 4 : if(!self_->claimed_.exchange(
136 : true, std::memory_order_acq_rel))
137 : {
138 3 : self_->canceled_ = true;
139 3 : self_->ex_.post(self_->cont_);
140 : }
141 4 : }
142 : };
143 :
144 : using stop_cb_t =
145 : std::stop_callback<cancel_fn>;
146 :
147 : // Aligned storage for stop_cb_t. Declared last:
148 : // its destructor may block while the callback
149 : // accesses the members above.
150 : BOOST_CAPY_MSVC_WARNING_PUSH
151 : BOOST_CAPY_MSVC_WARNING_DISABLE(4324) // padded due to alignas
152 : alignas(stop_cb_t)
153 : unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
154 : BOOST_CAPY_MSVC_WARNING_POP
155 :
156 20 : stop_cb_t& stop_cb_() noexcept
157 : {
158 : return *reinterpret_cast<stop_cb_t*>(
159 20 : stop_cb_buf_);
160 : }
161 :
162 : public:
163 : /** Destroy the awaiter, leaving the event unable to reach it.
164 :
165 : Destroys the stop callback if one is registered, and unlinks
166 : the awaiter from the event's wait queue if it is still linked.
167 : Both are necessary when the coroutine frame is torn down while
168 : suspended, so that neither `set()` nor the stop callback can
169 : reach a destroyed awaiter.
170 : */
171 52 : ~wait_awaiter()
172 : {
173 52 : if(active_)
174 1 : stop_cb_().~stop_cb_t();
175 52 : if(in_list_)
176 1 : e_->waiters_.remove(this);
177 52 : }
178 :
179 : /** Construct an awaiter for the given event.
180 :
181 : @param e The event to wait on. It must outlive the awaiter.
182 : */
183 25 : explicit wait_awaiter(async_event* e) noexcept
184 25 : : e_(e)
185 : {
186 25 : }
187 :
188 : /** Construct by moving.
189 :
190 : The moved-from awaiter is left inert: its destructor no longer
191 : destroys the stop callback and no longer unlinks from the
192 : event's wait queue.
193 :
194 : @param o The awaiter to move from.
195 : */
196 27 : wait_awaiter(wait_awaiter&& o) noexcept
197 54 : : e_(o.e_)
198 27 : , cont_(o.cont_)
199 27 : , ex_(o.ex_)
200 27 : , claimed_(o.claimed_.load(
201 : std::memory_order_relaxed))
202 27 : , canceled_(o.canceled_)
203 27 : , active_(std::exchange(o.active_, false))
204 54 : , in_list_(std::exchange(o.in_list_, false))
205 : {
206 27 : }
207 :
208 : /** Copy construction is disabled; a waiter is linked into the
209 : event's wait queue by address.
210 :
211 : @param other The awaiter that would be copied.
212 : */
213 : wait_awaiter(wait_awaiter const& other) = delete;
214 :
215 : /** Copy assignment is disabled; a waiter is linked into the
216 : event's wait queue by address.
217 :
218 : @param other The awaiter that would be assigned from.
219 :
220 : @return A reference to `*this`.
221 : */
222 : wait_awaiter& operator=(wait_awaiter const& other) = delete;
223 :
224 : /** Move assignment is disabled; a waiter is linked into the
225 : event's wait queue by address.
226 :
227 : @param other The awaiter that would be moved from.
228 :
229 : @return A reference to `*this`.
230 : */
231 : wait_awaiter& operator=(wait_awaiter&& other) = delete;
232 :
233 : /** Report whether the event is already set.
234 :
235 : @return `true` if the event is set, in which case the awaiting
236 : coroutine does not suspend; otherwise `false`.
237 : */
238 25 : bool await_ready() const noexcept
239 : {
240 25 : return e_->set_;
241 : }
242 :
243 : /** Enqueue the awaiting coroutine until the event is set.
244 :
245 : This is the @ref IoAwaitable overload of `await_suspend`.
246 :
247 : If a stop request is already pending on `env->stop_token`, the
248 : awaiter records the cancellation and does not enqueue.
249 :
250 : Otherwise it stores `h` and `env->executor`, links itself into
251 : the event's wait queue, and registers a stop callback on
252 : `env->stop_token`. Exactly one of `set()` and that callback posts
253 : `h` through the stored executor, whichever claims the waiter
254 : first. Only the post is subject to that race. A losing stop
255 : callback does nothing at all, but `set()` unlinks every waiter it
256 : pops whether it claims it or not. That is why @ref await_resume
257 : unlinks a canceled waiter only when it is still linked.
258 :
259 : @param h The awaiting coroutine, resumed when the event is set
260 : or the wait is canceled.
261 :
262 : @param env The execution environment. Its executor posts the
263 : resumption and its stop token is watched for the duration of
264 : the wait. It must outlive the wait.
265 :
266 : @return `h` if a stop request was already pending, which
267 : resumes the awaiting coroutine immediately without enqueuing
268 : it. Otherwise `std::noop_coroutine()`, which leaves the
269 : coroutine suspended and returns control to the resumer.
270 : */
271 : std::coroutine_handle<>
272 21 : await_suspend(
273 : std::coroutine_handle<> h,
274 : io_env const* env) noexcept
275 : {
276 21 : if(env->stop_token.stop_requested())
277 : {
278 1 : canceled_ = true;
279 1 : return h;
280 : }
281 20 : cont_.h = h;
282 20 : ex_ = env->executor;
283 20 : e_->waiters_.push_back(this);
284 20 : in_list_ = true;
285 60 : ::new(stop_cb_buf_) stop_cb_t(
286 20 : env->stop_token, cancel_fn{this});
287 20 : active_ = true;
288 20 : return std::noop_coroutine();
289 : }
290 :
291 : /** Complete the wait and report the outcome.
292 :
293 : Destroys the stop callback if one is registered. If the wait
294 : was canceled while still linked into the event's wait queue,
295 : unlinks it. `set()` pops every waiter, so a canceled waiter may
296 : or may not still be linked when it resumes.
297 :
298 : @return An empty `io_result<>` if the event was set, or one
299 : holding `error::canceled` if the stop token fired first.
300 : */
301 22 : io_result<> await_resume() noexcept
302 : {
303 22 : if(active_)
304 : {
305 19 : stop_cb_().~stop_cb_t();
306 19 : active_ = false;
307 : }
308 22 : if(canceled_)
309 : {
310 4 : if(in_list_)
311 : {
312 3 : e_->waiters_.remove(this);
313 3 : in_list_ = false;
314 : }
315 : return {make_error_code(
316 4 : error::canceled)};
317 : }
318 18 : return {{}};
319 : }
320 : };
321 :
322 : /// Construct an unset event.
323 : async_event() = default;
324 :
325 : /** Copy construction is disabled; suspended waiters point into the
326 : event's wait queue.
327 :
328 : @param other The event that would be copied.
329 : */
330 : async_event(async_event const& other) = delete;
331 :
332 : /** Copy assignment is disabled; suspended waiters point into the
333 : event's wait queue.
334 :
335 : @param other The event that would be assigned from.
336 :
337 : @return A reference to `*this`.
338 : */
339 : async_event& operator=(async_event const& other) = delete;
340 :
341 : /** Move construction is disabled; suspended waiters point into the
342 : event's wait queue.
343 :
344 : @param other The event that would be moved from.
345 : */
346 : async_event(async_event&& other) = delete;
347 :
348 : /** Move assignment is disabled; suspended waiters point into the
349 : event's wait queue.
350 :
351 : @param other The event that would be moved from.
352 :
353 : @return A reference to `*this`.
354 : */
355 : async_event& operator=(async_event&& other) = delete;
356 :
357 : /** Returns an awaiter that waits until the event is set.
358 :
359 : If the event is already set, completes immediately.
360 :
361 : @return An awaitable that await-returns `(error_code)`.
362 : */
363 25 : wait_awaiter wait() noexcept
364 : {
365 25 : return wait_awaiter{this};
366 : }
367 :
368 : /** Resumes every waiting coroutine and marks the event set for later `wait()` calls.
369 :
370 : All waiting coroutines are resumed. Canceled waiters
371 : are skipped. Subsequent calls to wait() complete
372 : immediately until clear() is called.
373 : */
374 17 : void set()
375 : {
376 17 : set_ = true;
377 : for(;;)
378 : {
379 33 : auto* w = waiters_.pop_front();
380 33 : if(!w)
381 17 : break;
382 16 : w->in_list_ = false;
383 16 : if(!w->claimed_.exchange(
384 : true, std::memory_order_acq_rel))
385 : {
386 16 : w->ex_.post(w->cont_);
387 : }
388 16 : }
389 17 : }
390 :
391 : /** Clears the event.
392 :
393 : Subsequent calls to wait() suspend until
394 : set() is called again.
395 : */
396 2 : void clear() noexcept
397 : {
398 2 : set_ = false;
399 2 : }
400 :
401 : /** Returns true if the event is currently set.
402 :
403 : @return `true` if the event is set; otherwise `false`.
404 : */
405 9 : bool is_set() const noexcept
406 : {
407 9 : return set_;
408 : }
409 : };
410 :
411 : } // namespace capy
412 : } // namespace boost
413 :
414 : #endif
|
