include/boost/corosio/detail/timer.hpp

98.5% Lines (64 / 65) 94.4% Functions (17 / 18)

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

timer.hpp

f(x) Functions (18)

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