include/boost/capy/ex/async_mutex.hpp

100.0% Lines (135 / 135) 100.0% Functions (23 / 23)

[ ] prev/next file

async_mutex.hpp

f(x) Functions (23)

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/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	8x	void operator()() const noexcept
183		{
184	8x	if(!self_->claimed_.exchange(
185		true, std::memory_order_acq_rel))
186		{
187	8x	self_->canceled_ = true;
188	8x	self_->ex_.post(self_->cont_);
189		}
190	8x	}
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	100026x	stop_cb_t& stop_cb_() noexcept
206		{
207		return reinterpret_cast<stop_cb_t>(
208	100026x	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	200099x	~lock_awaiter()
229		{
230	200099x	if(active_)
231		{
232	100009x	stop_cb_().~stop_cb_t();
233		}
234	200099x	if(in_list_)
235	3x	m_->waiters_.remove(this);
236		// Handed the lock but destroyed before resuming: pass it on.
237	200099x	if(granted_)
238		{
239	100002x	m_->granted_to_ = nullptr;
240	100002x	m_->unlock();
241		}
242	200099x	}
243
244		/** Construct an awaiter for the given mutex.
245
246		@param m The mutex to acquire. It must outlive the awaiter.
247		*/
248	100052x	explicit lock_awaiter(async_mutex* m) noexcept
249	100052x	: m_(m)
250		{
251	100052x	}
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	100047x	lock_awaiter(lock_awaiter&& o) noexcept
262	200094x	: m_(o.m_)
263	100047x	, cont_(o.cont_)
264	100047x	, ex_(o.ex_)
265	100047x	, claimed_(o.claimed_.load(
266		std::memory_order_relaxed))
267	100047x	, canceled_(o.canceled_)
268	100047x	, active_(std::exchange(o.active_, false))
269	100047x	, granted_(std::exchange(o.granted_, false))
270	200094x	, in_list_(std::exchange(o.in_list_, false))
271		{
272	100047x	}
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	100052x	bool await_ready() const noexcept
312		{
313	100052x	if(!m_->locked_)
314		{
315	24x	m_->locked_ = true;
316	24x	return true;
317		}
318	100028x	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	100028x	await_suspend(
349		std::coroutine_handle<> h,
350		await_context const* env) noexcept
351		{
352	100028x	if(env->stop_token.stop_requested())
353		{
354	2x	canceled_ = true;
355	2x	return h;
356		}
357	100026x	cont_.h = h;
358	100026x	cont_.owner = env->owner;
359	100026x	ex_ = env->executor;
360	100026x	m_->waiters_.push_back(this);
361	100026x	in_list_ = true;
362	300078x	::new(stop_cb_buf_) stop_cb_t(
363	100026x	env->stop_token, cancel_fn{this});
364	100026x	active_ = true;
365	100026x	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	38x	[[nodiscard]] results<> await_resume() noexcept
379		{
380	38x	if(std::exchange(granted_, false))
381	9x	m_->granted_to_ = nullptr;
382	38x	if(active_)
383		{
384	17x	stop_cb_().~stop_cb_t();
385	17x	if(canceled_)
386		{
387	8x	if(in_list_)
388		{
389	6x	m_->waiters_.remove(this);
390	6x	in_list_ = false;
391		}
392	8x	active_ = false;
393	16x	return {make_error_code(
394	8x	error::canceled)};
395		}
396	9x	active_ = false;
397		}
398	30x	if(canceled_)
399	4x	return {make_error_code(
400	2x	error::canceled)};
401	28x	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	11x	~lock_guard()
414		{
415	11x	if(m_)
416	3x	m_->unlock();
417	11x	}
418
419		/// Construct a guard that holds no mutex.
420	2x	lock_guard() noexcept
421	2x	: m_(nullptr)
422		{
423	2x	}
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	3x	explicit lock_guard(async_mutex* m) noexcept
433	3x	: m_(m)
434		{
435	3x	}
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	6x	lock_guard(lock_guard&& o) noexcept
445	6x	: m_(std::exchange(o.m_, nullptr))
446		{
447	6x	}
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	5x	explicit lock_guard_awaiter(async_mutex* m) noexcept
500	5x	: m_(m)
501	5x	, inner_(m)
502		{
503	5x	}
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	5x	bool await_ready() const noexcept
515		{
516	5x	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	2x	await_suspend(
539		std::coroutine_handle<> h,
540		await_context const* env) noexcept
541		{
542	2x	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	5x	[[nodiscard]] results<lock_guard> await_resume() noexcept
553		{
554	5x	auto r = inner_.await_resume();
555	5x	if(std::get<0>(r))
556	2x	return {std::get<0>(r), lock_guard()};
557	3x	return {std::error_code(), lock_guard(m_)};
558		}
559		};
560
561		/// Construct an unlocked mutex.
562	3x	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	24x	~async_mutex()
578		{
579	24x	if(unlocking_)
580	1x	unlocking_->gone = true;
581		// Already posted, so it is only told the lock no longer exists.
582	24x	if(granted_to_)
583		{
584	1x	granted_to_->granted_ = false;
585	1x	granted_to_->canceled_ = true;
586		}
587	28x	while(auto* w = waiters_.pop_front())
588		{
589	4x	w->in_list_ = false;
590	4x	if(!w->claimed_.exchange(true, std::memory_order_acq_rel))
591		{
592	3x	w->canceled_ = true;
593	3x	w->ex_.post(w->cont_);
594		}
595	4x	}
596	24x	}
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	100047x	lock_awaiter lock() noexcept
635		{
636	100047x	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	5x	lock_guard_awaiter scoped_lock() noexcept
644		{
645	5x	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	100032x	void unlock() noexcept
656		{
657	100032x	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	100002x	unlocking_->again = true;
662	100003x	return;
663		}
664	30x	unlock_scope scope;
665	30x	unlocking_ = &scope;
666		do
667		{
668	100031x	scope.again = false;
669	100031x	auto* waiter = grant_next();
670	100031x	if(!waiter)
671	19x	break;
672	100012x	waiter->ex_.post(waiter->cont_);
673	100012x	if(scope.gone)
674	1x	return;
675		}
676	100011x	while(scope.again);
677	29x	unlocking_ = nullptr;
678		}
679
680		private:
681		// Hand the lock to the next unclaimed waiter, or release it.
682	100031x	lock_awaiter* grant_next() noexcept
683		{
684	100032x	while(auto* waiter = waiters_.pop_front())
685		{
686	100013x	waiter->in_list_ = false;
687	100013x	if(!waiter->claimed_.exchange(
688		true, std::memory_order_acq_rel))
689		{
690	100012x	waiter->granted_ = true;
691	100012x	granted_to_ = waiter;
692	100012x	return waiter;
693		}
694	1x	}
695	19x	locked_ = false;
696	19x	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	29x	bool is_locked() const noexcept
705		{
706	29x	return locked_;
707		}
708		};
709
710		} // namespace capy
711		} // namespace boost
712
713		#endif
714