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