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