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