include/boost/corosio/detail/timer.hpp

96.8% Lines (61/63) 100.0% List of functions (15/15)

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

timer.hpp

f(x) Functions (15)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2025 Vinnie Falco ([email protected])
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. intrusive_list only stores
43		// waiter_node pointers, so this forward declaration suffices for
44		// implementation's 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		Multiple coroutines may wait concurrently on the same timer.
56		When the timer expires, all waiters complete with success. When
57		the timer is cancelled, all waiters complete with an error that
58		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, waiter list) 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 waiter list 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		/// True if `wait()` has been called since last cancel.
112		std::atomic<bool> might_have_pending_waits_{false};
113
114		/// The timer service that owns this implementation.
115		timer_service* svc_ = nullptr;
116
117		/// Coroutines currently waiting on this timer's expiry.
118		intrusive_list<waiter_node> waiters_;
119
120		/// Free list linkage, reused when this impl is recycled.
121		implementation* next_free_ = nullptr;
122
123		/// Construct bound to the given timer service.
124	458x	explicit implementation(timer_service& svc) noexcept : svc_(&svc) {}
125
126		/** Check whether the timer is expired and absent from the heap.
127
128		The single definition of the already-expired fast-path
129		predicate: `await_suspend` tests it inline and `wait()`
130		re-tests it because the expiry can elapse between the two
131		reads.
132		*/
133	27212x	bool already_expired() const noexcept
134		{
135	81636x	return heap_index_.load(std::memory_order_relaxed) == npos &&
136	27212x	(expiry_ ==
137	53428x	(std::chrono::steady_clock::time_point::min)() \|\|
138	53428x	expiry_ <= std::chrono::steady_clock::now());
139		}
140
141		/** Asynchronously wait for the timer to expire.
142
143		Publishes the waiter into the service's heap and waiter
144		list, after which it may complete on any thread. If the
145		timer is already expired and not in the heap, completes
146		by posting the continuation without publishing.
147
148		@par Preconditions
149		@p w is fully initialized, and its storage (the awaitable
150		on the suspended coroutine's frame) outlives the wait.
151
152		@param w The waiter to publish.
153		*/
154		// Exported at member level: dllexport on the enclosing timer
155		// class does not extend to nested classes, and header-inline
156		// callers (wait_awaitable::await_suspend) reference this
157		// symbol from outside the corosio DLL.
158		BOOST_COROSIO_DECL
159		std::coroutine_handle<> wait(waiter_node& w);
160		};
161
162		/// The clock type used for time operations.
163		using clock_type = std::chrono::steady_clock;
164
165		/// The time point type for absolute expiry times.
166		using time_point = clock_type::time_point;
167
168		/// The duration type for relative expiry times.
169		using duration = clock_type::duration;
170
171		/** Destructor.
172
173		Cancels any pending operations and releases timer resources.
174		*/
175		~timer() override;
176
177		/** Construct a timer from an execution context.
178
179		@param ctx The execution context that will own this timer. It
180		must be a corosio io_context; otherwise the constructor
181		throws (a timer service is required).
182
183		@throws std::logic_error if @p ctx is not an io_context.
184		*/
185		explicit timer(capy::execution_context& ctx);
186
187		/** Construct a timer with an initial absolute expiry time.
188
189		@param ctx The execution context that will own this timer. It
190		must be a corosio io_context; otherwise the constructor
191		throws (a timer service is required).
192		@param t The initial expiry time point.
193
194		@throws std::logic_error if @p ctx is not an io_context.
195		*/
196		timer(capy::execution_context& ctx, time_point t);
197
198		/** Construct a timer with an initial relative expiry time.
199
200		@param ctx The execution context that will own this timer. It
201		must be a corosio io_context; otherwise the constructor
202		throws (a timer service is required).
203		@param d The initial expiry duration relative to now.
204
205		@throws std::logic_error if @p ctx is not an io_context.
206		*/
207		template<class Rep, class Period>
208		timer(capy::execution_context& ctx, std::chrono::duration<Rep, Period> d)
209		: timer(ctx)
210		{
211		expires_after(d);
212		}
213
214		/** Construct a timer from an executor.
215
216		The timer is associated with the executor's context, which must
217		be a corosio io_context.
218
219		@param ex The executor whose context will own this timer.
220
221		@throws std::logic_error if the executor's context is not an
222		io_context.
223		*/
224		template<class Ex>
225		requires(!std::same_as<std::remove_cvref_t<Ex>, timer>) &&
226		capy::Executor<Ex>
227		explicit timer(Ex const& ex) : timer(ex.context())
228		{
229		}
230
231		/** Construct a timer from an executor with an absolute expiry time.
232
233		@param ex The executor whose context will own this timer.
234		@param t The initial expiry time point.
235
236		@throws std::logic_error if the executor's context is not an
237		io_context.
238		*/
239		template<class Ex>
240		requires capy::Executor<Ex>
241		timer(Ex const& ex, time_point t) : timer(ex.context(), t)
242		{
243		}
244
245		/** Construct a timer from an executor with a relative expiry time.
246
247		@param ex The executor whose context will own this timer.
248		@param d The initial expiry duration relative to now.
249
250		@throws std::logic_error if the executor's context is not an
251		io_context.
252		*/
253		template<class Ex, class Rep, class Period>
254		requires capy::Executor<Ex>
255		timer(Ex const& ex, std::chrono::duration<Rep, Period> d)
256		: timer(ex.context(), d)
257		{
258		}
259
260		/** Move constructor.
261
262		Transfers ownership of the timer resources.
263
264		@param other The timer to move from.
265
266		@pre No awaitables returned by @p other's methods exist.
267		@pre The execution context associated with @p other must
268		outlive this timer.
269		*/
270		timer(timer&& other) noexcept;
271
272		/** Move assignment operator.
273
274		Closes any existing timer and transfers ownership.
275
276		@param other The timer to move from.
277
278		@pre No awaitables returned by either `*this` or @p other's
279		methods exist.
280		@pre The execution context associated with @p other must
281		outlive this timer.
282
283		@return Reference to this timer.
284		*/
285		timer& operator=(timer&& other) noexcept;
286
287		timer(timer const&) = delete;
288		timer& operator=(timer const&) = delete;
289
290		/** Cancel all pending asynchronous wait operations.
291
292		All outstanding operations complete with an error code that
293		compares equal to `capy::cond::canceled`.
294
295		@return The number of operations that were cancelled.
296		*/
297		std::size_t cancel()
298		{
299		if (!get().might_have_pending_waits_.load(std::memory_order_relaxed))
300		return 0;
301		return do_cancel();
302		}
303
304		/** Cancel one pending asynchronous wait operation.
305
306		The oldest pending wait is cancelled (FIFO order). It
307		completes with an error code that compares equal to
308		`capy::cond::canceled`.
309
310		@return The number of operations that were cancelled (0 or 1).
311		*/
312		std::size_t cancel_one()
313		{
314		if (!get().might_have_pending_waits_.load(std::memory_order_relaxed))
315		return 0;
316		return do_cancel_one();
317		}
318
319		/** Return the timer's expiry time as an absolute time.
320
321		@return The expiry time point. If no expiry has been set,
322		returns a default-constructed time_point.
323		*/
324		time_point expiry() const noexcept
325		{
326		return get().expiry_;
327		}
328
329		/** Set the timer's expiry time as an absolute time.
330
331		Any pending asynchronous wait operations will be cancelled.
332
333		@param t The expiry time to be used for the timer.
334
335		@return The number of pending operations that were cancelled.
336		*/
337	9x	std::size_t expires_at(time_point t)
338		{
339	9x	auto& impl = get();
340	9x	impl.expiry_ = t;
341	9x	if (impl.heap_index_.load(std::memory_order_relaxed) ==
342	18x	implementation::npos &&
343	9x	!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
344	9x	return 0;
345	✗	return do_update_expiry();
346		}
347
348		/** Set the timer's expiry time relative to now.
349
350		Any pending asynchronous wait operations will be cancelled.
351
352		@param d The expiry time relative to now.
353
354		@return The number of pending operations that were cancelled.
355		*/
356	14226x	std::size_t expires_after(duration d)
357		{
358	14226x	auto& impl = get();
359	14226x	if (d <= duration::zero())
360	996x	impl.expiry_ = (time_point::min)();
361		else
362		{
363		// Saturate rather than overflow: a clamped near-max duration
364		// (e.g. delay(hours::max())) would wrap now() + d past the
365		// clock's range and appear already elapsed.
366	13230x	auto const now = clock_type::now();
367	13230x	impl.expiry_ = ((time_point::max)() - now < d)
368	26454x	? (time_point::max)()
369	13224x	: now + d;
370		}
371	14226x	if (impl.heap_index_.load(std::memory_order_relaxed) ==
372	28452x	implementation::npos &&
373	14226x	!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
374	14226x	return 0;
375	✗	return do_update_expiry();
376		}
377
378		/** Set the timer's expiry time relative to now.
379
380		This is a convenience overload that accepts any duration type
381		and converts it to the timer's native duration type. Any
382		pending asynchronous wait operations will be cancelled.
383
384		@param d The expiry time relative to now.
385
386		@return The number of pending operations that were cancelled.
387		*/
388		template<class Rep, class Period>
389		std::size_t expires_after(std::chrono::duration<Rep, Period> d)
390		{
391		return expires_after(std::chrono::duration_cast<duration>(d));
392		}
393
394		/** Wait for the timer to expire.
395
396		Multiple coroutines may wait on the same timer concurrently.
397		When the timer expires, all waiters complete with success.
398
399		The operation supports cancellation via `std::stop_token` through
400		the affine awaitable protocol. If the associated stop token is
401		triggered, only that waiter completes with an error that
402		compares equal to `capy::cond::canceled`; other waiters are
403		unaffected.
404
405		This timer must outlive the returned awaitable.
406
407		@return An awaitable that completes with `io_result<>`.
408		*/
409		// Defined below wait_awaitable, which needs timer complete.
410		wait_awaitable wait();
411
412		protected:
413		explicit timer(handle h) noexcept : io_object(std::move(h)) {}
414
415		private:
416		// Defined in src/corosio/src/timer.cpp, which includes both this
417		// header and timer_service.hpp, so the timer_service_* free
418		// functions are visible there.
419		std::size_t do_cancel();
420		std::size_t do_cancel_one();
421		std::size_t do_update_expiry();
422
423		/// Return the underlying implementation.
424	28470x	implementation& get() const noexcept
425		{
426	28470x	return static_cast<implementation>(h_.get());
427		}
428		};
429
430		/** Frame-resident per-wait state for a timer wait.
431
432		One node exists per `co_await` on a timer, embedded in the
433		awaitable on the suspended coroutine's frame — never allocated.
434		Once published by `implementation::wait()` the node may be
435		completed from any thread; every completion path finishes
436		touching the node before resuming or destroying the coroutine,
437		because either act may end the node's storage.
438
439		The node owns no resources: the stop token is borrowed from the
440		awaiting chain's `io_env` (which outlives the suspension) and
441		the stop callback is managed manually in `cb_buf_`, destroyed on
442		every completion path before the frame can die.
443		*/
444		struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node
445		: intrusive_list<waiter_node>::node
446		{
447		// Embedded completion op — avoids heap allocation per fire/cancel.
448		// Members are exported and defined non-inline in timer.cpp: the
449		// inline waiter_node constructor references do_complete and the
450		// vtable from translation units that reach this header through
451		// delay.hpp without ever including timer_service.hpp, so the one
452		// strong definition must live in a TU that is always linked.
453		struct BOOST_COROSIO_SYMBOL_VISIBLE completion_op final : scheduler_op
454		{
455		waiter_node* waiter_ = nullptr;
456
457		BOOST_COROSIO_DECL
458		static void do_complete(
459		void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);
460
461	28470x	completion_op() noexcept : scheduler_op(&do_complete) {}
462
463		BOOST_COROSIO_DECL void operator()() override;
464		BOOST_COROSIO_DECL void destroy() override;
465		};
466
467		// Per-waiter stop_token cancellation
468		struct canceller
469		{
470		waiter_node* waiter_;
471		BOOST_COROSIO_DECL void operator()() const;
472		};
473
474		using stop_cb_type = std::stop_callback<canceller>;
475
476		// nullptr once removed from timer's waiter list (concurrency marker)
477		/// The timer this waiter is published on, or `nullptr`.
478		timer::implementation* impl_ = nullptr;
479
480		/// The timer service that completes this waiter.
481		timer_service* svc_ = nullptr;
482
483		/// The suspended coroutine, destroyed by the shutdown drains.
484		std::coroutine_handle<> h_;
485
486		/// The continuation posted to resume the coroutine.
487		capy::continuation cont_;
488
489		/// The executor the continuation is posted through.
490		capy::executor_ref d_;
491
492		// Borrowed from the awaiting chain's io_env, which outlives the
493		// suspension; the node holds no owning state.
494		/// The stop token observed for cancellation.
495		std::stop_token const* token_ = nullptr;
496
497		/// The completion result read by `await_resume`.
498		std::error_code ec_;
499
500		/// The embedded completion op posted to the scheduler.
501		completion_op op_;
502
503		// stop_callback is neither movable nor assignable; construct it
504		// in place once the node is pinned on the coroutine frame, and
505		// destroy it manually on every completion path.
506		/// Storage for the armed stop callback.
507		alignas(stop_cb_type) unsigned char cb_buf_[sizeof(stop_cb_type)];
508
509		/// True while `cb_buf_` holds a live stop callback.
510		bool cb_active_ = false;
511
512	28470x	waiter_node() noexcept
513	28470x	{
514	28470x	op_.waiter_ = this;
515	28470x	}
516
517		// The embedded op self-points and the list hooks are published
518		// to other threads; the node never moves.
519		waiter_node(waiter_node const&) = delete;
520		waiter_node& operator=(waiter_node const&) = delete;
521
522		/** Arm the stop callback.
523
524		@par Preconditions
525		`token_` is set.
526		*/
527	2788x	void arm_stop_cb()
528		{
529	2788x	new (cb_buf_) stop_cb_type(*token_, canceller{this});
530	2788x	cb_active_ = true;
531	2788x	}
532
533		/// Destroy the stop callback if armed.
534	12977x	void reset_stop_cb() noexcept
535		{
536	12977x	if (cb_active_)
537		{
538	2788x	std::launder(reinterpret_cast<stop_cb_type*>(cb_buf_))
539	2788x	->~stop_cb_type();
540	2788x	cb_active_ = false;
541		}
542	12977x	}
543		};
544
545		/** Awaitable returned by `timer::wait()`.
546
547		Carries the waiter node so a wait performs no allocation. The
548		awaitable is movable only before `await_suspend` publishes the
549		node (a move builds a fresh, quiescent node); afterwards it is
550		pinned on the coroutine frame until the wait completes.
551		*/
552		struct wait_awaitable
553		{
554		timer& t_;
555		waiter_node w_;
556
557	14235x	explicit wait_awaitable(timer& t) noexcept : t_(t) {}
558
559	14235x	wait_awaitable(wait_awaitable&& o) noexcept : t_(o.t_) {}
560
561		wait_awaitable(wait_awaitable const&) = delete;
562		wait_awaitable& operator=(wait_awaitable const&) = delete;
563		wait_awaitable& operator=(wait_awaitable&&) = delete;
564
565	3060x	bool await_ready() const noexcept
566		{
567	3060x	return false;
568		}
569
570		// Cancellation surfaces through w_.ec_: the stop_token path in
571		// wait() completes the waiter with error::canceled written to
572		// it, so there is no separate token to consult here.
573	14193x	capy::io_result<> await_resume() const noexcept
574		{
575	14193x	return {w_.ec_};
576		}
577
578	14235x	auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
579		-> std::coroutine_handle<>
580		{
581	14235x	auto& impl = t_.get();
582	14235x	w_.h_ = h;
583	14235x	w_.cont_.h = h;
584	14235x	w_.d_ = env->executor;
585
586		// Inline fast path: already expired and not in the heap
587	14235x	if (impl.already_expired())
588		{
589	1258x	w_.ec_ = {};
590	1258x	w_.d_.post(w_.cont_);
591	1258x	return std::noop_coroutine();
592		}
593
594	12977x	w_.token_ = &env->stop_token;
595	12977x	return impl.wait(w_);
596		}
597		};
598
599		inline wait_awaitable
600	14235x	timer::wait()
601		{
602	14235x	return wait_awaitable(*this);
603		}
604
605		} // namespace boost::corosio::detail
606
607		#endif
608