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