include/boost/capy/ex/async_event.hpp

100.0% Lines (84 / 84) 100.0% Functions (15 / 15)

[ ] prev/next file

async_event.hpp

f(x) Functions (15)

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