include/boost/capy/ex/async_waker.hpp

93.0% Lines (66/71) 100.0% List of functions (11/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		//
4		// Distributed under the Boost Software License, Version 1.0. (See accompanying
5		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6		//
7		// Official repository: https://github.com/cppalliance/capy
8		//
9
10		#ifndef BOOST_CAPY_EX_ASYNC_WAKER_HPP
11		#define BOOST_CAPY_EX_ASYNC_WAKER_HPP
12
13		#include <boost/capy/detail/config.hpp>
14		#include <boost/capy/continuation.hpp>
15		#include <boost/capy/error.hpp>
16		#include <boost/capy/ex/executor_ref.hpp>
17		#include <boost/capy/ex/io_env.hpp>
18		#include <boost/capy/io_result.hpp>
19
20		#include <atomic>
21		#include <coroutine>
22		#include <new>
23		#include <stop_token>
24		#include <utility>
25
26		/* async_waker implementation notes
27		===================================
28
29		wake() must be callable from foreign threads (that is the whole
30		point: the user's thread provides the timing). A waiter-side
31		claimed_ flag is not enough there -- the
32		waker has to dereference the waiter, and nothing would pin the
33		waiter's frame between reading the pointer and claiming it.
34
35		So the three-state st_ atomic is the single arbiter:
36
37		empty --arm--> armed --wake/cancel CAS--> empty
38		empty --wake--> token --wait consumes--> empty
39
40		Whoever wins the armed->empty CAS owns the resume and may
41		dereference waiter_: the frame cannot die underneath the
42		winner because the coroutine only resumes when the winner
43		posts it. The loser never touches the waiter. When the stop
44		callback wins, a concurrent wake retries, finds empty, and
45		latches a token -- a racing wakeup is deferred, never lost.
46
47		Serialized resumption is required: await_suspend keeps
48		writing after the publishing armed-CAS (the stop_cb
49		placement-new and active_ = true), so a wake/cancel winner
50		can post the continuation while that tail is still running.
51		The posted resume must be ordered after await_suspend's
52		return, which holds on a single-threaded executor (the one
53		thread is still inside await_suspend) and on a strand (the
54		resume is a later turn, synchronized with the current one).
55		A raw multi-threaded executor lets another worker run
56		await_resume against those in-flight writes. async_event and
57		async_mutex make the same assumption; it is stated explicitly
58		here because wake() invites foreign threads into the picture.
59		*/
60
61		namespace boost {
62		namespace capy {
63
64		/** A single-slot waker that hands one wakeup to a waiting coroutine.
65
66		This is the escape hatch for timing and other external events:
67		the user provides the thread and the clock, capy provides the
68		suspension point. One coroutine suspends in `wait()`; any
69		thread wakes it with `wake()`.
70
71		A wakeup with no waiter present is latched as a single pending
72		token, and the next `wait()` consumes it immediately. This
73		makes the wake-before-wait race benign without any lock
74		protocol. Multiple wakes collapse into one token.
75
76		@par Cancellation
77
78		If the environment's stop token is triggered while suspended,
79		the wait completes with `error::canceled`. A wake that loses
80		the race against cancellation is latched for the next `wait()`
81		rather than dropped.
82
83		@par Zero Allocation
84
85		No heap allocation occurs for wait or wake operations.
86
87		@par Thread Safety
88
89		Distinct objects: Safe.@n
90		Shared objects: `wake()` may be called from any thread.
91		`wait()` must only be awaited by one coroutine at a time, and
92		only on an executor that never runs the coroutine's
93		continuations concurrently: a single-threaded executor or a
94		strand over a multi-threaded one (the same threading model as
95		`async_event` and `async_mutex`). Awaiting `wait()` directly
96		on a multi-threaded executor is undefined.
97
98		This type is non-copyable and non-movable because a suspended
99		waiter holds a pointer into the object.
100
101		@par Example
102		@code
103		async_waker waker;
104
105		// user-provided timing thread
106		std::thread th([&waker] {
107		std::this_thread::sleep_for(100ms);
108		waker.wake();
109		});
110
111		task<> waiter() {
112		auto [ec] = co_await waker.wait();
113		// resumed on the executor after ~100ms
114		}
115		// ... th.join() after the pool drains
116		@endcode
117		*/
118		class async_waker
119		{
120		public:
121		class wait_awaiter;
122
123		private:
124		static constexpr int state_empty = 0; // no token, no waiter
125		static constexpr int state_token = 1; // latched wakeup
126		static constexpr int state_armed = 2; // waiter suspended
127
128		std::atomic<int> st_{state_empty};
129		wait_awaiter* waiter_ = nullptr;
130
131		public:
132		/** Awaiter returned by wait().
133		*/
134		class wait_awaiter
135		{
136		friend class async_waker;
137
138		async_waker* waker_;
139		continuation cont_;
140		executor_ref ex_;
141
142		// Declared before stop_cb_buf_: the callback accesses
143		// these members, so they must still be alive if the
144		// stop_cb_ destructor blocks.
145		bool canceled_ = false;
146		bool active_ = false;
147		bool published_ = false;
148
149		struct cancel_fn
150		{
151		wait_awaiter* self_;
152
153	24x	void operator()() const noexcept
154		{
155	24x	int expected = state_armed;
156	48x	if(self_->waker_->st_.compare_exchange_strong(
157		expected, state_empty,
158		std::memory_order_acq_rel,
159		std::memory_order_acquire))
160		{
161	18x	self_->canceled_ = true;
162	18x	self_->ex_.post(self_->cont_);
163		}
164	24x	}
165		};
166
167		using stop_cb_t = std::stop_callback<cancel_fn>;
168
169		// Aligned storage for stop_cb_t. Declared last: its
170		// destructor may block while the callback accesses the
171		// members above.
172		BOOST_CAPY_MSVC_WARNING_PUSH
173		BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
174		alignas(stop_cb_t)
175		unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
176		BOOST_CAPY_MSVC_WARNING_POP
177
178	101x	stop_cb_t& stop_cb_() noexcept
179		{
180	101x	return reinterpret_cast<stop_cb_t>(stop_cb_buf_);
181		}
182
183		public:
184	256x	~wait_awaiter()
185		{
186	256x	if(active_)
187	1x	stop_cb_().~stop_cb_t();
188	256x	if(published_)
189		{
190		// Destroyed while still armed (frame torn down
191		// without resuming): deregister so a later
192		// wake cannot touch the dead frame.
193	1x	int expected = state_armed;
194	1x	waker_->st_.compare_exchange_strong(
195		expected, state_empty,
196		std::memory_order_acq_rel,
197		std::memory_order_acquire);
198		}
199	256x	}
200
201	128x	explicit wait_awaiter(async_waker* waker) noexcept
202	128x	: waker_(waker)
203		{
204	128x	}
205
206	128x	wait_awaiter(wait_awaiter&& o) noexcept
207	128x	: waker_(o.waker_)
208	128x	, cont_(o.cont_)
209	128x	, ex_(o.ex_)
210	128x	, canceled_(o.canceled_)
211	128x	, active_(std::exchange(o.active_, false))
212	128x	, published_(std::exchange(o.published_, false))
213		{
214	128x	}
215
216		wait_awaiter(wait_awaiter const&) = delete;
217		wait_awaiter& operator=(wait_awaiter const&) = delete;
218		wait_awaiter& operator=(wait_awaiter&&) = delete;
219
220		/// Consume a latched token, completing synchronously.
221	128x	bool await_ready() noexcept
222		{
223	128x	int expected = state_token;
224	128x	return waker_->st_.compare_exchange_strong(
225		expected, state_empty,
226		std::memory_order_acq_rel,
227	128x	std::memory_order_acquire);
228		}
229
230		/** IoAwaitable protocol overload. */
231		std::coroutine_handle<>
232	104x	await_suspend(
233		std::coroutine_handle<> h,
234		io_env const* env) noexcept
235		{
236	104x	if(env->stop_token.stop_requested())
237		{
238	3x	canceled_ = true;
239	3x	return h;
240		}
241	101x	cont_.h = h;
242	101x	ex_ = env->executor;
243	101x	waker_->waiter_ = this;
244
245	101x	int expected = state_empty;
246	202x	if(!waker_->st_.compare_exchange_strong(
247		expected, state_armed,
248		std::memory_order_acq_rel,
249		std::memory_order_acquire))
250		{
251		// Single-waiter precondition: a second concurrent
252		// wait would find the slot armed.
253	✗	BOOST_CAPY_ASSERT(expected == state_token);
254
255		// A wake latched between await_ready and here;
256		// consume it and resume inline.
257	✗	waker_->st_.store(
258		state_empty, std::memory_order_release);
259	✗	return h;
260		}
261	101x	published_ = true;
262
263	303x	::new(stop_cb_buf_) stop_cb_t(
264	101x	env->stop_token, cancel_fn{this});
265	101x	active_ = true;
266	101x	return std::noop_coroutine();
267		}
268
269	127x	io_result<> await_resume() noexcept
270		{
271	127x	if(active_)
272		{
273	100x	stop_cb_().~stop_cb_t();
274	100x	active_ = false;
275		}
276	127x	published_ = false;
277	127x	if(canceled_)
278	20x	return {make_error_code(error::canceled)};
279	107x	return {{}};
280		}
281		};
282
283		/// Construct with no token latched.
284	1x	async_waker() = default;
285
286		/// Copy constructor (deleted).
287		async_waker(async_waker const&) = delete;
288
289		/// Copy assignment (deleted).
290		async_waker& operator=(async_waker const&) = delete;
291
292		/// Move constructor (deleted).
293		async_waker(async_waker&&) = delete;
294
295		/// Move assignment (deleted).
296		async_waker& operator=(async_waker&&) = delete;
297
298		/** Asynchronously wait until woken.
299
300		If a token is latched, completes immediately and consumes
301		it. Otherwise suspends until `wake()` or the stop token
302		fires.
303
304		@par Preconditions
305		No other coroutine is currently waiting on this object.
306
307		@return An awaitable that await-returns `io_result<>`;
308		empty on wakeup, `error::canceled` if the stop
309		token wins.
310		*/
311	128x	wait_awaiter wait() noexcept
312		{
313	128x	return wait_awaiter{this};
314		}
315
316		/** Wake the waiter, or latch the wakeup if none waits.
317
318		Callable from any thread. The waiter's resumption is
319		posted through its executor; this call never resumes a
320		coroutine inline. Multiple calls without an intervening
321		`wait()` collapse into a single token.
322		*/
323	109x	void wake() noexcept
324		{
325		for(;;)
326		{
327	109x	int s = st_.load(std::memory_order_acquire);
328	109x	if(s == state_token)
329	109x	return;
330	107x	if(s == state_empty)
331		{
332	48x	if(st_.compare_exchange_weak(
333		s, state_token,
334		std::memory_order_acq_rel,
335		std::memory_order_acquire))
336	24x	return;
337	✗	continue;
338		}
339		// armed: winning this CAS claims the waiter, whose
340		// frame is pinned until we post its resumption.
341	166x	if(st_.compare_exchange_weak(
342		s, state_empty,
343		std::memory_order_acq_rel,
344		std::memory_order_acquire))
345		{
346	83x	auto* w = waiter_;
347	83x	w->ex_.post(w->cont_);
348	83x	return;
349		}
350	✗	}
351		}
352		};
353
354		} // namespace capy
355		} // namespace boost
356
357		#endif
358