include/boost/capy/ex/async_mutex.hpp

100.0% Lines (91/0/91) 100.0% List of functions (20/0/20)

[ ] prev/next file

async_mutex.hpp

f(x) Functions (20)

Function Calls Lines Blocks

Line	Hits	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	7x	void operator()() const noexcept
183		{
184	7x	if(!self_->claimed_.exchange(
185		true, std::memory_order_acq_rel))
186		{
187	7x	self_->canceled_ = true;
188	7x	self_->ex_.post(self_->cont_);
189		}
190	7x	}
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	19x	stop_cb_t& stop_cb_() noexcept
206		{
207		return reinterpret_cast<stop_cb_t>(
208	19x	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	76x	~lock_awaiter()
225		{
226	76x	if(active_)
227		{
228	3x	stop_cb_().~stop_cb_t();
229	3x	m_->waiters_.remove(this);
230		}
231	76x	}
232
233		/** Construct an awaiter for the given mutex.
234
235		@param m The mutex to acquire. It must outlive the awaiter.
236		*/
237	38x	explicit lock_awaiter(async_mutex* m) noexcept
238	38x	: m_(m)
239		{
240	38x	}
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	38x	lock_awaiter(lock_awaiter&& o) noexcept
251	76x	: m_(o.m_)
252	38x	, cont_(o.cont_)
253	38x	, ex_(o.ex_)
254	38x	, claimed_(o.claimed_.load(
255		std::memory_order_relaxed))
256	38x	, canceled_(o.canceled_)
257	76x	, active_(std::exchange(o.active_, false))
258		{
259	38x	}
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	38x	bool await_ready() const noexcept
299		{
300	38x	if(!m_->locked_)
301		{
302	17x	m_->locked_ = true;
303	17x	return true;
304		}
305	21x	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	21x	await_suspend(
336		std::coroutine_handle<> h,
337		io_env const* env) noexcept
338		{
339	21x	if(env->stop_token.stop_requested())
340		{
341	2x	canceled_ = true;
342	2x	return h;
343		}
344	19x	cont_.h = h;
345	19x	ex_ = env->executor;
346	19x	m_->waiters_.push_back(this);
347	57x	::new(stop_cb_buf_) stop_cb_t(
348	19x	env->stop_token, cancel_fn{this});
349	19x	active_ = true;
350	19x	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	35x	io_result<> await_resume() noexcept
364		{
365	35x	if(active_)
366		{
367	16x	stop_cb_().~stop_cb_t();
368	16x	if(canceled_)
369		{
370	7x	m_->waiters_.remove(this);
371	7x	active_ = false;
372		return {make_error_code(
373	7x	error::canceled)};
374		}
375	9x	active_ = false;
376		}
377	28x	if(canceled_)
378		return {make_error_code(
379	2x	error::canceled)};
380	26x	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	9x	~lock_guard()
393		{
394	9x	if(m_)
395	2x	m_->unlock();
396	9x	}
397
398		/// Construct a guard that holds no mutex.
399	2x	lock_guard() noexcept
400	2x	: m_(nullptr)
401		{
402	2x	}
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	2x	explicit lock_guard(async_mutex* m) noexcept
412	2x	: m_(m)
413		{
414	2x	}
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	5x	lock_guard(lock_guard&& o) noexcept
424	5x	: m_(std::exchange(o.m_, nullptr))
425		{
426	5x	}
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	4x	explicit lock_guard_awaiter(async_mutex* m) noexcept
479	4x	: m_(m)
480	4x	, inner_(m)
481		{
482	4x	}
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	4x	bool await_ready() const noexcept
494		{
495	4x	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	2x	await_suspend(
518		std::coroutine_handle<> h,
519		io_env const* env) noexcept
520		{
521	2x	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	4x	io_result<lock_guard> await_resume() noexcept
532		{
533	4x	auto r = inner_.await_resume();
534	4x	if(r.ec)
535	2x	return {r.ec, {}};
536	2x	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	34x	lock_awaiter lock() noexcept
580		{
581	34x	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	4x	lock_guard_awaiter scoped_lock() noexcept
589		{
590	4x	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	26x	void unlock() noexcept
601		{
602		for(;;)
603		{
604	27x	auto* waiter = waiters_.pop_front();
605	27x	if(!waiter)
606		{
607	17x	locked_ = false;
608	17x	return;
609		}
610	10x	if(!waiter->claimed_.exchange(
611		true, std::memory_order_acq_rel))
612		{
613	9x	waiter->ex_.post(waiter->cont_);
614	9x	return;
615		}
616	1x	}
617		}
618
619		/** Returns true if the mutex is currently locked.
620
621		@return `true` if the mutex is held; otherwise `false`.
622		*/
623	27x	bool is_locked() const noexcept
624		{
625	27x	return locked_;
626		}
627		};
628
629		} // namespace capy
630		} // namespace boost
631
632		#endif
633