. /include/boost/capy /ex /async_waker.hpp
92.96% Lines (66/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   @par !example example 103   @par !example example
104   104  
105   */ 105   */
106   class async_waker 106   class async_waker
107   { 107   {
108   public: 108   public:
109   class wait_awaiter; 109   class wait_awaiter;
110   110  
111   private: 111   private:
112   static constexpr int state_empty = 0; // no token, no waiter 112   static constexpr int state_empty = 0; // no token, no waiter
113   static constexpr int state_token = 1; // latched wakeup 113   static constexpr int state_token = 1; // latched wakeup
114   static constexpr int state_armed = 2; // waiter suspended 114   static constexpr int state_armed = 2; // waiter suspended
115   115  
116   std::atomic<int> st_{state_empty}; 116   std::atomic<int> st_{state_empty};
117   wait_awaiter* waiter_ = nullptr; 117   wait_awaiter* waiter_ = nullptr;
118   118  
119   public: 119   public:
120   /** Suspends the caller until `wake()` runs, or resumes it with `error::canceled` on a stop request. 120   /** Suspends the caller until `wake()` runs, or resumes it with `error::canceled` on a stop request.
121   */ 121   */
122   class wait_awaiter 122   class wait_awaiter
123   { 123   {
124   friend class async_waker; 124   friend class async_waker;
125   125  
126   async_waker* waker_; 126   async_waker* waker_;
127   continuation cont_; 127   continuation cont_;
128   executor_ref ex_; 128   executor_ref ex_;
129   129  
130   // Declared before stop_cb_buf_: the callback accesses 130   // Declared before stop_cb_buf_: the callback accesses
131   // these members, so they must still be alive if the 131   // these members, so they must still be alive if the
132   // stop_cb_ destructor blocks. 132   // stop_cb_ destructor blocks.
133   bool canceled_ = false; 133   bool canceled_ = false;
134   bool active_ = false; 134   bool active_ = false;
135   bool published_ = false; 135   bool published_ = false;
136   136  
137   struct cancel_fn 137   struct cancel_fn
138   { 138   {
139   wait_awaiter* self_; 139   wait_awaiter* self_;
140   140  
HITCBC 141   17 void operator()() const noexcept 141   7 void operator()() const noexcept
142   { 142   {
HITCBC 143   17 int expected = state_armed; 143   7 int expected = state_armed;
HITCBC 144   34 if(self_->waker_->st_.compare_exchange_strong( 144   14 if(self_->waker_->st_.compare_exchange_strong(
145   expected, state_empty, 145   expected, state_empty,
146   std::memory_order_acq_rel, 146   std::memory_order_acq_rel,
147   std::memory_order_acquire)) 147   std::memory_order_acquire))
148   { 148   {
HITCBC 149   7 self_->canceled_ = true; 149   7 self_->canceled_ = true;
HITCBC 150   7 self_->ex_.post(self_->cont_); 150   7 self_->ex_.post(self_->cont_);
151   } 151   }
HITCBC 152   17 } 152   7 }
153   }; 153   };
154   154  
155   using stop_cb_t = std::stop_callback<cancel_fn>; 155   using stop_cb_t = std::stop_callback<cancel_fn>;
156   156  
157   // Aligned storage for stop_cb_t. Declared last: its 157   // Aligned storage for stop_cb_t. Declared last: its
158   // destructor may block while the callback accesses the 158   // destructor may block while the callback accesses the
159   // members above. 159   // members above.
160   BOOST_CAPY_MSVC_WARNING_PUSH 160   BOOST_CAPY_MSVC_WARNING_PUSH
161   BOOST_CAPY_MSVC_WARNING_DISABLE(4324) 161   BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
162   alignas(stop_cb_t) 162   alignas(stop_cb_t)
163   unsigned char stop_cb_buf_[sizeof(stop_cb_t)]; 163   unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
164   BOOST_CAPY_MSVC_WARNING_POP 164   BOOST_CAPY_MSVC_WARNING_POP
165   165  
HITCBC 166   31 stop_cb_t& stop_cb_() noexcept 166   18 stop_cb_t& stop_cb_() noexcept
167   { 167   {
HITCBC 168   31 return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_); 168   18 return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_);
169   } 169   }
170   170  
171   public: 171   public:
172   /** Destroy the awaiter, leaving the waker unable to reach it. 172   /** Destroy the awaiter, leaving the waker unable to reach it.
173   173  
174   Destroys the stop callback if one is registered. If the awaiter 174   Destroys the stop callback if one is registered. If the awaiter
175   is still armed, it also returns the waker's slot to the empty 175   is still armed, it also returns the waker's slot to the empty
176   state, so a later `wake()` cannot dereference a destroyed 176   state, so a later `wake()` cannot dereference a destroyed
177   awaiter. That case means the frame is being torn down without 177   awaiter. That case means the frame is being torn down without
178   ever being resumed; a wake arriving afterward latches a token 178   ever being resumed; a wake arriving afterward latches a token
179   instead. 179   instead.
180   */ 180   */
HITCBC 181   294 ~wait_awaiter() 181   308 ~wait_awaiter()
182   { 182   {
HITCBC 183   294 if(active_) 183   308 if(active_)
HITCBC 184   1 stop_cb_().~stop_cb_t(); 184   1 stop_cb_().~stop_cb_t();
HITCBC 185   294 if(published_) 185   308 if(published_)
186   { 186   {
187   // Destroyed while still armed (frame torn down 187   // Destroyed while still armed (frame torn down
188   // without resuming): deregister so a later 188   // without resuming): deregister so a later
189   // wake cannot touch the dead frame. 189   // wake cannot touch the dead frame.
HITCBC 190   1 int expected = state_armed; 190   1 int expected = state_armed;
HITCBC 191   1 waker_->st_.compare_exchange_strong( 191   1 waker_->st_.compare_exchange_strong(
192   expected, state_empty, 192   expected, state_empty,
193   std::memory_order_acq_rel, 193   std::memory_order_acq_rel,
194   std::memory_order_acquire); 194   std::memory_order_acquire);
195   } 195   }
HITCBC 196   294 } 196   308 }
197   197  
198   /** Construct an awaiter for the given waker. 198   /** Construct an awaiter for the given waker.
199   199  
200   @param waker The waker to wait on. It must outlive the awaiter. 200   @param waker The waker to wait on. It must outlive the awaiter.
201   */ 201   */
HITCBC 202   147 explicit wait_awaiter(async_waker* waker) noexcept 202   154 explicit wait_awaiter(async_waker* waker) noexcept
HITCBC 203   147 : waker_(waker) 203   154 : waker_(waker)
204   { 204   {
HITCBC 205   147 } 205   154 }
206   206  
207   /** Construct by moving. 207   /** Construct by moving.
208   208  
209   The moved-from awaiter is left inert: its destructor no longer 209   The moved-from awaiter is left inert: its destructor no longer
210   destroys the stop callback and no longer deregisters from the 210   destroys the stop callback and no longer deregisters from the
211   waker. 211   waker.
212   212  
213   @param o The awaiter to move from. 213   @param o The awaiter to move from.
214   */ 214   */
HITCBC 215   147 wait_awaiter(wait_awaiter&& o) noexcept 215   154 wait_awaiter(wait_awaiter&& o) noexcept
HITCBC 216   147 : waker_(o.waker_) 216   154 : waker_(o.waker_)
HITCBC 217   147 , cont_(o.cont_) 217   154 , cont_(o.cont_)
HITCBC 218   147 , ex_(o.ex_) 218   154 , ex_(o.ex_)
HITCBC 219   147 , canceled_(o.canceled_) 219   154 , canceled_(o.canceled_)
HITCBC 220   147 , active_(std::exchange(o.active_, false)) 220   154 , active_(std::exchange(o.active_, false))
HITCBC 221   147 , published_(std::exchange(o.published_, false)) 221   154 , published_(std::exchange(o.published_, false))
222   { 222   {
HITCBC 223   147 } 223   154 }
224   224  
225   /** Copy construction is disabled; an armed waiter is registered 225   /** Copy construction is disabled; an armed waiter is registered
226   with the waker by address. 226   with the waker by address.
227   227  
228   @param other The awaiter that would be copied. 228   @param other The awaiter that would be copied.
229   */ 229   */
230   wait_awaiter(wait_awaiter const& other) = delete; 230   wait_awaiter(wait_awaiter const& other) = delete;
231   231  
232   /** Copy assignment is disabled; an armed waiter is registered 232   /** Copy assignment is disabled; an armed waiter is registered
233   with the waker by address. 233   with the waker by address.
234   234  
235   @param other The awaiter that would be assigned from. 235   @param other The awaiter that would be assigned from.
236   236  
237   @return A reference to `*this`. 237   @return A reference to `*this`.
238   */ 238   */
239   wait_awaiter& operator=(wait_awaiter const& other) = delete; 239   wait_awaiter& operator=(wait_awaiter const& other) = delete;
240   240  
241   /** Move assignment is disabled; an armed waiter is registered 241   /** Move assignment is disabled; an armed waiter is registered
242   with the waker by address. 242   with the waker by address.
243   243  
244   @param other The awaiter that would be moved from. 244   @param other The awaiter that would be moved from.
245   245  
246   @return A reference to `*this`. 246   @return A reference to `*this`.
247   */ 247   */
248   wait_awaiter& operator=(wait_awaiter&& other) = delete; 248   wait_awaiter& operator=(wait_awaiter&& other) = delete;
249   249  
250   /** Consume a latched token, completing synchronously. 250   /** Consume a latched token, completing synchronously.
251   251  
252   This is not a pure query: the check is a compare-exchange that 252   This is not a pure query: the check is a compare-exchange that
253   takes the token. Calling it twice is not idempotent: the second 253   takes the token. Calling it twice is not idempotent: the second
254   call reports `false`, because the first already consumed the 254   call reports `false`, because the first already consumed the
255   wakeup. 255   wakeup.
256   256  
257   @return `true` if a pending wakeup token was latched and has now 257   @return `true` if a pending wakeup token was latched and has now
258   been consumed, in which case the awaiting coroutine does not 258   been consumed, in which case the awaiting coroutine does not
259   suspend; otherwise `false`. 259   suspend; otherwise `false`.
260   */ 260   */
HITCBC 261   147 bool await_ready() noexcept 261   154 bool await_ready() noexcept
262   { 262   {
HITCBC 263   147 int expected = state_token; 263   154 int expected = state_token;
HITCBC 264   147 return waker_->st_.compare_exchange_strong( 264   154 return waker_->st_.compare_exchange_strong(
265   expected, state_empty, 265   expected, state_empty,
266   std::memory_order_acq_rel, 266   std::memory_order_acq_rel,
HITCBC 267   147 std::memory_order_acquire); 267   154 std::memory_order_acquire);
268   } 268   }
269   269  
270   /** Arm the waker with the awaiting coroutine. 270   /** Arm the waker with the awaiting coroutine.
271   271  
272   This is the @ref IoAwaitable overload of `await_suspend`. 272   This is the @ref IoAwaitable overload of `await_suspend`.
273   Unlike `async_event` and `async_mutex`, it has three outcomes, 273   Unlike `async_event` and `async_mutex`, it has three outcomes,
274   because a `wake()` from another thread can land in the window 274   because a `wake()` from another thread can land in the window
275   between `await_ready` and this call. 275   between `await_ready` and this call.
276   276  
277   @li A stop request is already pending on `env->stop_token`: the 277   @li A stop request is already pending on `env->stop_token`: the
278   awaiter records the cancellation and does not arm. 278   awaiter records the cancellation and does not arm.
279   279  
280   @li The waker's slot is no longer empty. Under the single-waiter 280   @li The waker's slot is no longer empty. Under the single-waiter
281   precondition that means a wakeup was latched after 281   precondition that means a wakeup was latched after
282   `await_ready` looked, so the token is consumed here instead 282   `await_ready` looked, so the token is consumed here instead
283   and the wait succeeds. 283   and the wait succeeds.
284   284  
285   @li Otherwise the slot moves to the armed state, publishing this 285   @li Otherwise the slot moves to the armed state, publishing this
286   awaiter to the waker, and a stop callback is registered on 286   awaiter to the waker, and a stop callback is registered on
287   `env->stop_token`. Whichever of `wake()` and that callback 287   `env->stop_token`. Whichever of `wake()` and that callback
288   wins the armed-to-empty transition posts `h` through 288   wins the armed-to-empty transition posts `h` through
289   `env->executor`. The loser does nothing, and a losing 289   `env->executor`. The loser does nothing, and a losing
290   `wake()` re-latches its token for the next `wait()`. 290   `wake()` re-latches its token for the next `wait()`.
291   291  
292   @param h The awaiting coroutine, resumed when the waker fires 292   @param h The awaiting coroutine, resumed when the waker fires
293   or the wait is canceled. 293   or the wait is canceled.
294   294  
295   @param env The execution environment. Its executor posts the 295   @param env The execution environment. Its executor posts the
296   resumption and its stop token is watched for the duration of 296   resumption and its stop token is watched for the duration of
297   the wait. It must outlive the wait. 297   the wait. It must outlive the wait.
298   298  
299   @return `h` in the first two cases, which resumes the awaiting 299   @return `h` in the first two cases, which resumes the awaiting
300   coroutine immediately; otherwise `std::noop_coroutine()`, which 300   coroutine immediately; otherwise `std::noop_coroutine()`, which
301   leaves the coroutine suspended and returns control to the 301   leaves the coroutine suspended and returns control to the
302   resumer. 302   resumer.
303   */ 303   */
304   std::coroutine_handle<> 304   std::coroutine_handle<>
HITCBC 305   64 await_suspend( 305   58 await_suspend(
306   std::coroutine_handle<> h, 306   std::coroutine_handle<> h,
307   io_env const* env) noexcept 307   io_env const* env) noexcept
308   { 308   {
HITCBC 309   64 if(env->stop_token.stop_requested()) 309   58 if(env->stop_token.stop_requested())
310   { 310   {
HITCBC 311   33 canceled_ = true; 311   40 canceled_ = true;
HITCBC 312   33 return h; 312   40 return h;
313   } 313   }
HITCBC 314   31 cont_.h = h; 314   18 cont_.h = h;
HITCBC 315   31 ex_ = env->executor; 315   18 ex_ = env->executor;
HITCBC 316   31 waker_->waiter_ = this; 316   18 waker_->waiter_ = this;
317   317  
HITCBC 318   31 int expected = state_empty; 318   18 int expected = state_empty;
HITCBC 319   62 if(!waker_->st_.compare_exchange_strong( 319   36 if(!waker_->st_.compare_exchange_strong(
320   expected, state_armed, 320   expected, state_armed,
321   std::memory_order_acq_rel, 321   std::memory_order_acq_rel,
322   std::memory_order_acquire)) 322   std::memory_order_acquire))
323   { 323   {
324   // Single-waiter precondition: a second concurrent 324   // Single-waiter precondition: a second concurrent
325   // wait would find the slot armed. 325   // wait would find the slot armed.
MISUBC 326   ✗ BOOST_CAPY_ASSERT(expected == state_token); 326   ✗ BOOST_CAPY_ASSERT(expected == state_token);
327   327  
328   // A wake latched between await_ready and here; 328   // A wake latched between await_ready and here;
329   // consume it and resume inline. 329   // consume it and resume inline.
MISUBC 330   ✗ waker_->st_.store( 330   ✗ waker_->st_.store(
331   state_empty, std::memory_order_release); 331   state_empty, std::memory_order_release);
MISUBC 332   ✗ return h; 332   ✗ return h;
333   } 333   }
HITCBC 334   31 published_ = true; 334   18 published_ = true;
335   335  
HITCBC 336   93 ::new(stop_cb_buf_) stop_cb_t( 336   54 ::new(stop_cb_buf_) stop_cb_t(
HITCBC 337   31 env->stop_token, cancel_fn{this}); 337   18 env->stop_token, cancel_fn{this});
HITCBC 338   31 active_ = true; 338   18 active_ = true;
HITCBC 339   31 return std::noop_coroutine(); 339   18 return std::noop_coroutine();
340   } 340   }
341   341  
342   /** Complete the wait and report the outcome. 342   /** Complete the wait and report the outcome.
343   343  
344   Destroys the stop callback if one is registered and clears the 344   Destroys the stop callback if one is registered and clears the
345   armed bookkeeping, so the destructor does not deregister a slot 345   armed bookkeeping, so the destructor does not deregister a slot
346   the resumption already consumed. 346   the resumption already consumed.
347   347  
348   @return An empty `io_result<>` if the wait was woken, whether by 348   @return An empty `io_result<>` if the wait was woken, whether by
349   `wake()` or by a token consumed inline. Otherwise one holding 349   `wake()` or by a token consumed inline. Otherwise one holding
350   `error::canceled`, which means the stop token won the race. 350   `error::canceled`, which means the stop token won the race.
351   */ 351   */
HITCBC 352   146 [[nodiscard]] io_result<> await_resume() noexcept 352   153 [[nodiscard]] io_result<> await_resume() noexcept
353   { 353   {
HITCBC 354   146 if(active_) 354   153 if(active_)
355   { 355   {
HITCBC 356   30 stop_cb_().~stop_cb_t(); 356   17 stop_cb_().~stop_cb_t();
HITCBC 357   30 active_ = false; 357   17 active_ = false;
358   } 358   }
HITCBC 359   146 published_ = false; 359   153 published_ = false;
HITCBC 360   146 if(canceled_) 360   153 if(canceled_)
HITCBC 361   39 return {make_error_code(error::canceled)}; 361   46 return {make_error_code(error::canceled)};
HITCBC 362   107 return {{}}; 362   107 return {{}};
363   } 363   }
364   }; 364   };
365   365  
366   /// Construct with no token latched. 366   /// Construct with no token latched.
HITCBC 367   1 async_waker() = default; 367   1 async_waker() = default;
368   368  
369   /** Copy construction is disabled; an armed waiter points into the 369   /** Copy construction is disabled; an armed waiter points into the
370   waker. 370   waker.
371   371  
372   @param other The waker that would be copied. 372   @param other The waker that would be copied.
373   */ 373   */
374   async_waker(async_waker const& other) = delete; 374   async_waker(async_waker const& other) = delete;
375   375  
376   /** Copy assignment is disabled; an armed waiter points into the waker. 376   /** Copy assignment is disabled; an armed waiter points into the waker.
377   377  
378   @param other The waker that would be assigned from. 378   @param other The waker that would be assigned from.
379   379  
380   @return A reference to `*this`. 380   @return A reference to `*this`.
381   */ 381   */
382   async_waker& operator=(async_waker const& other) = delete; 382   async_waker& operator=(async_waker const& other) = delete;
383   383  
384   /** Move construction is disabled; an armed waiter points into the 384   /** Move construction is disabled; an armed waiter points into the
385   waker. 385   waker.
386   386  
387   @param other The waker that would be moved from. 387   @param other The waker that would be moved from.
388   */ 388   */
389   async_waker(async_waker&& other) = delete; 389   async_waker(async_waker&& other) = delete;
390   390  
391   /** Move assignment is disabled; an armed waiter points into the waker. 391   /** Move assignment is disabled; an armed waiter points into the waker.
392   392  
393   @param other The waker that would be moved from. 393   @param other The waker that would be moved from.
394   394  
395   @return A reference to `*this`. 395   @return A reference to `*this`.
396   */ 396   */
397   async_waker& operator=(async_waker&& other) = delete; 397   async_waker& operator=(async_waker&& other) = delete;
398   398  
399   /** Asynchronously wait until woken. 399   /** Asynchronously wait until woken.
400   400  
401   If a token is latched, completes immediately and consumes 401   If a token is latched, completes immediately and consumes
402   it. Otherwise suspends until `wake()` or the stop token 402   it. Otherwise suspends until `wake()` or the stop token
403   fires. 403   fires.
404   404  
405   @par Preconditions 405   @par Preconditions
406   No other coroutine is currently waiting on this object. 406   No other coroutine is currently waiting on this object.
407   407  
408   @return An awaitable that await-returns `io_result<>`; 408   @return An awaitable that await-returns `io_result<>`;
409   empty on wakeup, `error::canceled` if the stop 409   empty on wakeup, `error::canceled` if the stop
410   token wins. 410   token wins.
411   */ 411   */
HITCBC 412   147 wait_awaiter wait() noexcept 412   154 wait_awaiter wait() noexcept
413   { 413   {
HITCBC 414   147 return wait_awaiter{this}; 414   154 return wait_awaiter{this};
415   } 415   }
416   416  
417   /** Wake the waiter, or latch the wakeup if none waits. 417   /** Wake the waiter, or latch the wakeup if none waits.
418   418  
419   Callable from any thread. The waiter's resumption is 419   Callable from any thread. The waiter's resumption is
420   posted through its executor; this call never resumes a 420   posted through its executor; this call never resumes a
421   coroutine inline. Multiple calls without an intervening 421   coroutine inline. Multiple calls without an intervening
422   `wait()` collapse into a single token. 422   `wait()` collapse into a single token.
423   */ 423   */
HITCBC 424   109 void wake() noexcept 424   109 void wake() noexcept
425   { 425   {
426   for(;;) 426   for(;;)
427   { 427   {
HITCBC 428   109 int s = st_.load(std::memory_order_acquire); 428   109 int s = st_.load(std::memory_order_acquire);
HITCBC 429   109 if(s == state_token) 429   109 if(s == state_token)
HITCBC 430   109 return; 430   109 return;
HITCBC 431   107 if(s == state_empty) 431   107 if(s == state_empty)
432   { 432   {
HITCBC 433   166 if(st_.compare_exchange_weak( 433   192 if(st_.compare_exchange_weak(
434   s, state_token, 434   s, state_token,
435   std::memory_order_acq_rel, 435   std::memory_order_acq_rel,
436   std::memory_order_acquire)) 436   std::memory_order_acquire))
HITCBC 437   83 return; 437   96 return;
MISUBC 438   ✗ continue; 438   ✗ continue;
439   } 439   }
440   // armed: winning this CAS claims the waiter, whose 440   // armed: winning this CAS claims the waiter, whose
441   // frame is pinned until we post its resumption. 441   // frame is pinned until we post its resumption.
HITCBC 442   48 if(st_.compare_exchange_weak( 442   22 if(st_.compare_exchange_weak(
443   s, state_empty, 443   s, state_empty,
444   std::memory_order_acq_rel, 444   std::memory_order_acq_rel,
445   std::memory_order_acquire)) 445   std::memory_order_acquire))
446   { 446   {
HITCBC 447   24 auto* w = waiter_; 447   11 auto* w = waiter_;
HITCBC 448   24 w->ex_.post(w->cont_); 448   11 w->ex_.post(w->cont_);
HITCBC 449   24 return; 449   11 return;
450   } 450   }
MISUBC 451   ✗ } 451   ✗ }
452   } 452   }
453   }; 453   };
454   454  
455   } // namespace capy 455   } // namespace capy
456   } // namespace boost 456   } // namespace boost
457   457  
458   #endif 458   #endif