include/boost/corosio/detail/timer.hpp

100.0% Lines (60/0/60) 100.0% List of functions (16/0/16)

[ ] prev/next file

timer.hpp

f(x) Functions (16)

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