. /include/boost/capy /ex /async_mutex.hpp
100.00% Lines (93/93) 100.00% Functions (20/20)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com) 2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
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_ASYNC_MUTEX_HPP 11   #ifndef BOOST_CAPY_ASYNC_MUTEX_HPP
12   #define BOOST_CAPY_ASYNC_MUTEX_HPP 12   #define BOOST_CAPY_ASYNC_MUTEX_HPP
13   13  
14   #include <boost/capy/detail/config.hpp> 14   #include <boost/capy/detail/config.hpp>
15   #include <boost/capy/detail/intrusive.hpp> 15   #include <boost/capy/detail/intrusive.hpp>
16   #include <boost/capy/continuation.hpp> 16   #include <boost/capy/continuation.hpp>
17   #include <boost/capy/concept/executor.hpp> 17   #include <boost/capy/concept/executor.hpp>
18   #include <boost/capy/error.hpp> 18   #include <boost/capy/error.hpp>
19   #include <boost/capy/ex/io_env.hpp> 19   #include <boost/capy/ex/io_env.hpp>
20   #include <boost/capy/io_result.hpp> 20   #include <boost/capy/io_result.hpp>
21   21  
22   #include <stop_token> 22   #include <stop_token>
23   23  
24   #include <atomic> 24   #include <atomic>
25   #include <coroutine> 25   #include <coroutine>
26   #include <new> 26   #include <new>
27   #include <utility> 27   #include <utility>
28   28  
29   /* async_mutex implementation notes 29   /* async_mutex implementation notes
30   ================================ 30   ================================
31   31  
32   Waiters form a doubly-linked intrusive list (fair FIFO). lock_awaiter 32   Waiters form a doubly-linked intrusive list (fair FIFO). lock_awaiter
33   inherits intrusive_list<lock_awaiter>::node; the list is owned by 33   inherits intrusive_list<lock_awaiter>::node; the list is owned by
34   async_mutex::waiters_. 34   async_mutex::waiters_.
35   35  
36   Cancellation via stop_token 36   Cancellation via stop_token
37   --------------------------- 37   ---------------------------
38   A std::stop_callback is registered in await_suspend. Two actors can 38   A std::stop_callback is registered in await_suspend. Two actors can
39   race to resume the suspended coroutine: unlock() and the stop callback. 39   race to resume the suspended coroutine: unlock() and the stop callback.
40   An atomic bool `claimed_` resolves the race -- whoever does 40   An atomic bool `claimed_` resolves the race -- whoever does
41   claimed_.exchange(true) and reads false wins. The loser does nothing. 41   claimed_.exchange(true) and reads false wins. The loser does nothing.
42   42  
43   The stop callback calls ex_.post(h_). The stop_callback is 43   The stop callback calls ex_.post(h_). The stop_callback is
44   destroyed later in await_resume. cancel_fn touches no members 44   destroyed later in await_resume. cancel_fn touches no members
45   after post returns (same pattern as delete-this). 45   after post returns (same pattern as delete-this).
46   46  
47   unlock() pops waiters from the front. If the popped waiter was 47   unlock() pops waiters from the front. If the popped waiter was
48   already claimed by the stop callback, unlock() skips it and tries 48   already claimed by the stop callback, unlock() skips it and tries
49   the next. await_resume removes the (still-linked) canceled waiter 49   the next. await_resume removes the (still-linked) canceled waiter
50   via waiters_.remove(this). 50   via waiters_.remove(this).
51   51  
52   The stop_callback lives in a union to suppress automatic 52   The stop_callback lives in a union to suppress automatic
53   construction/destruction. Placement new in await_suspend, explicit 53   construction/destruction. Placement new in await_suspend, explicit
54   destructor call in await_resume and ~lock_awaiter. 54   destructor call in await_resume and ~lock_awaiter.
55   55  
56   Member ordering constraint 56   Member ordering constraint
57   -------------------------- 57   --------------------------
58   The union containing stop_cb_ must be declared AFTER the members 58   The union containing stop_cb_ must be declared AFTER the members
59   the callback accesses (h_, ex_, claimed_, canceled_). If the 59   the callback accesses (h_, ex_, claimed_, canceled_). If the
60   stop_cb_ destructor blocks waiting for a concurrent callback, those 60   stop_cb_ destructor blocks waiting for a concurrent callback, those
61   members must still be alive (C++ destroys in reverse declaration 61   members must still be alive (C++ destroys in reverse declaration
62   order). 62   order).
63   63  
64   active_ flag 64   active_ flag
65   ------------ 65   ------------
66   Tracks both list membership and stop_cb_ lifetime (they are always 66   Tracks both list membership and stop_cb_ lifetime (they are always
67   set and cleared together). Used by the destructor to clean up if the 67   set and cleared together). Used by the destructor to clean up if the
68   coroutine is destroyed while suspended (e.g. execution_context 68   coroutine is destroyed while suspended (e.g. execution_context
69   shutdown). 69   shutdown).
70   70  
71   Cancellation scope 71   Cancellation scope
72   ------------------ 72   ------------------
73   Cancellation only takes effect while the coroutine is suspended in 73   Cancellation only takes effect while the coroutine is suspended in
74   the wait queue. If the mutex is unlocked, await_ready acquires it 74   the wait queue. If the mutex is unlocked, await_ready acquires it
75   immediately without checking the stop token. This is intentional: 75   immediately without checking the stop token. This is intentional:
76   the fast path has no token access and no overhead. 76   the fast path has no token access and no overhead.
77   77  
78   Threading assumptions 78   Threading assumptions
79   --------------------- 79   ---------------------
80   - All list mutations happen on the executor thread (await_suspend, 80   - All list mutations happen on the executor thread (await_suspend,
81   await_resume, unlock, ~lock_awaiter). 81   await_resume, unlock, ~lock_awaiter).
82   - The stop callback may fire from any thread, but only touches 82   - The stop callback may fire from any thread, but only touches
83   claimed_ (atomic) and then calls post. It never touches the 83   claimed_ (atomic) and then calls post. It never touches the
84   list. 84   list.
85   - ~lock_awaiter must be called from the executor thread. This is 85   - ~lock_awaiter must be called from the executor thread. This is
86   guaranteed during normal shutdown but NOT if the coroutine frame 86   guaranteed during normal shutdown but NOT if the coroutine frame
87   is destroyed from another thread while a stop callback could 87   is destroyed from another thread while a stop callback could
88   fire (precondition violation, same as cppcoro/folly). 88   fire (precondition violation, same as cppcoro/folly).
89   */ 89   */
90   90  
91   namespace boost { 91   namespace boost {
92   namespace capy { 92   namespace capy {
93   93  
94   /** Queues coroutines in `lock()` and resumes exactly one when the mutex is free. 94   /** Queues coroutines in `lock()` and resumes exactly one when the mutex is free.
95   95  
96   This mutex provides mutual exclusion for coroutines without blocking. 96   This mutex provides mutual exclusion for coroutines without blocking.
97   When a coroutine attempts to acquire a locked mutex, it suspends and 97   When a coroutine attempts to acquire a locked mutex, it suspends and
98   is added to an intrusive wait queue. When the holder unlocks, the next 98   is added to an intrusive wait queue. When the holder unlocks, the next
99   waiter is resumed with the lock held. 99   waiter is resumed with the lock held.
100   100  
101   @par Cancellation 101   @par Cancellation
102   102  
103   When a coroutine is suspended waiting for the mutex and its stop 103   When a coroutine is suspended waiting for the mutex and its stop
104   token is triggered, the waiter completes with `error::canceled` 104   token is triggered, the waiter completes with `error::canceled`
105   instead of acquiring the lock. 105   instead of acquiring the lock.
106   106  
107   Cancellation only applies while the coroutine is suspended in the 107   Cancellation only applies while the coroutine is suspended in the
108   wait queue. If the mutex is unlocked when `lock()` is called, the 108   wait queue. If the mutex is unlocked when `lock()` is called, the
109   lock is acquired immediately even if the stop token is already 109   lock is acquired immediately even if the stop token is already
110   signaled. 110   signaled.
111   111  
112   @par Zero Allocation 112   @par Zero Allocation
113   113  
114   No heap allocation occurs for lock operations. 114   No heap allocation occurs for lock operations.
115   115  
116   @par Thread Safety 116   @par Thread Safety
117   117  
118   Distinct objects: Safe.@n 118   Distinct objects: Safe.@n
119   Shared objects: Unsafe. 119   Shared objects: Unsafe.
120   120  
121   The mutex operations are designed for single-threaded use on one 121   The mutex operations are designed for single-threaded use on one
122   executor. The stop callback may fire from any thread. 122   executor. The stop callback may fire from any thread.
123   123  
124   This type is non-copyable and non-movable because suspended 124   This type is non-copyable and non-movable because suspended
125   waiters hold intrusive pointers into the mutex's internal list. 125   waiters hold intrusive pointers into the mutex's internal list.
126   126  
127   @par Example 127   @par Example
128 - @code 128 + @par !example example
129 - async_mutex cm;  
130 -  
131 - task<> protected_operation() {  
132 - auto [ec] = co_await cm.lock();  
133 - if(ec)  
134 - co_return;  
135 - // ... critical section ...  
136 - cm.unlock();  
137 - }  
138 - // Or with RAII:  
139 - task<> protected_operation_raii() {  
140 - auto [ec, guard] = co_await cm.scoped_lock();  
141 - if(ec)  
142 - co_return;  
143 - // ... critical section ...  
144 - // unlocks automatically  
145 - }  
146 - @endcode  
147   129  
148   */ 130   */
149   class async_mutex 131   class async_mutex
150   { 132   {
151   public: 133   public:
152   class lock_awaiter; 134   class lock_awaiter;
153   class lock_guard; 135   class lock_guard;
154   class lock_guard_awaiter; 136   class lock_guard_awaiter;
155   137  
156   private: 138   private:
157   bool locked_ = false; 139   bool locked_ = false;
158   detail::intrusive_list<lock_awaiter> waiters_; 140   detail::intrusive_list<lock_awaiter> waiters_;
159   141  
160   public: 142   public:
161   /** Suspends the caller until the mutex is free, or resumes it with `error::canceled` on a stop request. 143   /** Suspends the caller until the mutex is free, or resumes it with `error::canceled` on a stop request.
162   */ 144   */
163   class lock_awaiter 145   class lock_awaiter
164   : public detail::intrusive_list<lock_awaiter>::node 146   : public detail::intrusive_list<lock_awaiter>::node
165   { 147   {
166   friend class async_mutex; 148   friend class async_mutex;
167   149  
168   async_mutex* m_; 150   async_mutex* m_;
169   continuation cont_; 151   continuation cont_;
170   executor_ref ex_; 152   executor_ref ex_;
171   153  
172   // These members must be declared before stop_cb_ 154   // These members must be declared before stop_cb_
173   // (see comment on the union below). 155   // (see comment on the union below).
174   std::atomic<bool> claimed_{false}; 156   std::atomic<bool> claimed_{false};
175   bool canceled_ = false; 157   bool canceled_ = false;
176   bool active_ = false; 158   bool active_ = false;
177   159  
178   struct cancel_fn 160   struct cancel_fn
179   { 161   {
180   lock_awaiter* self_; 162   lock_awaiter* self_;
181   163  
HITCBC 182   7 void operator()() const noexcept 164   7 void operator()() const noexcept
183   { 165   {
HITCBC 184   7 if(!self_->claimed_.exchange( 166   7 if(!self_->claimed_.exchange(
185   true, std::memory_order_acq_rel)) 167   true, std::memory_order_acq_rel))
186   { 168   {
HITCBC 187   7 self_->canceled_ = true; 169   7 self_->canceled_ = true;
HITCBC 188   7 self_->ex_.post(self_->cont_); 170   7 self_->ex_.post(self_->cont_);
189   } 171   }
HITCBC 190   7 } 172   7 }
191   }; 173   };
192   174  
193   using stop_cb_t = 175   using stop_cb_t =
194   std::stop_callback<cancel_fn>; 176   std::stop_callback<cancel_fn>;
195   177  
196   // Aligned storage for stop_cb_t. Declared last: 178   // Aligned storage for stop_cb_t. Declared last:
197   // its destructor may block while the callback 179   // its destructor may block while the callback
198   // accesses the members above. 180   // accesses the members above.
199   BOOST_CAPY_MSVC_WARNING_PUSH 181   BOOST_CAPY_MSVC_WARNING_PUSH
200   BOOST_CAPY_MSVC_WARNING_DISABLE(4324) // padded due to alignas 182   BOOST_CAPY_MSVC_WARNING_DISABLE(4324) // padded due to alignas
201   alignas(stop_cb_t) 183   alignas(stop_cb_t)
202   unsigned char stop_cb_buf_[sizeof(stop_cb_t)]; 184   unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
203   BOOST_CAPY_MSVC_WARNING_POP 185   BOOST_CAPY_MSVC_WARNING_POP
204   186  
HITCBC 205   19 stop_cb_t& stop_cb_() noexcept 187   19 stop_cb_t& stop_cb_() noexcept
206   { 188   {
207   return *reinterpret_cast<stop_cb_t*>( 189   return *reinterpret_cast<stop_cb_t*>(
HITCBC 208   19 stop_cb_buf_); 190   19 stop_cb_buf_);
209   } 191   }
210   192  
211   public: 193   public:
212   /** Destroy the awaiter, leaving the mutex unable to reach it. 194   /** Destroy the awaiter, leaving the mutex unable to reach it.
213   195  
214   If the awaiter is suspended in the wait queue, destroys the 196   If the awaiter is suspended in the wait queue, destroys the
215   stop callback and unlinks the awaiter. Neither `unlock()` nor 197   stop callback and unlinks the awaiter. Neither `unlock()` nor
216   the stop callback can then reach a destroyed awaiter when the 198   the stop callback can then reach a destroyed awaiter when the
217   coroutine frame is torn down while suspended. 199   coroutine frame is torn down while suspended.
218   200  
219   @par Preconditions 201   @par Preconditions
220   Called on the executor thread. The stop callback may fire from 202   Called on the executor thread. The stop callback may fire from
221   any thread, so destroying a still-suspended awaiter from 203   any thread, so destroying a still-suspended awaiter from
222   another thread is undefined. 204   another thread is undefined.
223   */ 205   */
HITCBC 224   76 ~lock_awaiter() 206   76 ~lock_awaiter()
225   { 207   {
HITCBC 226   76 if(active_) 208   76 if(active_)
227   { 209   {
HITCBC 228   3 stop_cb_().~stop_cb_t(); 210   3 stop_cb_().~stop_cb_t();
HITCBC 229   3 m_->waiters_.remove(this); 211   3 m_->waiters_.remove(this);
230   } 212   }
HITCBC 231   76 } 213   76 }
232   214  
233   /** Construct an awaiter for the given mutex. 215   /** Construct an awaiter for the given mutex.
234   216  
235   @param m The mutex to acquire. It must outlive the awaiter. 217   @param m The mutex to acquire. It must outlive the awaiter.
236   */ 218   */
HITCBC 237   38 explicit lock_awaiter(async_mutex* m) noexcept 219   38 explicit lock_awaiter(async_mutex* m) noexcept
HITCBC 238   38 : m_(m) 220   38 : m_(m)
239   { 221   {
HITCBC 240   38 } 222   38 }
241   223  
242   /** Construct by moving. 224   /** Construct by moving.
243   225  
244   The moved-from awaiter is left inert: its destructor no longer 226   The moved-from awaiter is left inert: its destructor no longer
245   destroys the stop callback and no longer unlinks from the 227   destroys the stop callback and no longer unlinks from the
246   mutex's wait queue. 228   mutex's wait queue.
247   229  
248   @param o The awaiter to move from. 230   @param o The awaiter to move from.
249   */ 231   */
HITCBC 250   38 lock_awaiter(lock_awaiter&& o) noexcept 232   38 lock_awaiter(lock_awaiter&& o) noexcept
HITCBC 251   76 : m_(o.m_) 233   76 : m_(o.m_)
HITCBC 252   38 , cont_(o.cont_) 234   38 , cont_(o.cont_)
HITCBC 253   38 , ex_(o.ex_) 235   38 , ex_(o.ex_)
HITCBC 254   38 , claimed_(o.claimed_.load( 236   38 , claimed_(o.claimed_.load(
255   std::memory_order_relaxed)) 237   std::memory_order_relaxed))
HITCBC 256   38 , canceled_(o.canceled_) 238   38 , canceled_(o.canceled_)
HITCBC 257   76 , active_(std::exchange(o.active_, false)) 239   76 , active_(std::exchange(o.active_, false))
258   { 240   {
HITCBC 259   38 } 241   38 }
260   242  
261   /** Copy construction is disabled; a waiter is linked into the 243   /** Copy construction is disabled; a waiter is linked into the
262   mutex's wait queue by address. 244   mutex's wait queue by address.
263   245  
264   @param other The awaiter that would be copied. 246   @param other The awaiter that would be copied.
265   */ 247   */
266   lock_awaiter(lock_awaiter const& other) = delete; 248   lock_awaiter(lock_awaiter const& other) = delete;
267   249  
268   /** Copy assignment is disabled; a waiter is linked into the 250   /** Copy assignment is disabled; a waiter is linked into the
269   mutex's wait queue by address. 251   mutex's wait queue by address.
270   252  
271   @param other The awaiter that would be assigned from. 253   @param other The awaiter that would be assigned from.
272   254  
273   @return A reference to `*this`. 255   @return A reference to `*this`.
274   */ 256   */
275   lock_awaiter& operator=(lock_awaiter const& other) = delete; 257   lock_awaiter& operator=(lock_awaiter const& other) = delete;
276   258  
277   /** Move assignment is disabled; a waiter is linked into the 259   /** Move assignment is disabled; a waiter is linked into the
278   mutex's wait queue by address. 260   mutex's wait queue by address.
279   261  
280   @param other The awaiter that would be moved from. 262   @param other The awaiter that would be moved from.
281   263  
282   @return A reference to `*this`. 264   @return A reference to `*this`.
283   */ 265   */
284   lock_awaiter& operator=(lock_awaiter&& other) = delete; 266   lock_awaiter& operator=(lock_awaiter&& other) = delete;
285   267  
286   /** Acquire the mutex if it is free, reporting whether to suspend. 268   /** Acquire the mutex if it is free, reporting whether to suspend.
287   269  
288   This is not a pure query: on the fast path it takes the lock. 270   This is not a pure query: on the fast path it takes the lock.
289   When the mutex is unlocked, it marks the mutex locked and 271   When the mutex is unlocked, it marks the mutex locked and
290   reports that no suspension is needed. The stop token is not 272   reports that no suspension is needed. The stop token is not
291   consulted, so an uncontended `lock()` succeeds even when stop 273   consulted, so an uncontended `lock()` succeeds even when stop
292   has already been requested. 274   has already been requested.
293   275  
294   @return `true` if the mutex was free and is now held by the 276   @return `true` if the mutex was free and is now held by the
295   awaiting coroutine. `false` if the mutex is held elsewhere, in 277   awaiting coroutine. `false` if the mutex is held elsewhere, in
296   which case the coroutine suspends. 278   which case the coroutine suspends.
297   */ 279   */
HITCBC 298   38 bool await_ready() const noexcept 280   38 bool await_ready() const noexcept
299   { 281   {
HITCBC 300   38 if(!m_->locked_) 282   38 if(!m_->locked_)
301   { 283   {
HITCBC 302   17 m_->locked_ = true; 284   17 m_->locked_ = true;
HITCBC 303   17 return true; 285   17 return true;
304   } 286   }
HITCBC 305   21 return false; 287   21 return false;
306   } 288   }
307   289  
308   /** Enqueue the awaiting coroutine until the mutex is released. 290   /** Enqueue the awaiting coroutine until the mutex is released.
309   291  
310   This is the @ref IoAwaitable overload of `await_suspend`. 292   This is the @ref IoAwaitable overload of `await_suspend`.
311   293  
312   If a stop request is already pending on `env->stop_token`, the 294   If a stop request is already pending on `env->stop_token`, the
313   awaiter records the cancellation and does not enqueue. The 295   awaiter records the cancellation and does not enqueue. The
314   mutex is not acquired. 296   mutex is not acquired.
315   297  
316   Otherwise it stores `h` and `env->executor`, links itself into 298   Otherwise it stores `h` and `env->executor`, links itself into
317   the back of the mutex's wait queue, and registers a stop 299   the back of the mutex's wait queue, and registers a stop
318   callback on `env->stop_token`. Whichever of `unlock()` and that 300   callback on `env->stop_token`. Whichever of `unlock()` and that
319   callback claims the awaiter first posts `h` through the stored 301   callback claims the awaiter first posts `h` through the stored
320   executor; the other skips it. 302   executor; the other skips it.
321   303  
322   @param h The awaiting coroutine, resumed when the mutex is 304   @param h The awaiting coroutine, resumed when the mutex is
323   acquired or the wait is canceled. 305   acquired or the wait is canceled.
324   306  
325   @param env The execution environment. Its executor posts the 307   @param env The execution environment. Its executor posts the
326   resumption and its stop token is watched for the duration of 308   resumption and its stop token is watched for the duration of
327   the wait. It must outlive the wait. 309   the wait. It must outlive the wait.
328   310  
329   @return `h` if a stop request was already pending, which 311   @return `h` if a stop request was already pending, which
330   resumes the awaiting coroutine immediately without enqueuing 312   resumes the awaiting coroutine immediately without enqueuing
331   it. Otherwise `std::noop_coroutine()`, which leaves the 313   it. Otherwise `std::noop_coroutine()`, which leaves the
332   coroutine suspended and returns control to the resumer. 314   coroutine suspended and returns control to the resumer.
333   */ 315   */
334   std::coroutine_handle<> 316   std::coroutine_handle<>
HITCBC 335   21 await_suspend( 317   21 await_suspend(
336   std::coroutine_handle<> h, 318   std::coroutine_handle<> h,
337   io_env const* env) noexcept 319   io_env const* env) noexcept
338   { 320   {
HITCBC 339   21 if(env->stop_token.stop_requested()) 321   21 if(env->stop_token.stop_requested())
340   { 322   {
HITCBC 341   2 canceled_ = true; 323   2 canceled_ = true;
HITCBC 342   2 return h; 324   2 return h;
343   } 325   }
HITCBC 344   19 cont_.h = h; 326   19 cont_.h = h;
HITCBC 345   19 ex_ = env->executor; 327   19 ex_ = env->executor;
HITCBC 346   19 m_->waiters_.push_back(this); 328   19 m_->waiters_.push_back(this);
HITCBC 347   57 ::new(stop_cb_buf_) stop_cb_t( 329   57 ::new(stop_cb_buf_) stop_cb_t(
HITCBC 348   19 env->stop_token, cancel_fn{this}); 330   19 env->stop_token, cancel_fn{this});
HITCBC 349   19 active_ = true; 331   19 active_ = true;
HITCBC 350   19 return std::noop_coroutine(); 332   19 return std::noop_coroutine();
351   } 333   }
352   334  
353   /** Complete the acquisition and report the outcome. 335   /** Complete the acquisition and report the outcome.
354   336  
355   Destroys the stop callback if one is registered, and unlinks a 337   Destroys the stop callback if one is registered, and unlinks a
356   canceled awaiter from the wait queue. 338   canceled awaiter from the wait queue.
357   339  
358   @return An empty `io_result<>` if the mutex is now held by the 340   @return An empty `io_result<>` if the mutex is now held by the
359   awaiting coroutine. Otherwise one holding `error::canceled`, 341   awaiting coroutine. Otherwise one holding `error::canceled`,
360   which means the stop token won the race and the mutex is not 342   which means the stop token won the race and the mutex is not
361   held. 343   held.
362   */ 344   */
HITCBC 363   35 [[nodiscard]] io_result<> await_resume() noexcept 345   35 [[nodiscard]] io_result<> await_resume() noexcept
364   { 346   {
HITCBC 365   35 if(active_) 347   35 if(active_)
366   { 348   {
HITCBC 367   16 stop_cb_().~stop_cb_t(); 349   16 stop_cb_().~stop_cb_t();
HITCBC 368   16 if(canceled_) 350   16 if(canceled_)
369   { 351   {
HITCBC 370   7 m_->waiters_.remove(this); 352   7 m_->waiters_.remove(this);
HITCBC 371   7 active_ = false; 353   7 active_ = false;
HITCBC 372   14 return {make_error_code( 354   14 return {make_error_code(
HITCBC 373   7 error::canceled)}; 355   7 error::canceled)};
374   } 356   }
HITCBC 375   9 active_ = false; 357   9 active_ = false;
376   } 358   }
HITCBC 377   28 if(canceled_) 359   28 if(canceled_)
HITCBC 378   4 return {make_error_code( 360   4 return {make_error_code(
HITCBC 379   2 error::canceled)}; 361   2 error::canceled)};
HITCBC 380   26 return {{}}; 362   26 return {{}};
381   } 363   }
382   }; 364   };
383   365  
384   /** Unlocks the mutex automatically when destroyed. 366   /** Unlocks the mutex automatically when destroyed.
385   */ 367   */
386   class [[nodiscard]] lock_guard 368   class [[nodiscard]] lock_guard
387   { 369   {
388   async_mutex* m_; 370   async_mutex* m_;
389   371  
390   public: 372   public:
391   /// Unlock the mutex, if this guard holds one. 373   /// Unlock the mutex, if this guard holds one.
HITCBC 392   9 ~lock_guard() 374   9 ~lock_guard()
393   { 375   {
HITCBC 394   9 if(m_) 376   9 if(m_)
HITCBC 395   2 m_->unlock(); 377   2 m_->unlock();
HITCBC 396   9 } 378   9 }
397   379  
398   /// Construct a guard that holds no mutex. 380   /// Construct a guard that holds no mutex.
HITCBC 399   2 lock_guard() noexcept 381   2 lock_guard() noexcept
HITCBC 400   2 : m_(nullptr) 382   2 : m_(nullptr)
401   { 383   {
HITCBC 402   2 } 384   2 }
403   385  
404   /** Construct a guard that releases the given mutex on destruction. 386   /** Construct a guard that releases the given mutex on destruction.
405   387  
406   Adopts an already-held lock; it does not acquire one. 388   Adopts an already-held lock; it does not acquire one.
407   389  
408   @param m The mutex to unlock on destruction. It must outlive 390   @param m The mutex to unlock on destruction. It must outlive
409   the guard. 391   the guard.
410   */ 392   */
HITCBC 411   2 explicit lock_guard(async_mutex* m) noexcept 393   2 explicit lock_guard(async_mutex* m) noexcept
HITCBC 412   2 : m_(m) 394   2 : m_(m)
413   { 395   {
HITCBC 414   2 } 396   2 }
415   397  
416   /** Construct by moving, transferring the lock. 398   /** Construct by moving, transferring the lock.
417   399  
418   @par Postconditions 400   @par Postconditions
419   `o` holds no mutex, and its destructor unlocks nothing. 401   `o` holds no mutex, and its destructor unlocks nothing.
420   402  
421   @param o The guard to move from. 403   @param o The guard to move from.
422   */ 404   */
HITCBC 423   5 lock_guard(lock_guard&& o) noexcept 405   5 lock_guard(lock_guard&& o) noexcept
HITCBC 424   5 : m_(std::exchange(o.m_, nullptr)) 406   5 : m_(std::exchange(o.m_, nullptr))
425   { 407   {
HITCBC 426   5 } 408   5 }
427   409  
428   /** Assign by moving, transferring the lock. 410   /** Assign by moving, transferring the lock.
429   411  
430   If this guard already holds a mutex, that mutex is unlocked 412   If this guard already holds a mutex, that mutex is unlocked
431   first. Self-assignment is a no-op. 413   first. Self-assignment is a no-op.
432   414  
433   @par Postconditions 415   @par Postconditions
434   `o` holds no mutex, and its destructor unlocks nothing. 416   `o` holds no mutex, and its destructor unlocks nothing.
435   417  
436   @param o The guard to move from. 418   @param o The guard to move from.
437   419  
438   @return A reference to `*this`. 420   @return A reference to `*this`.
439   */ 421   */
440   lock_guard& operator=(lock_guard&& o) noexcept 422   lock_guard& operator=(lock_guard&& o) noexcept
441   { 423   {
442   if(this != &o) 424   if(this != &o)
443   { 425   {
444   if(m_) 426   if(m_)
445   m_->unlock(); 427   m_->unlock();
446   m_ = std::exchange(o.m_, nullptr); 428   m_ = std::exchange(o.m_, nullptr);
447   } 429   }
448   return *this; 430   return *this;
449   } 431   }
450   432  
451   /** Copy construction is disabled; a guard uniquely owns the lock. 433   /** Copy construction is disabled; a guard uniquely owns the lock.
452   434  
453   @param other The guard that would be copied. 435   @param other The guard that would be copied.
454   */ 436   */
455   lock_guard(lock_guard const& other) = delete; 437   lock_guard(lock_guard const& other) = delete;
456   438  
457   /** Copy assignment is disabled; a guard uniquely owns the lock. 439   /** Copy assignment is disabled; a guard uniquely owns the lock.
458   440  
459   @param other The guard that would be assigned from. 441   @param other The guard that would be assigned from.
460   442  
461   @return A reference to `*this`. 443   @return A reference to `*this`.
462   */ 444   */
463   lock_guard& operator=(lock_guard const& other) = delete; 445   lock_guard& operator=(lock_guard const& other) = delete;
464   }; 446   };
465   447  
466   /** Acquires the mutex like `lock_awaiter`, then resumes with a `lock_guard` that unlocks it. 448   /** Acquires the mutex like `lock_awaiter`, then resumes with a `lock_guard` that unlocks it.
467   */ 449   */
468   class lock_guard_awaiter 450   class lock_guard_awaiter
469   { 451   {
470   async_mutex* m_; 452   async_mutex* m_;
471   lock_awaiter inner_; 453   lock_awaiter inner_;
472   454  
473   public: 455   public:
474   /** Construct an awaiter for the given mutex. 456   /** Construct an awaiter for the given mutex.
475   457  
476   @param m The mutex to acquire. It must outlive the awaiter. 458   @param m The mutex to acquire. It must outlive the awaiter.
477   */ 459   */
HITCBC 478   4 explicit lock_guard_awaiter(async_mutex* m) noexcept 460   4 explicit lock_guard_awaiter(async_mutex* m) noexcept
HITCBC 479   4 : m_(m) 461   4 : m_(m)
HITCBC 480   4 , inner_(m) 462   4 , inner_(m)
481   { 463   {
HITCBC 482   4 } 464   4 }
483   465  
484   /** Acquire the mutex if it is free, reporting whether to suspend. 466   /** Acquire the mutex if it is free, reporting whether to suspend.
485   467  
486   Delegates to @ref lock_awaiter::await_ready, so as there this is 468   Delegates to @ref lock_awaiter::await_ready, so as there this is
487   not a pure query: on the fast path it takes the lock. 469   not a pure query: on the fast path it takes the lock.
488   470  
489   @return `true` if the mutex was free and is now held by the 471   @return `true` if the mutex was free and is now held by the
490   awaiting coroutine. `false` if the mutex is held elsewhere, in 472   awaiting coroutine. `false` if the mutex is held elsewhere, in
491   which case the coroutine suspends. 473   which case the coroutine suspends.
492   */ 474   */
HITCBC 493   4 bool await_ready() const noexcept 475   4 bool await_ready() const noexcept
494   { 476   {
HITCBC 495   4 return inner_.await_ready(); 477   4 return inner_.await_ready();
496   } 478   }
497   479  
498   /** Enqueue the awaiting coroutine until the mutex is released. 480   /** Enqueue the awaiting coroutine until the mutex is released.
499   481  
500   This is the @ref IoAwaitable overload of `await_suspend`. It 482   This is the @ref IoAwaitable overload of `await_suspend`. It
501   delegates to @ref lock_awaiter::await_suspend on the wrapped 483   delegates to @ref lock_awaiter::await_suspend on the wrapped
502   awaiter, so it has that function's contract. 484   awaiter, so it has that function's contract.
503   485  
504   @param h The awaiting coroutine, resumed when the mutex is 486   @param h The awaiting coroutine, resumed when the mutex is
505   acquired or the wait is canceled. 487   acquired or the wait is canceled.
506   488  
507   @param env The execution environment. Its executor posts the 489   @param env The execution environment. Its executor posts the
508   resumption and its stop token is watched for the duration of 490   resumption and its stop token is watched for the duration of
509   the wait. It must outlive the wait. 491   the wait. It must outlive the wait.
510   492  
511   @return `h` if a stop request was already pending, which 493   @return `h` if a stop request was already pending, which
512   resumes the awaiting coroutine immediately without enqueuing 494   resumes the awaiting coroutine immediately without enqueuing
513   it. Otherwise `std::noop_coroutine()`, which leaves the 495   it. Otherwise `std::noop_coroutine()`, which leaves the
514   coroutine suspended and returns control to the resumer. 496   coroutine suspended and returns control to the resumer.
515   */ 497   */
516   std::coroutine_handle<> 498   std::coroutine_handle<>
HITCBC 517   2 await_suspend( 499   2 await_suspend(
518   std::coroutine_handle<> h, 500   std::coroutine_handle<> h,
519   io_env const* env) noexcept 501   io_env const* env) noexcept
520   { 502   {
HITCBC 521   2 return inner_.await_suspend(h, env); 503   2 return inner_.await_suspend(h, env);
522   } 504   }
523   505  
524   /** Complete the acquisition and report the outcome. 506   /** Complete the acquisition and report the outcome.
525   507  
526   @return An `io_result<lock_guard>` destructuring as 508   @return An `io_result<lock_guard>` destructuring as
527   `[ec, guard]`. On success `ec` is empty and `guard` holds the 509   `[ec, guard]`. On success `ec` is empty and `guard` holds the
528   mutex, releasing it when destroyed. If the wait was canceled, 510   mutex, releasing it when destroyed. If the wait was canceled,
529   `ec` is `error::canceled` and `guard` holds no mutex. 511   `ec` is `error::canceled` and `guard` holds no mutex.
530   */ 512   */
HITCBC 531   4 [[nodiscard]] io_result<lock_guard> await_resume() noexcept 513   4 [[nodiscard]] io_result<lock_guard> await_resume() noexcept
532   { 514   {
HITCBC 533   4 auto r = inner_.await_resume(); 515   4 auto r = inner_.await_resume();
HITCBC 534   4 if(std::get<0>(r)) 516   4 if(std::get<0>(r))
HITCBC 535   2 return {std::get<0>(r), lock_guard()}; 517   2 return {std::get<0>(r), lock_guard()};
HITCBC 536   2 return {std::error_code(), lock_guard(m_)}; 518   2 return {std::error_code(), lock_guard(m_)};
537   } 519   }
538   }; 520   };
539   521  
540   /// Construct an unlocked mutex. 522   /// Construct an unlocked mutex.
541   async_mutex() = default; 523   async_mutex() = default;
542   524  
543   /** Copy construction is disabled; suspended waiters point into the 525   /** Copy construction is disabled; suspended waiters point into the
544   mutex's wait queue. 526   mutex's wait queue.
545   527  
546   @param other The mutex that would be copied. 528   @param other The mutex that would be copied.
547   */ 529   */
548   async_mutex(async_mutex const& other) = delete; 530   async_mutex(async_mutex const& other) = delete;
549   531  
550   /** Copy assignment is disabled; suspended waiters point into the 532   /** Copy assignment is disabled; suspended waiters point into the
551   mutex's wait queue. 533   mutex's wait queue.
552   534  
553   @param other The mutex that would be assigned from. 535   @param other The mutex that would be assigned from.
554   536  
555   @return A reference to `*this`. 537   @return A reference to `*this`.
556   */ 538   */
557   async_mutex& operator=(async_mutex const& other) = delete; 539   async_mutex& operator=(async_mutex const& other) = delete;
558   540  
559   /** Move construction is disabled; suspended waiters point into the 541   /** Move construction is disabled; suspended waiters point into the
560   mutex's wait queue. 542   mutex's wait queue.
561   543  
562   @param other The mutex that would be moved from. 544   @param other The mutex that would be moved from.
563   */ 545   */
564   async_mutex(async_mutex&& other) = delete; 546   async_mutex(async_mutex&& other) = delete;
565   547  
566   /** Move assignment is disabled; suspended waiters point into the 548   /** Move assignment is disabled; suspended waiters point into the
567   mutex's wait queue. 549   mutex's wait queue.
568   550  
569   @param other The mutex that would be moved from. 551   @param other The mutex that would be moved from.
570   552  
571   @return A reference to `*this`. 553   @return A reference to `*this`.
572   */ 554   */
573   async_mutex& operator=(async_mutex&& other) = delete; 555   async_mutex& operator=(async_mutex&& other) = delete;
574   556  
575   /** Returns an awaiter that acquires the mutex. 557   /** Returns an awaiter that acquires the mutex.
576   558  
577   @return An awaitable that await-returns `(error_code)`. 559   @return An awaitable that await-returns `(error_code)`.
578   */ 560   */
HITCBC 579   34 lock_awaiter lock() noexcept 561   34 lock_awaiter lock() noexcept
580   { 562   {
HITCBC 581   34 return lock_awaiter{this}; 563   34 return lock_awaiter{this};
582   } 564   }
583   565  
584   /** Returns an awaiter that acquires the mutex with RAII. 566   /** Returns an awaiter that acquires the mutex with RAII.
585   567  
586   @return An awaitable that await-returns `(error_code,lock_guard)`. 568   @return An awaitable that await-returns `(error_code,lock_guard)`.
587   */ 569   */
HITCBC 588   4 lock_guard_awaiter scoped_lock() noexcept 570   4 lock_guard_awaiter scoped_lock() noexcept
589   { 571   {
HITCBC 590   4 return lock_guard_awaiter(this); 572   4 return lock_guard_awaiter(this);
591   } 573   }
592   574  
593   /** Releases the mutex. 575   /** Releases the mutex.
594   576  
595   If waiters are queued, the next eligible waiter is 577   If waiters are queued, the next eligible waiter is
596   resumed with the lock held. Canceled waiters are 578   resumed with the lock held. Canceled waiters are
597   skipped. If no eligible waiter remains, the mutex 579   skipped. If no eligible waiter remains, the mutex
598   becomes unlocked. 580   becomes unlocked.
599   */ 581   */
HITCBC 600   26 void unlock() noexcept 582   26 void unlock() noexcept
601   { 583   {
602   for(;;) 584   for(;;)
603   { 585   {
HITCBC 604   27 auto* waiter = waiters_.pop_front(); 586   27 auto* waiter = waiters_.pop_front();
HITCBC 605   27 if(!waiter) 587   27 if(!waiter)
606   { 588   {
HITCBC 607   17 locked_ = false; 589   17 locked_ = false;
HITCBC 608   17 return; 590   17 return;
609   } 591   }
HITCBC 610   10 if(!waiter->claimed_.exchange( 592   10 if(!waiter->claimed_.exchange(
611   true, std::memory_order_acq_rel)) 593   true, std::memory_order_acq_rel))
612   { 594   {
HITCBC 613   9 waiter->ex_.post(waiter->cont_); 595   9 waiter->ex_.post(waiter->cont_);
HITCBC 614   9 return; 596   9 return;
615   } 597   }
HITCBC 616   1 } 598   1 }
617   } 599   }
618   600  
619   /** Returns true if the mutex is currently locked. 601   /** Returns true if the mutex is currently locked.
620   602  
621   @return `true` if the mutex is held; otherwise `false`. 603   @return `true` if the mutex is held; otherwise `false`.
622   */ 604   */
HITCBC 623   27 bool is_locked() const noexcept 605   27 bool is_locked() const noexcept
624   { 606   {
HITCBC 625   27 return locked_; 607   27 return locked_;
626   } 608   }
627   }; 609   };
628   610  
629   } // namespace capy 611   } // namespace capy
630   } // namespace boost 612   } // namespace boost
631   613  
632   #endif 614   #endif