include/boost/capy/ex/async_waker.hpp

100.0% Lines (96 / 96, 13 excl) 100.0% Functions (13 / 13)

[ ] prev/next file

async_waker.hpp

f(x) Functions (13)

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