include/boost/corosio/detail/timer.hpp

98.3% Lines (58/0/59) 94.1% List of functions (16/0/17)

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

timer.hpp

f(x) Functions (17)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3		// Copyright (c) 2026 Steve Gerbino
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/corosio
9		//
10
11		#ifndef BOOST_COROSIO_DETAIL_TIMER_HPP
12		#define BOOST_COROSIO_DETAIL_TIMER_HPP
13
14		#include <boost/corosio/detail/config.hpp>
15		#include <boost/corosio/detail/intrusive.hpp>
16		#include <boost/corosio/detail/scheduler_op.hpp>
17		#include <boost/corosio/io/io_object.hpp>
18		#include <boost/capy/continuation.hpp>
19		#include <boost/capy/io_result.hpp>
20		#include <boost/capy/error.hpp>
21		#include <boost/capy/ex/executor_ref.hpp>
22		#include <boost/capy/ex/execution_context.hpp>
23		#include <boost/capy/ex/io_env.hpp>
24
25		#include <atomic>
26		#include <chrono>
27		#include <coroutine>
28		#include <cstddef>
29		#include <limits>
30		#include <new>
31		#include <stop_token>
32		#include <system_error>
33
34		namespace boost::corosio::detail {
35
36		// timer_service is defined in timer_service.hpp, which includes this
37		// header. waiter_node and wait_awaitable are defined below the timer
38		// class: waiter_node stores a timer::implementation*, which cannot be
39		// forward-declared as a nested type. implementation stores only a
40		// waiter_node pointer, so this forward declaration suffices for its
41		// data layout.
42		class timer_service;
43		struct waiter_node;
44		struct wait_awaitable;
45
46		/** An asynchronous timer for coroutine I/O.
47
48		This class provides asynchronous timer operations that return
49		awaitable types. The timer can be used to schedule operations
50		to occur after a specified duration or at a specific time point.
51
52		Each timer carries at most one wait: `delay` and `timeout` own a
53		private timer per `co_await`. When the timer expires the waiter
54		completes with success; a cancelled wait completes with an error
55		that compares equal to `capy::cond::canceled`.
56
57		Each timer operation participates in the affine awaitable protocol,
58		ensuring coroutines resume on the correct executor.
59
60		@par Thread Safety
61		Distinct objects: Safe.@n
62		Shared objects: Unsafe.
63
64		@par Semantics
65		Timers are not backed by per-timer kernel objects. The io_context's
66		timer service keeps a process-side min-heap of pending expirations;
67		the nearest expiry drives the reactor's poll timeout, and expirations
68		are processed in the run loop.
69		*/
70		class BOOST_COROSIO_DECL timer : public io_object
71		{
72		friend struct wait_awaitable;
73
74		public:
75		/** Backend state and wait entry point for a timer.
76
77		Holds per-timer state ( expiry, heap position, the single waiter ) and
78		the `wait` entry point used by the awaitable returned from
79		@ref timer::wait. There is exactly one concrete timer backend,
80		so `wait` is a plain member function rather than a virtual
81		dispatch point.
82		*/
83		struct implementation : io_object::implementation
84		{
85		/// Sentinel value indicating the timer is not in the heap.
86		static constexpr std::size_t npos =
87		(std::numeric_limits<std::size_t>::max)();
88
89		// Only mutated by the owning thread (expires_at/expires_after)
90		// before a wait is published; cross-thread consumers read the
91		// heap entry's copied time_, never this field, so it needs no
92		// atomicity.
93		/// The absolute expiry time point.
94		std::chrono::steady_clock::time_point expiry_{};
95
96		// heap_index_ and might_have_pending_waits_ are cross-thread
97		// hints, not authoritative state: the real state lives in the
98		// heap and the published waiter under timer_service::mutex_. Every
99		// unlocked fast-out that reads them is either re-validated under
100		// the mutex or safe under a stale value in both directions, and
101		// any locked writer / locked reader pair is already ordered by
102		// the mutex. All accesses therefore use memory_order_relaxed,
103		// which keeps the lock-free fast paths fence-free while making
104		// the concurrent reads well-defined.
105		/// Index in the timer service's min-heap, or `npos`.
106		std::atomic<std::size_t> heap_index_{npos};
107
108		// false implies waiter_ is null: both are cleared together
109		// under the service mutex.
110		/// True if `wait()` has been called since last cancel.
111		std::atomic<bool> might_have_pending_waits_{false};
112
113		/// The timer service that owns this implementation.
114		timer_service* svc_ = nullptr;
115
116		// Exactly one wait may be outstanding: delay and timeout own
117		// a private timer per co_await, and the service's drains rely
118		// on the one-to-one pairing.
119		/// The waiter published on this timer, or `nullptr`.
120		waiter_node* waiter_ = nullptr;
121
122		/// Free list linkage, reused when this impl is recycled.
123		implementation* next_free_ = nullptr;
124
125		/// Construct bound to the given timer service.
126	1399x	explicit implementation(timer_service& svc) noexcept : svc_(&svc) {}
127
128		/** Check whether the timer is expired and absent from the heap.
129
130		The single definition of the already-expired fast-path
131		predicate: `await_suspend` tests it inline and `wait()`
132		re-tests it because the expiry can elapse between the two
133		reads.
134		*/
135	28763x	bool already_expired() const noexcept
136		{
137	86289x	return heap_index_.load(std::memory_order_relaxed) == npos &&
138	56862x	(expiry_ == (std::chrono::steady_clock::time_point::min)() \|\|
139	56862x	expiry_ <= std::chrono::steady_clock::now());
140		}
141
142		/** Asynchronously wait for the timer to expire.
143
144		Publishes the waiter into the service's heap and the
145		timer's waiter slot, after which it may complete on any
146		thread. If the timer is already expired and not in the
147		heap, completes by posting the continuation without
148		publishing.
149
150		@par Preconditions
151		@p w is fully initialized, and its storage (the awaitable
152		on the suspended coroutine's frame) outlives the wait.
153
154		@param w The waiter to publish.
155		*/
156		// Exported at member level: dllexport on the enclosing timer
157		// class does not extend to nested classes, and header-inline
158		// callers (wait_awaitable::await_suspend) reference this
159		// symbol from outside the corosio DLL.
160		BOOST_COROSIO_DECL
161		std::coroutine_handle<> wait(waiter_node& w);
162
163		/** Publish a waiter unconditionally.
164
165		Like `wait`, but never takes the elapsed fast path. The
166		fast path posts the continuation directly, bypassing the
167		embedded op; hook-driven waits must observe every
168		completion through the op, where the re-arm hook runs.
169
170		@par Preconditions
171		Same as `wait`.
172
173		@param w The waiter to publish.
174		*/
175		std::coroutine_handle<> publish(waiter_node& w);
176		};
177
178		/// The clock type used for time operations.
179		using clock_type = std::chrono::steady_clock;
180
181		/// The time point type for absolute expiry times.
182		using time_point = clock_type::time_point;
183
184		/// The duration type for relative expiry times.
185		using duration = clock_type::duration;
186
187		/** Destructor.
188
189		Cancels any pending operations and releases timer resources.
190		*/
191		~timer() override;
192
193		/** Construct a timer from an execution context.
194
195		@param ctx The execution context that will own this timer. It
196		must be a corosio io_context; otherwise the constructor
197		throws (a timer service is required).
198
199		@throws std::logic_error if @p ctx is not an io_context.
200		*/
201		explicit timer(capy::execution_context& ctx);
202
203		/** Move constructor.
204
205		Transfers ownership of the timer resources. Required so a
206		disengaged `std::optional<timer>` is movable; a timer is never
207		moved while a wait is published.
208
209		@pre No awaitables returned by @p other's methods exist.
210		*/
211	✗	timer(timer&&) noexcept = default;
212
213		/** Move assignment operator.
214
215		Closes any existing timer and transfers ownership.
216
217		@pre No awaitables returned by either `*this` or @p other's
218		methods exist.
219		*/
220		timer& operator=(timer&&) noexcept = default;
221
222		timer(timer const&) = delete;
223		timer& operator=(timer const&) = delete;
224
225		/** Return the timer's expiry time as an absolute time.
226
227		@return The expiry time point. If no expiry has been set,
228		returns a default-constructed time_point.
229		*/
230		time_point expiry() const noexcept
231		{
232		return get().expiry_;
233		}
234
235		/** Set the timer's expiry time as an absolute time.
236
237		@par Preconditions
238		No wait is published on this timer.
239
240		@param t The expiry time to be used for the timer.
241		*/
242	16x	void expires_at(time_point t)
243		{
244	16x	auto& impl = get();
245	32x	BOOST_COROSIO_ASSERT(
246		impl.heap_index_.load(std::memory_order_relaxed) ==
247		implementation::npos);
248	16x	impl.expiry_ = t;
249	16x	}
250
251		/** Set the timer's expiry time relative to now.
252
253		@par Preconditions
254		No wait is published on this timer.
255
256		@param d The expiry time relative to now.
257		*/
258	19281x	void expires_after(duration d)
259		{
260	19281x	auto& impl = get();
261	38562x	BOOST_COROSIO_ASSERT(
262		impl.heap_index_.load(std::memory_order_relaxed) ==
263		implementation::npos);
264	19281x	if (d <= duration::zero())
265	681x	impl.expiry_ = (time_point::min)();
266		else
267		{
268		// Saturate rather than overflow: a clamped near-max duration
269		// (e.g. delay(hours::max())) would wrap now() + d past the
270		// clock's range and appear already elapsed.
271	18600x	auto const now = clock_type::now();
272	18600x	impl.expiry_ =
273	18600x	((time_point::max)() - now < d) ? (time_point::max)() : now + d;
274		}
275	19281x	}
276
277		/** Set the timer's expiry time relative to now.
278
279		This is a convenience overload that accepts any duration type
280		and converts it to the timer's native duration type.
281
282		@param d The expiry time relative to now.
283		*/
284		template<class Rep, class Period>
285		void expires_after(std::chrono::duration<Rep, Period> d)
286		{
287		expires_after(std::chrono::duration_cast<duration>(d));
288		}
289
290		/** Wait for the timer to expire.
291
292		At most one wait may be outstanding at a time.
293
294		The operation supports cancellation via `std::stop_token` through
295		the affine awaitable protocol. If the associated stop token is
296		triggered, only that waiter completes with an error that
297		compares equal to `capy::cond::canceled`.
298
299		This timer must outlive the returned awaitable.
300
301		@return An awaitable that completes with `io_result<>`.
302		*/
303		// Defined below wait_awaitable, which needs timer complete.
304		wait_awaitable wait();
305
306		/** Publish a hook-driven wait.
307
308		Bypasses the elapsed fast path so every completion is
309		delivered through the waiter's embedded op, where the
310		re-arm hook is consulted. Used by awaitables that
311		re-publish the waiter to continue a logical wait across
312		several timer expirations.
313
314		@par Preconditions
315		@p w is fully initialized ( handle, executor, stop token,
316		hook fields ) and its storage outlives the wait.
317
318		@param w The waiter to publish.
319
320		@return `std::noop_coroutine()`.
321		*/
322		std::coroutine_handle<> publish_wait(waiter_node& w);
323
324		/** Re-arm an already-fired waiter with a new relative expiry.
325
326		Stores the ( saturated ) expiry and re-publishes @p w. The
327		waiter's original work count and stop callback remain in
328		effect. Must only be called from the waiter's re-arm hook,
329		where the waiter has been popped from the service but not
330		yet resumed.
331
332		@par Preconditions
333		The timer has no other waiters — this is what makes the
334		unlocked expiry write race-free.
335
336		Re-publication needs heap capacity and can fail under
337		allocation pressure. On failure the waiter is left exactly as
338		the hook received it, so the caller completes the wait through
339		the normal resume path instead of re-arming.
340
341		@param w The waiter to re-publish.
342		@param d The next expiry relative to now.
343
344		@return `true` if re-published; `false` if allocation failed.
345		*/
346		[[nodiscard]] bool rearm_wait(waiter_node& w, duration d) noexcept;
347
348		protected:
349		explicit timer(handle h) noexcept : io_object(std::move(h)) {}
350
351		private:
352		/// Return the underlying implementation.
353	38594x	implementation& get() const noexcept
354		{
355	38594x	return static_cast<implementation>(h_.get());
356		}
357		};
358
359		/** Frame-resident per-wait state for a timer wait.
360
361		One node exists per `co_await` on a timer, embedded in the
362		awaitable on the suspended coroutine's frame — never allocated.
363		Once published by `implementation::wait()` the node may be
364		completed from any thread; every completion path finishes
365		touching the node before resuming or destroying the coroutine,
366		because either act may end the node's storage.
367
368		The node owns no resources: the stop token is borrowed from the
369		awaiting chain's `io_env` (which outlives the suspension) and
370		the stop callback is managed manually in `cb_buf_`, destroyed on
371		every completion path before the frame can die.
372		*/
373		struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node
374		: intrusive_list<waiter_node>::node
375		{
376		// Embedded completion op — avoids heap allocation per fire/cancel.
377		// Members are exported and defined non-inline in timer.cpp: the
378		// inline waiter_node constructor references do_complete and the
379		// vtable from translation units that reach this header through
380		// delay.hpp without ever including timer_service.hpp, so the one
381		// strong definition must live in a TU that is always linked.
382		struct BOOST_COROSIO_SYMBOL_VISIBLE completion_op final : scheduler_op
383		{
384		waiter_node* waiter_ = nullptr;
385
386		BOOST_COROSIO_DECL
387		static void do_complete(
388		void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);
389
390	32122x	completion_op() noexcept : scheduler_op(&do_complete) {}
391
392		BOOST_COROSIO_DECL void operator()() override;
393		BOOST_COROSIO_DECL void destroy() override;
394		};
395
396		// Per-waiter stop_token cancellation
397		struct canceller
398		{
399		waiter_node* waiter_;
400		BOOST_COROSIO_DECL void operator()() const;
401		};
402
403		using stop_cb_type = std::stop_callback<canceller>;
404
405		// nullptr once unpublished from the timer ( concurrency marker )
406		/// The timer this waiter is published on, or `nullptr`.
407		timer::implementation* impl_ = nullptr;
408
409		/// The timer service that completes this waiter.
410		timer_service* svc_ = nullptr;
411
412		/// The suspended coroutine, destroyed by the shutdown drains.
413		std::coroutine_handle<> h_;
414
415		/// The continuation posted to resume the coroutine.
416		capy::continuation cont_;
417
418		/// The executor the continuation is posted through.
419		capy::executor_ref d_;
420
421		// Borrowed from the awaiting chain's io_env, which outlives the
422		// suspension; the node holds no owning state.
423		/// The stop token observed for cancellation.
424		std::stop_token const* token_ = nullptr;
425
426		/// The completion result read by `await_resume`.
427		std::error_code ec_;
428
429		// Consulted by the completion op before resuming; lets a
430		// clock-facade wait re-publish itself instead of completing.
431		// Never consulted on the shutdown destroy path. Consulted on
432		// every completion, including cancellation ( `ec_` set ) — the
433		// hook must inspect `w`'s `ec_` and must not re-arm a canceled
434		// waiter. Runs inside the completion path; must not throw.
435		/// Re-arm hook: return true to skip resumption ( wait continues ).
436		bool (on_fire_)(void) noexcept = nullptr;
437
438		/// Context passed to `on_fire_` ( the owning awaitable ).
439		void* on_fire_ctx_ = nullptr;
440
441		/// The embedded completion op posted to the scheduler.
442		completion_op op_;
443
444		// stop_callback is neither movable nor assignable; construct it
445		// in place once the node is pinned on the coroutine frame, and
446		// destroy it manually on every completion path.
447		/// Storage for the armed stop callback.
448		alignas(stop_cb_type) unsigned char cb_buf_[sizeof(stop_cb_type)];
449
450		/// True while `cb_buf_` holds a live stop callback.
451		bool cb_active_ = false;
452
453	32122x	waiter_node() noexcept
454	32122x	{
455	32122x	op_.waiter_ = this;
456	32122x	}
457
458		// The embedded op self-points and the list hooks are published
459		// to other threads; the node never moves.
460		waiter_node(waiter_node const&) = delete;
461		waiter_node& operator=(waiter_node const&) = delete;
462
463		/** Bind the coroutine and its environment before publication.
464
465		The single definition of the fields every wait must populate
466		before the node is published; hook-driven waits additionally
467		set `on_fire_` / `on_fire_ctx_`.
468
469		@param h The coroutine to resume on completion.
470		@param env The awaiting chain's environment; must outlive
471		the suspension.
472		*/
473	15056x	void bind(std::coroutine_handle<> h, capy::io_env const& env) noexcept
474		{
475	15056x	h_ = h;
476	15056x	cont_.h = h;
477	15056x	d_ = env.executor;
478	15056x	token_ = &env.stop_token;
479	15056x	}
480
481		/** Arm the stop callback.
482
483		@par Preconditions
484		`token_` is set.
485		*/
486	1584x	void arm_stop_cb()
487		{
488	1584x	new (cb_buf_) stop_cb_type(*token_, canceller{this});
489	1584x	cb_active_ = true;
490	1584x	}
491
492		/// Destroy the stop callback if armed.
493	14203x	void reset_stop_cb() noexcept
494		{
495	14203x	if (cb_active_)
496		{
497	1584x	std::launder(reinterpret_cast<stop_cb_type*>(cb_buf_))
498	1584x	->~stop_cb_type();
499	1584x	cb_active_ = false;
500		}
501	14203x	}
502		};
503
504		/** Awaitable returned by `timer::wait()`.
505
506		Carries the waiter node so a wait performs no allocation. The
507		awaitable is movable only before `await_suspend` publishes the
508		node (a move builds a fresh, quiescent node); afterwards it is
509		pinned on the coroutine frame until the wait completes.
510		*/
511		struct wait_awaitable
512		{
513		timer& t_;
514		waiter_node w_;
515
516	14808x	explicit wait_awaitable(timer& t) noexcept : t_(t) {}
517
518	14808x	wait_awaitable(wait_awaitable&& o) noexcept : t_(o.t_) {}
519
520		wait_awaitable(wait_awaitable const&) = delete;
521		wait_awaitable& operator=(wait_awaitable const&) = delete;
522		wait_awaitable& operator=(wait_awaitable&&) = delete;
523
524	2053x	bool await_ready() const noexcept
525		{
526	2053x	return false;
527		}
528
529		// Cancellation surfaces through w_.ec_: the stop_token path in
530		// wait() completes the waiter with error::canceled written to
531		// it, so there is no separate token to consult here.
532	14779x	[[nodiscard]] capy::io_result<> await_resume() const noexcept
533		{
534	14779x	return {w_.ec_};
535		}
536
537	14808x	auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
538		-> std::coroutine_handle<>
539		{
540	14808x	auto& impl = t_.get();
541	14808x	w_.bind(h, *env);
542
543		// Inline fast path: already expired and not in the heap.
544		// Post instead of dispatch so the coroutine yields to the
545		// scheduler, allowing other queued work to run.
546	14808x	if (impl.already_expired())
547		{
548	853x	w_.ec_ = {};
549	853x	w_.d_.post(w_.cont_);
550	853x	return std::noop_coroutine();
551		}
552
553	13955x	return impl.wait(w_);
554		}
555		};
556
557		inline wait_awaitable
558	14808x	timer::wait()
559		{
560	14808x	return wait_awaitable(*this);
561		}
562
563		} // namespace boost::corosio::detail
564
565		#endif
566