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