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