include/boost/capy/ex/async_event.hpp

100.0% Lines (66/0/66) 100.0% List of functions (12/0/12)

[ ] prev/next file

async_event.hpp

f(x) Functions (12)

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_EVENT_HPP
12		#define BOOST_CAPY_ASYNC_EVENT_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_event implementation notes
30		=================================
31
32		Same cancellation pattern as async_mutex (see that file for the
33		full discussion on claimed_, stop_cb lifetime, member ordering,
34		and threading assumptions).
35
36		Key difference: set() wakes ALL waiters (broadcast), not one.
37		It pops every waiter from the list and posts the ones it
38		claims. Waiters already claimed by a stop callback are skipped.
39
40		Because set() pops all waiters, a canceled waiter may have been
41		removed from the list by set() before its await_resume runs.
42		This requires a separate in_list_ flag (unlike async_mutex where
43		active_ served double duty). await_resume only calls remove()
44		when in_list_ is true.
45		*/
46
47		namespace boost {
48		namespace capy {
49
50		/** Queues coroutines in `wait()` and resumes all of them when `set()` is called.
51
52		This event provides a way to notify multiple coroutines that some
53		condition has occurred. When a coroutine awaits an unset event, it
54		suspends and is added to a wait queue. When the event is set, all
55		waiting coroutines are resumed.
56
57		@par Cancellation
58
59		When a coroutine is suspended waiting for the event and its stop
60		token is triggered, the waiter completes with `error::canceled`
61		instead of waiting for `set()`.
62
63		Cancellation only applies while the coroutine is suspended in the
64		wait queue. If the event is already set when `wait()` is called,
65		the wait completes immediately even if the stop token is already
66		signaled.
67
68		@par Zero Allocation
69
70		No heap allocation occurs for wait operations.
71
72		@par Thread Safety
73
74		Distinct objects: Safe.@n
75		Shared objects: Unsafe.
76
77		The event operations are designed for single-threaded use on one
78		executor. The stop callback may fire from any thread.
79
80		This type is non-copyable and non-movable because suspended
81		waiters hold intrusive pointers into the event's internal list.
82
83		@par Example
84		@code
85		async_event event;
86
87		task<> waiter() {
88		auto [ec] = co_await event.wait();
89		if(ec)
90		co_return;
91		// ... event was set ...
92		}
93
94		task<> notifier() {
95		// ... do some work ...
96		event.set(); // Wake all waiters
97		}
98		@endcode
99		*/
100		class async_event
101		{
102		public:
103		class wait_awaiter;
104
105		private:
106		bool set_ = false;
107		detail::intrusive_list<wait_awaiter> waiters_;
108
109		public:
110		/** Suspends the caller until `set()` runs, or resumes it with `error::canceled` on a stop request.
111		*/
112		class wait_awaiter
113		: public detail::intrusive_list<wait_awaiter>::node
114		{
115		friend class async_event;
116
117		async_event* e_;
118		continuation cont_;
119		executor_ref ex_;
120
121		// Declared before stop_cb_buf_: the callback
122		// accesses these members, so they must still be
123		// alive if the stop_cb_ destructor blocks.
124		std::atomic<bool> claimed_{false};
125		bool canceled_ = false;
126		bool active_ = false;
127		bool in_list_ = false;
128
129		struct cancel_fn
130		{
131		wait_awaiter* self_;
132
133	4x	void operator()() const noexcept
134		{
135	4x	if(!self_->claimed_.exchange(
136		true, std::memory_order_acq_rel))
137		{
138	3x	self_->canceled_ = true;
139	3x	self_->ex_.post(self_->cont_);
140		}
141	4x	}
142		};
143
144		using stop_cb_t =
145		std::stop_callback<cancel_fn>;
146
147		// Aligned storage for stop_cb_t. Declared last:
148		// its destructor may block while the callback
149		// accesses the members above.
150		BOOST_CAPY_MSVC_WARNING_PUSH
151		BOOST_CAPY_MSVC_WARNING_DISABLE(4324) // padded due to alignas
152		alignas(stop_cb_t)
153		unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
154		BOOST_CAPY_MSVC_WARNING_POP
155
156	20x	stop_cb_t& stop_cb_() noexcept
157		{
158		return reinterpret_cast<stop_cb_t>(
159	20x	stop_cb_buf_);
160		}
161
162		public:
163		/** Destroy the awaiter, leaving the event unable to reach it.
164
165		Destroys the stop callback if one is registered, and unlinks
166		the awaiter from the event's wait queue if it is still linked.
167		Both are necessary when the coroutine frame is torn down while
168		suspended, so that neither `set()` nor the stop callback can
169		reach a destroyed awaiter.
170		*/
171	52x	~wait_awaiter()
172		{
173	52x	if(active_)
174	1x	stop_cb_().~stop_cb_t();
175	52x	if(in_list_)
176	1x	e_->waiters_.remove(this);
177	52x	}
178
179		/** Construct an awaiter for the given event.
180
181		@param e The event to wait on. It must outlive the awaiter.
182		*/
183	25x	explicit wait_awaiter(async_event* e) noexcept
184	25x	: e_(e)
185		{
186	25x	}
187
188		/** Construct by moving.
189
190		The moved-from awaiter is left inert: its destructor no longer
191		destroys the stop callback and no longer unlinks from the
192		event's wait queue.
193
194		@param o The awaiter to move from.
195		*/
196	27x	wait_awaiter(wait_awaiter&& o) noexcept
197	54x	: e_(o.e_)
198	27x	, cont_(o.cont_)
199	27x	, ex_(o.ex_)
200	27x	, claimed_(o.claimed_.load(
201		std::memory_order_relaxed))
202	27x	, canceled_(o.canceled_)
203	27x	, active_(std::exchange(o.active_, false))
204	54x	, in_list_(std::exchange(o.in_list_, false))
205		{
206	27x	}
207
208		/** Copy construction is disabled; a waiter is linked into the
209		event's wait queue by address.
210
211		@param other The awaiter that would be copied.
212		*/
213		wait_awaiter(wait_awaiter const& other) = delete;
214
215		/** Copy assignment is disabled; a waiter is linked into the
216		event's wait queue by address.
217
218		@param other The awaiter that would be assigned from.
219
220		@return A reference to `*this`.
221		*/
222		wait_awaiter& operator=(wait_awaiter const& other) = delete;
223
224		/** Move assignment is disabled; a waiter is linked into the
225		event's wait queue by address.
226
227		@param other The awaiter that would be moved from.
228
229		@return A reference to `*this`.
230		*/
231		wait_awaiter& operator=(wait_awaiter&& other) = delete;
232
233		/** Report whether the event is already set.
234
235		@return `true` if the event is set, in which case the awaiting
236		coroutine does not suspend; otherwise `false`.
237		*/
238	25x	bool await_ready() const noexcept
239		{
240	25x	return e_->set_;
241		}
242
243		/** Enqueue the awaiting coroutine until the event is set.
244
245		This is the @ref IoAwaitable overload of `await_suspend`.
246
247		If a stop request is already pending on `env->stop_token`, the
248		awaiter records the cancellation and does not enqueue.
249
250		Otherwise it stores `h` and `env->executor`, links itself into
251		the event's wait queue, and registers a stop callback on
252		`env->stop_token`. Exactly one of `set()` and that callback posts
253		`h` through the stored executor, whichever claims the waiter
254		first. Only the post is subject to that race. A losing stop
255		callback does nothing at all, but `set()` unlinks every waiter it
256		pops whether it claims it or not. That is why @ref await_resume
257		unlinks a canceled waiter only when it is still linked.
258
259		@param h The awaiting coroutine, resumed when the event is set
260		or the wait is canceled.
261
262		@param env The execution environment. Its executor posts the
263		resumption and its stop token is watched for the duration of
264		the wait. It must outlive the wait.
265
266		@return `h` if a stop request was already pending, which
267		resumes the awaiting coroutine immediately without enqueuing
268		it. Otherwise `std::noop_coroutine()`, which leaves the
269		coroutine suspended and returns control to the resumer.
270		*/
271		std::coroutine_handle<>
272	21x	await_suspend(
273		std::coroutine_handle<> h,
274		io_env const* env) noexcept
275		{
276	21x	if(env->stop_token.stop_requested())
277		{
278	1x	canceled_ = true;
279	1x	return h;
280		}
281	20x	cont_.h = h;
282	20x	ex_ = env->executor;
283	20x	e_->waiters_.push_back(this);
284	20x	in_list_ = true;
285	60x	::new(stop_cb_buf_) stop_cb_t(
286	20x	env->stop_token, cancel_fn{this});
287	20x	active_ = true;
288	20x	return std::noop_coroutine();
289		}
290
291		/** Complete the wait and report the outcome.
292
293		Destroys the stop callback if one is registered. If the wait
294		was canceled while still linked into the event's wait queue,
295		unlinks it. `set()` pops every waiter, so a canceled waiter may
296		or may not still be linked when it resumes.
297
298		@return An empty `io_result<>` if the event was set, or one
299		holding `error::canceled` if the stop token fired first.
300		*/
301	22x	io_result<> await_resume() noexcept
302		{
303	22x	if(active_)
304		{
305	19x	stop_cb_().~stop_cb_t();
306	19x	active_ = false;
307		}
308	22x	if(canceled_)
309		{
310	4x	if(in_list_)
311		{
312	3x	e_->waiters_.remove(this);
313	3x	in_list_ = false;
314		}
315		return {make_error_code(
316	4x	error::canceled)};
317		}
318	18x	return {{}};
319		}
320		};
321
322		/// Construct an unset event.
323		async_event() = default;
324
325		/** Copy construction is disabled; suspended waiters point into the
326		event's wait queue.
327
328		@param other The event that would be copied.
329		*/
330		async_event(async_event const& other) = delete;
331
332		/** Copy assignment is disabled; suspended waiters point into the
333		event's wait queue.
334
335		@param other The event that would be assigned from.
336
337		@return A reference to `*this`.
338		*/
339		async_event& operator=(async_event const& other) = delete;
340
341		/** Move construction is disabled; suspended waiters point into the
342		event's wait queue.
343
344		@param other The event that would be moved from.
345		*/
346		async_event(async_event&& other) = delete;
347
348		/** Move assignment is disabled; suspended waiters point into the
349		event's wait queue.
350
351		@param other The event that would be moved from.
352
353		@return A reference to `*this`.
354		*/
355		async_event& operator=(async_event&& other) = delete;
356
357		/** Returns an awaiter that waits until the event is set.
358
359		If the event is already set, completes immediately.
360
361		@return An awaitable that await-returns `(error_code)`.
362		*/
363	25x	wait_awaiter wait() noexcept
364		{
365	25x	return wait_awaiter{this};
366		}
367
368		/** Resumes every waiting coroutine and marks the event set for later `wait()` calls.
369
370		All waiting coroutines are resumed. Canceled waiters
371		are skipped. Subsequent calls to wait() complete
372		immediately until clear() is called.
373		*/
374	17x	void set()
375		{
376	17x	set_ = true;
377		for(;;)
378		{
379	33x	auto* w = waiters_.pop_front();
380	33x	if(!w)
381	17x	break;
382	16x	w->in_list_ = false;
383	16x	if(!w->claimed_.exchange(
384		true, std::memory_order_acq_rel))
385		{
386	16x	w->ex_.post(w->cont_);
387		}
388	16x	}
389	17x	}
390
391		/** Clears the event.
392
393		Subsequent calls to wait() suspend until
394		set() is called again.
395		*/
396	2x	void clear() noexcept
397		{
398	2x	set_ = false;
399	2x	}
400
401		/** Returns true if the event is currently set.
402
403		@return `true` if the event is set; otherwise `false`.
404		*/
405	9x	bool is_set() const noexcept
406		{
407	9x	return set_;
408		}
409		};
410
411		} // namespace capy
412		} // namespace boost
413
414		#endif
415