include/boost/capy/ex/async_event.hpp

100.0% Lines (67/0/67) 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		@par !example example
85
86		*/
87		class async_event
88		{
89		public:
90		class wait_awaiter;
91
92		private:
93		bool set_ = false;
94		detail::intrusive_list<wait_awaiter> waiters_;
95
96		public:
97		/** Suspends the caller until `set()` runs, or resumes it with `error::canceled` on a stop request.
98		*/
99		class wait_awaiter
100		: public detail::intrusive_list<wait_awaiter>::node
101		{
102		friend class async_event;
103
104		async_event* e_;
105		continuation cont_;
106		executor_ref ex_;
107
108		// Declared before stop_cb_buf_: the callback
109		// accesses these members, so they must still be
110		// alive if the stop_cb_ destructor blocks.
111		std::atomic<bool> claimed_{false};
112		bool canceled_ = false;
113		bool active_ = false;
114		bool in_list_ = false;
115
116		struct cancel_fn
117		{
118		wait_awaiter* self_;
119
120	4x	void operator()() const noexcept
121		{
122	4x	if(!self_->claimed_.exchange(
123		true, std::memory_order_acq_rel))
124		{
125	3x	self_->canceled_ = true;
126	3x	self_->ex_.post(self_->cont_);
127		}
128	4x	}
129		};
130
131		using stop_cb_t =
132		std::stop_callback<cancel_fn>;
133
134		// Aligned storage for stop_cb_t. Declared last:
135		// its destructor may block while the callback
136		// accesses the members above.
137		BOOST_CAPY_MSVC_WARNING_PUSH
138		BOOST_CAPY_MSVC_WARNING_DISABLE(4324) // padded due to alignas
139		alignas(stop_cb_t)
140		unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
141		BOOST_CAPY_MSVC_WARNING_POP
142
143	20x	stop_cb_t& stop_cb_() noexcept
144		{
145		return reinterpret_cast<stop_cb_t>(
146	20x	stop_cb_buf_);
147		}
148
149		public:
150		/** Destroy the awaiter, leaving the event unable to reach it.
151
152		Destroys the stop callback if one is registered, and unlinks
153		the awaiter from the event's wait queue if it is still linked.
154		Both are necessary when the coroutine frame is torn down while
155		suspended, so that neither `set()` nor the stop callback can
156		reach a destroyed awaiter.
157		*/
158	52x	~wait_awaiter()
159		{
160	52x	if(active_)
161	1x	stop_cb_().~stop_cb_t();
162	52x	if(in_list_)
163	1x	e_->waiters_.remove(this);
164	52x	}
165
166		/** Construct an awaiter for the given event.
167
168		@param e The event to wait on. It must outlive the awaiter.
169		*/
170	25x	explicit wait_awaiter(async_event* e) noexcept
171	25x	: e_(e)
172		{
173	25x	}
174
175		/** Construct by moving.
176
177		The moved-from awaiter is left inert: its destructor no longer
178		destroys the stop callback and no longer unlinks from the
179		event's wait queue.
180
181		@param o The awaiter to move from.
182		*/
183	27x	wait_awaiter(wait_awaiter&& o) noexcept
184	54x	: e_(o.e_)
185	27x	, cont_(o.cont_)
186	27x	, ex_(o.ex_)
187	27x	, claimed_(o.claimed_.load(
188		std::memory_order_relaxed))
189	27x	, canceled_(o.canceled_)
190	27x	, active_(std::exchange(o.active_, false))
191	54x	, in_list_(std::exchange(o.in_list_, false))
192		{
193	27x	}
194
195		/** Copy construction is disabled; a waiter is linked into the
196		event's wait queue by address.
197
198		@param other The awaiter that would be copied.
199		*/
200		wait_awaiter(wait_awaiter const& other) = delete;
201
202		/** Copy assignment is disabled; a waiter is linked into the
203		event's wait queue by address.
204
205		@param other The awaiter that would be assigned from.
206
207		@return A reference to `*this`.
208		*/
209		wait_awaiter& operator=(wait_awaiter const& other) = delete;
210
211		/** Move assignment is disabled; a waiter is linked into the
212		event's wait queue by address.
213
214		@param other The awaiter that would be moved from.
215
216		@return A reference to `*this`.
217		*/
218		wait_awaiter& operator=(wait_awaiter&& other) = delete;
219
220		/** Report whether the event is already set.
221
222		@return `true` if the event is set, in which case the awaiting
223		coroutine does not suspend; otherwise `false`.
224		*/
225	25x	bool await_ready() const noexcept
226		{
227	25x	return e_->set_;
228		}
229
230		/** Enqueue the awaiting coroutine until the event is set.
231
232		This is the @ref IoAwaitable overload of `await_suspend`.
233
234		If a stop request is already pending on `env->stop_token`, the
235		awaiter records the cancellation and does not enqueue.
236
237		Otherwise it stores `h` and `env->executor`, links itself into
238		the event's wait queue, and registers a stop callback on
239		`env->stop_token`. Exactly one of `set()` and that callback posts
240		`h` through the stored executor, whichever claims the waiter
241		first. Only the post is subject to that race. A losing stop
242		callback does nothing at all, but `set()` unlinks every waiter it
243		pops whether it claims it or not. That is why @ref await_resume
244		unlinks a canceled waiter only when it is still linked.
245
246		@param h The awaiting coroutine, resumed when the event is set
247		or the wait is canceled.
248
249		@param env The execution environment. Its executor posts the
250		resumption and its stop token is watched for the duration of
251		the wait. It must outlive the wait.
252
253		@return `h` if a stop request was already pending, which
254		resumes the awaiting coroutine immediately without enqueuing
255		it. Otherwise `std::noop_coroutine()`, which leaves the
256		coroutine suspended and returns control to the resumer.
257		*/
258		std::coroutine_handle<>
259	21x	await_suspend(
260		std::coroutine_handle<> h,
261		io_env const* env) noexcept
262		{
263	21x	if(env->stop_token.stop_requested())
264		{
265	1x	canceled_ = true;
266	1x	return h;
267		}
268	20x	cont_.h = h;
269	20x	ex_ = env->executor;
270	20x	e_->waiters_.push_back(this);
271	20x	in_list_ = true;
272	60x	::new(stop_cb_buf_) stop_cb_t(
273	20x	env->stop_token, cancel_fn{this});
274	20x	active_ = true;
275	20x	return std::noop_coroutine();
276		}
277
278		/** Complete the wait and report the outcome.
279
280		Destroys the stop callback if one is registered. If the wait
281		was canceled while still linked into the event's wait queue,
282		unlinks it. `set()` pops every waiter, so a canceled waiter may
283		or may not still be linked when it resumes.
284
285		@return An empty `io_result<>` if the event was set, or one
286		holding `error::canceled` if the stop token fired first.
287		*/
288	22x	[[nodiscard]] io_result<> await_resume() noexcept
289		{
290	22x	if(active_)
291		{
292	19x	stop_cb_().~stop_cb_t();
293	19x	active_ = false;
294		}
295	22x	if(canceled_)
296		{
297	4x	if(in_list_)
298		{
299	3x	e_->waiters_.remove(this);
300	3x	in_list_ = false;
301		}
302	8x	return {make_error_code(
303	4x	error::canceled)};
304		}
305	18x	return {{}};
306		}
307		};
308
309		/// Construct an unset event.
310		async_event() = default;
311
312		/** Copy construction is disabled; suspended waiters point into the
313		event's wait queue.
314
315		@param other The event that would be copied.
316		*/
317		async_event(async_event const& other) = delete;
318
319		/** Copy assignment is disabled; suspended waiters point into the
320		event's wait queue.
321
322		@param other The event that would be assigned from.
323
324		@return A reference to `*this`.
325		*/
326		async_event& operator=(async_event const& other) = delete;
327
328		/** Move construction is disabled; suspended waiters point into the
329		event's wait queue.
330
331		@param other The event that would be moved from.
332		*/
333		async_event(async_event&& other) = delete;
334
335		/** Move assignment is disabled; suspended waiters point into the
336		event's wait queue.
337
338		@param other The event that would be moved from.
339
340		@return A reference to `*this`.
341		*/
342		async_event& operator=(async_event&& other) = delete;
343
344		/** Returns an awaiter that waits until the event is set.
345
346		If the event is already set, completes immediately.
347
348		@return An awaitable that await-returns `(error_code)`.
349		*/
350	25x	wait_awaiter wait() noexcept
351		{
352	25x	return wait_awaiter{this};
353		}
354
355		/** Resumes every waiting coroutine and marks the event set for later `wait()` calls.
356
357		All waiting coroutines are resumed. Canceled waiters
358		are skipped. Subsequent calls to wait() complete
359		immediately until clear() is called.
360		*/
361	17x	void set()
362		{
363	17x	set_ = true;
364		for(;;)
365		{
366	33x	auto* w = waiters_.pop_front();
367	33x	if(!w)
368	17x	break;
369	16x	w->in_list_ = false;
370	16x	if(!w->claimed_.exchange(
371		true, std::memory_order_acq_rel))
372		{
373	16x	w->ex_.post(w->cont_);
374		}
375	16x	}
376	17x	}
377
378		/** Clears the event.
379
380		Subsequent calls to wait() suspend until
381		set() is called again.
382		*/
383	2x	void clear() noexcept
384		{
385	2x	set_ = false;
386	2x	}
387
388		/** Returns true if the event is currently set.
389
390		@return `true` if the event is set; otherwise `false`.
391		*/
392	9x	bool is_set() const noexcept
393		{
394	9x	return set_;
395		}
396		};
397
398		} // namespace capy
399		} // namespace boost
400
401		#endif
402