include/boost/capy/ex/async_mutex.hpp

100.0% Lines (93/0/93) 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		@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	7x	void operator()() const noexcept
165		{
166	7x	if(!self_->claimed_.exchange(
167		true, std::memory_order_acq_rel))
168		{
169	7x	self_->canceled_ = true;
170	7x	self_->ex_.post(self_->cont_);
171		}
172	7x	}
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	19x	stop_cb_t& stop_cb_() noexcept
188		{
189		return reinterpret_cast<stop_cb_t>(
190	19x	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	76x	~lock_awaiter()
207		{
208	76x	if(active_)
209		{
210	3x	stop_cb_().~stop_cb_t();
211	3x	m_->waiters_.remove(this);
212		}
213	76x	}
214
215		/** Construct an awaiter for the given mutex.
216
217		@param m The mutex to acquire. It must outlive the awaiter.
218		*/
219	38x	explicit lock_awaiter(async_mutex* m) noexcept
220	38x	: m_(m)
221		{
222	38x	}
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	38x	lock_awaiter(lock_awaiter&& o) noexcept
233	76x	: m_(o.m_)
234	38x	, cont_(o.cont_)
235	38x	, ex_(o.ex_)
236	38x	, claimed_(o.claimed_.load(
237		std::memory_order_relaxed))
238	38x	, canceled_(o.canceled_)
239	76x	, active_(std::exchange(o.active_, false))
240		{
241	38x	}
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	38x	bool await_ready() const noexcept
281		{
282	38x	if(!m_->locked_)
283		{
284	17x	m_->locked_ = true;
285	17x	return true;
286		}
287	21x	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	21x	await_suspend(
318		std::coroutine_handle<> h,
319		io_env const* env) noexcept
320		{
321	21x	if(env->stop_token.stop_requested())
322		{
323	2x	canceled_ = true;
324	2x	return h;
325		}
326	19x	cont_.h = h;
327	19x	ex_ = env->executor;
328	19x	m_->waiters_.push_back(this);
329	57x	::new(stop_cb_buf_) stop_cb_t(
330	19x	env->stop_token, cancel_fn{this});
331	19x	active_ = true;
332	19x	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	35x	[[nodiscard]] io_result<> await_resume() noexcept
346		{
347	35x	if(active_)
348		{
349	16x	stop_cb_().~stop_cb_t();
350	16x	if(canceled_)
351		{
352	7x	m_->waiters_.remove(this);
353	7x	active_ = false;
354	14x	return {make_error_code(
355	7x	error::canceled)};
356		}
357	9x	active_ = false;
358		}
359	28x	if(canceled_)
360	4x	return {make_error_code(
361	2x	error::canceled)};
362	26x	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	9x	~lock_guard()
375		{
376	9x	if(m_)
377	2x	m_->unlock();
378	9x	}
379
380		/// Construct a guard that holds no mutex.
381	2x	lock_guard() noexcept
382	2x	: m_(nullptr)
383		{
384	2x	}
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	2x	explicit lock_guard(async_mutex* m) noexcept
394	2x	: m_(m)
395		{
396	2x	}
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	5x	lock_guard(lock_guard&& o) noexcept
406	5x	: m_(std::exchange(o.m_, nullptr))
407		{
408	5x	}
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	4x	explicit lock_guard_awaiter(async_mutex* m) noexcept
461	4x	: m_(m)
462	4x	, inner_(m)
463		{
464	4x	}
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	4x	bool await_ready() const noexcept
476		{
477	4x	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	2x	await_suspend(
500		std::coroutine_handle<> h,
501		io_env const* env) noexcept
502		{
503	2x	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	4x	[[nodiscard]] io_result<lock_guard> await_resume() noexcept
514		{
515	4x	auto r = inner_.await_resume();
516	4x	if(std::get<0>(r))
517	2x	return {std::get<0>(r), lock_guard()};
518	2x	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	34x	lock_awaiter lock() noexcept
562		{
563	34x	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	4x	lock_guard_awaiter scoped_lock() noexcept
571		{
572	4x	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	26x	void unlock() noexcept
583		{
584		for(;;)
585		{
586	27x	auto* waiter = waiters_.pop_front();
587	27x	if(!waiter)
588		{
589	17x	locked_ = false;
590	17x	return;
591		}
592	10x	if(!waiter->claimed_.exchange(
593		true, std::memory_order_acq_rel))
594		{
595	9x	waiter->ex_.post(waiter->cont_);
596	9x	return;
597		}
598	1x	}
599		}
600
601		/** Returns true if the mutex is currently locked.
602
603		@return `true` if the mutex is held; otherwise `false`.
604		*/
605	27x	bool is_locked() const noexcept
606		{
607	27x	return locked_;
608		}
609		};
610
611		} // namespace capy
612		} // namespace boost
613
614		#endif
615