include/boost/corosio/detail/timer.hpp

98.5% Lines (67 / 68) 94.7% Functions (18 / 19)

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

timer.hpp

f(x) Functions (19)

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