TLA Line data 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 HIT 317 : 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 20820 : bool already_expired() const noexcept
139 : {
140 62460 : return heap_index_.load(std::memory_order_relaxed) == npos &&
141 20820 : (expiry_ ==
142 40976 : (std::chrono::steady_clock::time_point::min)() ||
143 40976 : 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 16 : void expires_at(time_point t)
328 : {
329 16 : auto& impl = get();
330 32 : BOOST_COROSIO_ASSERT(
331 : impl.heap_index_.load(std::memory_order_relaxed) ==
332 : implementation::npos);
333 16 : impl.expiry_ = t;
334 16 : }
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 10849 : void expires_after(duration d)
344 : {
345 10849 : auto& impl = get();
346 21698 : BOOST_COROSIO_ASSERT(
347 : impl.heap_index_.load(std::memory_order_relaxed) ==
348 : implementation::npos);
349 10849 : if (d <= duration::zero())
350 680 : 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 10169 : auto const now = clock_type::now();
357 10169 : impl.expiry_ = ((time_point::max)() - now < d)
358 20334 : ? (time_point::max)()
359 10165 : : now + d;
360 : }
361 10849 : }
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 21730 : implementation& get() const noexcept
440 : {
441 21730 : 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 23718 : 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 23718 : waiter_node() noexcept
540 23718 : {
541 23718 : op_.waiter_ = this;
542 23718 : }
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 10853 : void bind(std::coroutine_handle<> h, capy::io_env const& env) noexcept
560 : {
561 10853 : h_ = h;
562 10853 : cont_.h = h;
563 10853 : d_ = env.executor;
564 10853 : token_ = &env.stop_token;
565 10853 : }
566 :
567 : /** Arm the stop callback.
568 :
569 : @par Preconditions
570 : `token_` is set.
571 : */
572 1431 : void arm_stop_cb()
573 : {
574 1431 : new (cb_buf_) stop_cb_type(*token_, canceller{this});
575 1431 : cb_active_ = true;
576 1431 : }
577 :
578 : /// Destroy the stop callback if armed.
579 9986 : void reset_stop_cb() noexcept
580 : {
581 9986 : if (cb_active_)
582 : {
583 1431 : std::launder(reinterpret_cast<stop_cb_type*>(cb_buf_))
584 1431 : ->~stop_cb_type();
585 1431 : cb_active_ = false;
586 : }
587 9986 : }
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 10843 : explicit wait_awaitable(timer& t) noexcept : t_(t) {}
603 :
604 10843 : 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 2053 : bool await_ready() const noexcept
611 : {
612 2053 : 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 10815 : capy::io_result<> await_resume() const noexcept
619 : {
620 10815 : return {w_.ec_};
621 : }
622 :
623 10843 : auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
624 : -> std::coroutine_handle<>
625 : {
626 10843 : auto& impl = t_.get();
627 10843 : 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 10843 : if (impl.already_expired())
633 : {
634 866 : w_.ec_ = {};
635 866 : w_.d_.post(w_.cont_);
636 866 : return std::noop_coroutine();
637 : }
638 :
639 9977 : return impl.wait(w_);
640 : }
641 : };
642 :
643 : inline wait_awaitable
644 10843 : timer::wait()
645 : {
646 10843 : return wait_awaitable(*this);
647 : }
648 :
649 : } // namespace boost::corosio::detail
650 :
651 : #endif
|
