. /include/boost/capy /ex /async_waker.hpp
92.86% Lines (65/70) 100.00% Functions (10/10)
[ ] 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   // 3   //
4   // Distributed under the Boost Software License, Version 1.0. (See accompanying 4   // Distributed under the Boost Software License, Version 1.0. (See accompanying
5   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) 5   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6   // 6   //
7   // Official repository: https://github.com/cppalliance/capy 7   // Official repository: https://github.com/cppalliance/capy
8   // 8   //
9   9  
10   #ifndef BOOST_CAPY_EX_ASYNC_WAKER_HPP 10   #ifndef BOOST_CAPY_EX_ASYNC_WAKER_HPP
11   #define BOOST_CAPY_EX_ASYNC_WAKER_HPP 11   #define BOOST_CAPY_EX_ASYNC_WAKER_HPP
12   12  
13   #include <boost/capy/detail/config.hpp> 13   #include <boost/capy/detail/config.hpp>
14   #include <boost/capy/continuation.hpp> 14   #include <boost/capy/continuation.hpp>
15   #include <boost/capy/error.hpp> 15   #include <boost/capy/error.hpp>
16   #include <boost/capy/ex/executor_ref.hpp> 16   #include <boost/capy/ex/executor_ref.hpp>
17   #include <boost/capy/ex/io_env.hpp> 17   #include <boost/capy/ex/io_env.hpp>
18   #include <boost/capy/io_result.hpp> 18   #include <boost/capy/io_result.hpp>
19   19  
20   #include <atomic> 20   #include <atomic>
21   #include <coroutine> 21   #include <coroutine>
22   #include <new> 22   #include <new>
23   #include <stop_token> 23   #include <stop_token>
24   #include <utility> 24   #include <utility>
25   25  
26   /* async_waker implementation notes 26   /* async_waker implementation notes
27   =================================== 27   ===================================
28   28  
29   wake() must be callable from foreign threads (that is the whole 29   wake() must be callable from foreign threads (that is the whole
30   point: the user's thread provides the timing). A waiter-side 30   point: the user's thread provides the timing). A waiter-side
31   claimed_ flag is not enough there -- the 31   claimed_ flag is not enough there -- the
32   waker has to dereference the waiter, and nothing would pin the 32   waker has to dereference the waiter, and nothing would pin the
33   waiter's frame between reading the pointer and claiming it. 33   waiter's frame between reading the pointer and claiming it.
34   34  
35   So the three-state st_ atomic is the single arbiter: 35   So the three-state st_ atomic is the single arbiter:
36   36  
37   empty --arm--> armed --wake/cancel CAS--> empty 37   empty --arm--> armed --wake/cancel CAS--> empty
38   empty --wake--> token --wait consumes--> empty 38   empty --wake--> token --wait consumes--> empty
39   39  
40   Whoever wins the armed->empty CAS owns the resume and may 40   Whoever wins the armed->empty CAS owns the resume and may
41   dereference waiter_: the frame cannot die underneath the 41   dereference waiter_: the frame cannot die underneath the
42   winner because the coroutine only resumes when the winner 42   winner because the coroutine only resumes when the winner
43   posts it. The loser never touches the waiter. When the stop 43   posts it. The loser never touches the waiter. When the stop
44   callback wins, a concurrent wake retries, finds empty, and 44   callback wins, a concurrent wake retries, finds empty, and
45   latches a token -- a racing wakeup is deferred, never lost. 45   latches a token -- a racing wakeup is deferred, never lost.
46   46  
47   Serialized resumption is required: await_suspend keeps 47   Serialized resumption is required: await_suspend keeps
48   writing after the publishing armed-CAS (the stop_cb 48   writing after the publishing armed-CAS (the stop_cb
49   placement-new and active_ = true), so a wake/cancel winner 49   placement-new and active_ = true), so a wake/cancel winner
50   can post the continuation while that tail is still running. 50   can post the continuation while that tail is still running.
51   The posted resume must be ordered after await_suspend's 51   The posted resume must be ordered after await_suspend's
52   return, which holds on a single-threaded executor (the one 52   return, which holds on a single-threaded executor (the one
53   thread is still inside await_suspend) and on a strand (the 53   thread is still inside await_suspend) and on a strand (the
54   resume is a later turn, synchronized with the current one). 54   resume is a later turn, synchronized with the current one).
55   A raw multi-threaded executor lets another worker run 55   A raw multi-threaded executor lets another worker run
56   await_resume against those in-flight writes. async_event and 56   await_resume against those in-flight writes. async_event and
57   async_mutex make the same assumption; it is stated explicitly 57   async_mutex make the same assumption; it is stated explicitly
58   here because wake() invites foreign threads into the picture. 58   here because wake() invites foreign threads into the picture.
59   */ 59   */
60   60  
61   namespace boost { 61   namespace boost {
62   namespace capy { 62   namespace capy {
63   63  
64   /** A single-slot waker that hands one wakeup to a waiting coroutine. 64   /** A single-slot waker that hands one wakeup to a waiting coroutine.
65   65  
66   This is the escape hatch for timing and other external events: 66   This is the escape hatch for timing and other external events:
67   the user provides the thread and the clock, capy provides the 67   the user provides the thread and the clock, capy provides the
68   suspension point. One coroutine suspends in `wait()`; any 68   suspension point. One coroutine suspends in `wait()`; any
69   thread wakes it with `wake()`. 69   thread wakes it with `wake()`.
70   70  
71   A wakeup with no waiter present is latched as a single pending 71   A wakeup with no waiter present is latched as a single pending
72   token, and the next `wait()` consumes it immediately. This 72   token, and the next `wait()` consumes it immediately. This
73   makes the wake-before-wait race benign without any lock 73   makes the wake-before-wait race benign without any lock
74   protocol. Multiple wakes collapse into one token. 74   protocol. Multiple wakes collapse into one token.
75   75  
76   @par Cancellation 76   @par Cancellation
77   77  
78   If the environment's stop token is triggered while suspended, 78   If the environment's stop token is triggered while suspended,
79   the wait completes with `error::canceled`. A wake that loses 79   the wait completes with `error::canceled`. A wake that loses
80   the race against cancellation is latched for the next `wait()` 80   the race against cancellation is latched for the next `wait()`
81   rather than dropped. 81   rather than dropped.
82   82  
83   @par Zero Allocation 83   @par Zero Allocation
84   84  
85   No heap allocation occurs for wait or wake operations. 85   No heap allocation occurs for wait or wake operations.
86   86  
87   @par Thread Safety 87   @par Thread Safety
88   88  
89   Distinct objects: Safe.@n 89   Distinct objects: Safe.@n
90   Shared objects: `wake()` may be called from any thread. 90   Shared objects: `wake()` may be called from any thread.
91   `wait()` must only be awaited by one coroutine at a time, and 91   `wait()` must only be awaited by one coroutine at a time, and
92   only on an executor that never runs the coroutine's 92   only on an executor that never runs the coroutine's
93   continuations concurrently: a single-threaded executor or a 93   continuations concurrently: a single-threaded executor or a
94   strand over a multi-threaded one (the same threading model as 94   strand over a multi-threaded one (the same threading model as
95   `async_event` and `async_mutex`). Awaiting `wait()` directly 95   `async_event` and `async_mutex`). Awaiting `wait()` directly
96   on a multi-threaded executor is undefined. 96   on a multi-threaded executor is undefined.
97   97  
98   This type is non-copyable and non-movable because a suspended 98   This type is non-copyable and non-movable because a suspended
99   waiter holds a pointer into the object. 99   waiter holds a pointer into the object.
100   100  
101   @par Example 101   @par Example
102   @code 102   @code
103   async_waker waker; 103   async_waker waker;
104   104  
105   // user-provided timing thread 105   // user-provided timing thread
106   std::thread th([&waker] { 106   std::thread th([&waker] {
107   std::this_thread::sleep_for(100ms); 107   std::this_thread::sleep_for(100ms);
108   waker.wake(); 108   waker.wake();
109   }); 109   });
110   110  
111   task<> waiter() { 111   task<> waiter() {
112   auto [ec] = co_await waker.wait(); 112   auto [ec] = co_await waker.wait();
113   // resumed on the executor after ~100ms 113   // resumed on the executor after ~100ms
114   } 114   }
115   // ... th.join() after the pool drains 115   // ... th.join() after the pool drains
116   @endcode 116   @endcode
117   */ 117   */
118   class async_waker 118   class async_waker
119   { 119   {
120   public: 120   public:
121   class wait_awaiter; 121   class wait_awaiter;
122   122  
123   private: 123   private:
124   static constexpr int state_empty = 0; // no token, no waiter 124   static constexpr int state_empty = 0; // no token, no waiter
125   static constexpr int state_token = 1; // latched wakeup 125   static constexpr int state_token = 1; // latched wakeup
126   static constexpr int state_armed = 2; // waiter suspended 126   static constexpr int state_armed = 2; // waiter suspended
127   127  
128   std::atomic<int> st_{state_empty}; 128   std::atomic<int> st_{state_empty};
129   wait_awaiter* waiter_ = nullptr; 129   wait_awaiter* waiter_ = nullptr;
130   130  
131   public: 131   public:
132   /** Awaiter returned by wait(). 132   /** Awaiter returned by wait().
133   */ 133   */
134   class wait_awaiter 134   class wait_awaiter
135   { 135   {
136   friend class async_waker; 136   friend class async_waker;
137   137  
138   async_waker* waker_; 138   async_waker* waker_;
139   continuation cont_; 139   continuation cont_;
140   executor_ref ex_; 140   executor_ref ex_;
141   141  
142   // Declared before stop_cb_buf_: the callback accesses 142   // Declared before stop_cb_buf_: the callback accesses
143   // these members, so they must still be alive if the 143   // these members, so they must still be alive if the
144   // stop_cb_ destructor blocks. 144   // stop_cb_ destructor blocks.
145   bool canceled_ = false; 145   bool canceled_ = false;
146   bool active_ = false; 146   bool active_ = false;
147   bool published_ = false; 147   bool published_ = false;
148   148  
149   struct cancel_fn 149   struct cancel_fn
150   { 150   {
151   wait_awaiter* self_; 151   wait_awaiter* self_;
152   152  
HITCBC 153   11 void operator()() const noexcept 153   4 void operator()() const noexcept
154   { 154   {
HITCBC 155   11 int expected = state_armed; 155   4 int expected = state_armed;
HITCBC 156   22 if(self_->waker_->st_.compare_exchange_strong( 156   8 if(self_->waker_->st_.compare_exchange_strong(
157   expected, state_empty, 157   expected, state_empty,
158   std::memory_order_acq_rel, 158   std::memory_order_acq_rel,
159   std::memory_order_acquire)) 159   std::memory_order_acquire))
160   { 160   {
HITCBC 161   10 self_->canceled_ = true; 161   3 self_->canceled_ = true;
HITCBC 162   10 self_->ex_.post(self_->cont_); 162   3 self_->ex_.post(self_->cont_);
163   } 163   }
HITCBC 164   11 } 164   4 }
165   }; 165   };
166   166  
167   using stop_cb_t = std::stop_callback<cancel_fn>; 167   using stop_cb_t = std::stop_callback<cancel_fn>;
168   168  
169   // Aligned storage for stop_cb_t. Declared last: its 169   // Aligned storage for stop_cb_t. Declared last: its
170   // destructor may block while the callback accesses the 170   // destructor may block while the callback accesses the
171   // members above. 171   // members above.
172   BOOST_CAPY_MSVC_WARNING_PUSH 172   BOOST_CAPY_MSVC_WARNING_PUSH
173   BOOST_CAPY_MSVC_WARNING_DISABLE(4324) 173   BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
174   alignas(stop_cb_t) 174   alignas(stop_cb_t)
175   unsigned char stop_cb_buf_[sizeof(stop_cb_t)]; 175   unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
176   BOOST_CAPY_MSVC_WARNING_POP 176   BOOST_CAPY_MSVC_WARNING_POP
177   177  
HITCBC 178   31 stop_cb_t& stop_cb_() noexcept 178   8 stop_cb_t& stop_cb_() noexcept
179   { 179   {
HITCBC 180   31 return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_); 180   8 return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_);
181   } 181   }
182   182  
183   public: 183   public:
HITCBC 184   296 ~wait_awaiter() 184   262 ~wait_awaiter()
185   { 185   {
HITCBC 186   296 if(active_) 186   262 if(active_)
HITCBC 187   1 stop_cb_().~stop_cb_t(); 187   1 stop_cb_().~stop_cb_t();
HITCBC 188   296 if(published_) 188   262 if(published_)
189   { 189   {
190   // Destroyed while still armed (frame torn down 190   // Destroyed while still armed (frame torn down
191   // without resuming): deregister so a later 191   // without resuming): deregister so a later
192   // wake cannot touch the dead frame. 192   // wake cannot touch the dead frame.
HITCBC 193   1 int expected = state_armed; 193   1 int expected = state_armed;
HITCBC 194   1 waker_->st_.compare_exchange_strong( 194   1 waker_->st_.compare_exchange_strong(
195   expected, state_empty, 195   expected, state_empty,
196   std::memory_order_acq_rel, 196   std::memory_order_acq_rel,
197   std::memory_order_acquire); 197   std::memory_order_acquire);
198   } 198   }
HITCBC 199   296 } 199   262 }
200   200  
HITCBC 201   148 explicit wait_awaiter(async_waker* waker) noexcept 201   131 explicit wait_awaiter(async_waker* waker) noexcept
HITCBC 202   148 : waker_(waker) 202   131 : waker_(waker)
203   { 203   {
HITCBC 204   148 } 204   131 }
205   205  
HITCBC 206   148 wait_awaiter(wait_awaiter&& o) noexcept 206   131 wait_awaiter(wait_awaiter&& o) noexcept
HITCBC 207   148 : waker_(o.waker_) 207   131 : waker_(o.waker_)
HITCBC 208   148 , cont_(o.cont_) 208   131 , cont_(o.cont_)
HITCBC 209   148 , ex_(o.ex_) 209   131 , ex_(o.ex_)
HITCBC 210   148 , canceled_(o.canceled_) 210   131 , canceled_(o.canceled_)
HITCBC 211   148 , active_(std::exchange(o.active_, false)) 211   131 , active_(std::exchange(o.active_, false))
HITCBC 212   148 , published_(std::exchange(o.published_, false)) 212   131 , published_(std::exchange(o.published_, false))
213   { 213   {
HITCBC 214   148 } 214   131 }
215   215  
216   wait_awaiter(wait_awaiter const&) = delete; 216   wait_awaiter(wait_awaiter const&) = delete;
217   wait_awaiter& operator=(wait_awaiter const&) = delete; 217   wait_awaiter& operator=(wait_awaiter const&) = delete;
218   wait_awaiter& operator=(wait_awaiter&&) = delete; 218   wait_awaiter& operator=(wait_awaiter&&) = delete;
219   219  
220   /// Consume a latched token, completing synchronously. 220   /// Consume a latched token, completing synchronously.
HITCBC 221   148 bool await_ready() noexcept 221   131 bool await_ready() noexcept
222   { 222   {
HITCBC 223   148 int expected = state_token; 223   131 int expected = state_token;
HITCBC 224   148 return waker_->st_.compare_exchange_strong( 224   131 return waker_->st_.compare_exchange_strong(
225   expected, state_empty, 225   expected, state_empty,
226   std::memory_order_acq_rel, 226   std::memory_order_acq_rel,
HITCBC 227   148 std::memory_order_acquire); 227   131 std::memory_order_acquire);
228   } 228   }
229   229  
230   /** IoAwaitable protocol overload. */ 230   /** IoAwaitable protocol overload. */
231   std::coroutine_handle<> 231   std::coroutine_handle<>
HITCBC 232   63 await_suspend( 232   30 await_suspend(
233   std::coroutine_handle<> h, 233   std::coroutine_handle<> h,
234   io_env const* env) noexcept 234   io_env const* env) noexcept
235   { 235   {
HITCBC 236   63 if(env->stop_token.stop_requested()) 236   30 if(env->stop_token.stop_requested())
237   { 237   {
HITCBC 238   32 canceled_ = true; 238   22 canceled_ = true;
HITCBC 239   32 return h; 239   22 return h;
240   } 240   }
HITCBC 241   31 cont_.h = h; 241   8 cont_.h = h;
HITCBC 242   31 ex_ = env->executor; 242   8 ex_ = env->executor;
HITCBC 243   31 waker_->waiter_ = this; 243   8 waker_->waiter_ = this;
244   244  
HITCBC 245   31 int expected = state_empty; 245   8 int expected = state_empty;
HITCBC 246   62 if(!waker_->st_.compare_exchange_strong( 246   16 if(!waker_->st_.compare_exchange_strong(
247   expected, state_armed, 247   expected, state_armed,
248   std::memory_order_acq_rel, 248   std::memory_order_acq_rel,
249   std::memory_order_acquire)) 249   std::memory_order_acquire))
250   { 250   {
251   // Single-waiter precondition: a second concurrent 251   // Single-waiter precondition: a second concurrent
252   // wait would find the slot armed. 252   // wait would find the slot armed.
MISUBC 253   ✗ BOOST_CAPY_ASSERT(expected == state_token); 253   ✗ BOOST_CAPY_ASSERT(expected == state_token);
254   254  
255   // A wake latched between await_ready and here; 255   // A wake latched between await_ready and here;
256   // consume it and resume inline. 256   // consume it and resume inline.
MISUBC 257   ✗ waker_->st_.store( 257   ✗ waker_->st_.store(
258   state_empty, std::memory_order_release); 258   state_empty, std::memory_order_release);
MISUBC 259   ✗ return h; 259   ✗ return h;
260   } 260   }
HITCBC 261   31 published_ = true; 261   8 published_ = true;
262   262  
HITCBC 263   93 ::new(stop_cb_buf_) stop_cb_t( 263   24 ::new(stop_cb_buf_) stop_cb_t(
HITCBC 264   31 env->stop_token, cancel_fn{this}); 264   8 env->stop_token, cancel_fn{this});
HITCBC 265   31 active_ = true; 265   8 active_ = true;
HITCBC 266   31 return std::noop_coroutine(); 266   8 return std::noop_coroutine();
267   } 267   }
268   268  
HITCBC 269   147 io_result<> await_resume() noexcept 269   130 io_result<> await_resume() noexcept
270   { 270   {
HITCBC 271   147 if(active_) 271   130 if(active_)
272   { 272   {
HITCBC 273   30 stop_cb_().~stop_cb_t(); 273   7 stop_cb_().~stop_cb_t();
HITCBC 274   30 active_ = false; 274   7 active_ = false;
275   } 275   }
HITCBC 276   147 published_ = false; 276   130 published_ = false;
HITCBC 277   147 if(canceled_) 277   130 if(canceled_)
HITCBC 278   41 return {make_error_code(error::canceled)}; 278   24 return {make_error_code(error::canceled)};
HITCBC 279   106 return {{}}; 279   106 return {{}};
280   } 280   }
281   }; 281   };
282   282  
283   /// Construct with no token latched. 283   /// Construct with no token latched.
284   async_waker() = default; 284   async_waker() = default;
285   285  
286   /// Copy constructor (deleted). 286   /// Copy constructor (deleted).
287   async_waker(async_waker const&) = delete; 287   async_waker(async_waker const&) = delete;
288   288  
289   /// Copy assignment (deleted). 289   /// Copy assignment (deleted).
290   async_waker& operator=(async_waker const&) = delete; 290   async_waker& operator=(async_waker const&) = delete;
291   291  
292   /// Move constructor (deleted). 292   /// Move constructor (deleted).
293   async_waker(async_waker&&) = delete; 293   async_waker(async_waker&&) = delete;
294   294  
295   /// Move assignment (deleted). 295   /// Move assignment (deleted).
296   async_waker& operator=(async_waker&&) = delete; 296   async_waker& operator=(async_waker&&) = delete;
297   297  
298   /** Asynchronously wait until woken. 298   /** Asynchronously wait until woken.
299   299  
300   If a token is latched, completes immediately and consumes 300   If a token is latched, completes immediately and consumes
301   it. Otherwise suspends until `wake()` or the stop token 301   it. Otherwise suspends until `wake()` or the stop token
302   fires. 302   fires.
303   303  
304   @par Preconditions 304   @par Preconditions
305   No other coroutine is currently waiting on this object. 305   No other coroutine is currently waiting on this object.
306   306  
307   @return An awaitable that await-returns `io_result<>`; 307   @return An awaitable that await-returns `io_result<>`;
308   empty on wakeup, `error::canceled` if the stop 308   empty on wakeup, `error::canceled` if the stop
309   token wins. 309   token wins.
310   */ 310   */
HITCBC 311   148 wait_awaiter wait() noexcept 311   131 wait_awaiter wait() noexcept
312   { 312   {
HITCBC 313   148 return wait_awaiter{this}; 313   131 return wait_awaiter{this};
314   } 314   }
315   315  
316   /** Wake the waiter, or latch the wakeup if none waits. 316   /** Wake the waiter, or latch the wakeup if none waits.
317   317  
318   Callable from any thread. The waiter's resumption is 318   Callable from any thread. The waiter's resumption is
319   posted through its executor; this call never resumes a 319   posted through its executor; this call never resumes a
320   coroutine inline. Multiple calls without an intervening 320   coroutine inline. Multiple calls without an intervening
321   `wait()` collapse into a single token. 321   `wait()` collapse into a single token.
322   */ 322   */
HITCBC 323   108 void wake() noexcept 323   108 void wake() noexcept
324   { 324   {
325   for(;;) 325   for(;;)
326   { 326   {
HITCBC 327   108 int s = st_.load(std::memory_order_acquire); 327   108 int s = st_.load(std::memory_order_acquire);
HITCBC 328   108 if(s == state_token) 328   108 if(s == state_token)
HITCBC 329   108 return; 329   108 return;
HITCBC 330   106 if(s == state_empty) 330   106 if(s == state_empty)
331   { 331   {
HITCBC 332   170 if(st_.compare_exchange_weak( 332   202 if(st_.compare_exchange_weak(
333   s, state_token, 333   s, state_token,
334   std::memory_order_acq_rel, 334   std::memory_order_acq_rel,
335   std::memory_order_acquire)) 335   std::memory_order_acquire))
HITCBC 336   85 return; 336   101 return;
MISUBC 337   ✗ continue; 337   ✗ continue;
338   } 338   }
339   // armed: winning this CAS claims the waiter, whose 339   // armed: winning this CAS claims the waiter, whose
340   // frame is pinned until we post its resumption. 340   // frame is pinned until we post its resumption.
HITCBC 341   42 if(st_.compare_exchange_weak( 341   10 if(st_.compare_exchange_weak(
342   s, state_empty, 342   s, state_empty,
343   std::memory_order_acq_rel, 343   std::memory_order_acq_rel,
344   std::memory_order_acquire)) 344   std::memory_order_acquire))
345   { 345   {
HITCBC 346   21 auto* w = waiter_; 346   5 auto* w = waiter_;
HITCBC 347   21 w->ex_.post(w->cont_); 347   5 w->ex_.post(w->cont_);
HITCBC 348   21 return; 348   5 return;
349   } 349   }
MISUBC 350   ✗ } 350   ✗ }
351   } 351   }
352   }; 352   };
353   353  
354   } // namespace capy 354   } // namespace capy
355   } // namespace boost 355   } // namespace boost
356   356  
357   #endif 357   #endif