include/boost/corosio/detail/timer_service.hpp

72.3% Lines (214/296) 84.4% List of functions (27/32)
timer_service.hpp
f(x) Functions (32)
Function Calls Lines Blocks
boost::corosio::detail::timer_service::callback::callback() :98 1219x 100.0% 100.0% boost::corosio::detail::timer_service::callback::callback(void*, void (*)(void*)) :101 1219x 100.0% 100.0% boost::corosio::detail::timer_service::callback::operator()() const :110 8422x 100.0% 100.0% boost::corosio::detail::timer_service::timer_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :139 1219x 100.0% 100.0% boost::corosio::detail::timer_service::get_scheduler() :145 16984x 100.0% 100.0% boost::corosio::detail::timer_service::~timer_service() :151 2438x 100.0% 100.0% boost::corosio::detail::timer_service::set_on_earliest_changed(boost::corosio::detail::timer_service::callback) :157 1219x 100.0% 100.0% boost::corosio::detail::timer_service::nearest_expiry() const :170 220457x 100.0% 73.0% boost::corosio::detail::timer_service::refresh_cached_nearest() :208 250840x 100.0% 70.0% boost::corosio::detail::tl_cache_owner::~tl_cache_owner() :236 35x 100.0% 100.0% boost::corosio::detail::arm_tl_cache_cleanup() :244 9290x 100.0% 100.0% boost::corosio::detail::try_pop_tl_cache(boost::corosio::detail::timer_service*) :251 9360x 87.5% 78.0% boost::corosio::detail::try_push_tl_cache(boost::corosio::detail::timer::implementation*) :266 9334x 100.0% 100.0% boost::corosio::detail::timer_service_invalidate_cache() :278 1219x 100.0% 100.0% boost::corosio::detail::timer_service::shutdown() :287 1219x 100.0% 75.0% boost::corosio::detail::timer_service::construct() :338 9360x 65.0% 44.0% boost::corosio::detail::timer_service::destroy(boost::corosio::io_object::implementation*) :375 9360x 100.0% 100.0% boost::corosio::detail::timer_service::destroy_impl(boost::corosio::detail::timer::implementation&) :389 9334x 75.0% 62.0% boost::corosio::detail::timer_service::update_timer(boost::corosio::detail::timer::implementation&, std::chrono::time_point<std::chrono::_V2::steady_clock, std::chrono::duration<long, std::ratio<1l, 1000000000l> > >) :417 0 0.0% 0.0% boost::corosio::detail::timer_service::insert_waiter(boost::corosio::detail::timer::implementation&, boost::corosio::detail::waiter_node*) :476 8505x 66.7% 62.0% boost::corosio::detail::timer_service::cancel_timer(boost::corosio::detail::timer::implementation&) :525 9334x 100.0% 88.0% boost::corosio::detail::timer_service::cancel_waiter(boost::corosio::detail::waiter_node*) :565 1386x 92.9% 80.0% boost::corosio::detail::timer_service::cancel_one_waiter(boost::corosio::detail::timer::implementation&) :591 0 0.0% 0.0% boost::corosio::detail::timer_service::process_expired() :619 240947x 100.0% 91.0% boost::corosio::detail::timer_service::remove_timer_impl(boost::corosio::detail::timer::implementation&) :655 8479x 84.6% 69.0% boost::corosio::detail::timer_service::up_heap(unsigned long) :686 8505x 100.0% 100.0% boost::corosio::detail::timer_service::down_heap(unsigned long) :699 6857x 92.3% 71.0% boost::corosio::detail::timer_service::swap_heap(unsigned long, unsigned long) :719 13692x 100.0% 66.0% boost::corosio::detail::timer_service_update_expiry(boost::corosio::detail::timer::implementation&) :745 0 0.0% 0.0% boost::corosio::detail::timer_service_cancel(boost::corosio::detail::timer::implementation&) :751 0 0.0% 0.0% boost::corosio::detail::timer_service_cancel_one(boost::corosio::detail::timer::implementation&) :757 0 0.0% 0.0% boost::corosio::detail::get_timer_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :763 1219x 100.0% 100.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/intrusive.hpp>
18 #include <boost/corosio/detail/thread_local_ptr.hpp>
19 #include <boost/capy/error.hpp>
20 #include <boost/capy/ex/execution_context.hpp>
21 #include <boost/capy/ex/executor_ref.hpp>
22 #include <system_error>
23
24 #include <atomic>
25 #include <chrono>
26 #include <coroutine>
27 #include <cstddef>
28 #include <limits>
29 #include <mutex>
30 #include <stop_token>
31 #include <utility>
32 #include <vector>
33
34 namespace boost::corosio::detail {
35
36 struct scheduler;
37
38 /*
39 Timer Service
40 =============
41
42 Data Structures
43 ---------------
44 waiter_node (defined in timer.hpp) holds per-waiter state:
45 coroutine handle, executor, error output, embedded
46 completion_op. Each concurrent co_await t.wait() embeds one
47 waiter_node in the awaitable on the suspended coroutine's
48 frame — waits perform no allocation.
49
50 timer::implementation holds per-timer state: expiry, heap
51 index, and an intrusive_list of waiter_nodes. Multiple
52 coroutines can wait on the same timer simultaneously.
53
54 timer_service owns a min-heap of active timers and a free list
55 of recycled impls. The heap is ordered by expiry time; the
56 scheduler queries nearest_expiry() to set the epoll/timerfd
57 timeout.
58
59 Optimization Strategy
60 ---------------------
61 1. Deferred heap insertion — expires_after() stores the expiry
62 but does not insert into the heap. Insertion happens in wait().
63 2. Thread-local impl cache — single-slot per-thread cache.
64 3. Frame-resident waiter_node with embedded completion_op —
65 eliminates heap allocation per wait/fire/cancel.
66 4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().
67 5. might_have_pending_waits_ flag — skips lock when no wait issued.
68
69 Concurrency
70 -----------
71 stop_token callbacks can fire from any thread. The impl_
72 pointer on waiter_node is used as a "still in list" marker.
73 A waiter_node's storage is the suspended coroutine's frame:
74 every completion path must finish touching the node before
75 posting the continuation or destroying the handle.
76 */
77
78 inline void timer_service_invalidate_cache() noexcept;
79
80 // timer_service class body — member function definitions are
81 // out-of-class (after implementation and waiter_node are complete)
82 class BOOST_COROSIO_DECL timer_service final
83 : public capy::execution_context::service
84 , public io_object::io_service
85 {
86 public:
87 using clock_type = std::chrono::steady_clock;
88 using time_point = clock_type::time_point;
89
90 /// Type-erased callback for earliest-expiry-changed notifications.
91 class callback
92 {
93 void* ctx_ = nullptr;
94 void (*fn_)(void*) = nullptr;
95
96 public:
97 /// Construct an empty callback.
98 1219x callback() = default;
99
100 /// Construct a callback with the given context and function.
101 1219x callback(void* ctx, void (*fn)(void*)) noexcept : ctx_(ctx), fn_(fn) {}
102
103 /// Return true if the callback is non-empty.
104 explicit operator bool() const noexcept
105 {
106 return fn_ != nullptr;
107 }
108
109 /// Invoke the callback.
110 8422x void operator()() const
111 {
112 8422x if (fn_)
113 8422x fn_(ctx_);
114 8422x }
115 };
116
117 private:
118 struct heap_entry
119 {
120 time_point time_;
121 timer::implementation* timer_;
122 };
123
124 scheduler* sched_ = nullptr;
125 BOOST_COROSIO_MSVC_WARNING_PUSH
126 BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std:: members, dll-interface
127 mutable std::mutex mutex_;
128 std::vector<heap_entry> heap_;
129 timer::implementation* free_list_ = nullptr;
130 callback on_earliest_changed_;
131 bool shutting_down_ = false;
132 // Avoids mutex in nearest_expiry() and empty()
133 mutable std::atomic<std::int64_t> cached_nearest_ns_{
134 (std::numeric_limits<std::int64_t>::max)()};
135 BOOST_COROSIO_MSVC_WARNING_POP
136
137 public:
138 /// Construct the timer service bound to a scheduler.
139 1219x inline timer_service(capy::execution_context&, scheduler& sched)
140 1219x : sched_(&sched)
141 {
142 1219x }
143
144 /// Return the associated scheduler.
145 16984x inline scheduler& get_scheduler() noexcept
146 {
147 16984x return *sched_;
148 }
149
150 /// Destroy the timer service.
151 2438x ~timer_service() override = default;
152
153 timer_service(timer_service const&) = delete;
154 timer_service& operator=(timer_service const&) = delete;
155
156 /// Register a callback invoked when the earliest expiry changes.
157 1219x inline void set_on_earliest_changed(callback cb)
158 {
159 1219x on_earliest_changed_ = cb;
160 1219x }
161
162 /// Return true if no timers are in the heap.
163 inline bool empty() const noexcept
164 {
165 return cached_nearest_ns_.load(std::memory_order_acquire) ==
166 (std::numeric_limits<std::int64_t>::max)();
167 }
168
169 /// Return the nearest timer expiry without acquiring the mutex.
170 220457x inline time_point nearest_expiry() const noexcept
171 {
172 220457x auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
173 220457x return time_point(time_point::duration(ns));
174 }
175
176 /// Cancel all pending timers and free cached resources.
177 inline void shutdown() override;
178
179 /// Construct a new timer implementation.
180 inline io_object::implementation* construct() override;
181
182 /// Destroy a timer implementation, cancelling pending waiters.
183 inline void destroy(io_object::implementation* p) override;
184
185 /// Cancel and recycle a timer implementation.
186 inline void destroy_impl(timer::implementation& impl);
187
188 /// Update the timer expiry, cancelling existing waiters.
189 inline std::size_t update_timer(
190 timer::implementation& impl, time_point new_time);
191
192 /// Insert a waiter into the timer's waiter list and the heap.
193 inline void insert_waiter(timer::implementation& impl, waiter_node* w);
194
195 /// Cancel all waiters on a timer.
196 inline std::size_t cancel_timer(timer::implementation& impl);
197
198 /// Cancel one specific waiter ( stop_token callback path ).
199 inline void cancel_waiter(waiter_node* w);
200
201 /// Cancel the oldest pending waiter on a timer ( FIFO ).
202 inline std::size_t cancel_one_waiter(timer::implementation& impl);
203
204 /// Complete all waiters whose timers have expired.
205 inline std::size_t process_expired();
206
207 private:
208 250840x inline void refresh_cached_nearest() noexcept
209 {
210 250840x auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
211 247655x : heap_[0].time_.time_since_epoch().count();
212 250840x cached_nearest_ns_.store(ns, std::memory_order_release);
213 250840x }
214
215 inline void remove_timer_impl(timer::implementation& impl);
216 inline void up_heap(std::size_t index);
217 inline void down_heap(std::size_t index);
218 inline void swap_heap(std::size_t i1, std::size_t i2);
219 };
220
221 // Thread-local cache avoids hot-path mutex acquisitions:
222 // single-slot impl cache, validated by comparing svc_. Cleared by
223 // timer_service_invalidate_cache() during shutdown.
224
225 inline thread_local_ptr<timer::implementation> tl_cached_impl;
226
227 // The POD TLS slot above never runs destructors, so a short-lived
228 // run() thread would leak its cached impl. Each push arms this
229 // owner, whose destructor frees the slot at thread exit. A cached
230 // entry is a quiescent heap object (nothing in the heap or free
231 // list) and deletion touches no service state, so it is safe after
232 // the owning service is gone (the stale-entry path in
233 // try_pop_tl_cache deletes the same way).
234 struct tl_cache_owner
235 {
236 35x ~tl_cache_owner()
237 {
238 35x delete tl_cached_impl.get();
239 35x tl_cached_impl.set(nullptr);
240 35x }
241 };
242
243 inline void
244 9290x arm_tl_cache_cleanup() noexcept
245 {
246 9290x thread_local tl_cache_owner owner;
247 (void)owner;
248 9290x }
249
250 inline timer::implementation*
251 9360x try_pop_tl_cache(timer_service* svc) noexcept
252 {
253 9360x auto* impl = tl_cached_impl.get();
254 9360x if (impl)
255 {
256 9070x tl_cached_impl.set(nullptr);
257 9070x if (impl->svc_ == svc)
258 9070x return impl;
259 // Stale impl from a destroyed service
260 delete impl;
261 }
262 290x return nullptr;
263 }
264
265 inline bool
266 9334x try_push_tl_cache(timer::implementation* impl) noexcept
267 {
268 9334x if (!tl_cached_impl.get())
269 {
270 9290x arm_tl_cache_cleanup();
271 9290x tl_cached_impl.set(impl);
272 9290x return true;
273 }
274 44x return false;
275 }
276
277 inline void
278 1219x timer_service_invalidate_cache() noexcept
279 {
280 1219x delete tl_cached_impl.get();
281 1219x tl_cached_impl.set(nullptr);
282 1219x }
283
284 // timer_service out-of-class member function definitions
285
286 inline void
287 1219x timer_service::shutdown()
288 {
289 1219x timer_service_invalidate_cache();
290 1219x shutting_down_ = true;
291
292 // Snapshot impls and detach them from the heap so that
293 // coroutine-owned timer destructors (triggered by h.destroy()
294 // below) cannot re-enter remove_timer_impl() and mutate the
295 // vector during iteration.
296 1219x std::vector<timer::implementation*> impls;
297 1219x impls.reserve(heap_.size());
298 1245x for (auto& entry : heap_)
299 {
300 26x entry.timer_->heap_index_.store(
301 (std::numeric_limits<std::size_t>::max)(),
302 std::memory_order_relaxed);
303 26x impls.push_back(entry.timer_);
304 }
305 1219x heap_.clear();
306 1219x cached_nearest_ns_.store(
307 (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);
308
309 // Cancel waiting timers. Each waiter called work_started()
310 // in implementation::wait(). On IOCP the scheduler shutdown
311 // loop exits when outstanding_work_ reaches zero, so we must
312 // call work_finished() here to balance it. On other backends
313 // this is harmless.
314 1245x for (auto* impl : impls)
315 {
316 52x while (auto* w = impl->waiters_.pop_front())
317 {
318 26x w->reset_stop_cb();
319 26x auto h = std::exchange(w->h_, {});
320 26x sched_->work_finished();
321 // Destroying the frame also ends the node's storage
322 26x if (h)
323 26x h.destroy();
324 26x }
325 26x delete impl;
326 }
327
328 // Delete free-listed impls
329 1263x while (free_list_)
330 {
331 44x auto* next = free_list_->next_free_;
332 44x delete free_list_;
333 44x free_list_ = next;
334 }
335 1219x }
336
337 inline io_object::implementation*
338 9360x timer_service::construct()
339 {
340 9360x timer::implementation* impl = try_pop_tl_cache(this);
341 9360x if (impl)
342 {
343 9070x impl->svc_ = this;
344 // Reset expiry_ too: a recycled impl must behave like a fresh
345 // one, whose default expiry reads as already elapsed
346 9070x impl->expiry_ = {};
347 9070x impl->heap_index_.store(
348 (std::numeric_limits<std::size_t>::max)(),
349 std::memory_order_relaxed);
350 9070x impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
351 9070x return impl;
352 }
353
354 290x std::lock_guard lock(mutex_);
355 290x if (free_list_)
356 {
357 impl = free_list_;
358 free_list_ = impl->next_free_;
359 impl->next_free_ = nullptr;
360 impl->svc_ = this;
361 impl->expiry_ = {};
362 impl->heap_index_.store(
363 (std::numeric_limits<std::size_t>::max)(),
364 std::memory_order_relaxed);
365 impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
366 }
367 else
368 {
369 290x impl = new timer::implementation(*this);
370 }
371 290x return impl;
372 290x }
373
374 inline void
375 9360x timer_service::destroy(io_object::implementation* p)
376 {
377 // During shutdown the drain loop owns every impl and deletes
378 // them directly. A frame destroyed by that loop can unwind a
379 // handle whose impl was freed in an earlier iteration (a
380 // timeout's parent frame owns the timeout timer while
381 // suspended on the inner delay's timer), so bail out before
382 // even downcasting the pointer.
383 9360x if (shutting_down_)
384 26x return;
385 9334x destroy_impl(static_cast<timer::implementation&>(*p));
386 }
387
388 inline void
389 9334x timer_service::destroy_impl(timer::implementation& impl)
390 {
391 // During shutdown the impl is owned by the shutdown loop.
392 // Re-entering here (from a coroutine-owned timer destructor
393 // triggered by h.destroy()) must not modify the heap or
394 // recycle the impl — shutdown deletes it directly.
395 9334x if (shutting_down_)
396 9290x return;
397
398 9334x cancel_timer(impl);
399
400 18668x if (impl.heap_index_.load(std::memory_order_relaxed) !=
401 9334x (std::numeric_limits<std::size_t>::max)())
402 {
403 std::lock_guard lock(mutex_);
404 remove_timer_impl(impl);
405 refresh_cached_nearest();
406 }
407
408 9334x if (try_push_tl_cache(&impl))
409 9290x return;
410
411 44x std::lock_guard lock(mutex_);
412 44x impl.next_free_ = free_list_;
413 44x free_list_ = &impl;
414 44x }
415
416 inline std::size_t
417 timer_service::update_timer(timer::implementation& impl, time_point new_time)
418 {
419 // Gate on the flag, not waiters_: reading the non-atomic list
420 // here would race a concurrent drain. A false flag is safe to
421 // trust pre-lock: wait() stores it true before publishing, and
422 // it is cleared only under the mutex when the waiter list is
423 // empty, so false implies no published waiters.
424 bool in_heap =
425 (impl.heap_index_.load(std::memory_order_relaxed) !=
426 (std::numeric_limits<std::size_t>::max)());
427 if (!in_heap &&
428 !impl.might_have_pending_waits_.load(std::memory_order_relaxed))
429 return 0;
430
431 bool notify = false;
432 intrusive_list<waiter_node> canceled;
433
434 {
435 std::lock_guard lock(mutex_);
436
437 while (auto* w = impl.waiters_.pop_front())
438 {
439 w->impl_ = nullptr;
440 canceled.push_back(w);
441 }
442
443 std::size_t idx = impl.heap_index_.load(std::memory_order_relaxed);
444 if (idx < heap_.size())
445 {
446 time_point old_time = heap_[idx].time_;
447 heap_[idx].time_ = new_time;
448
449 if (new_time < old_time)
450 up_heap(idx);
451 else
452 down_heap(idx);
453
454 notify =
455 (impl.heap_index_.load(std::memory_order_relaxed) == 0);
456 }
457
458 refresh_cached_nearest();
459 }
460
461 std::size_t count = 0;
462 while (auto* w = canceled.pop_front())
463 {
464 w->ec_ = make_error_code(capy::error::canceled);
465 sched_->post(&w->op_);
466 ++count;
467 }
468
469 if (notify)
470 on_earliest_changed_();
471
472 return count;
473 }
474
475 inline void
476 8505x timer_service::insert_waiter(timer::implementation& impl, waiter_node* w)
477 {
478 8505x bool notify = false;
479 8505x bool lost_cancel = false;
480 {
481 8505x std::lock_guard lock(mutex_);
482 // Publish: from here the waiter is visible to the fire path and
483 // to its own stop callback (impl_ non-null enables cancel_waiter).
484 8505x w->impl_ = &impl;
485 17010x if (impl.heap_index_.load(std::memory_order_relaxed) ==
486 8505x (std::numeric_limits<std::size_t>::max)())
487 {
488 8505x impl.heap_index_.store(heap_.size(), std::memory_order_relaxed);
489 8505x heap_.push_back({impl.expiry_, &impl});
490 8505x up_heap(heap_.size() - 1);
491 8505x notify =
492 8505x (impl.heap_index_.load(std::memory_order_relaxed) == 0);
493 8505x refresh_cached_nearest();
494 }
495 8505x impl.waiters_.push_back(w);
496
497 // Lost-cancel re-check: a stop requested after the canceller was
498 // armed in wait() but before this publication found impl_ null
499 // and returned a no-op. Observe it now and undo the insertion.
500 8505x if (w->token_->stop_requested())
501 {
502 w->impl_ = nullptr;
503 impl.waiters_.remove(w);
504 if (impl.waiters_.empty())
505 {
506 remove_timer_impl(impl);
507 impl.might_have_pending_waits_.store(
508 false, std::memory_order_relaxed);
509 }
510 refresh_cached_nearest();
511 lost_cancel = true;
512 notify = false; // insertion undone; nearest unchanged
513 }
514 8505x }
515 8505x if (notify)
516 8422x on_earliest_changed_();
517 8505x if (lost_cancel)
518 {
519 w->ec_ = make_error_code(capy::error::canceled);
520 sched_->post(&w->op_);
521 }
522 8505x }
523
524 inline std::size_t
525 9334x timer_service::cancel_timer(timer::implementation& impl)
526 {
527 9334x if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
528 9332x return 0;
529
530 // No unlocked already-done fast-out here: it would need the
531 // non-atomic waiters_ (a race with concurrent drains), and an
532 // index-only check is lifetime-unsafe because npos is stored
533 // before the drain finishes touching the impl. A stale-true
534 // flag is rare with the stateless API; the locked path below
535 // re-validates.
536
537 2x intrusive_list<waiter_node> canceled;
538
539 {
540 2x std::lock_guard lock(mutex_);
541 2x remove_timer_impl(impl);
542 4x while (auto* w = impl.waiters_.pop_front())
543 {
544 2x w->impl_ = nullptr;
545 2x canceled.push_back(w);
546 2x }
547 // Store false as the final touch of the impl under the lock so
548 // update_timer's pre-lock false-flag trust holds unqualified.
549 2x impl.might_have_pending_waits_.store(false, std::memory_order_relaxed);
550 2x refresh_cached_nearest();
551 2x }
552
553 2x std::size_t count = 0;
554 4x while (auto* w = canceled.pop_front())
555 {
556 2x w->ec_ = make_error_code(capy::error::canceled);
557 2x sched_->post(&w->op_);
558 2x ++count;
559 2x }
560
561 2x return count;
562 }
563
564 inline void
565 1386x timer_service::cancel_waiter(waiter_node* w)
566 {
567 {
568 1386x std::lock_guard lock(mutex_);
569 // Already removed by another drain: cancel_timer,
570 // cancel_one_waiter, update_timer, process_expired, or
571 // insert_waiter's lost-cancel recheck
572 1386x if (!w->impl_)
573 return;
574 1386x auto* impl = w->impl_;
575 1386x w->impl_ = nullptr;
576 1386x impl->waiters_.remove(w);
577 1386x if (impl->waiters_.empty())
578 {
579 1386x remove_timer_impl(*impl);
580 1386x impl->might_have_pending_waits_.store(
581 false, std::memory_order_relaxed);
582 }
583 1386x refresh_cached_nearest();
584 1386x }
585
586 1386x w->ec_ = make_error_code(capy::error::canceled);
587 1386x sched_->post(&w->op_);
588 }
589
590 inline std::size_t
591 timer_service::cancel_one_waiter(timer::implementation& impl)
592 {
593 if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
594 return 0;
595
596 waiter_node* w = nullptr;
597
598 {
599 std::lock_guard lock(mutex_);
600 w = impl.waiters_.pop_front();
601 if (!w)
602 return 0;
603 w->impl_ = nullptr;
604 if (impl.waiters_.empty())
605 {
606 remove_timer_impl(impl);
607 impl.might_have_pending_waits_.store(
608 false, std::memory_order_relaxed);
609 }
610 refresh_cached_nearest();
611 }
612
613 w->ec_ = make_error_code(capy::error::canceled);
614 sched_->post(&w->op_);
615 return 1;
616 }
617
618 inline std::size_t
619 240947x timer_service::process_expired()
620 {
621 240947x intrusive_list<waiter_node> expired;
622
623 {
624 240947x std::lock_guard lock(mutex_);
625 240947x auto now = clock_type::now();
626
627 248038x while (!heap_.empty() && heap_[0].time_ <= now)
628 {
629 7091x timer::implementation* t = heap_[0].timer_;
630 7091x remove_timer_impl(*t);
631 14182x while (auto* w = t->waiters_.pop_front())
632 {
633 7091x w->impl_ = nullptr;
634 7091x w->ec_ = {};
635 7091x expired.push_back(w);
636 7091x }
637 7091x t->might_have_pending_waits_.store(
638 false, std::memory_order_relaxed);
639 }
640
641 240947x refresh_cached_nearest();
642 240947x }
643
644 240947x std::size_t count = 0;
645 248038x while (auto* w = expired.pop_front())
646 {
647 7091x sched_->post(&w->op_);
648 7091x ++count;
649 7091x }
650
651 240947x return count;
652 }
653
654 inline void
655 8479x timer_service::remove_timer_impl(timer::implementation& impl)
656 {
657 8479x std::size_t index = impl.heap_index_.load(std::memory_order_relaxed);
658 8479x if (index >= heap_.size())
659 return; // Not in heap
660
661 8479x if (index == heap_.size() - 1)
662 {
663 // Last element, just pop
664 1622x impl.heap_index_.store(
665 (std::numeric_limits<std::size_t>::max)(),
666 std::memory_order_relaxed);
667 1622x heap_.pop_back();
668 }
669 else
670 {
671 // Swap with last and reheapify
672 6857x swap_heap(index, heap_.size() - 1);
673 6857x impl.heap_index_.store(
674 (std::numeric_limits<std::size_t>::max)(),
675 std::memory_order_relaxed);
676 6857x heap_.pop_back();
677
678 6857x if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
679 up_heap(index);
680 else
681 6857x down_heap(index);
682 }
683 }
684
685 inline void
686 8505x timer_service::up_heap(std::size_t index)
687 {
688 15338x while (index > 0)
689 {
690 6916x std::size_t parent = (index - 1) / 2;
691 6916x if (!(heap_[index].time_ < heap_[parent].time_))
692 83x break;
693 6833x swap_heap(index, parent);
694 6833x index = parent;
695 }
696 8505x }
697
698 inline void
699 6857x timer_service::down_heap(std::size_t index)
700 {
701 6857x std::size_t child = index * 2 + 1;
702 6859x while (child < heap_.size())
703 {
704 4x std::size_t min_child = (child + 1 == heap_.size() ||
705 heap_[child].time_ < heap_[child + 1].time_)
706 4x ? child
707 4x : child + 1;
708
709 4x if (heap_[index].time_ < heap_[min_child].time_)
710 2x break;
711
712 2x swap_heap(index, min_child);
713 2x index = min_child;
714 2x child = index * 2 + 1;
715 }
716 6857x }
717
718 inline void
719 13692x timer_service::swap_heap(std::size_t i1, std::size_t i2)
720 {
721 13692x heap_entry tmp = heap_[i1];
722 13692x heap_[i1] = heap_[i2];
723 13692x heap_[i2] = tmp;
724 13692x heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed);
725 13692x heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed);
726 13692x }
727
728 // waiter_node's completion_op and canceller members are defined in
729 // timer.cpp alongside implementation::wait(), for the same reason
730 // wait() lives there (see below).
731
732 // timer::implementation::wait() is defined in timer.cpp, not here.
733 // It must be a non-inline definition in a translation unit that is
734 // always pulled into the link whenever detail::timer is used (every
735 // consumer needs timer's constructors from that same object file).
736 // An inline definition in this header would only be emitted in
737 // translation units that happen to also include this header, which
738 // is not guaranteed for every caller of wait_awaitable::await_suspend
739 // in timer.hpp (e.g. code that only reaches timer.hpp through
740 // delay.hpp, without transitively including a scheduler header).
741
742 // Free functions
743
744 inline std::size_t
745 timer_service_update_expiry(timer::implementation& impl)
746 {
747 return impl.svc_->update_timer(impl, impl.expiry_);
748 }
749
750 inline std::size_t
751 timer_service_cancel(timer::implementation& impl) noexcept
752 {
753 return impl.svc_->cancel_timer(impl);
754 }
755
756 inline std::size_t
757 timer_service_cancel_one(timer::implementation& impl) noexcept
758 {
759 return impl.svc_->cancel_one_waiter(impl);
760 }
761
762 inline timer_service&
763 1219x get_timer_service(capy::execution_context& ctx, scheduler& sched)
764 {
765 1219x return ctx.make_service<timer_service>(sched);
766 }
767
768 } // namespace boost::corosio::detail
769
770 #endif
771