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