include/boost/capy/ex/async_waker.hpp

93.0% Lines (66/0/71) 100.0% List of functions (11/0/11)

[ ] prev/next file n p prev/next uncovered line

async_waker.hpp

f(x) Functions (11)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
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_EX_ASYNC_WAKER_HPP
12		#define BOOST_CAPY_EX_ASYNC_WAKER_HPP
13
14		#include <boost/capy/detail/config.hpp>
15		#include <boost/capy/continuation.hpp>
16		#include <boost/capy/error.hpp>
17		#include <boost/capy/ex/executor_ref.hpp>
18		#include <boost/capy/ex/io_env.hpp>
19		#include <boost/capy/io_result.hpp>
20
21		#include <atomic>
22		#include <coroutine>
23		#include <new>
24		#include <stop_token>
25		#include <utility>
26
27		/* async_waker implementation notes
28		===================================
29
30		wake() must be callable from foreign threads (that is the whole
31		point: the user's thread provides the timing). A waiter-side
32		claimed_ flag is not enough there -- the
33		waker has to dereference the waiter, and nothing would pin the
34		waiter's frame between reading the pointer and claiming it.
35
36		So the three-state st_ atomic is the single arbiter:
37
38		empty --arm--> armed --wake/cancel CAS--> empty
39		empty --wake--> token --wait consumes--> empty
40
41		Whoever wins the armed->empty CAS owns the resume and may
42		dereference waiter_: the frame cannot die underneath the
43		winner because the coroutine only resumes when the winner
44		posts it. The loser never touches the waiter. When the stop
45		callback wins, a concurrent wake retries, finds empty, and
46		latches a token -- a racing wakeup is deferred, never lost.
47
48		Serialized resumption is required: await_suspend keeps
49		writing after the publishing armed-CAS (the stop_cb
50		placement-new and active_ = true), so a wake/cancel winner
51		can post the continuation while that tail is still running.
52		The posted resume must be ordered after await_suspend's
53		return, which holds on a single-threaded executor (the one
54		thread is still inside await_suspend) and on a strand (the
55		resume is a later turn, synchronized with the current one).
56		A raw multi-threaded executor lets another worker run
57		await_resume against those in-flight writes. async_event and
58		async_mutex make the same assumption; it is stated explicitly
59		here because wake() invites foreign threads into the picture.
60		*/
61
62		namespace boost {
63		namespace capy {
64
65		/** A single-slot waker that hands one wakeup to a waiting coroutine.
66
67		This is the escape hatch for timing and other external events:
68		the user provides the thread and the clock, capy provides the
69		suspension point. One coroutine suspends in `wait()`; any
70		thread wakes it with `wake()`.
71
72		A wakeup with no waiter present is latched as a single pending
73		token, and the next `wait()` consumes it immediately. This
74		makes the wake-before-wait race benign without any lock
75		protocol. Multiple wakes collapse into one token.
76
77		@par Cancellation
78
79		If the environment's stop token is triggered while suspended,
80		the wait completes with `error::canceled`. A wake that loses
81		the race against cancellation is latched for the next `wait()`
82		rather than dropped.
83
84		@par Zero Allocation
85
86		No heap allocation occurs for wait or wake operations.
87
88		@par Thread Safety
89
90		Distinct objects: Safe.@n
91		Shared objects: `wake()` may be called from any thread.
92		`wait()` must only be awaited by one coroutine at a time. The
93		executor must never run the coroutine's continuations
94		concurrently: use a single-threaded executor, or a strand over
95		a multi-threaded one. That is the same threading model as
96		`async_event` and `async_mutex`. Awaiting `wait()` directly
97		on a multi-threaded executor is undefined.
98
99		This type is non-copyable and non-movable because a suspended
100		waiter holds a pointer into the object.
101
102		@par Example
103		@par !example example
104
105		*/
106		class async_waker
107		{
108		public:
109		class wait_awaiter;
110
111		private:
112		static constexpr int state_empty = 0; // no token, no waiter
113		static constexpr int state_token = 1; // latched wakeup
114		static constexpr int state_armed = 2; // waiter suspended
115
116		std::atomic<int> st_{state_empty};
117		wait_awaiter* waiter_ = nullptr;
118
119		public:
120		/** Suspends the caller until `wake()` runs, or resumes it with `error::canceled` on a stop request.
121		*/
122		class wait_awaiter
123		{
124		friend class async_waker;
125
126		async_waker* waker_;
127		continuation cont_;
128		executor_ref ex_;
129
130		// Declared before stop_cb_buf_: the callback accesses
131		// these members, so they must still be alive if the
132		// stop_cb_ destructor blocks.
133		bool canceled_ = false;
134		bool active_ = false;
135		bool published_ = false;
136
137		struct cancel_fn
138		{
139		wait_awaiter* self_;
140
141	9x	void operator()() const noexcept
142		{
143	9x	int expected = state_armed;
144	18x	if(self_->waker_->st_.compare_exchange_strong(
145		expected, state_empty,
146		std::memory_order_acq_rel,
147		std::memory_order_acquire))
148		{
149	8x	self_->canceled_ = true;
150	8x	self_->ex_.post(self_->cont_);
151		}
152	9x	}
153		};
154
155		using stop_cb_t = std::stop_callback<cancel_fn>;
156
157		// Aligned storage for stop_cb_t. Declared last: its
158		// destructor may block while the callback accesses the
159		// members above.
160		BOOST_CAPY_MSVC_WARNING_PUSH
161		BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
162		alignas(stop_cb_t)
163		unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
164		BOOST_CAPY_MSVC_WARNING_POP
165
166	19x	stop_cb_t& stop_cb_() noexcept
167		{
168	19x	return reinterpret_cast<stop_cb_t>(stop_cb_buf_);
169		}
170
171		public:
172		/** Destroy the awaiter, leaving the waker unable to reach it.
173
174		Destroys the stop callback if one is registered. If the awaiter
175		is still armed, it also returns the waker's slot to the empty
176		state, so a later `wake()` cannot dereference a destroyed
177		awaiter. That case means the frame is being torn down without
178		ever being resumed; a wake arriving afterward latches a token
179		instead.
180		*/
181	294x	~wait_awaiter()
182		{
183	294x	if(active_)
184	1x	stop_cb_().~stop_cb_t();
185	294x	if(published_)
186		{
187		// Destroyed while still armed (frame torn down
188		// without resuming): deregister so a later
189		// wake cannot touch the dead frame.
190	1x	int expected = state_armed;
191	1x	waker_->st_.compare_exchange_strong(
192		expected, state_empty,
193		std::memory_order_acq_rel,
194		std::memory_order_acquire);
195		}
196	294x	}
197
198		/** Construct an awaiter for the given waker.
199
200		@param waker The waker to wait on. It must outlive the awaiter.
201		*/
202	147x	explicit wait_awaiter(async_waker* waker) noexcept
203	147x	: waker_(waker)
204		{
205	147x	}
206
207		/** Construct by moving.
208
209		The moved-from awaiter is left inert: its destructor no longer
210		destroys the stop callback and no longer deregisters from the
211		waker.
212
213		@param o The awaiter to move from.
214		*/
215	147x	wait_awaiter(wait_awaiter&& o) noexcept
216	147x	: waker_(o.waker_)
217	147x	, cont_(o.cont_)
218	147x	, ex_(o.ex_)
219	147x	, canceled_(o.canceled_)
220	147x	, active_(std::exchange(o.active_, false))
221	147x	, published_(std::exchange(o.published_, false))
222		{
223	147x	}
224
225		/** Copy construction is disabled; an armed waiter is registered
226		with the waker by address.
227
228		@param other The awaiter that would be copied.
229		*/
230		wait_awaiter(wait_awaiter const& other) = delete;
231
232		/** Copy assignment is disabled; an armed waiter is registered
233		with the waker by address.
234
235		@param other The awaiter that would be assigned from.
236
237		@return A reference to `*this`.
238		*/
239		wait_awaiter& operator=(wait_awaiter const& other) = delete;
240
241		/** Move assignment is disabled; an armed waiter is registered
242		with the waker by address.
243
244		@param other The awaiter that would be moved from.
245
246		@return A reference to `*this`.
247		*/
248		wait_awaiter& operator=(wait_awaiter&& other) = delete;
249
250		/** Consume a latched token, completing synchronously.
251
252		This is not a pure query: the check is a compare-exchange that
253		takes the token. Calling it twice is not idempotent: the second
254		call reports `false`, because the first already consumed the
255		wakeup.
256
257		@return `true` if a pending wakeup token was latched and has now
258		been consumed, in which case the awaiting coroutine does not
259		suspend; otherwise `false`.
260		*/
261	147x	bool await_ready() noexcept
262		{
263	147x	int expected = state_token;
264	147x	return waker_->st_.compare_exchange_strong(
265		expected, state_empty,
266		std::memory_order_acq_rel,
267	147x	std::memory_order_acquire);
268		}
269
270		/** Arm the waker with the awaiting coroutine.
271
272		This is the @ref IoAwaitable overload of `await_suspend`.
273		Unlike `async_event` and `async_mutex`, it has three outcomes,
274		because a `wake()` from another thread can land in the window
275		between `await_ready` and this call.
276
277		@li A stop request is already pending on `env->stop_token`: the
278		awaiter records the cancellation and does not arm.
279
280		@li The waker's slot is no longer empty. Under the single-waiter
281		precondition that means a wakeup was latched after
282		`await_ready` looked, so the token is consumed here instead
283		and the wait succeeds.
284
285		@li Otherwise the slot moves to the armed state, publishing this
286		awaiter to the waker, and a stop callback is registered on
287		`env->stop_token`. Whichever of `wake()` and that callback
288		wins the armed-to-empty transition posts `h` through
289		`env->executor`. The loser does nothing, and a losing
290		`wake()` re-latches its token for the next `wait()`.
291
292		@param h The awaiting coroutine, resumed when the waker fires
293		or the wait is canceled.
294
295		@param env The execution environment. Its executor posts the
296		resumption and its stop token is watched for the duration of
297		the wait. It must outlive the wait.
298
299		@return `h` in the first two cases, which resumes the awaiting
300		coroutine immediately; otherwise `std::noop_coroutine()`, which
301		leaves the coroutine suspended and returns control to the
302		resumer.
303		*/
304		std::coroutine_handle<>
305	51x	await_suspend(
306		std::coroutine_handle<> h,
307		io_env const* env) noexcept
308		{
309	51x	if(env->stop_token.stop_requested())
310		{
311	32x	canceled_ = true;
312	32x	return h;
313		}
314	19x	cont_.h = h;
315	19x	ex_ = env->executor;
316	19x	waker_->waiter_ = this;
317
318	19x	int expected = state_empty;
319	38x	if(!waker_->st_.compare_exchange_strong(
320		expected, state_armed,
321		std::memory_order_acq_rel,
322		std::memory_order_acquire))
323		{
324		// Single-waiter precondition: a second concurrent
325		// wait would find the slot armed.
326	✗	BOOST_CAPY_ASSERT(expected == state_token);
327
328		// A wake latched between await_ready and here;
329		// consume it and resume inline.
330	✗	waker_->st_.store(
331		state_empty, std::memory_order_release);
332	✗	return h;
333		}
334	19x	published_ = true;
335
336	57x	::new(stop_cb_buf_) stop_cb_t(
337	19x	env->stop_token, cancel_fn{this});
338	19x	active_ = true;
339	19x	return std::noop_coroutine();
340		}
341
342		/** Complete the wait and report the outcome.
343
344		Destroys the stop callback if one is registered and clears the
345		armed bookkeeping, so the destructor does not deregister a slot
346		the resumption already consumed.
347
348		@return An empty `io_result<>` if the wait was woken, whether by
349		`wake()` or by a token consumed inline. Otherwise one holding
350		`error::canceled`, which means the stop token won the race.
351		*/
352	146x	[[nodiscard]] io_result<> await_resume() noexcept
353		{
354	146x	if(active_)
355		{
356	18x	stop_cb_().~stop_cb_t();
357	18x	active_ = false;
358		}
359	146x	published_ = false;
360	146x	if(canceled_)
361	39x	return {make_error_code(error::canceled)};
362	107x	return {{}};
363		}
364		};
365
366		/// Construct with no token latched.
367	1x	async_waker() = default;
368
369		/** Copy construction is disabled; an armed waiter points into the
370		waker.
371
372		@param other The waker that would be copied.
373		*/
374		async_waker(async_waker const& other) = delete;
375
376		/** Copy assignment is disabled; an armed waiter points into the waker.
377
378		@param other The waker that would be assigned from.
379
380		@return A reference to `*this`.
381		*/
382		async_waker& operator=(async_waker const& other) = delete;
383
384		/** Move construction is disabled; an armed waiter points into the
385		waker.
386
387		@param other The waker that would be moved from.
388		*/
389		async_waker(async_waker&& other) = delete;
390
391		/** Move assignment is disabled; an armed waiter points into the waker.
392
393		@param other The waker that would be moved from.
394
395		@return A reference to `*this`.
396		*/
397		async_waker& operator=(async_waker&& other) = delete;
398
399		/** Asynchronously wait until woken.
400
401		If a token is latched, completes immediately and consumes
402		it. Otherwise suspends until `wake()` or the stop token
403		fires.
404
405		@par Preconditions
406		No other coroutine is currently waiting on this object.
407
408		@return An awaitable that await-returns `io_result<>`;
409		empty on wakeup, `error::canceled` if the stop
410		token wins.
411		*/
412	147x	wait_awaiter wait() noexcept
413		{
414	147x	return wait_awaiter{this};
415		}
416
417		/** Wake the waiter, or latch the wakeup if none waits.
418
419		Callable from any thread. The waiter's resumption is
420		posted through its executor; this call never resumes a
421		coroutine inline. Multiple calls without an intervening
422		`wait()` collapse into a single token.
423		*/
424	109x	void wake() noexcept
425		{
426		for(;;)
427		{
428	109x	int s = st_.load(std::memory_order_acquire);
429	109x	if(s == state_token)
430	109x	return;
431	107x	if(s == state_empty)
432		{
433	192x	if(st_.compare_exchange_weak(
434		s, state_token,
435		std::memory_order_acq_rel,
436		std::memory_order_acquire))
437	96x	return;
438	✗	continue;
439		}
440		// armed: winning this CAS claims the waiter, whose
441		// frame is pinned until we post its resumption.
442	22x	if(st_.compare_exchange_weak(
443		s, state_empty,
444		std::memory_order_acq_rel,
445		std::memory_order_acquire))
446		{
447	11x	auto* w = waiter_;
448	11x	w->ex_.post(w->cont_);
449	11x	return;
450		}
451	✗	}
452		}
453		};
454
455		} // namespace capy
456		} // namespace boost
457
458		#endif
459