. /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/await_context.hpp> 18   #include <boost/capy/ex/await_context.hpp>
19   #include <boost/capy/results.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   618 bool claim_armed() noexcept 124   591 bool claim_armed() noexcept
125   { 125   {
HITCBC 126   618 int expected = state_armed; 126   591 int expected = state_armed;
127   for(;;) 127   for(;;)
128   { 128   {
HITCBC 129   1238 if(st_.compare_exchange_weak( 129   1182 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   31 return true; 133   33 return true;
HITCBC 134   588 if(expected == state_armed) 134   558 if(expected == state_armed)
135   continue; // LCOV_EXCL_LINE spurious weak CAS failure 135   continue; // LCOV_EXCL_LINE spurious weak CAS failure
HITCBC 136   588 if(expected != state_arming) 136   558 if(expected != state_arming)
HITCBC 137   587 return false; 137   558 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   200 void operator()() const noexcept 171   173 void operator()() const noexcept
172   { 172   {
HITCBC 173   200 if(self_->waker_->claim_armed()) 173   173 if(self_->waker_->claim_armed())
174   { 174   {
HITCBC 175   29 self_->canceled_ = true; 175   31 self_->canceled_ = true;
HITCBC 176   29 self_->claimed_.store( 176   31 self_->claimed_.store(
177   true, std::memory_order_release); 177   true, std::memory_order_release);
HITCBC 178   29 self_->ex_.post(self_->cont_); 178   31 self_->ex_.post(self_->cont_);
179   } 179   }
HITCBC 180   200 } 180   173 }
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   521 stop_cb_t& stop_cb_() noexcept 194   502 stop_cb_t& stop_cb_() noexcept
195   { 195   {
HITCBC 196   521 return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_); 196   502 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   4900 ~wait_awaiter()
210   { 210   {
HITCBC 211   4884 if(active_) 211   4900 if(active_)
HITCBC 212   3 stop_cb_().~stop_cb_t(); 212   3 stop_cb_().~stop_cb_t();
HITCBC 213   4889 if(published_ && 213   4905 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   4900 }
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   2450 explicit wait_awaiter(async_waker* waker) noexcept
HITCBC 232   2442 : waker_(waker) 232   2450 : waker_(waker)
233   { 233   {
HITCBC 234   2442 } 234   2450 }
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   2450 wait_awaiter(wait_awaiter&& o) noexcept
HITCBC 245   2442 : waker_(o.waker_) 245   2450 : waker_(o.waker_)
HITCBC 246   2442 , cont_(o.cont_) 246   2450 , cont_(o.cont_)
HITCBC 247   2442 , ex_(o.ex_) 247   2450 , ex_(o.ex_)
HITCBC 248   2442 , canceled_(o.canceled_) 248   2450 , canceled_(o.canceled_)
HITCBC 249   2442 , active_(std::exchange(o.active_, false)) 249   2450 , active_(std::exchange(o.active_, false))
HITCBC 250   2442 , published_(std::exchange(o.published_, false)) 250   2450 , published_(std::exchange(o.published_, false))
HITCBC 251   2442 , claimed_(o.claimed_.load(std::memory_order_relaxed)) 251   2450 , claimed_(o.claimed_.load(std::memory_order_relaxed))
252   { 252   {
HITCBC 253   2442 } 253   2450 }
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   2450 bool await_ready() noexcept
292   { 292   {
HITCBC 293   2442 int expected = state_token; 293   2450 int expected = state_token;
HITCBC 294   2442 return waker_->st_.compare_exchange_strong( 294   2450 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   2450 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 ContextAwaitable 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   2349 await_suspend( 338   2366 await_suspend(
339   std::coroutine_handle<> h, 339   std::coroutine_handle<> h,
340   await_context const* env) noexcept 340   await_context const* env) noexcept
341   { 341   {
HITCBC 342   2349 if(env->stop_token.stop_requested()) 342   2366 if(env->stop_token.stop_requested())
343   { 343   {
HITCBC 344   1828 canceled_ = true; 344   1864 canceled_ = true;
HITCBC 345   1828 return h; 345   1864 return h;
346   } 346   }
HITCBC 347   521 cont_.h = h; 347   502 cont_.h = h;
HITCBC 348   521 cont_.owner = env->owner; 348   502 cont_.owner = env->owner;
HITCBC 349   521 ex_ = env->executor; 349   502 ex_ = env->executor;
HITCBC 350   521 waker_->waiter_ = this; 350   502 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   1563 ::new(stop_cb_buf_) stop_cb_t( 354   1506 ::new(stop_cb_buf_) stop_cb_t(
HITCBC 355   521 env->stop_token, cancel_fn{this}); 355   502 env->stop_token, cancel_fn{this});
HITCBC 356   521 active_ = true; 356   502 active_ = true;
HITCBC 357   521 published_ = true; 357   502 published_ = true;
358   358  
HITCBC 359   521 int expected = state_empty; 359   502 int expected = state_empty;
HITCBC 360   1042 if(!waker_->st_.compare_exchange_strong( 360   1004 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   // LCOV_EXCL_START race: a wake latched between await_ready and here
366   // Single-waiter precondition: a second concurrent 366   // Single-waiter precondition: a second concurrent
367   // wait would find the slot armed. 367   // wait would find the slot armed.
368   BOOST_CAPY_ASSERT(expected == state_token); 368   BOOST_CAPY_ASSERT(expected == state_token);
369   369  
370   // A wake latched between await_ready and here; 370   // A wake latched between await_ready and here;
371   // consume it and resume inline. 371   // consume it and resume inline.
372   published_ = false; 372   published_ = false;
373   stop_cb_().~stop_cb_t(); 373   stop_cb_().~stop_cb_t();
374   active_ = false; 374   active_ = false;
375   waker_->st_.store( 375   waker_->st_.store(
376   state_empty, std::memory_order_release); 376   state_empty, std::memory_order_release);
377   return h; 377   return h;
378   // LCOV_EXCL_STOP 378   // LCOV_EXCL_STOP
379   } 379   }
380   380  
381   // While arming nothing else can claim this waiter, so a stop 381   // While arming nothing else can claim this waiter, so a stop
382   // requested as the callback registered is settled here, and 382   // requested as the callback registered is settled here, and
383   // the waker is not touched again once the waiter is armed. 383   // the waker is not touched again once the waiter is armed.
HITCBC 384   521 if(env->stop_token.stop_requested()) 384   501 if(env->stop_token.stop_requested())
385   { 385   {
HITCBC 386   171 published_ = false; 386   141 published_ = false;
HITCBC 387   171 waker_->st_.store( 387   141 waker_->st_.store(
388   state_empty, std::memory_order_release); 388   state_empty, std::memory_order_release);
HITCBC 389   171 canceled_ = true; 389   141 canceled_ = true;
HITCBC 390   171 return h; 390   141 return h;
391   } 391   }
HITCBC 392   350 waker_->st_.store(state_armed, std::memory_order_release); 392   360 waker_->st_.store(state_armed, std::memory_order_release);
HITCBC 393   350 return std::noop_coroutine(); 393   360 return std::noop_coroutine();
394   } 394   }
395   395  
396   /** Complete the wait and report the outcome. 396   /** Complete the wait and report the outcome.
397   397  
398   Destroys the stop callback if one is registered and clears the 398   Destroys the stop callback if one is registered and clears the
399   armed bookkeeping, so the destructor does not deregister a slot 399   armed bookkeeping, so the destructor does not deregister a slot
400   the resumption already consumed. 400   the resumption already consumed.
401   401  
402   @return An empty `results<>` if the wait was woken, whether by 402   @return An empty `results<>` if the wait was woken, whether by
403   `wake()` or by a token consumed inline. Otherwise one holding 403   `wake()` or by a token consumed inline. Otherwise one holding
404   `error::canceled`, which means the stop token won the race or 404   `error::canceled`, which means the stop token won the race or
405   the waker was destroyed. 405   the waker was destroyed.
406   */ 406   */
HITCBC 407   2437 [[nodiscard]] results<> await_resume() noexcept 407   2445 [[nodiscard]] results<> await_resume() noexcept
408   { 408   {
HITCBC 409   2437 if(active_) 409   2445 if(active_)
410   { 410   {
HITCBC 411   198 stop_cb_().~stop_cb_t(); 411   170 stop_cb_().~stop_cb_t();
HITCBC 412   198 active_ = false; 412   170 active_ = false;
413   } 413   }
HITCBC 414   2437 published_ = false; 414   2445 published_ = false;
HITCBC 415   2437 if(canceled_) 415   2445 if(canceled_)
HITCBC 416   2028 return {make_error_code(error::canceled)}; 416   2036 return {make_error_code(error::canceled)};
HITCBC 417   409 return {{}}; 417   409 return {{}};
418   } 418   }
419   }; 419   };
420   420  
421   /// Construct with no token latched. 421   /// Construct with no token latched.
HITCBC 422   307 async_waker() = default; 422   307 async_waker() = default;
423   423  
424   /** Destroy the waker, canceling a coroutine still waiting. 424   /** Destroy the waker, canceling a coroutine still waiting.
425   425  
426   The waiter is posted to its executor to resume with 426   The waiter is posted to its executor to resume with
427   `error::canceled`, as if its stop token had been triggered. An 427   `error::canceled`, as if its stop token had been triggered. An
428   executor that can no longer run it abandons it instead, possibly 428   executor that can no longer run it abandons it instead, possibly
429   before this destructor returns. Either way the waiter no longer 429   before this destructor returns. Either way the waiter no longer
430   refers to the waker. 430   refers to the waker.
431   431  
432   @pre No `wake()` call is in progress or made afterward. 432   @pre No `wake()` call is in progress or made afterward.
433   */ 433   */
HITCBC 434   418 ~async_waker() 434   418 ~async_waker()
435   { 435   {
HITCBC 436   418 if(claim_armed()) 436   418 if(claim_armed())
437   { 437   {
HITCBC 438   2 auto* w = waiter_; 438   2 auto* w = waiter_;
439   // The stop callback reaches into the waker; deregistering 439   // The stop callback reaches into the waker; deregistering
440   // it waits out one already running. 440   // it waits out one already running.
HITCBC 441   2 if(w->active_) 441   2 if(w->active_)
442   { 442   {
443   using cb_t = wait_awaiter::stop_cb_t; 443   using cb_t = wait_awaiter::stop_cb_t;
HITCBC 444   2 w->stop_cb_().~cb_t(); 444   2 w->stop_cb_().~cb_t();
HITCBC 445   2 w->active_ = false; 445   2 w->active_ = false;
446   } 446   }
HITCBC 447   2 w->canceled_ = true; 447   2 w->canceled_ = true;
HITCBC 448   2 w->claimed_.store(true, std::memory_order_release); 448   2 w->claimed_.store(true, std::memory_order_release);
HITCBC 449   2 w->ex_.post(w->cont_); 449   2 w->ex_.post(w->cont_);
450   } 450   }
HITCBC 451   418 } 451   418 }
452   452  
453   /** Copy construction is disabled; an armed waiter points into the 453   /** Copy construction is disabled; an armed waiter points into the
454   waker. 454   waker.
455   455  
456   @param other The waker that would be copied. 456   @param other The waker that would be copied.
457   */ 457   */
458   async_waker(async_waker const& other) = delete; 458   async_waker(async_waker const& other) = delete;
459   459  
460   /** Copy assignment is disabled; an armed waiter points into the waker. 460   /** Copy assignment is disabled; an armed waiter points into the waker.
461   461  
462   @param other The waker that would be assigned from. 462   @param other The waker that would be assigned from.
463   463  
464   @return A reference to `*this`. 464   @return A reference to `*this`.
465   */ 465   */
466   async_waker& operator=(async_waker const& other) = delete; 466   async_waker& operator=(async_waker const& other) = delete;
467   467  
468   /** Move construction is disabled; an armed waiter points into the 468   /** Move construction is disabled; an armed waiter points into the
469   waker. 469   waker.
470   470  
471   @param other The waker that would be moved from. 471   @param other The waker that would be moved from.
472   */ 472   */
473   async_waker(async_waker&& other) = delete; 473   async_waker(async_waker&& other) = delete;
474   474  
475   /** Move assignment is disabled; an armed waiter points into the waker. 475   /** Move assignment is disabled; an armed waiter points into the waker.
476   476  
477   @param other The waker that would be moved from. 477   @param other The waker that would be moved from.
478   478  
479   @return A reference to `*this`. 479   @return A reference to `*this`.
480   */ 480   */
481   async_waker& operator=(async_waker&& other) = delete; 481   async_waker& operator=(async_waker&& other) = delete;
482   482  
483   /** Asynchronously wait until woken. 483   /** Asynchronously wait until woken.
484   484  
485   If a token is latched, completes immediately and consumes 485   If a token is latched, completes immediately and consumes
486   it. Otherwise suspends until `wake()` or the stop token 486   it. Otherwise suspends until `wake()` or the stop token
487   fires. 487   fires.
488   488  
489   @par Preconditions 489   @par Preconditions
490   No other coroutine is currently waiting on this object; a 490   No other coroutine is currently waiting on this object; a
491   coroutine woken or canceled but not yet resumed still is. 491   coroutine woken or canceled but not yet resumed still is.
492   492  
493   @return An awaitable that await-returns `results<>`; 493   @return An awaitable that await-returns `results<>`;
494   empty on wakeup, `error::canceled` if the stop 494   empty on wakeup, `error::canceled` if the stop
495   token wins. 495   token wins.
496   */ 496   */
HITCBC 497   2442 wait_awaiter wait() noexcept 497   2450 wait_awaiter wait() noexcept
498   { 498   {
HITCBC 499   2442 return wait_awaiter{this}; 499   2450 return wait_awaiter{this};
500   } 500   }
501   501  
502   /** Wake the waiter, or latch the wakeup if none waits. 502   /** Wake the waiter, or latch the wakeup if none waits.
503   503  
504   Callable from any thread. The waiter's resumption is 504   Callable from any thread. The waiter's resumption is
505   posted through its executor; this call never resumes a 505   posted through its executor; this call never resumes a
506   coroutine inline. Multiple calls without an intervening 506   coroutine inline. Multiple calls without an intervening
507   `wait()` collapse into a single token. It may wait briefly 507   `wait()` collapse into a single token. It may wait briefly
508   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
509   of the waiter that is running on another thread. 509   of the waiter that is running on another thread.
510   */ 510   */
HITCBC 511   413 void wake() noexcept 511   413 void wake() noexcept
512   { 512   {
513   for(;;) 513   for(;;)
514   { 514   {
HITCBC 515   413 int s = st_.load(std::memory_order_acquire); 515   413 int s = st_.load(std::memory_order_acquire);
HITCBC 516   413 if(s == state_token) 516   413 if(s == state_token)
HITCBC 517   413 return; 517   413 return;
HITCBC 518   411 if(s == state_empty) 518   411 if(s == state_empty)
519   { 519   {
HITCBC 520   186 if(st_.compare_exchange_weak( 520   170 if(st_.compare_exchange_weak(
521   s, state_token, 521   s, state_token,
522   std::memory_order_acq_rel, 522   std::memory_order_acq_rel,
523   std::memory_order_acquire)) 523   std::memory_order_acquire))
HITCBC 524   93 return; 524   85 return;
525   continue; // LCOV_EXCL_LINE spurious weak CAS failure 525   continue; // LCOV_EXCL_LINE spurious weak CAS failure
526   } 526   }
527   // The waiter is still publishing itself; it settles in a 527   // The waiter is still publishing itself; it settles in a
528   // few instructions. 528   // few instructions.
HITCBC 529   318 if(s == state_arming) 529   326 if(s == state_arming)
530   { 530   {
531   std::this_thread::yield(); // LCOV_EXCL_LINE race with a waiter still publishing itself 531   std::this_thread::yield(); // LCOV_EXCL_LINE race with a waiter still publishing itself
532   continue; // LCOV_EXCL_LINE race with a waiter still publishing itself 532   continue; // LCOV_EXCL_LINE race with a waiter still publishing itself
533   } 533   }
534   // armed: winning this CAS claims the waiter, whose 534   // armed: winning this CAS claims the waiter, whose
535   // frame is pinned until we post its resumption. 535   // frame is pinned until we post its resumption.
HITCBC 536   636 if(st_.compare_exchange_weak( 536   652 if(st_.compare_exchange_weak(
537   s, state_empty, 537   s, state_empty,
538   std::memory_order_acq_rel, 538   std::memory_order_acq_rel,
539   std::memory_order_acquire)) 539   std::memory_order_acquire))
540   { 540   {
HITCBC 541   318 auto* w = waiter_; 541   326 auto* w = waiter_;
542   // Nothing may reach this waker once the waiter is 542   // Nothing may reach this waker once the waiter is
543   // posted; this waits out a stop callback already 543   // posted; this waits out a stop callback already
544   // running, which then finds the slot empty. 544   // running, which then finds the slot empty.
545   using cb_t = wait_awaiter::stop_cb_t; 545   using cb_t = wait_awaiter::stop_cb_t;
HITCBC 546   318 w->stop_cb_().~cb_t(); 546   326 w->stop_cb_().~cb_t();
HITCBC 547   318 w->active_ = false; 547   326 w->active_ = false;
HITCBC 548   318 w->claimed_.store(true, std::memory_order_release); 548   326 w->claimed_.store(true, std::memory_order_release);
HITCBC 549   318 w->ex_.post(w->cont_); 549   326 w->ex_.post(w->cont_);
HITCBC 550   318 return; 550   326 return;
551   } 551   }
552   } // LCOV_EXCL_LINE armed CAS lost to a stop callback or failed spuriously 552   } // LCOV_EXCL_LINE armed CAS lost to a stop callback or failed spuriously
553   } 553   }
554   }; 554   };
555   555  
556   } // namespace capy 556   } // namespace capy
557   } // namespace boost 557   } // namespace boost
558   558  
559   #endif 559   #endif