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. 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 HIT 290 : 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 17865 : bool already_expired() const noexcept
134 : {
135 53595 : return heap_index_.load(std::memory_order_relaxed) == npos &&
136 17865 : (expiry_ ==
137 35066 : (std::chrono::steady_clock::time_point::min)() ||
138 35066 : 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 6 : std::size_t expires_at(time_point t)
338 : {
339 6 : auto& impl = get();
340 6 : impl.expiry_ = t;
341 6 : if (impl.heap_index_.load(std::memory_order_relaxed) ==
342 12 : implementation::npos &&
343 6 : !impl.might_have_pending_waits_.load(std::memory_order_relaxed))
344 6 : return 0;
345 MIS 0 : 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 HIT 9354 : std::size_t expires_after(duration d)
357 : {
358 9354 : auto& impl = get();
359 9354 : if (d <= duration::zero())
360 664 : 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 8690 : auto const now = clock_type::now();
367 8690 : impl.expiry_ = ((time_point::max)() - now < d)
368 17376 : ? (time_point::max)()
369 8686 : : now + d;
370 : }
371 9354 : if (impl.heap_index_.load(std::memory_order_relaxed) ==
372 18708 : implementation::npos &&
373 9354 : !impl.might_have_pending_waits_.load(std::memory_order_relaxed))
374 9354 : return 0;
375 MIS 0 : 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 HIT 18720 : implementation& get() const noexcept
425 : {
426 18720 : 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 18720 : 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 18720 : waiter_node() noexcept
513 18720 : {
514 18720 : op_.waiter_ = this;
515 18720 : }
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 1418 : void arm_stop_cb()
528 : {
529 1418 : new (cb_buf_) stop_cb_type(*token_, canceller{this});
530 1418 : cb_active_ = true;
531 1418 : }
532 :
533 : /// Destroy the stop callback if armed.
534 8505 : void reset_stop_cb() noexcept
535 : {
536 8505 : if (cb_active_)
537 : {
538 1418 : std::launder(reinterpret_cast<stop_cb_type*>(cb_buf_))
539 1418 : ->~stop_cb_type();
540 1418 : cb_active_ = false;
541 : }
542 8505 : }
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 9360 : explicit wait_awaitable(timer& t) noexcept : t_(t) {}
558 :
559 9360 : 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 2040 : bool await_ready() const noexcept
566 : {
567 2040 : 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 9332 : capy::io_result<> await_resume() const noexcept
574 : {
575 9332 : return {w_.ec_};
576 : }
577 :
578 9360 : auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
579 : -> std::coroutine_handle<>
580 : {
581 9360 : auto& impl = t_.get();
582 9360 : w_.h_ = h;
583 9360 : w_.cont_.h = h;
584 9360 : w_.d_ = env->executor;
585 :
586 : // Inline fast path: already expired and not in the heap
587 9360 : if (impl.already_expired())
588 : {
589 855 : w_.ec_ = {};
590 855 : w_.d_.post(w_.cont_);
591 855 : return std::noop_coroutine();
592 : }
593 :
594 8505 : w_.token_ = &env->stop_token;
595 8505 : return impl.wait(w_);
596 : }
597 : };
598 :
599 : inline wait_awaitable
600 9360 : timer::wait()
601 : {
602 9360 : return wait_awaitable(*this);
603 : }
604 :
605 : } // namespace boost::corosio::detail
606 :
607 : #endif
|
