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