include/boost/capy/ex/async_waker.hpp

92.9% Lines (65/70) 100.0% List of functions (10/10)

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

async_waker.hpp

f(x) Functions (10)

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	4x	void operator()() const noexcept
154		{
155	4x	int expected = state_armed;
156	8x	if(self_->waker_->st_.compare_exchange_strong(
157		expected, state_empty,
158		std::memory_order_acq_rel,
159		std::memory_order_acquire))
160		{
161	3x	self_->canceled_ = true;
162	3x	self_->ex_.post(self_->cont_);
163		}
164	4x	}
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	8x	stop_cb_t& stop_cb_() noexcept
179		{
180	8x	return reinterpret_cast<stop_cb_t>(stop_cb_buf_);
181		}
182
183		public:
184	262x	~wait_awaiter()
185		{
186	262x	if(active_)
187	1x	stop_cb_().~stop_cb_t();
188	262x	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	262x	}
200
201	131x	explicit wait_awaiter(async_waker* waker) noexcept
202	131x	: waker_(waker)
203		{
204	131x	}
205
206	131x	wait_awaiter(wait_awaiter&& o) noexcept
207	131x	: waker_(o.waker_)
208	131x	, cont_(o.cont_)
209	131x	, ex_(o.ex_)
210	131x	, canceled_(o.canceled_)
211	131x	, active_(std::exchange(o.active_, false))
212	131x	, published_(std::exchange(o.published_, false))
213		{
214	131x	}
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	131x	bool await_ready() noexcept
222		{
223	131x	int expected = state_token;
224	131x	return waker_->st_.compare_exchange_strong(
225		expected, state_empty,
226		std::memory_order_acq_rel,
227	131x	std::memory_order_acquire);
228		}
229
230		/** IoAwaitable protocol overload. */
231		std::coroutine_handle<>
232	30x	await_suspend(
233		std::coroutine_handle<> h,
234		io_env const* env) noexcept
235		{
236	30x	if(env->stop_token.stop_requested())
237		{
238	22x	canceled_ = true;
239	22x	return h;
240		}
241	8x	cont_.h = h;
242	8x	ex_ = env->executor;
243	8x	waker_->waiter_ = this;
244
245	8x	int expected = state_empty;
246	16x	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	8x	published_ = true;
262
263	24x	::new(stop_cb_buf_) stop_cb_t(
264	8x	env->stop_token, cancel_fn{this});
265	8x	active_ = true;
266	8x	return std::noop_coroutine();
267		}
268
269	130x	io_result<> await_resume() noexcept
270		{
271	130x	if(active_)
272		{
273	7x	stop_cb_().~stop_cb_t();
274	7x	active_ = false;
275		}
276	130x	published_ = false;
277	130x	if(canceled_)
278	24x	return {make_error_code(error::canceled)};
279	106x	return {{}};
280		}
281		};
282
283		/// Construct with no token latched.
284		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	131x	wait_awaiter wait() noexcept
312		{
313	131x	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	108x	void wake() noexcept
324		{
325		for(;;)
326		{
327	108x	int s = st_.load(std::memory_order_acquire);
328	108x	if(s == state_token)
329	108x	return;
330	106x	if(s == state_empty)
331		{
332	202x	if(st_.compare_exchange_weak(
333		s, state_token,
334		std::memory_order_acq_rel,
335		std::memory_order_acquire))
336	101x	return;
337	✗	continue;
338		}
339		// armed: winning this CAS claims the waiter, whose
340		// frame is pinned until we post its resumption.
341	10x	if(st_.compare_exchange_weak(
342		s, state_empty,
343		std::memory_order_acq_rel,
344		std::memory_order_acquire))
345		{
346	5x	auto* w = waiter_;
347	5x	w->ex_.post(w->cont_);
348	5x	return;
349		}
350	✗	}
351		}
352		};
353
354		} // namespace capy
355		} // namespace boost
356
357		#endif
358