. /include/boost/capy /ex /async_mutex.hpp
100.00% Lines (135/135) 100.00% Functions (23/23)
[ ] 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/await_context.hpp> 19   #include <boost/capy/ex/await_context.hpp>
20   #include <boost/capy/results.hpp> 20   #include <boost/capy/results.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   @par !example example 128   @par !example example
129   129  
130   */ 130   */
131   class async_mutex 131   class async_mutex
132   { 132   {
133   public: 133   public:
134   class lock_awaiter; 134   class lock_awaiter;
135   class lock_guard; 135   class lock_guard;
136   class lock_guard_awaiter; 136   class lock_guard_awaiter;
137   137  
138   private: 138   private:
139   bool locked_ = false; 139   bool locked_ = false;
140   detail::intrusive_list<lock_awaiter> waiters_; 140   detail::intrusive_list<lock_awaiter> waiters_;
141   // Handed the lock but not yet resumed; the destructor must reach it. 141   // Handed the lock but not yet resumed; the destructor must reach it.
142   lock_awaiter* granted_to_ = nullptr; 142   lock_awaiter* granted_to_ = nullptr;
143   143  
144   // State of the unlock() in progress, on its own stack: a post can 144   // State of the unlock() in progress, on its own stack: a post can
145   // destroy the granted waiter, re-entering unlock(), or this mutex. 145   // destroy the granted waiter, re-entering unlock(), or this mutex.
146   struct unlock_scope 146   struct unlock_scope
147   { 147   {
148   bool again = false; 148   bool again = false;
149   bool gone = false; 149   bool gone = false;
150   }; 150   };
151   unlock_scope* unlocking_ = nullptr; 151   unlock_scope* unlocking_ = nullptr;
152   152  
153   public: 153   public:
154   /** Suspends the caller until the mutex is free, or resumes it with `error::canceled` on a stop request. 154   /** Suspends the caller until the mutex is free, or resumes it with `error::canceled` on a stop request.
155   */ 155   */
156   class lock_awaiter 156   class lock_awaiter
157   : public detail::intrusive_list<lock_awaiter>::node 157   : public detail::intrusive_list<lock_awaiter>::node
158   { 158   {
159   friend class async_mutex; 159   friend class async_mutex;
160   160  
161   async_mutex* m_; 161   async_mutex* m_;
162   continuation cont_; 162   continuation cont_;
163   executor_ref ex_; 163   executor_ref ex_;
164   164  
165   // These members must be declared before stop_cb_ 165   // These members must be declared before stop_cb_
166   // (see comment on the union below). 166   // (see comment on the union below).
167   std::atomic<bool> claimed_{false}; 167   std::atomic<bool> claimed_{false};
168   bool canceled_ = false; 168   bool canceled_ = false;
169   bool active_ = false; 169   bool active_ = false;
170   170  
171   // Set by unlock() when it hands the lock to this waiter. 171   // Set by unlock() when it hands the lock to this waiter.
172   bool granted_ = false; 172   bool granted_ = false;
173   173  
174   // Cleared by whoever pops the node, so a waiter outliving 174   // Cleared by whoever pops the node, so a waiter outliving
175   // its mutex never reaches through m_. 175   // its mutex never reaches through m_.
176   bool in_list_ = false; 176   bool in_list_ = false;
177   177  
178   struct cancel_fn 178   struct cancel_fn
179   { 179   {
180   lock_awaiter* self_; 180   lock_awaiter* self_;
181   181  
HITCBC 182   8 void operator()() const noexcept 182   8 void operator()() const noexcept
183   { 183   {
HITCBC 184   8 if(!self_->claimed_.exchange( 184   8 if(!self_->claimed_.exchange(
185   true, std::memory_order_acq_rel)) 185   true, std::memory_order_acq_rel))
186   { 186   {
HITCBC 187   8 self_->canceled_ = true; 187   8 self_->canceled_ = true;
HITCBC 188   8 self_->ex_.post(self_->cont_); 188   8 self_->ex_.post(self_->cont_);
189   } 189   }
HITCBC 190   8 } 190   8 }
191   }; 191   };
192   192  
193   using stop_cb_t = 193   using stop_cb_t =
194   std::stop_callback<cancel_fn>; 194   std::stop_callback<cancel_fn>;
195   195  
196   // Aligned storage for stop_cb_t. Declared last: 196   // Aligned storage for stop_cb_t. Declared last:
197   // its destructor may block while the callback 197   // its destructor may block while the callback
198   // accesses the members above. 198   // accesses the members above.
199   BOOST_CAPY_MSVC_WARNING_PUSH 199   BOOST_CAPY_MSVC_WARNING_PUSH
200   BOOST_CAPY_MSVC_WARNING_DISABLE(4324) // padded due to alignas 200   BOOST_CAPY_MSVC_WARNING_DISABLE(4324) // padded due to alignas
201   alignas(stop_cb_t) 201   alignas(stop_cb_t)
202   unsigned char stop_cb_buf_[sizeof(stop_cb_t)]; 202   unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
203   BOOST_CAPY_MSVC_WARNING_POP 203   BOOST_CAPY_MSVC_WARNING_POP
204   204  
HITCBC 205   100026 stop_cb_t& stop_cb_() noexcept 205   100026 stop_cb_t& stop_cb_() noexcept
206   { 206   {
207   return *reinterpret_cast<stop_cb_t*>( 207   return *reinterpret_cast<stop_cb_t*>(
HITCBC 208   100026 stop_cb_buf_); 208   100026 stop_cb_buf_);
209   } 209   }
210   210  
211   public: 211   public:
212   /** Destroy the awaiter, leaving the mutex unable to reach it. 212   /** Destroy the awaiter, leaving the mutex unable to reach it.
213   213  
214   If the awaiter is suspended in the wait queue, destroys the 214   If the awaiter is suspended in the wait queue, destroys the
215   stop callback and unlinks the awaiter. Neither `unlock()` nor 215   stop callback and unlinks the awaiter. Neither `unlock()` nor
216   the stop callback can then reach a destroyed awaiter when the 216   the stop callback can then reach a destroyed awaiter when the
217   coroutine frame is torn down while suspended. 217   coroutine frame is torn down while suspended.
218   218  
219   If the lock was handed to this awaiter and the coroutine is 219   If the lock was handed to this awaiter and the coroutine is
220   destroyed before it resumes, the lock is released to the next 220   destroyed before it resumes, the lock is released to the next
221   waiter. 221   waiter.
222   222  
223   @par Preconditions 223   @par Preconditions
224   Called on the executor thread. The stop callback may fire from 224   Called on the executor thread. The stop callback may fire from
225   any thread, so destroying a still-suspended awaiter from 225   any thread, so destroying a still-suspended awaiter from
226   another thread is undefined. 226   another thread is undefined.
227   */ 227   */
HITCBC 228   200099 ~lock_awaiter() 228   200099 ~lock_awaiter()
229   { 229   {
HITCBC 230   200099 if(active_) 230   200099 if(active_)
231   { 231   {
HITCBC 232   100009 stop_cb_().~stop_cb_t(); 232   100009 stop_cb_().~stop_cb_t();
233   } 233   }
HITCBC 234   200099 if(in_list_) 234   200099 if(in_list_)
HITCBC 235   3 m_->waiters_.remove(this); 235   3 m_->waiters_.remove(this);
236   // Handed the lock but destroyed before resuming: pass it on. 236   // Handed the lock but destroyed before resuming: pass it on.
HITCBC 237   200099 if(granted_) 237   200099 if(granted_)
238   { 238   {
HITCBC 239   100002 m_->granted_to_ = nullptr; 239   100002 m_->granted_to_ = nullptr;
HITCBC 240   100002 m_->unlock(); 240   100002 m_->unlock();
241   } 241   }
HITCBC 242   200099 } 242   200099 }
243   243  
244   /** Construct an awaiter for the given mutex. 244   /** Construct an awaiter for the given mutex.
245   245  
246   @param m The mutex to acquire. It must outlive the awaiter. 246   @param m The mutex to acquire. It must outlive the awaiter.
247   */ 247   */
HITCBC 248   100052 explicit lock_awaiter(async_mutex* m) noexcept 248   100052 explicit lock_awaiter(async_mutex* m) noexcept
HITCBC 249   100052 : m_(m) 249   100052 : m_(m)
250   { 250   {
HITCBC 251   100052 } 251   100052 }
252   252  
253   /** Construct by moving. 253   /** Construct by moving.
254   254  
255   The moved-from awaiter is left inert: its destructor no longer 255   The moved-from awaiter is left inert: its destructor no longer
256   destroys the stop callback and no longer unlinks from the 256   destroys the stop callback and no longer unlinks from the
257   mutex's wait queue. 257   mutex's wait queue.
258   258  
259   @param o The awaiter to move from. 259   @param o The awaiter to move from.
260   */ 260   */
HITCBC 261   100047 lock_awaiter(lock_awaiter&& o) noexcept 261   100047 lock_awaiter(lock_awaiter&& o) noexcept
HITCBC 262   200094 : m_(o.m_) 262   200094 : m_(o.m_)
HITCBC 263   100047 , cont_(o.cont_) 263   100047 , cont_(o.cont_)
HITCBC 264   100047 , ex_(o.ex_) 264   100047 , ex_(o.ex_)
HITCBC 265   100047 , claimed_(o.claimed_.load( 265   100047 , claimed_(o.claimed_.load(
266   std::memory_order_relaxed)) 266   std::memory_order_relaxed))
HITCBC 267   100047 , canceled_(o.canceled_) 267   100047 , canceled_(o.canceled_)
HITCBC 268   100047 , active_(std::exchange(o.active_, false)) 268   100047 , active_(std::exchange(o.active_, false))
HITCBC 269   100047 , granted_(std::exchange(o.granted_, false)) 269   100047 , granted_(std::exchange(o.granted_, false))
HITCBC 270   200094 , in_list_(std::exchange(o.in_list_, false)) 270   200094 , in_list_(std::exchange(o.in_list_, false))
271   { 271   {
HITCBC 272   100047 } 272   100047 }
273   273  
274   /** Copy construction is disabled; a waiter is linked into the 274   /** Copy construction is disabled; a waiter is linked into the
275   mutex's wait queue by address. 275   mutex's wait queue by address.
276   276  
277   @param other The awaiter that would be copied. 277   @param other The awaiter that would be copied.
278   */ 278   */
279   lock_awaiter(lock_awaiter const& other) = delete; 279   lock_awaiter(lock_awaiter const& other) = delete;
280   280  
281   /** Copy assignment is disabled; a waiter is linked into the 281   /** Copy assignment is disabled; a waiter is linked into the
282   mutex's wait queue by address. 282   mutex's wait queue by address.
283   283  
284   @param other The awaiter that would be assigned from. 284   @param other The awaiter that would be assigned from.
285   285  
286   @return A reference to `*this`. 286   @return A reference to `*this`.
287   */ 287   */
288   lock_awaiter& operator=(lock_awaiter const& other) = delete; 288   lock_awaiter& operator=(lock_awaiter const& other) = delete;
289   289  
290   /** Move assignment is disabled; a waiter is linked into the 290   /** Move assignment is disabled; a waiter is linked into the
291   mutex's wait queue by address. 291   mutex's wait queue by address.
292   292  
293   @param other The awaiter that would be moved from. 293   @param other The awaiter that would be moved from.
294   294  
295   @return A reference to `*this`. 295   @return A reference to `*this`.
296   */ 296   */
297   lock_awaiter& operator=(lock_awaiter&& other) = delete; 297   lock_awaiter& operator=(lock_awaiter&& other) = delete;
298   298  
299   /** Acquire the mutex if it is free, reporting whether to suspend. 299   /** Acquire the mutex if it is free, reporting whether to suspend.
300   300  
301   This is not a pure query: on the fast path it takes the lock. 301   This is not a pure query: on the fast path it takes the lock.
302   When the mutex is unlocked, it marks the mutex locked and 302   When the mutex is unlocked, it marks the mutex locked and
303   reports that no suspension is needed. The stop token is not 303   reports that no suspension is needed. The stop token is not
304   consulted, so an uncontended `lock()` succeeds even when stop 304   consulted, so an uncontended `lock()` succeeds even when stop
305   has already been requested. 305   has already been requested.
306   306  
307   @return `true` if the mutex was free and is now held by the 307   @return `true` if the mutex was free and is now held by the
308   awaiting coroutine. `false` if the mutex is held elsewhere, in 308   awaiting coroutine. `false` if the mutex is held elsewhere, in
309   which case the coroutine suspends. 309   which case the coroutine suspends.
310   */ 310   */
HITCBC 311   100052 bool await_ready() const noexcept 311   100052 bool await_ready() const noexcept
312   { 312   {
HITCBC 313   100052 if(!m_->locked_) 313   100052 if(!m_->locked_)
314   { 314   {
HITCBC 315   24 m_->locked_ = true; 315   24 m_->locked_ = true;
HITCBC 316   24 return true; 316   24 return true;
317   } 317   }
HITCBC 318   100028 return false; 318   100028 return false;
319   } 319   }
320   320  
321   /** Enqueue the awaiting coroutine until the mutex is released. 321   /** Enqueue the awaiting coroutine until the mutex is released.
322   322  
323   This is the @ref ContextAwaitable overload of `await_suspend`. 323   This is the @ref ContextAwaitable overload of `await_suspend`.
324   324  
325   If a stop request is already pending on `env->stop_token`, the 325   If a stop request is already pending on `env->stop_token`, the
326   awaiter records the cancellation and does not enqueue. The 326   awaiter records the cancellation and does not enqueue. The
327   mutex is not acquired. 327   mutex is not acquired.
328   328  
329   Otherwise it stores `h` and `env->executor`, links itself into 329   Otherwise it stores `h` and `env->executor`, links itself into
330   the back of the mutex's wait queue, and registers a stop 330   the back of the mutex's wait queue, and registers a stop
331   callback on `env->stop_token`. Whichever of `unlock()` and that 331   callback on `env->stop_token`. Whichever of `unlock()` and that
332   callback claims the awaiter first posts `h` through the stored 332   callback claims the awaiter first posts `h` through the stored
333   executor; the other skips it. 333   executor; the other skips it.
334   334  
335   @param h The awaiting coroutine, resumed when the mutex is 335   @param h The awaiting coroutine, resumed when the mutex is
336   acquired or the wait is canceled. 336   acquired or the wait is canceled.
337   337  
338   @param env The execution environment. Its executor posts the 338   @param env The execution environment. Its executor posts the
339   resumption and its stop token is watched for the duration of 339   resumption and its stop token is watched for the duration of
340   the wait. It must outlive the wait. 340   the wait. It must outlive the wait.
341   341  
342   @return `h` if a stop request was already pending, which 342   @return `h` if a stop request was already pending, which
343   resumes the awaiting coroutine immediately without enqueuing 343   resumes the awaiting coroutine immediately without enqueuing
344   it. Otherwise `std::noop_coroutine()`, which leaves the 344   it. Otherwise `std::noop_coroutine()`, which leaves the
345   coroutine suspended and returns control to the resumer. 345   coroutine suspended and returns control to the resumer.
346   */ 346   */
347   std::coroutine_handle<> 347   std::coroutine_handle<>
HITCBC 348   100028 await_suspend( 348   100028 await_suspend(
349   std::coroutine_handle<> h, 349   std::coroutine_handle<> h,
350   await_context const* env) noexcept 350   await_context const* env) noexcept
351   { 351   {
HITCBC 352   100028 if(env->stop_token.stop_requested()) 352   100028 if(env->stop_token.stop_requested())
353   { 353   {
HITCBC 354   2 canceled_ = true; 354   2 canceled_ = true;
HITCBC 355   2 return h; 355   2 return h;
356   } 356   }
HITCBC 357   100026 cont_.h = h; 357   100026 cont_.h = h;
HITCBC 358   100026 cont_.owner = env->owner; 358   100026 cont_.owner = env->owner;
HITCBC 359   100026 ex_ = env->executor; 359   100026 ex_ = env->executor;
HITCBC 360   100026 m_->waiters_.push_back(this); 360   100026 m_->waiters_.push_back(this);
HITCBC 361   100026 in_list_ = true; 361   100026 in_list_ = true;
HITCBC 362   300078 ::new(stop_cb_buf_) stop_cb_t( 362   300078 ::new(stop_cb_buf_) stop_cb_t(
HITCBC 363   100026 env->stop_token, cancel_fn{this}); 363   100026 env->stop_token, cancel_fn{this});
HITCBC 364   100026 active_ = true; 364   100026 active_ = true;
HITCBC 365   100026 return std::noop_coroutine(); 365   100026 return std::noop_coroutine();
366   } 366   }
367   367  
368   /** Complete the acquisition and report the outcome. 368   /** Complete the acquisition and report the outcome.
369   369  
370   Destroys the stop callback if one is registered, and unlinks a 370   Destroys the stop callback if one is registered, and unlinks a
371   canceled awaiter from the wait queue. 371   canceled awaiter from the wait queue.
372   372  
373   @return An empty `results<>` if the mutex is now held by the 373   @return An empty `results<>` if the mutex is now held by the
374   awaiting coroutine. Otherwise one holding `error::canceled`, 374   awaiting coroutine. Otherwise one holding `error::canceled`,
375   which means the stop token won the race or the mutex was 375   which means the stop token won the race or the mutex was
376   destroyed, and the mutex is not held. 376   destroyed, and the mutex is not held.
377   */ 377   */
HITCBC 378   38 [[nodiscard]] results<> await_resume() noexcept 378   38 [[nodiscard]] results<> await_resume() noexcept
379   { 379   {
HITCBC 380   38 if(std::exchange(granted_, false)) 380   38 if(std::exchange(granted_, false))
HITCBC 381   9 m_->granted_to_ = nullptr; 381   9 m_->granted_to_ = nullptr;
HITCBC 382   38 if(active_) 382   38 if(active_)
383   { 383   {
HITCBC 384   17 stop_cb_().~stop_cb_t(); 384   17 stop_cb_().~stop_cb_t();
HITCBC 385   17 if(canceled_) 385   17 if(canceled_)
386   { 386   {
HITCBC 387   8 if(in_list_) 387   8 if(in_list_)
388   { 388   {
HITCBC 389   6 m_->waiters_.remove(this); 389   6 m_->waiters_.remove(this);
HITCBC 390   6 in_list_ = false; 390   6 in_list_ = false;
391   } 391   }
HITCBC 392   8 active_ = false; 392   8 active_ = false;
HITCBC 393   16 return {make_error_code( 393   16 return {make_error_code(
HITCBC 394   8 error::canceled)}; 394   8 error::canceled)};
395   } 395   }
HITCBC 396   9 active_ = false; 396   9 active_ = false;
397   } 397   }
HITCBC 398   30 if(canceled_) 398   30 if(canceled_)
HITCBC 399   4 return {make_error_code( 399   4 return {make_error_code(
HITCBC 400   2 error::canceled)}; 400   2 error::canceled)};
HITCBC 401   28 return {{}}; 401   28 return {{}};
402   } 402   }
403   }; 403   };
404   404  
405   /** Unlocks the mutex automatically when destroyed. 405   /** Unlocks the mutex automatically when destroyed.
406   */ 406   */
407   class [[nodiscard]] lock_guard 407   class [[nodiscard]] lock_guard
408   { 408   {
409   async_mutex* m_; 409   async_mutex* m_;
410   410  
411   public: 411   public:
412   /// Unlock the mutex, if this guard holds one. 412   /// Unlock the mutex, if this guard holds one.
HITCBC 413   11 ~lock_guard() 413   11 ~lock_guard()
414   { 414   {
HITCBC 415   11 if(m_) 415   11 if(m_)
HITCBC 416   3 m_->unlock(); 416   3 m_->unlock();
HITCBC 417   11 } 417   11 }
418   418  
419   /// Construct a guard that holds no mutex. 419   /// Construct a guard that holds no mutex.
HITCBC 420   2 lock_guard() noexcept 420   2 lock_guard() noexcept
HITCBC 421   2 : m_(nullptr) 421   2 : m_(nullptr)
422   { 422   {
HITCBC 423   2 } 423   2 }
424   424  
425   /** Construct a guard that releases the given mutex on destruction. 425   /** Construct a guard that releases the given mutex on destruction.
426   426  
427   Adopts an already-held lock; it does not acquire one. 427   Adopts an already-held lock; it does not acquire one.
428   428  
429   @param m The mutex to unlock on destruction. It must outlive 429   @param m The mutex to unlock on destruction. It must outlive
430   the guard. 430   the guard.
431   */ 431   */
HITCBC 432   3 explicit lock_guard(async_mutex* m) noexcept 432   3 explicit lock_guard(async_mutex* m) noexcept
HITCBC 433   3 : m_(m) 433   3 : m_(m)
434   { 434   {
HITCBC 435   3 } 435   3 }
436   436  
437   /** Construct by moving, transferring the lock. 437   /** Construct by moving, transferring the lock.
438   438  
439   @par Postconditions 439   @par Postconditions
440   `o` holds no mutex, and its destructor unlocks nothing. 440   `o` holds no mutex, and its destructor unlocks nothing.
441   441  
442   @param o The guard to move from. 442   @param o The guard to move from.
443   */ 443   */
HITCBC 444   6 lock_guard(lock_guard&& o) noexcept 444   6 lock_guard(lock_guard&& o) noexcept
HITCBC 445   6 : m_(std::exchange(o.m_, nullptr)) 445   6 : m_(std::exchange(o.m_, nullptr))
446   { 446   {
HITCBC 447   6 } 447   6 }
448   448  
449   /** Assign by moving, transferring the lock. 449   /** Assign by moving, transferring the lock.
450   450  
451   If this guard already holds a mutex, that mutex is unlocked 451   If this guard already holds a mutex, that mutex is unlocked
452   first. Self-assignment is a no-op. 452   first. Self-assignment is a no-op.
453   453  
454   @par Postconditions 454   @par Postconditions
455   `o` holds no mutex, and its destructor unlocks nothing. 455   `o` holds no mutex, and its destructor unlocks nothing.
456   456  
457   @param o The guard to move from. 457   @param o The guard to move from.
458   458  
459   @return A reference to `*this`. 459   @return A reference to `*this`.
460   */ 460   */
461   lock_guard& operator=(lock_guard&& o) noexcept 461   lock_guard& operator=(lock_guard&& o) noexcept
462   { 462   {
463   if(this != &o) 463   if(this != &o)
464   { 464   {
465   if(m_) 465   if(m_)
466   m_->unlock(); 466   m_->unlock();
467   m_ = std::exchange(o.m_, nullptr); 467   m_ = std::exchange(o.m_, nullptr);
468   } 468   }
469   return *this; 469   return *this;
470   } 470   }
471   471  
472   /** Copy construction is disabled; a guard uniquely owns the lock. 472   /** Copy construction is disabled; a guard uniquely owns the lock.
473   473  
474   @param other The guard that would be copied. 474   @param other The guard that would be copied.
475   */ 475   */
476   lock_guard(lock_guard const& other) = delete; 476   lock_guard(lock_guard const& other) = delete;
477   477  
478   /** Copy assignment is disabled; a guard uniquely owns the lock. 478   /** Copy assignment is disabled; a guard uniquely owns the lock.
479   479  
480   @param other The guard that would be assigned from. 480   @param other The guard that would be assigned from.
481   481  
482   @return A reference to `*this`. 482   @return A reference to `*this`.
483   */ 483   */
484   lock_guard& operator=(lock_guard const& other) = delete; 484   lock_guard& operator=(lock_guard const& other) = delete;
485   }; 485   };
486   486  
487   /** Acquires the mutex like `lock_awaiter`, then resumes with a `lock_guard` that unlocks it. 487   /** Acquires the mutex like `lock_awaiter`, then resumes with a `lock_guard` that unlocks it.
488   */ 488   */
489   class lock_guard_awaiter 489   class lock_guard_awaiter
490   { 490   {
491   async_mutex* m_; 491   async_mutex* m_;
492   lock_awaiter inner_; 492   lock_awaiter inner_;
493   493  
494   public: 494   public:
495   /** Construct an awaiter for the given mutex. 495   /** Construct an awaiter for the given mutex.
496   496  
497   @param m The mutex to acquire. It must outlive the awaiter. 497   @param m The mutex to acquire. It must outlive the awaiter.
498   */ 498   */
HITCBC 499   5 explicit lock_guard_awaiter(async_mutex* m) noexcept 499   5 explicit lock_guard_awaiter(async_mutex* m) noexcept
HITCBC 500   5 : m_(m) 500   5 : m_(m)
HITCBC 501   5 , inner_(m) 501   5 , inner_(m)
502   { 502   {
HITCBC 503   5 } 503   5 }
504   504  
505   /** Acquire the mutex if it is free, reporting whether to suspend. 505   /** Acquire the mutex if it is free, reporting whether to suspend.
506   506  
507   Delegates to @ref lock_awaiter::await_ready, so as there this is 507   Delegates to @ref lock_awaiter::await_ready, so as there this is
508   not a pure query: on the fast path it takes the lock. 508   not a pure query: on the fast path it takes the lock.
509   509  
510   @return `true` if the mutex was free and is now held by the 510   @return `true` if the mutex was free and is now held by the
511   awaiting coroutine. `false` if the mutex is held elsewhere, in 511   awaiting coroutine. `false` if the mutex is held elsewhere, in
512   which case the coroutine suspends. 512   which case the coroutine suspends.
513   */ 513   */
HITCBC 514   5 bool await_ready() const noexcept 514   5 bool await_ready() const noexcept
515   { 515   {
HITCBC 516   5 return inner_.await_ready(); 516   5 return inner_.await_ready();
517   } 517   }
518   518  
519   /** Enqueue the awaiting coroutine until the mutex is released. 519   /** Enqueue the awaiting coroutine until the mutex is released.
520   520  
521   This is the @ref ContextAwaitable overload of `await_suspend`. It 521   This is the @ref ContextAwaitable overload of `await_suspend`. It
522   delegates to @ref lock_awaiter::await_suspend on the wrapped 522   delegates to @ref lock_awaiter::await_suspend on the wrapped
523   awaiter, so it has that function's contract. 523   awaiter, so it has that function's contract.
524   524  
525   @param h The awaiting coroutine, resumed when the mutex is 525   @param h The awaiting coroutine, resumed when the mutex is
526   acquired or the wait is canceled. 526   acquired or the wait is canceled.
527   527  
528   @param env The execution environment. Its executor posts the 528   @param env The execution environment. Its executor posts the
529   resumption and its stop token is watched for the duration of 529   resumption and its stop token is watched for the duration of
530   the wait. It must outlive the wait. 530   the wait. It must outlive the wait.
531   531  
532   @return `h` if a stop request was already pending, which 532   @return `h` if a stop request was already pending, which
533   resumes the awaiting coroutine immediately without enqueuing 533   resumes the awaiting coroutine immediately without enqueuing
534   it. Otherwise `std::noop_coroutine()`, which leaves the 534   it. Otherwise `std::noop_coroutine()`, which leaves the
535   coroutine suspended and returns control to the resumer. 535   coroutine suspended and returns control to the resumer.
536   */ 536   */
537   std::coroutine_handle<> 537   std::coroutine_handle<>
HITCBC 538   2 await_suspend( 538   2 await_suspend(
539   std::coroutine_handle<> h, 539   std::coroutine_handle<> h,
540   await_context const* env) noexcept 540   await_context const* env) noexcept
541   { 541   {
HITCBC 542   2 return inner_.await_suspend(h, env); 542   2 return inner_.await_suspend(h, env);
543   } 543   }
544   544  
545   /** Complete the acquisition and report the outcome. 545   /** Complete the acquisition and report the outcome.
546   546  
547   @return A `results<lock_guard>` destructuring as 547   @return A `results<lock_guard>` destructuring as
548   `[ec, guard]`. On success `ec` is empty and `guard` holds the 548   `[ec, guard]`. On success `ec` is empty and `guard` holds the
549   mutex, releasing it when destroyed. If the wait was canceled, 549   mutex, releasing it when destroyed. If the wait was canceled,
550   `ec` is `error::canceled` and `guard` holds no mutex. 550   `ec` is `error::canceled` and `guard` holds no mutex.
551   */ 551   */
HITCBC 552   5 [[nodiscard]] results<lock_guard> await_resume() noexcept 552   5 [[nodiscard]] results<lock_guard> await_resume() noexcept
553   { 553   {
HITCBC 554   5 auto r = inner_.await_resume(); 554   5 auto r = inner_.await_resume();
HITCBC 555   5 if(std::get<0>(r)) 555   5 if(std::get<0>(r))
HITCBC 556   2 return {std::get<0>(r), lock_guard()}; 556   2 return {std::get<0>(r), lock_guard()};
HITCBC 557   3 return {std::error_code(), lock_guard(m_)}; 557   3 return {std::error_code(), lock_guard(m_)};
558   } 558   }
559   }; 559   };
560   560  
561   /// Construct an unlocked mutex. 561   /// Construct an unlocked mutex.
HITCBC 562   3 async_mutex() = default; 562   3 async_mutex() = default;
563   563  
564   /** Destroy the mutex, canceling every coroutine still waiting. 564   /** Destroy the mutex, canceling every coroutine still waiting.
565   565  
566   Each waiter is posted to its executor to resume with 566   Each waiter is posted to its executor to resume with
567   `error::canceled`, as if its stop token had been triggered. An 567   `error::canceled`, as if its stop token had been triggered. An
568   executor that can no longer run it abandons it instead, possibly 568   executor that can no longer run it abandons it instead, possibly
569   before this destructor returns. Either way the waiter no longer 569   before this destructor returns. Either way the waiter no longer
570   refers to the mutex. 570   refers to the mutex.
571   571  
572   A waiter handed the lock but not yet resumed resumes with 572   A waiter handed the lock but not yet resumed resumes with
573   `error::canceled` instead, holding nothing. 573   `error::canceled` instead, holding nothing.
574   574  
575   @pre No coroutine holds the lock. 575   @pre No coroutine holds the lock.
576   */ 576   */
HITCBC 577   24 ~async_mutex() 577   24 ~async_mutex()
578   { 578   {
HITCBC 579   24 if(unlocking_) 579   24 if(unlocking_)
HITCBC 580   1 unlocking_->gone = true; 580   1 unlocking_->gone = true;
581   // Already posted, so it is only told the lock no longer exists. 581   // Already posted, so it is only told the lock no longer exists.
HITCBC 582   24 if(granted_to_) 582   24 if(granted_to_)
583   { 583   {
HITCBC 584   1 granted_to_->granted_ = false; 584   1 granted_to_->granted_ = false;
HITCBC 585   1 granted_to_->canceled_ = true; 585   1 granted_to_->canceled_ = true;
586   } 586   }
HITCBC 587   28 while(auto* w = waiters_.pop_front()) 587   28 while(auto* w = waiters_.pop_front())
588   { 588   {
HITCBC 589   4 w->in_list_ = false; 589   4 w->in_list_ = false;
HITCBC 590   4 if(!w->claimed_.exchange(true, std::memory_order_acq_rel)) 590   4 if(!w->claimed_.exchange(true, std::memory_order_acq_rel))
591   { 591   {
HITCBC 592   3 w->canceled_ = true; 592   3 w->canceled_ = true;
HITCBC 593   3 w->ex_.post(w->cont_); 593   3 w->ex_.post(w->cont_);
594   } 594   }
HITCBC 595   4 } 595   4 }
HITCBC 596   24 } 596   24 }
597   597  
598   /** Copy construction is disabled; suspended waiters point into the 598   /** Copy construction is disabled; suspended waiters point into the
599   mutex's wait queue. 599   mutex's wait queue.
600   600  
601   @param other The mutex that would be copied. 601   @param other The mutex that would be copied.
602   */ 602   */
603   async_mutex(async_mutex const& other) = delete; 603   async_mutex(async_mutex const& other) = delete;
604   604  
605   /** Copy assignment is disabled; suspended waiters point into the 605   /** Copy assignment is disabled; suspended waiters point into the
606   mutex's wait queue. 606   mutex's wait queue.
607   607  
608   @param other The mutex that would be assigned from. 608   @param other The mutex that would be assigned from.
609   609  
610   @return A reference to `*this`. 610   @return A reference to `*this`.
611   */ 611   */
612   async_mutex& operator=(async_mutex const& other) = delete; 612   async_mutex& operator=(async_mutex const& other) = delete;
613   613  
614   /** Move construction is disabled; suspended waiters point into the 614   /** Move construction is disabled; suspended waiters point into the
615   mutex's wait queue. 615   mutex's wait queue.
616   616  
617   @param other The mutex that would be moved from. 617   @param other The mutex that would be moved from.
618   */ 618   */
619   async_mutex(async_mutex&& other) = delete; 619   async_mutex(async_mutex&& other) = delete;
620   620  
621   /** Move assignment is disabled; suspended waiters point into the 621   /** Move assignment is disabled; suspended waiters point into the
622   mutex's wait queue. 622   mutex's wait queue.
623   623  
624   @param other The mutex that would be moved from. 624   @param other The mutex that would be moved from.
625   625  
626   @return A reference to `*this`. 626   @return A reference to `*this`.
627   */ 627   */
628   async_mutex& operator=(async_mutex&& other) = delete; 628   async_mutex& operator=(async_mutex&& other) = delete;
629   629  
630   /** Returns an awaiter that acquires the mutex. 630   /** Returns an awaiter that acquires the mutex.
631   631  
632   @return An awaitable that await-returns `(error_code)`. 632   @return An awaitable that await-returns `(error_code)`.
633   */ 633   */
HITCBC 634   100047 lock_awaiter lock() noexcept 634   100047 lock_awaiter lock() noexcept
635   { 635   {
HITCBC 636   100047 return lock_awaiter{this}; 636   100047 return lock_awaiter{this};
637   } 637   }
638   638  
639   /** Returns an awaiter that acquires the mutex with RAII. 639   /** Returns an awaiter that acquires the mutex with RAII.
640   640  
641   @return An awaitable that await-returns `(error_code,lock_guard)`. 641   @return An awaitable that await-returns `(error_code,lock_guard)`.
642   */ 642   */
HITCBC 643   5 lock_guard_awaiter scoped_lock() noexcept 643   5 lock_guard_awaiter scoped_lock() noexcept
644   { 644   {
HITCBC 645   5 return lock_guard_awaiter(this); 645   5 return lock_guard_awaiter(this);
646   } 646   }
647   647  
648   /** Releases the mutex. 648   /** Releases the mutex.
649   649  
650   If waiters are queued, the next eligible waiter is 650   If waiters are queued, the next eligible waiter is
651   resumed with the lock held. Canceled waiters are 651   resumed with the lock held. Canceled waiters are
652   skipped. If no eligible waiter remains, the mutex 652   skipped. If no eligible waiter remains, the mutex
653   becomes unlocked. 653   becomes unlocked.
654   */ 654   */
HITCBC 655   100032 void unlock() noexcept 655   100032 void unlock() noexcept
656   { 656   {
HITCBC 657   100032 if(unlocking_) 657   100032 if(unlocking_)
658   { 658   {
659   // Reached from a granted waiter destroyed by the post below; 659   // Reached from a granted waiter destroyed by the post below;
660   // that call hands the lock on once the post returns. 660   // that call hands the lock on once the post returns.
HITCBC 661   100002 unlocking_->again = true; 661   100002 unlocking_->again = true;
HITCBC 662   100003 return; 662   100003 return;
663   } 663   }
HITCBC 664   30 unlock_scope scope; 664   30 unlock_scope scope;
HITCBC 665   30 unlocking_ = &scope; 665   30 unlocking_ = &scope;
666   do 666   do
667   { 667   {
HITCBC 668   100031 scope.again = false; 668   100031 scope.again = false;
HITCBC 669   100031 auto* waiter = grant_next(); 669   100031 auto* waiter = grant_next();
HITCBC 670   100031 if(!waiter) 670   100031 if(!waiter)
HITCBC 671   19 break; 671   19 break;
HITCBC 672   100012 waiter->ex_.post(waiter->cont_); 672   100012 waiter->ex_.post(waiter->cont_);
HITCBC 673   100012 if(scope.gone) 673   100012 if(scope.gone)
HITCBC 674   1 return; 674   1 return;
675   } 675   }
HITCBC 676   100011 while(scope.again); 676   100011 while(scope.again);
HITCBC 677   29 unlocking_ = nullptr; 677   29 unlocking_ = nullptr;
678   } 678   }
679   679  
680   private: 680   private:
681   // Hand the lock to the next unclaimed waiter, or release it. 681   // Hand the lock to the next unclaimed waiter, or release it.
HITCBC 682   100031 lock_awaiter* grant_next() noexcept 682   100031 lock_awaiter* grant_next() noexcept
683   { 683   {
HITCBC 684   100032 while(auto* waiter = waiters_.pop_front()) 684   100032 while(auto* waiter = waiters_.pop_front())
685   { 685   {
HITCBC 686   100013 waiter->in_list_ = false; 686   100013 waiter->in_list_ = false;
HITCBC 687   100013 if(!waiter->claimed_.exchange( 687   100013 if(!waiter->claimed_.exchange(
688   true, std::memory_order_acq_rel)) 688   true, std::memory_order_acq_rel))
689   { 689   {
HITCBC 690   100012 waiter->granted_ = true; 690   100012 waiter->granted_ = true;
HITCBC 691   100012 granted_to_ = waiter; 691   100012 granted_to_ = waiter;
HITCBC 692   100012 return waiter; 692   100012 return waiter;
693   } 693   }
HITCBC 694   1 } 694   1 }
HITCBC 695   19 locked_ = false; 695   19 locked_ = false;
HITCBC 696   19 return nullptr; 696   19 return nullptr;
697   } 697   }
698   698  
699   public: 699   public:
700   /** Returns true if the mutex is currently locked. 700   /** Returns true if the mutex is currently locked.
701   701  
702   @return `true` if the mutex is held; otherwise `false`. 702   @return `true` if the mutex is held; otherwise `false`.
703   */ 703   */
HITCBC 704   29 bool is_locked() const noexcept 704   29 bool is_locked() const noexcept
705   { 705   {
HITCBC 706   29 return locked_; 706   29 return locked_;
707   } 707   }
708   }; 708   };
709   709  
710   } // namespace capy 710   } // namespace capy
711   } // namespace boost 711   } // namespace boost
712   712  
713   #endif 713   #endif