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