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