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