. /include/boost/capy /ex /async_waker.hpp
86.79% Lines (92/106) 100.00% Functions (13/13)
[ ] 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 <thread>
25   #include <utility> 26   #include <utility>
26   27  
27   /* async_waker implementation notes 28   /* async_waker implementation notes
28   =================================== 29   ===================================
29   30  
30   wake() must be callable from foreign threads (that is the whole 31   wake() must be callable from foreign threads (that is the whole
31   point: the user's thread provides the timing). A waiter-side 32   point: the user's thread provides the timing). A waiter-side
32   claimed_ flag is not enough there -- the 33   claimed_ flag is not enough there -- the
33   waker has to dereference the waiter, and nothing would pin the 34   waker has to dereference the waiter, and nothing would pin the
34   waiter's frame between reading the pointer and claiming it. 35   waiter's frame between reading the pointer and claiming it.
35   36  
36 - So the three-state st_ atomic is the single arbiter: 37 + So the st_ atomic is the single arbiter:
37   38  
38 - empty --arm--> armed --wake/cancel CAS--> empty 39 + empty --arm--> arming --publish--> armed --wake/cancel CAS--> empty
  40 + arming --stop already requested--> empty
39   empty --wake--> token --wait consumes--> empty 41   empty --wake--> token --wait consumes--> empty
40   42  
41   Whoever wins the armed->empty CAS owns the resume and may 43   Whoever wins the armed->empty CAS owns the resume and may
42   dereference waiter_: the frame cannot die underneath the 44   dereference waiter_: the frame cannot die underneath the
43   winner because the coroutine only resumes when the winner 45   winner because the coroutine only resumes when the winner
44   posts it. The loser never touches the waiter. When the stop 46   posts it. The loser never touches the waiter. When the stop
45   callback wins, a concurrent wake retries, finds empty, and 47   callback wins, a concurrent wake retries, finds empty, and
46   latches a token -- a racing wakeup is deferred, never lost. 48   latches a token -- a racing wakeup is deferred, never lost.
47   49  
48 - Serialized resumption is required: await_suspend keeps 50 + The stop callback is registered before the waiter enters the
49 - writing after the publishing armed-CAS (the stop_cb 51 + arming state, and a wake() winner destroys it before posting.
50 - placement-new and active_ = true), so a wake/cancel winner 52 + While arming, nothing can claim the waiter: wake(), the stop
51 - can post the continuation while that tail is still running. 53 + callback and the destructor wait the few instructions it takes.
52 - The posted resume must be ordered after await_suspend's 54 + So the waiter settles a stop requested during registration
53 - return, which holds on a single-threaded executor (the one 55 + itself, its last touch of the waker is the publishing store,
54 - thread is still inside await_suspend) and on a strand (the 56 + and the waker can die right after the wake() that claimed it.
55 - resume is a later turn, synchronized with the current one). 57 +
56 - A raw multi-threaded executor lets another worker run 58 + Serialized resumption is still required: the executor must not
57 - await_resume against those in-flight writes. async_event and 59 + run the waiter's continuations concurrently, since a winner can
58 - async_mutex make the same assumption; it is stated explicitly 60 + post the waiter before its await_suspend has returned. async_event
  61 + and async_mutex make the same assumption; it is stated explicitly
59   here because wake() invites foreign threads into the picture. 62   here because wake() invites foreign threads into the picture.
60   */ 63   */
61   64  
62   namespace boost { 65   namespace boost {
63   namespace capy { 66   namespace capy {
64   67  
65   /** A single-slot waker that hands one wakeup to a waiting coroutine. 68   /** A single-slot waker that hands one wakeup to a waiting coroutine.
66   69  
67   This is the escape hatch for timing and other external events: 70   This is the escape hatch for timing and other external events:
68   the user provides the thread and the clock, capy provides the 71   the user provides the thread and the clock, capy provides the
69   suspension point. One coroutine suspends in `wait()`; any 72   suspension point. One coroutine suspends in `wait()`; any
70   thread wakes it with `wake()`. 73   thread wakes it with `wake()`.
71   74  
72   A wakeup with no waiter present is latched as a single pending 75   A wakeup with no waiter present is latched as a single pending
73   token, and the next `wait()` consumes it immediately. This 76   token, and the next `wait()` consumes it immediately. This
74   makes the wake-before-wait race benign without any lock 77   makes the wake-before-wait race benign without any lock
75   protocol. Multiple wakes collapse into one token. 78   protocol. Multiple wakes collapse into one token.
76   79  
77   @par Cancellation 80   @par Cancellation
78   81  
79   If the environment's stop token is triggered while suspended, 82   If the environment's stop token is triggered while suspended,
80   the wait completes with `error::canceled`. A wake that loses 83   the wait completes with `error::canceled`. A wake that loses
81   the race against cancellation is latched for the next `wait()` 84   the race against cancellation is latched for the next `wait()`
82   rather than dropped. 85   rather than dropped.
83   86  
84   @par Zero Allocation 87   @par Zero Allocation
85   88  
86   No heap allocation occurs for wait or wake operations. 89   No heap allocation occurs for wait or wake operations.
87   90  
88   @par Thread Safety 91   @par Thread Safety
89   92  
90   Distinct objects: Safe.@n 93   Distinct objects: Safe.@n
91   Shared objects: `wake()` may be called from any thread. 94   Shared objects: `wake()` may be called from any thread.
92   `wait()` must only be awaited by one coroutine at a time. The 95   `wait()` must only be awaited by one coroutine at a time. The
93   executor must never run the coroutine's continuations 96   executor must never run the coroutine's continuations
94   concurrently: use a single-threaded executor, or a strand over 97   concurrently: use a single-threaded executor, or a strand over
95   a multi-threaded one. That is the same threading model as 98   a multi-threaded one. That is the same threading model as
96   `async_event` and `async_mutex`. Awaiting `wait()` directly 99   `async_event` and `async_mutex`. Awaiting `wait()` directly
97   on a multi-threaded executor is undefined. 100   on a multi-threaded executor is undefined.
98   101  
99   This type is non-copyable and non-movable because a suspended 102   This type is non-copyable and non-movable because a suspended
100   waiter holds a pointer into the object. 103   waiter holds a pointer into the object.
101   104  
102   @par Example 105   @par Example
103   @par !example example 106   @par !example example
104   107  
105   */ 108   */
106   class async_waker 109   class async_waker
107   { 110   {
108   public: 111   public:
109   class wait_awaiter; 112   class wait_awaiter;
110   113  
111   private: 114   private:
112 - static constexpr int state_empty = 0; // no token, no waiter 115 + static constexpr int state_empty = 0; // no token, no waiter
113 - static constexpr int state_token = 1; // latched wakeup 116 + static constexpr int state_token = 1; // latched wakeup
114 - static constexpr int state_armed = 2; // waiter suspended 117 + static constexpr int state_armed = 2; // waiter suspended
  118 + static constexpr int state_arming = 3; // waiter publishing itself
115   119  
116   std::atomic<int> st_{state_empty}; 120   std::atomic<int> st_{state_empty};
117   wait_awaiter* waiter_ = nullptr; 121   wait_awaiter* waiter_ = nullptr;
118   122  
  123 + // Take the armed waiter, waiting out one still publishing itself.
HITGNC   124 + 2427 bool claim_armed() noexcept
  125 + {
HITGNC   126 + 2427 int expected = state_armed;
  127 + for(;;)
  128 + {
HITGNC   129 + 4854 if(st_.compare_exchange_weak(
  130 + expected, state_empty,
  131 + std::memory_order_acq_rel,
  132 + std::memory_order_acquire))
HITGNC   133 + 2011 return true;
HITGNC   134 + 416 if(expected == state_armed)
  135 + continue; // LCOV_EXCL_LINE spurious weak CAS failure
HITGNC   136 + 416 if(expected != state_arming)
HITGNC   137 + 416 return false;
  138 + std::this_thread::yield(); // LCOV_EXCL_LINE race with a waiter still publishing itself
  139 + expected = state_armed; // LCOV_EXCL_LINE race with a waiter still publishing itself
  140 + }
  141 + }
  142 +
119   public: 143   public:
120   /** Suspends the caller until `wake()` runs, or resumes it with `error::canceled` on a stop request. 144   /** Suspends the caller until `wake()` runs, or resumes it with `error::canceled` on a stop request.
121   */ 145   */
122   class wait_awaiter 146   class wait_awaiter
123   { 147   {
124   friend class async_waker; 148   friend class async_waker;
125   149  
126   async_waker* waker_; 150   async_waker* waker_;
127   continuation cont_; 151   continuation cont_;
128   executor_ref ex_; 152   executor_ref ex_;
129   153  
130   // Declared before stop_cb_buf_: the callback accesses 154   // Declared before stop_cb_buf_: the callback accesses
131   // these members, so they must still be alive if the 155   // these members, so they must still be alive if the
132   // stop_cb_ destructor blocks. 156   // stop_cb_ destructor blocks.
133   bool canceled_ = false; 157   bool canceled_ = false;
134   bool active_ = false; 158   bool active_ = false;
135   bool published_ = false; 159   bool published_ = false;
136   160  
  161 + // Set by whoever wins the armed->empty transition, before it
  162 + // posts. A claimed waiter belongs to the executor's queue and
  163 + // may be discarded after the waker is gone, so its destructor
  164 + // must not touch the waker.
  165 + std::atomic<bool> claimed_{false};
  166 +
137   struct cancel_fn 167   struct cancel_fn
138   { 168   {
139   wait_awaiter* self_; 169   wait_awaiter* self_;
140   170  
HITCBC 141   20 void operator()() const noexcept 171   2009 void operator()() const noexcept
142   { 172   {
HITCBC 143 - 20 int expected = state_armed; 173 + 2009 if(self_->waker_->claim_armed())
DCB 144 - 40 if(self_->waker_->st_.compare_exchange_strong(  
145 - expected, state_empty,  
146 - std::memory_order_acq_rel,  
147 - std::memory_order_acquire))  
148   { 174   {
HITCBC 149   17 self_->canceled_ = true; 175   2009 self_->canceled_ = true;
HITGNC   176 + 2009 self_->claimed_.store(
  177 + true, std::memory_order_release);
HITCBC 150   17 self_->ex_.post(self_->cont_); 178   2009 self_->ex_.post(self_->cont_);
151   } 179   }
HITCBC 152   20 } 180   2009 }
153   }; 181   };
154   182  
155   using stop_cb_t = std::stop_callback<cancel_fn>; 183   using stop_cb_t = std::stop_callback<cancel_fn>;
156   184  
157   // Aligned storage for stop_cb_t. Declared last: its 185   // Aligned storage for stop_cb_t. Declared last: its
158   // destructor may block while the callback accesses the 186   // destructor may block while the callback accesses the
159   // members above. 187   // members above.
160   BOOST_CAPY_MSVC_WARNING_PUSH 188   BOOST_CAPY_MSVC_WARNING_PUSH
161   BOOST_CAPY_MSVC_WARNING_DISABLE(4324) 189   BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
162   alignas(stop_cb_t) 190   alignas(stop_cb_t)
163   unsigned char stop_cb_buf_[sizeof(stop_cb_t)]; 191   unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
164   BOOST_CAPY_MSVC_WARNING_POP 192   BOOST_CAPY_MSVC_WARNING_POP
165   193  
HITCBC 166   49 stop_cb_t& stop_cb_() noexcept 194   2325 stop_cb_t& stop_cb_() noexcept
167   { 195   {
HITCBC 168   49 return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_); 196   2325 return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_);
169   } 197   }
170   198  
171   public: 199   public:
172   /** Destroy the awaiter, leaving the waker unable to reach it. 200   /** Destroy the awaiter, leaving the waker unable to reach it.
173   201  
174   Destroys the stop callback if one is registered. If the awaiter 202   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 203   is still armed, it also returns the waker's slot to the empty
176   state, so a later `wake()` cannot dereference a destroyed 204   state, so a later `wake()` cannot dereference a destroyed
177   awaiter. That case means the frame is being torn down without 205   awaiter. That case means the frame is being torn down without
178   ever being resumed; a wake arriving afterward latches a token 206   ever being resumed; a wake arriving afterward latches a token
179   instead. 207   instead.
180   */ 208   */
HITCBC 181   294 ~wait_awaiter() 209   4918 ~wait_awaiter()
182   { 210   {
HITCBC 183   294 if(active_) 211   4918 if(active_)
HITCBC 184   1 stop_cb_().~stop_cb_t(); 212   3 stop_cb_().~stop_cb_t();
HITCBC 185 - 294 if(published_) 213 + 4923 if(published_ &&
HITGNC   214 + 5 !claimed_.load(std::memory_order_acquire))
186   { 215   {
187   // Destroyed while still armed (frame torn down 216   // Destroyed while still armed (frame torn down
188   // without resuming): deregister so a later 217   // without resuming): deregister so a later
189   // wake cannot touch the dead frame. 218   // wake cannot touch the dead frame.
HITCBC 190   1 int expected = state_armed; 219   1 int expected = state_armed;
HITCBC 191   1 waker_->st_.compare_exchange_strong( 220   1 waker_->st_.compare_exchange_strong(
192   expected, state_empty, 221   expected, state_empty,
193   std::memory_order_acq_rel, 222   std::memory_order_acq_rel,
194   std::memory_order_acquire); 223   std::memory_order_acquire);
195   } 224   }
HITCBC 196   294 } 225   4918 }
197   226  
198   /** Construct an awaiter for the given waker. 227   /** Construct an awaiter for the given waker.
199   228  
200   @param waker The waker to wait on. It must outlive the awaiter. 229   @param waker The waker to wait on. It must outlive the awaiter.
201   */ 230   */
HITCBC 202   147 explicit wait_awaiter(async_waker* waker) noexcept 231   2459 explicit wait_awaiter(async_waker* waker) noexcept
HITCBC 203   147 : waker_(waker) 232   2459 : waker_(waker)
204   { 233   {
HITCBC 205   147 } 234   2459 }
206   235  
207   /** Construct by moving. 236   /** Construct by moving.
208   237  
209   The moved-from awaiter is left inert: its destructor no longer 238   The moved-from awaiter is left inert: its destructor no longer
210   destroys the stop callback and no longer deregisters from the 239   destroys the stop callback and no longer deregisters from the
211   waker. 240   waker.
212   241  
213   @param o The awaiter to move from. 242   @param o The awaiter to move from.
214   */ 243   */
HITCBC 215   147 wait_awaiter(wait_awaiter&& o) noexcept 244   2459 wait_awaiter(wait_awaiter&& o) noexcept
HITCBC 216   147 : waker_(o.waker_) 245   2459 : waker_(o.waker_)
HITCBC 217   147 , cont_(o.cont_) 246   2459 , cont_(o.cont_)
HITCBC 218   147 , ex_(o.ex_) 247   2459 , ex_(o.ex_)
HITCBC 219   147 , canceled_(o.canceled_) 248   2459 , canceled_(o.canceled_)
HITCBC 220   147 , active_(std::exchange(o.active_, false)) 249   2459 , active_(std::exchange(o.active_, false))
HITCBC 221   147 , published_(std::exchange(o.published_, false)) 250   2459 , published_(std::exchange(o.published_, false))
HITGNC   251 + 2459 , claimed_(o.claimed_.load(std::memory_order_relaxed))
222   { 252   {
HITCBC 223   147 } 253   2459 }
224   254  
225   /** Copy construction is disabled; an armed waiter is registered 255   /** Copy construction is disabled; an armed waiter is registered
226   with the waker by address. 256   with the waker by address.
227   257  
228   @param other The awaiter that would be copied. 258   @param other The awaiter that would be copied.
229   */ 259   */
230   wait_awaiter(wait_awaiter const& other) = delete; 260   wait_awaiter(wait_awaiter const& other) = delete;
231   261  
232   /** Copy assignment is disabled; an armed waiter is registered 262   /** Copy assignment is disabled; an armed waiter is registered
233   with the waker by address. 263   with the waker by address.
234   264  
235   @param other The awaiter that would be assigned from. 265   @param other The awaiter that would be assigned from.
236   266  
237   @return A reference to `*this`. 267   @return A reference to `*this`.
238   */ 268   */
239   wait_awaiter& operator=(wait_awaiter const& other) = delete; 269   wait_awaiter& operator=(wait_awaiter const& other) = delete;
240   270  
241   /** Move assignment is disabled; an armed waiter is registered 271   /** Move assignment is disabled; an armed waiter is registered
242   with the waker by address. 272   with the waker by address.
243   273  
244   @param other The awaiter that would be moved from. 274   @param other The awaiter that would be moved from.
245   275  
246   @return A reference to `*this`. 276   @return A reference to `*this`.
247   */ 277   */
248   wait_awaiter& operator=(wait_awaiter&& other) = delete; 278   wait_awaiter& operator=(wait_awaiter&& other) = delete;
249   279  
250   /** Consume a latched token, completing synchronously. 280   /** Consume a latched token, completing synchronously.
251   281  
252   This is not a pure query: the check is a compare-exchange that 282   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 283   takes the token. Calling it twice is not idempotent: the second
254   call reports `false`, because the first already consumed the 284   call reports `false`, because the first already consumed the
255   wakeup. 285   wakeup.
256   286  
257   @return `true` if a pending wakeup token was latched and has now 287   @return `true` if a pending wakeup token was latched and has now
258   been consumed, in which case the awaiting coroutine does not 288   been consumed, in which case the awaiting coroutine does not
259   suspend; otherwise `false`. 289   suspend; otherwise `false`.
260   */ 290   */
HITCBC 261   147 bool await_ready() noexcept 291   2459 bool await_ready() noexcept
262   { 292   {
HITCBC 263   147 int expected = state_token; 293   2459 int expected = state_token;
HITCBC 264   147 return waker_->st_.compare_exchange_strong( 294   2459 return waker_->st_.compare_exchange_strong(
265   expected, state_empty, 295   expected, state_empty,
266   std::memory_order_acq_rel, 296   std::memory_order_acq_rel,
HITCBC 267   147 std::memory_order_acquire); 297   2459 std::memory_order_acquire);
268   } 298   }
269   299  
270   /** Arm the waker with the awaiting coroutine. 300   /** Arm the waker with the awaiting coroutine.
271   301  
272   This is the @ref IoAwaitable overload of `await_suspend`. 302   This is the @ref IoAwaitable overload of `await_suspend`.
273   Unlike `async_event` and `async_mutex`, it has three outcomes, 303   Unlike `async_event` and `async_mutex`, it has three outcomes,
274   because a `wake()` from another thread can land in the window 304   because a `wake()` from another thread can land in the window
275   between `await_ready` and this call. 305   between `await_ready` and this call.
276   306  
277   @li A stop request is already pending on `env->stop_token`: the 307   @li A stop request is already pending on `env->stop_token`: the
278   awaiter records the cancellation and does not arm. 308   awaiter records the cancellation and does not arm.
279   309  
280   @li The waker's slot is no longer empty. Under the single-waiter 310   @li The waker's slot is no longer empty. Under the single-waiter
281   precondition that means a wakeup was latched after 311   precondition that means a wakeup was latched after
282   `await_ready` looked, so the token is consumed here instead 312   `await_ready` looked, so the token is consumed here instead
283   and the wait succeeds. 313   and the wait succeeds.
284   314  
285 - @li Otherwise the slot moves to the armed state, publishing this 315 + @li Otherwise a stop callback is registered on
286 - awaiter to the waker, and a stop callback is registered on 316 + `env->stop_token` and the slot moves to the armed state,
287 - `env->stop_token`. Whichever of `wake()` and that callback 317 + publishing this awaiter to the waker. A stop requested
288 - wins the armed-to-empty transition posts `h` through 318 + while the callback registered completes the wait canceled
289 - `env->executor`. The loser does nothing, and a losing 319 + instead. Whichever of `wake()` and that callback wins the
290 - `wake()` re-latches its token for the next `wait()`. 320 + armed-to-empty transition posts `h` through
  321 + `env->executor`; a winning `wake()` first deregisters the
  322 + callback. The loser does nothing, and a losing `wake()`
  323 + re-latches its token for the next `wait()`.
291   324  
292   @param h The awaiting coroutine, resumed when the waker fires 325   @param h The awaiting coroutine, resumed when the waker fires
293   or the wait is canceled. 326   or the wait is canceled.
294   327  
295   @param env The execution environment. Its executor posts the 328   @param env The execution environment. Its executor posts the
296   resumption and its stop token is watched for the duration of 329   resumption and its stop token is watched for the duration of
297   the wait. It must outlive the wait. 330   the wait. It must outlive the wait.
298   331  
299   @return `h` in the first two cases, which resumes the awaiting 332   @return `h` in the first two cases, which resumes the awaiting
300   coroutine immediately; otherwise `std::noop_coroutine()`, which 333   coroutine immediately; otherwise `std::noop_coroutine()`, which
301   leaves the coroutine suspended and returns control to the 334   leaves the coroutine suspended and returns control to the
302   resumer. 335   resumer.
303   */ 336   */
304   std::coroutine_handle<> 337   std::coroutine_handle<>
HITCBC 305   72 await_suspend( 338   2361 await_suspend(
306   std::coroutine_handle<> h, 339   std::coroutine_handle<> h,
307   io_env const* env) noexcept 340   io_env const* env) noexcept
308   { 341   {
HITCBC 309   72 if(env->stop_token.stop_requested()) 342   2361 if(env->stop_token.stop_requested())
310   { 343   {
HITCBC 311   23 canceled_ = true; 344   36 canceled_ = true;
HITCBC 312   23 return h; 345   36 return h;
313   } 346   }
HITCBC 314   49 cont_.h = h; 347   2325 cont_.h = h;
HITGNC   348 + 2325 cont_.owner = env->owner;
HITCBC 315   49 ex_ = env->executor; 349   2325 ex_ = env->executor;
HITCBC 316   49 waker_->waiter_ = this; 350   2325 waker_->waiter_ = this;
317   351  
  352 + // Registered before publishing, so the wake() that claims
  353 + // this waiter can deregister it before posting.
HITGNC   354 + 6975 ::new(stop_cb_buf_) stop_cb_t(
HITGNC   355 + 2325 env->stop_token, cancel_fn{this});
HITGNC   356 + 2325 active_ = true;
HITGNC   357 + 2325 published_ = true;
  358 +
HITCBC 318   49 int expected = state_empty; 359   2325 int expected = state_empty;
HITCBC 319   98 if(!waker_->st_.compare_exchange_strong( 360   4650 if(!waker_->st_.compare_exchange_strong(
320 - expected, state_armed, 361 + expected, state_arming,
321   std::memory_order_acq_rel, 362   std::memory_order_acq_rel,
322   std::memory_order_acquire)) 363   std::memory_order_acquire))
323   { 364   {
324   // Single-waiter precondition: a second concurrent 365   // Single-waiter precondition: a second concurrent
325   // wait would find the slot armed. 366   // wait would find the slot armed.
MISUBC 326   ✗ BOOST_CAPY_ASSERT(expected == state_token); 367   ✗ BOOST_CAPY_ASSERT(expected == state_token);
327   368  
328   // A wake latched between await_ready and here; 369   // A wake latched between await_ready and here;
329   // consume it and resume inline. 370   // consume it and resume inline.
MISUNC   371 + ✗ published_ = false;
MISUNC   372 + ✗ stop_cb_().~stop_cb_t();
MISUNC   373 + ✗ active_ = false;
MISUBC 330   ✗ waker_->st_.store( 374   ✗ waker_->st_.store(
331   state_empty, std::memory_order_release); 375   state_empty, std::memory_order_release);
MISUBC 332   ✗ return h; 376   ✗ return h;
333 - published_ = true;  
ECB 334   49 } 377   }
335   378  
ECB 336 - 147 ::new(stop_cb_buf_) stop_cb_t( 379 + // While arming nothing else can claim this waiter, so a stop
ECB 337 - 49 env->stop_token, cancel_fn{this}); 380 + // requested as the callback registered is settled here, and
ECB 338 - 49 active_ = true; 381 + // the waker is not touched again once the waiter is armed.
HITGNC   382 + 2325 if(env->stop_token.stop_requested())
  383 + {
MISUNC   384 + ✗ published_ = false;
MISUNC   385 + ✗ waker_->st_.store(
  386 + state_empty, std::memory_order_release);
MISUNC   387 + ✗ canceled_ = true;
MISUNC   388 + ✗ return h;
  389 + }
HITGNC   390 + 2325 waker_->st_.store(state_armed, std::memory_order_release);
HITCBC 339   49 return std::noop_coroutine(); 391   2325 return std::noop_coroutine();
340   } 392   }
341   393  
342   /** Complete the wait and report the outcome. 394   /** Complete the wait and report the outcome.
343   395  
344   Destroys the stop callback if one is registered and clears the 396   Destroys the stop callback if one is registered and clears the
345   armed bookkeeping, so the destructor does not deregister a slot 397   armed bookkeeping, so the destructor does not deregister a slot
346   the resumption already consumed. 398   the resumption already consumed.
347   399  
348   @return An empty `io_result<>` if the wait was woken, whether by 400   @return An empty `io_result<>` if the wait was woken, whether by
349   `wake()` or by a token consumed inline. Otherwise one holding 401   `wake()` or by a token consumed inline. Otherwise one holding
350 - `error::canceled`, which means the stop token won the race. 402 + `error::canceled`, which means the stop token won the race or
  403 + the waker was destroyed.
351   */ 404   */
HITCBC 352   146 [[nodiscard]] io_result<> await_resume() noexcept 405   2454 [[nodiscard]] io_result<> await_resume() noexcept
353   { 406   {
HITCBC 354   146 if(active_) 407   2454 if(active_)
355   { 408   {
HITCBC 356   48 stop_cb_().~stop_cb_t(); 409   2007 stop_cb_().~stop_cb_t();
HITCBC 357   48 active_ = false; 410   2007 active_ = false;
358   } 411   }
HITCBC 359   146 published_ = false; 412   2454 published_ = false;
HITCBC 360   146 if(canceled_) 413   2454 if(canceled_)
HITCBC 361   39 return {make_error_code(error::canceled)}; 414   2045 return {make_error_code(error::canceled)};
HITCBC 362   107 return {{}}; 415   409 return {{}};
363   } 416   }
364   }; 417   };
365   418  
366   /// Construct with no token latched. 419   /// Construct with no token latched.
HITCBC 367   1 async_waker() = default; 420   307 async_waker() = default;
368   421  
  422 + /** Destroy the waker, canceling a coroutine still waiting.
  423 +
  424 + The waiter is posted to its executor to resume with
  425 + `error::canceled`, as if its stop token had been triggered. An
  426 + executor that can no longer run it abandons it instead, possibly
  427 + before this destructor returns. Either way the waiter no longer
  428 + refers to the waker.
  429 +
  430 + @pre No `wake()` call is in progress or made afterward.
  431 + */
HITGNC   432 + 418 ~async_waker()
  433 + {
HITGNC   434 + 418 if(claim_armed())
  435 + {
HITGNC   436 + 2 auto* w = waiter_;
  437 + // The stop callback reaches into the waker; deregistering
  438 + // it waits out one already running.
HITGNC   439 + 2 if(w->active_)
  440 + {
  441 + using cb_t = wait_awaiter::stop_cb_t;
HITGNC   442 + 2 w->stop_cb_().~cb_t();
HITGNC   443 + 2 w->active_ = false;
  444 + }
HITGNC   445 + 2 w->canceled_ = true;
HITGNC   446 + 2 w->claimed_.store(true, std::memory_order_release);
HITGNC   447 + 2 w->ex_.post(w->cont_);
  448 + }
HITGNC   449 + 418 }
  450 +
369   /** Copy construction is disabled; an armed waiter points into the 451   /** Copy construction is disabled; an armed waiter points into the
370   waker. 452   waker.
371   453  
372   @param other The waker that would be copied. 454   @param other The waker that would be copied.
373   */ 455   */
374   async_waker(async_waker const& other) = delete; 456   async_waker(async_waker const& other) = delete;
375   457  
376   /** Copy assignment is disabled; an armed waiter points into the waker. 458   /** Copy assignment is disabled; an armed waiter points into the waker.
377   459  
378   @param other The waker that would be assigned from. 460   @param other The waker that would be assigned from.
379   461  
380   @return A reference to `*this`. 462   @return A reference to `*this`.
381   */ 463   */
382   async_waker& operator=(async_waker const& other) = delete; 464   async_waker& operator=(async_waker const& other) = delete;
383   465  
384   /** Move construction is disabled; an armed waiter points into the 466   /** Move construction is disabled; an armed waiter points into the
385   waker. 467   waker.
386   468  
387   @param other The waker that would be moved from. 469   @param other The waker that would be moved from.
388   */ 470   */
389   async_waker(async_waker&& other) = delete; 471   async_waker(async_waker&& other) = delete;
390   472  
391   /** Move assignment is disabled; an armed waiter points into the waker. 473   /** Move assignment is disabled; an armed waiter points into the waker.
392   474  
393   @param other The waker that would be moved from. 475   @param other The waker that would be moved from.
394   476  
395   @return A reference to `*this`. 477   @return A reference to `*this`.
396   */ 478   */
397   async_waker& operator=(async_waker&& other) = delete; 479   async_waker& operator=(async_waker&& other) = delete;
398   480  
399   /** Asynchronously wait until woken. 481   /** Asynchronously wait until woken.
400   482  
401   If a token is latched, completes immediately and consumes 483   If a token is latched, completes immediately and consumes
402   it. Otherwise suspends until `wake()` or the stop token 484   it. Otherwise suspends until `wake()` or the stop token
403   fires. 485   fires.
404   486  
405   @par Preconditions 487   @par Preconditions
406 - No other coroutine is currently waiting on this object. 488 + No other coroutine is currently waiting on this object; a
  489 + coroutine woken or canceled but not yet resumed still is.
407   490  
408   @return An awaitable that await-returns `io_result<>`; 491   @return An awaitable that await-returns `io_result<>`;
409   empty on wakeup, `error::canceled` if the stop 492   empty on wakeup, `error::canceled` if the stop
410   token wins. 493   token wins.
411   */ 494   */
HITCBC 412   147 wait_awaiter wait() noexcept 495   2459 wait_awaiter wait() noexcept
413   { 496   {
HITCBC 414   147 return wait_awaiter{this}; 497   2459 return wait_awaiter{this};
415   } 498   }
416   499  
417   /** Wake the waiter, or latch the wakeup if none waits. 500   /** Wake the waiter, or latch the wakeup if none waits.
418   501  
419   Callable from any thread. The waiter's resumption is 502   Callable from any thread. The waiter's resumption is
420   posted through its executor; this call never resumes a 503   posted through its executor; this call never resumes a
421   coroutine inline. Multiple calls without an intervening 504   coroutine inline. Multiple calls without an intervening
422 - `wait()` collapse into a single token. 505 + `wait()` collapse into a single token. It may wait briefly
  506 + for a waiter that is still suspending, or for a stop callback
  507 + of the waiter that is running on another thread.
423   */ 508   */
HITCBC 424   109 void wake() noexcept 509   413 void wake() noexcept
425   { 510   {
426   for(;;) 511   for(;;)
427   { 512   {
HITCBC 428   109 int s = st_.load(std::memory_order_acquire); 513   413 int s = st_.load(std::memory_order_acquire);
HITCBC 429   109 if(s == state_token) 514   413 if(s == state_token)
HITCBC 430   109 return; 515   413 return;
HITCBC 431   107 if(s == state_empty) 516   411 if(s == state_empty)
432   { 517   {
HITCBC 433   150 if(st_.compare_exchange_weak( 518   196 if(st_.compare_exchange_weak(
434   s, state_token, 519   s, state_token,
435   std::memory_order_acq_rel, 520   std::memory_order_acq_rel,
436   std::memory_order_acquire)) 521   std::memory_order_acquire))
HITCBC 437   75 return; 522   98 return;
MISUBC 438   ✗ continue; 523   ✗ continue;
439   } 524   }
  525 + // The waiter is still publishing itself; it settles in a
  526 + // few instructions.
HITGNC   527 + 313 if(s == state_arming)
  528 + {
MISUNC   529 + ✗ std::this_thread::yield();
MISUNC   530 + ✗ continue;
  531 + }
440   // armed: winning this CAS claims the waiter, whose 532   // armed: winning this CAS claims the waiter, whose
441   // frame is pinned until we post its resumption. 533   // frame is pinned until we post its resumption.
HITCBC 442   64 if(st_.compare_exchange_weak( 534   626 if(st_.compare_exchange_weak(
443   s, state_empty, 535   s, state_empty,
444   std::memory_order_acq_rel, 536   std::memory_order_acq_rel,
445   std::memory_order_acquire)) 537   std::memory_order_acquire))
446   { 538   {
HITCBC 447   32 auto* w = waiter_; 539   313 auto* w = waiter_;
  540 + // Nothing may reach this waker once the waiter is
  541 + // posted; this waits out a stop callback already
  542 + // running, which then finds the slot empty.
  543 + using cb_t = wait_awaiter::stop_cb_t;
HITGNC   544 + 313 w->stop_cb_().~cb_t();
HITGNC   545 + 313 w->active_ = false;
HITGNC   546 + 313 w->claimed_.store(true, std::memory_order_release);
HITCBC 448   32 w->ex_.post(w->cont_); 547   313 w->ex_.post(w->cont_);
HITCBC 449   32 return; 548   313 return;
450   } 549   }
MISUBC 451   ✗ } 550   ✗ }
452   } 551   }
453   }; 552   };
454   553  
455   } // namespace capy 554   } // namespace capy
456   } // namespace boost 555   } // namespace boost
457   556  
458   #endif 557   #endif