include/boost/corosio/detail/timer_service.hpp

95.6% Lines (241 / 252) 100.0% Functions (28 / 28)
timer_service.hpp
f(x) Functions (28)
Function Calls Lines Blocks
boost::corosio::detail::timer_service::callback::callback() :125 925x 100.0% 100.0% boost::corosio::detail::timer_service::callback::callback(void*, void (*)(void*)) :128 920x 100.0% 100.0% boost::corosio::detail::timer_service::callback::operator()() const :137 7931x 100.0% 100.0% boost::corosio::detail::timer_service::timer_service(boost::capy::execution_context&) :181 925x 100.0% 100.0% boost::corosio::detail::timer_service::get_scheduler() :190 33737x 100.0% 100.0% boost::corosio::detail::timer_service::~timer_service() :196 1842x 100.0% 100.0% boost::corosio::detail::timer_service::attach() :211 19416x 88.9% 84.0% boost::corosio::detail::timer_service::set_on_earliest_changed(boost::corosio::detail::timer_service::callback) :226 920x 100.0% 100.0% boost::corosio::detail::timer_service::nearest_expiry() const :239 18068x 100.0% 73.0% boost::corosio::detail::timer_service::refresh_cached_nearest() :270 62271x 100.0% 70.0% boost::corosio::detail::tl_cache_owner::~tl_cache_owner() :299 48x 100.0% 100.0% boost::corosio::detail::arm_tl_cache_cleanup() :307 15104x 100.0% 100.0% boost::corosio::detail::try_pop_tl_cache(boost::corosio::detail::timer_service*) :313 19416x 87.5% 78.0% boost::corosio::detail::try_push_tl_cache(boost::corosio::detail::timer::implementation*) :328 16165x 100.0% 100.0% boost::corosio::detail::timer_service_invalidate_cache() :340 920x 100.0% 100.0% boost::corosio::detail::timer_service::shutdown() :349 920x 100.0% 70.0% boost::corosio::detail::timer_service::shutdown()::{lambda(boost::corosio::detail::timer::implementation*)#1}::operator()(boost::corosio::detail::timer::implementation*) const :355 3251x 93.8% 100.0% boost::corosio::detail::timer_service::construct() :432 19416x 100.0% 71.0% boost::corosio::detail::timer_service::destroy(boost::corosio::io_object::implementation*) :451 19416x 100.0% 100.0% boost::corosio::detail::timer_service::destroy_impl(boost::corosio::detail::timer::implementation&) :465 16165x 66.7% 61.0% boost::corosio::detail::timer_service::insert_waiter(boost::corosio::detail::timer::implementation&, boost::corosio::detail::waiter_node*) :502 22622x 100.0% 73.0% boost::corosio::detail::timer_service::cancel_timer(boost::corosio::detail::timer::implementation&) :555 16165x 100.0% 89.0% boost::corosio::detail::timer_service::cancel_waiter(boost::corosio::detail::waiter_node*) :592 1753x 100.0% 86.0% boost::corosio::detail::timer_service::process_expired() :617 38062x 100.0% 91.0% boost::corosio::detail::timer_service::remove_timer_impl(boost::corosio::detail::timer::implementation&) :653 19383x 84.6% 69.0% boost::corosio::detail::timer_service::up_heap(unsigned long) :684 22622x 100.0% 100.0% boost::corosio::detail::timer_service::down_heap(unsigned long) :697 16626x 100.0% 100.0% boost::corosio::detail::timer_service::swap_heap(unsigned long, unsigned long) :717 42812x 100.0% 66.0%
Line TLA Hits 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_SERVICE_HPP
12 #define BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP
13
14 #include <boost/corosio/detail/timer.hpp>
15 #include <boost/corosio/detail/scheduler.hpp>
16 #include <boost/corosio/detail/scheduler_op.hpp>
17 #include <boost/corosio/detail/object_pool.hpp>
18 #include <boost/corosio/detail/object_ref.hpp>
19 #include <boost/corosio/detail/intrusive.hpp>
20 #include <boost/corosio/detail/thread_local_ptr.hpp>
21 #include <boost/capy/error.hpp>
22 #include <boost/capy/ex/execution_context.hpp>
23 #include <boost/capy/ex/abandon.hpp>
24 #include <boost/capy/ex/executor_ref.hpp>
25 #include <system_error>
26
27 #include <atomic>
28 #include <chrono>
29 #include <coroutine>
30 #include <cstddef>
31 #include <limits>
32 #include <mutex>
33 #include <thread>
34 #include <stop_token>
35 #include <utility>
36 #include <vector>
37
38 namespace boost::corosio::detail {
39
40 class scheduler;
41
42 /*
43 Timer Service
44 =============
45
46 Data Structures
47 ---------------
48 waiter_node (defined in timer.hpp) holds per-waiter state:
49 coroutine handle, executor, error output, embedded
50 completion_op. Each concurrent co_await t.wait() embeds one
51 waiter_node in the awaitable on the suspended coroutine's
52 frame — waits perform no allocation.
53
54 timer::implementation holds per-timer state: expiry, heap
55 index, and the single published waiter. Each timer holds
56 at most one waiter; process_expired's local cross-timer drain
57 list still threads waiters through their intrusive hooks when
58 collecting several timers' waiters past the lock.
59
60 timer_service owns a min-heap of active timers and recycles
61 retired impls through a per-service object_pool, fronted by a
62 thread-local single-slot cache. The heap is ordered by expiry
63 time; the scheduler queries nearest_expiry() to set the
64 epoll/timerfd timeout.
65
66 Optimization Strategy
67 ---------------------
68 1. Deferred heap insertion — expires_after() stores the expiry
69 but does not insert into the heap. Insertion happens in wait().
70 2. Thread-local impl cache — single-slot per-thread cache, in
71 front of the object_pool fallback; keeps a thread's hot impl
72 out of the shared free list.
73 3. Frame-resident waiter_node with embedded completion_op —
74 eliminates heap allocation per wait/fire/cancel.
75 4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().
76 5. might_have_pending_waits_ flag — skips lock when no wait issued.
77
78 Concurrency
79 -----------
80 stop_token callbacks can fire from any thread. The impl_
81 pointer on waiter_node is used as a "still in list" marker.
82 A waiter_node's storage is the suspended coroutine's frame:
83 every completion path must finish touching the node before
84 dispatching the continuation or destroying the handle.
85 */
86
87 inline void timer_service_invalidate_cache() noexcept;
88
89 /** The timer service's recycling pool.
90
91 Adds no behavior of its own — it exists only to re-expose
92 `adopt()` and `remove()` as public, so the timer service's
93 thread-local cache can transfer a cached impl's `live_` tracking
94 on both sides of the handoff: `construct()` re-adopts a
95 TL-cached impl as live, `shutdown()` detaches one before its
96 direct delete.
97 */
98 class timer_object_pool : public object_pool<timer::implementation>
99 {
100 public:
101 using object_pool::adopt;
102 using object_pool::remove;
103 };
104
105 // timer_service class body — member function definitions are
106 // out-of-class (after implementation and waiter_node are complete)
107 class BOOST_COROSIO_DECL timer_service final
108 : public capy::execution_context::service
109 , public io_object::io_service
110 {
111 friend struct timer::implementation;
112
113 public:
114 using clock_type = std::chrono::steady_clock;
115 using time_point = clock_type::time_point;
116
117 /// Type-erased callback for earliest-expiry-changed notifications.
118 class callback
119 {
120 void* ctx_ = nullptr;
121 void (*fn_)(void*) = nullptr;
122
123 public:
124 /// Construct an empty callback.
125 925x callback() = default;
126
127 /// Construct a callback with the given context and function.
128 920x callback(void* ctx, void (*fn)(void*)) noexcept : ctx_(ctx), fn_(fn) {}
129
130 /// Return true if the callback is non-empty.
131 explicit operator bool() const noexcept
132 {
133 return fn_ != nullptr;
134 }
135
136 /// Invoke the callback.
137 7931x void operator()() const
138 {
139 7931x if (fn_)
140 7931x fn_(ctx_);
141 7931x }
142 };
143
144 private:
145 struct heap_entry
146 {
147 time_point time_;
148 timer::implementation* timer_;
149 };
150
151 scheduler* sched_ = nullptr;
152 BOOST_COROSIO_MSVC_WARNING_PUSH
153 BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std:: members, dll-interface
154 mutable std::mutex mutex_;
155 std::vector<heap_entry> heap_;
156 timer_object_pool pool_;
157 callback on_earliest_changed_;
158 bool shutting_down_ = false;
159 // Set once the scheduler waits on this service; read without the
160 // mutex by every timer construction after the first.
161 std::atomic<bool> attached_{false};
162
163 // Cancellers that claimed a waiter and have yet to post it;
164 // shutdown waits them out so the scheduler's drain finds the post.
165 std::size_t cancels_posting_ = 0; // guarded by mutex_
166 // Avoids mutex in nearest_expiry() and empty()
167 mutable std::atomic<std::int64_t> cached_nearest_ns_{
168 (std::numeric_limits<std::int64_t>::max)()};
169 BOOST_COROSIO_MSVC_WARNING_POP
170
171 public:
172 /** Construct the timer service for a context's scheduler.
173
174 Created with the context's first timer. It does not attach to
175 the scheduler here: a concurrent first use may construct a
176 second service that the context then discards. See
177 @ref attach.
178
179 @throws std::logic_error if the context has no scheduler.
180 */
181 925x inline explicit timer_service(capy::execution_context& ctx)
182 925x : sched_(ctx.find_service<scheduler>())
183 {
184 925x if (!sched_)
185 4x detail::throw_logic_error(
186 "timer_service: the context has no scheduler");
187 937x }
188
189 /// Return the associated scheduler.
190 33737x inline scheduler& get_scheduler() noexcept
191 {
192 33737x return *sched_;
193 }
194
195 /// Destroy the timer service, detaching it from the scheduler.
196 1842x ~timer_service() override
197 921x {
198 921x if (attached_.load(std::memory_order_acquire))
199 920x sched_->detach_timer_service();
200 1842x }
201
202 /** Attach this service to the scheduler, if not yet attached.
203
204 Call once the context has settled which service it keeps; no
205 timer of this service can wait before this returns.
206
207 @par Exception Safety
208 Strong guarantee: on failure the service stays detached and
209 the next call retries.
210 */
211 19416x inline void attach()
212 {
213 19416x if (attached_.load(std::memory_order_acquire))
214 18496x return;
215 920x std::lock_guard lock(mutex_);
216 920x if (attached_.load(std::memory_order_relaxed))
217 ✗ return;
218 920x sched_->attach_timer_service(*this);
219 920x attached_.store(true, std::memory_order_release);
220 920x }
221
222 timer_service(timer_service const&) = delete;
223 timer_service& operator=(timer_service const&) = delete;
224
225 /// Register a callback invoked when the earliest expiry changes.
226 920x inline void set_on_earliest_changed(callback cb)
227 {
228 920x on_earliest_changed_ = cb;
229 920x }
230
231 /// Return true if no timers are in the heap.
232 inline bool empty() const noexcept
233 {
234 return cached_nearest_ns_.load(std::memory_order_acquire) ==
235 (std::numeric_limits<std::int64_t>::max)();
236 }
237
238 /// Return the nearest timer expiry without acquiring the mutex.
239 18068x inline time_point nearest_expiry() const noexcept
240 {
241 18068x auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
242 18068x return time_point(time_point::duration(ns));
243 }
244
245 /// Cancel all pending timers and free cached resources.
246 inline void shutdown() override;
247
248 /// Construct a new timer implementation.
249 inline io_object::implementation* construct() override;
250
251 /// Destroy a timer implementation, cancelling pending waiters.
252 inline void destroy(io_object::implementation* p) override;
253
254 /// Cancel and recycle a timer implementation.
255 inline void destroy_impl(timer::implementation& impl);
256
257 /// Publish the timer's waiter and insert the timer into the heap.
258 inline void insert_waiter(timer::implementation& impl, waiter_node* w);
259
260 /// Cancel the timer's published waiter, if any.
261 inline void cancel_timer(timer::implementation& impl);
262
263 /// Cancel one specific waiter ( stop_token callback path ).
264 inline void cancel_waiter(waiter_node* w);
265
266 /// Complete all waiters whose timers have expired.
267 inline std::size_t process_expired();
268
269 private:
270 62271x inline void refresh_cached_nearest() noexcept
271 {
272 62271x auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
273 59125x : heap_[0].time_.time_since_epoch().count();
274 62271x cached_nearest_ns_.store(ns, std::memory_order_release);
275 62271x }
276
277 inline void remove_timer_impl(timer::implementation& impl);
278 inline void up_heap(std::size_t index);
279 inline void down_heap(std::size_t index);
280 inline void swap_heap(std::size_t i1, std::size_t i2);
281 };
282
283 // Thread-local single-slot impl cache, validated by comparing svc_.
284 // Keeps a hot impl out of the shared free list; both sides still
285 // take the pool lock to move it in and out of live tracking.
286 // Cleared by timer_service_invalidate_cache() during shutdown.
287
288 inline thread_local_ptr<timer::implementation> tl_cached_impl;
289
290 // The POD TLS slot above never runs destructors, so a short-lived
291 // run() thread would leak its cached impl. Each push arms this
292 // owner, whose destructor frees the slot at thread exit. A cached
293 // entry is a quiescent heap object (nothing in the heap or free
294 // list) and deletion touches no service state, so it is safe after
295 // the owning service is gone (the stale-entry path in
296 // try_pop_tl_cache deletes the same way).
297 struct tl_cache_owner
298 {
299 48x ~tl_cache_owner()
300 {
301 48x delete tl_cached_impl.get();
302 48x tl_cached_impl.set(nullptr);
303 48x }
304 };
305
306 inline void
307 15104x arm_tl_cache_cleanup() noexcept
308 {
309 15104x [[maybe_unused]] thread_local tl_cache_owner owner;
310 15104x }
311
312 inline timer::implementation*
313 19416x try_pop_tl_cache(timer_service* svc) noexcept
314 {
315 19416x auto* impl = tl_cached_impl.get();
316 19416x if (impl)
317 {
318 14605x tl_cached_impl.set(nullptr);
319 14605x if (impl->svc_ == svc)
320 14605x return impl;
321 // Stale impl from a destroyed service
322 ✗ delete impl;
323 }
324 4811x return nullptr;
325 }
326
327 inline bool
328 16165x try_push_tl_cache(timer::implementation* impl) noexcept
329 {
330 16165x if (!tl_cached_impl.get())
331 {
332 15104x arm_tl_cache_cleanup();
333 15104x tl_cached_impl.set(impl);
334 15104x return true;
335 }
336 1061x return false;
337 }
338
339 inline void
340 920x timer_service_invalidate_cache() noexcept
341 {
342 920x delete tl_cached_impl.get();
343 920x tl_cached_impl.set(nullptr);
344 920x }
345
346 // timer_service out-of-class member function definitions
347
348 inline void
349 920x timer_service::shutdown()
350 {
351 920x timer_service_invalidate_cache();
352 920x shutting_down_ = true;
353 // Flag only; the real drain walks the expiry heap below, not
354 // object_pool's live_ list, so the visit callback is a no-op.
355 920x pool_.shutdown([](timer::implementation*) {});
356
357 // Snapshot impls and detach them from the heap so that
358 // coroutine-owned timer destructors (run by the unwinds the
359 // abandons below start) cannot re-enter remove_timer_impl() and
360 // mutate the vector during iteration. A stop requested on another thread
361 // reaches cancel_waiter() under the same lock: whichever side
362 // claims a waiter first owns it, and a waiter its canceller
363 // claimed is posted, so the scheduler's drain abandons it.
364 920x std::vector<timer::implementation*> impls;
365 920x std::vector<waiter_node*> waiters;
366 {
367 920x std::lock_guard lock(mutex_);
368 920x impls.reserve(heap_.size());
369 920x waiters.reserve(heap_.size());
370 4159x for (auto& entry : heap_)
371 {
372 3239x entry.timer_->heap_index_.store(
373 (std::numeric_limits<std::size_t>::max)(),
374 std::memory_order_relaxed);
375 3239x impls.push_back(entry.timer_);
376 3239x if (auto* w = std::exchange(entry.timer_->waiter_, nullptr))
377 {
378 3239x w->impl_ = nullptr;
379 3239x waiters.push_back(w);
380 }
381 }
382 920x heap_.clear();
383 920x cached_nearest_ns_.store(
384 (std::numeric_limits<std::int64_t>::max)(),
385 std::memory_order_release);
386 920x }
387
388 // A stop on another thread may have claimed its waiter before the
389 // pass above and still be posting it; the scheduler drains next,
390 // and must find that post.
391 for (;;)
392 {
393 {
394 920x std::lock_guard lock(mutex_);
395 920x if (cancels_posting_ == 0)
396 920x break;
397 920x }
398 ✗ std::this_thread::yield();
399 ✗ }
400
401 // Cancel waiting timers. Each waiter called work_started()
402 // in implementation::wait(); a waiter given up here never
403 // completes, so its work is finished here to keep the count
404 // balanced. Resetting the stop callback outside the lock
405 // waits out a canceller already running, which finds the waiter
406 // claimed.
407 4159x for (auto* w : waiters)
408 {
409 3239x w->reset_stop_cb();
410 3239x auto c = std::exchange(w->cont_, {});
411 3239x sched_->work_finished();
412 // Unwinding the frame also ends the node's storage
413 3239x if (c.h)
414 3239x capy::abandon(c);
415 }
416
417 4159x for (auto* impl : impls)
418 {
419 // Unlink from the pool's live_ list before the direct delete
420 // below, or ~object_pool()'s unconditional sweep would delete
421 // this impl a second time.
422 3239x [[maybe_unused]] bool const removed = pool_.remove(impl);
423 3239x BOOST_COROSIO_ASSERT(removed);
424 3239x delete impl;
425 }
426
427 // Anything still parked in pool_ (free-listed or live with no
428 // pending wait at shutdown) is freed by object_pool's own destructor.
429 920x }
430
431 inline io_object::implementation*
432 19416x timer_service::construct()
433 {
434 19416x timer::implementation* impl = try_pop_tl_cache(this);
435 19416x if (impl)
436 {
437 14605x impl->svc_ = this;
438 14605x impl->reuse();
439 14605x impl->refs_.store(1, std::memory_order_relaxed);
440 // The TL push that parked impl here also removed it from
441 // live_ (see destroy_impl) — re-adopt it now that it is live
442 // again.
443 14605x pool_.adopt(impl);
444 14605x return impl;
445 }
446
447 4811x return pool_.acquire(*this);
448 }
449
450 inline void
451 19416x timer_service::destroy(io_object::implementation* p)
452 {
453 // During shutdown the drain loop owns every impl and deletes
454 // them directly. A frame destroyed by that loop can unwind a
455 // handle whose impl was freed in an earlier iteration (a
456 // timeout's parent frame owns the timeout timer while
457 // suspended on the inner delay's timer), so bail out before
458 // even downcasting the pointer.
459 19416x if (shutting_down_)
460 3251x return;
461 16165x destroy_impl(static_cast<timer::implementation&>(*p));
462 }
463
464 inline void
465 16165x timer_service::destroy_impl(timer::implementation& impl)
466 {
467 // During shutdown the impl is owned by the shutdown loop.
468 // Re-entering here (from a coroutine-owned timer destructor
469 // triggered by h.destroy()) must not modify the heap or
470 // recycle the impl — shutdown deletes it directly.
471 16165x if (shutting_down_)
472 ✗ return;
473
474 16165x cancel_timer(impl);
475
476 32330x if (impl.heap_index_.load(std::memory_order_relaxed) !=
477 16165x (std::numeric_limits<std::size_t>::max)())
478 {
479 ✗ std::lock_guard lock(mutex_);
480 ✗ remove_timer_impl(impl);
481 ✗ refresh_cached_nearest();
482 ✗ }
483
484 16165x if (try_push_tl_cache(&impl))
485 {
486 // The TL slot takes sole ownership: it is deleted directly by
487 // tl_cache_owner / the stale-service path, never through this
488 // pool, so it must leave live_ now or ~object_pool() would see
489 // it as abandoned (and, at the wrong moment, double-delete it
490 // after a later construct() re-adopts it).
491 15104x [[maybe_unused]] bool const removed = pool_.remove(&impl);
492 15104x BOOST_COROSIO_ASSERT(removed);
493 15104x return;
494 }
495
496 // No op keepalives on a timer impl: this is the only reference,
497 // so release() always recycles immediately via retire().
498 1061x release(&impl);
499 }
500
501 inline void
502 22622x timer_service::insert_waiter(timer::implementation& impl, waiter_node* w)
503 {
504 22622x bool notify = false;
505 22622x bool lost_cancel = false;
506 {
507 22622x std::lock_guard lock(mutex_);
508 // Grow before publishing anything, so the push_back below
509 // cannot throw: a failure here leaves the waiter untouched,
510 // the strong guarantee rearm_wait's recovery relies on.
511 22622x if (impl.heap_index_.load(std::memory_order_relaxed) ==
512 45244x (std::numeric_limits<std::size_t>::max)() &&
513 22622x heap_.size() == heap_.capacity())
514 921x heap_.reserve(heap_.capacity() == 0 ? 16 : 2 * heap_.capacity());
515 // Publish: from here the waiter is visible to the fire path and
516 // to its own stop callback (impl_ non-null enables cancel_waiter).
517 22622x w->impl_ = &impl;
518 45244x if (impl.heap_index_.load(std::memory_order_relaxed) ==
519 22622x (std::numeric_limits<std::size_t>::max)())
520 {
521 22622x impl.heap_index_.store(heap_.size(), std::memory_order_relaxed);
522 22622x heap_.push_back({impl.expiry_, &impl});
523 22622x up_heap(heap_.size() - 1);
524 22622x notify = (impl.heap_index_.load(std::memory_order_relaxed) == 0);
525 22622x refresh_cached_nearest();
526 }
527 22622x BOOST_COROSIO_ASSERT(impl.waiter_ == nullptr);
528 22622x impl.waiter_ = w;
529
530 // Lost-cancel re-check: a stop requested after the canceller was
531 // armed in wait() but before this publication found impl_ null
532 // and returned a no-op. Observe it now and undo the insertion.
533 22622x if (w->token_->stop_requested())
534 {
535 1x w->impl_ = nullptr;
536 1x impl.waiter_ = nullptr;
537 1x remove_timer_impl(impl);
538 1x impl.might_have_pending_waits_.store(
539 false, std::memory_order_release);
540 1x refresh_cached_nearest();
541 1x lost_cancel = true;
542 1x notify = false; // insertion undone; nearest unchanged
543 }
544 22622x }
545 22622x if (notify)
546 7931x on_earliest_changed_();
547 22622x if (lost_cancel)
548 {
549 1x w->ec_ = make_error_code(capy::error::canceled);
550 1x sched_->post(&w->op_);
551 }
552 22622x }
553
554 inline void
555 16165x timer_service::cancel_timer(timer::implementation& impl)
556 {
557 // Acquire pairs with the drains' release: a caller that skips the
558 // lock may recycle the impl next, so the drain's writes to it must
559 // happen before.
560 16165x if (!impl.might_have_pending_waits_.load(std::memory_order_acquire))
561 16163x return;
562
563 // No unlocked already-done fast-out here: it would need the
564 // non-atomic waiter_ (a race with concurrent drains), and an
565 // index-only check is lifetime-unsafe because npos is stored
566 // before the drain finishes touching the impl. A stale-true
567 // flag is rare with the stateless API; the locked path below
568 // re-validates.
569
570 2x waiter_node* canceled = nullptr;
571
572 {
573 2x std::lock_guard lock(mutex_);
574 2x remove_timer_impl(impl);
575 2x canceled = std::exchange(impl.waiter_, nullptr);
576 2x if (canceled)
577 2x canceled->impl_ = nullptr;
578 // Store false as the final touch of the impl under the lock so
579 // a pre-lock false-flag check trusts it unqualified.
580 2x impl.might_have_pending_waits_.store(false, std::memory_order_release);
581 2x refresh_cached_nearest();
582 2x }
583
584 2x if (canceled)
585 {
586 2x canceled->ec_ = make_error_code(capy::error::canceled);
587 2x sched_->post(&canceled->op_);
588 }
589 }
590
591 inline void
592 1753x timer_service::cancel_waiter(waiter_node* w)
593 {
594 {
595 1753x std::lock_guard lock(mutex_);
596 // Already removed by another drain: cancel_timer,
597 // process_expired, or insert_waiter's lost-cancel recheck
598 1753x if (!w->impl_)
599 169x return;
600 1584x auto* impl = w->impl_;
601 1584x w->impl_ = nullptr;
602 1584x impl->waiter_ = nullptr;
603 1584x remove_timer_impl(*impl);
604 1584x impl->might_have_pending_waits_.store(false, std::memory_order_release);
605 1584x refresh_cached_nearest();
606 1584x ++cancels_posting_;
607 1753x }
608
609 1584x w->ec_ = make_error_code(capy::error::canceled);
610 1584x sched_->post(&w->op_);
611
612 1584x std::lock_guard lock(mutex_);
613 1584x --cancels_posting_;
614 1584x }
615
616 inline std::size_t
617 38062x timer_service::process_expired()
618 {
619 38062x intrusive_list<waiter_node> expired;
620
621 {
622 38062x std::lock_guard lock(mutex_);
623 38062x auto now = clock_type::now();
624
625 55858x while (!heap_.empty() && heap_[0].time_ <= now)
626 {
627 17796x timer::implementation* t = heap_[0].timer_;
628 17796x remove_timer_impl(*t);
629 17796x if (auto* w = std::exchange(t->waiter_, nullptr))
630 {
631 17796x w->impl_ = nullptr;
632 17796x w->ec_ = {};
633 17796x expired.push_back(w);
634 }
635 17796x t->might_have_pending_waits_.store(
636 false, std::memory_order_release);
637 }
638
639 38062x refresh_cached_nearest();
640 38062x }
641
642 38062x std::size_t count = 0;
643 55858x while (auto* w = expired.pop_front())
644 {
645 17796x sched_->post(&w->op_);
646 17796x ++count;
647 17796x }
648
649 38062x return count;
650 }
651
652 inline void
653 19383x timer_service::remove_timer_impl(timer::implementation& impl)
654 {
655 19383x std::size_t index = impl.heap_index_.load(std::memory_order_relaxed);
656 19383x if (index >= heap_.size())
657 ✗ return; // Not in heap
658
659 19383x if (index == heap_.size() - 1)
660 {
661 // Last element, just pop
662 2757x impl.heap_index_.store(
663 (std::numeric_limits<std::size_t>::max)(),
664 std::memory_order_relaxed);
665 2757x heap_.pop_back();
666 }
667 else
668 {
669 // Swap with last and reheapify
670 16626x swap_heap(index, heap_.size() - 1);
671 16626x impl.heap_index_.store(
672 (std::numeric_limits<std::size_t>::max)(),
673 std::memory_order_relaxed);
674 16626x heap_.pop_back();
675
676 16626x if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
677 ✗ up_heap(index);
678 else
679 16626x down_heap(index);
680 }
681 }
682
683 inline void
684 22622x timer_service::up_heap(std::size_t index)
685 {
686 28884x while (index > 0)
687 {
688 20952x std::size_t parent = (index - 1) / 2;
689 20952x if (!(heap_[index].time_ < heap_[parent].time_))
690 14690x break;
691 6262x swap_heap(index, parent);
692 6262x index = parent;
693 }
694 22622x }
695
696 inline void
697 16626x timer_service::down_heap(std::size_t index)
698 {
699 16626x std::size_t child = index * 2 + 1;
700 36550x while (child < heap_.size())
701 {
702 22405x std::size_t min_child = (child + 1 == heap_.size() ||
703 18257x heap_[child].time_ < heap_[child + 1].time_)
704 40662x ? child
705 22405x : child + 1;
706
707 22405x if (heap_[index].time_ < heap_[min_child].time_)
708 2481x break;
709
710 19924x swap_heap(index, min_child);
711 19924x index = min_child;
712 19924x child = index * 2 + 1;
713 }
714 16626x }
715
716 inline void
717 42812x timer_service::swap_heap(std::size_t i1, std::size_t i2)
718 {
719 42812x heap_entry tmp = heap_[i1];
720 42812x heap_[i1] = heap_[i2];
721 42812x heap_[i2] = tmp;
722 42812x heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed);
723 42812x heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed);
724 42812x }
725
726 // waiter_node's completion_op and canceller members are defined in
727 // timer.cpp alongside implementation::wait(), for the same reason
728 // wait() lives there (see below).
729
730 // timer::implementation::wait() is defined in timer.cpp, not here.
731 // It must be a non-inline definition in a translation unit that is
732 // always pulled into the link whenever detail::timer is used (every
733 // consumer needs timer's constructors from that same object file).
734 // An inline definition in this header would only be emitted in
735 // translation units that happen to also include this header, which
736 // is not guaranteed for every caller of wait_awaitable::await_suspend
737 // in timer.hpp (e.g. code that only reaches timer.hpp through
738 // delay.hpp, without transitively including a scheduler header).
739
740 } // namespace boost::corosio::detail
741
742 #endif
743