include/boost/corosio/detail/timer_service.hpp

93.0% Lines (344/370) 97.7% List of functions (42/43)
timer_service.hpp
f(x) Functions (43)
Function Calls Lines Blocks
boost::corosio::detail::timer_service::callback::callback() :96 1875x 100.0% 100.0% boost::corosio::detail::timer_service::callback::callback(void*, void (*)(void*)) :99 1875x 100.0% 100.0% boost::corosio::detail::timer_service::callback::operator()() const :108 14895x 100.0% 100.0% boost::corosio::detail::timer_service::timer_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :137 1875x 100.0% 100.0% boost::corosio::detail::timer_service::get_scheduler() :143 29946x 100.0% 100.0% boost::corosio::detail::timer_service::~timer_service() :149 3750x 100.0% 100.0% boost::corosio::detail::timer_service::set_on_earliest_changed(boost::corosio::detail::timer_service::callback) :155 1875x 100.0% 100.0% boost::corosio::detail::timer_service::empty() const :161 47932x 100.0% 70.0% boost::corosio::detail::timer_service::nearest_expiry() const :168 387273x 100.0% 73.0% boost::corosio::detail::timer_service::refresh_cached_nearest() :211 466727x 100.0% 70.0% boost::corosio::detail::waiter_node::completion_op::completion_op() :235 478x 100.0% 100.0% boost::corosio::detail::waiter_node::waiter_node() :261 478x 100.0% 100.0% boost::corosio::detail::try_pop_tl_cache(boost::corosio::detail::timer_service*) :298 15284x 87.5% 78.0% boost::corosio::detail::try_push_tl_cache(boost::corosio::detail::timer_service::implementation*) :313 15272x 100.0% 100.0% boost::corosio::detail::try_pop_waiter_tl_cache() :324 14979x 100.0% 100.0% boost::corosio::detail::try_push_waiter_tl_cache(boost::corosio::detail::waiter_node*) :336 14955x 100.0% 100.0% boost::corosio::detail::timer_service_invalidate_cache() :347 1875x 100.0% 100.0% boost::corosio::detail::timer_service::implementation::implementation(boost::corosio::detail::timer_service&) :358 598x 100.0% 100.0% boost::corosio::detail::timer_service::shutdown() :365 1875x 100.0% 82.0% boost::corosio::detail::timer_service::construct() :422 15284x 66.7% 76.0% boost::corosio::detail::timer_service::destroy(boost::corosio::io_object::implementation*) :451 15281x 100.0% 100.0% boost::corosio::detail::timer_service::destroy_impl(boost::corosio::detail::timer_service::implementation&) :457 15281x 73.3% 64.0% boost::corosio::detail::timer_service::create_waiter() :484 14979x 100.0% 87.0% boost::corosio::detail::timer_service::destroy_waiter(boost::corosio::detail::waiter_node*) :502 14955x 100.0% 100.0% boost::corosio::detail::timer_service::update_timer(boost::corosio::detail::timer_service::implementation&, std::chrono::time_point<std::chrono::_V2::steady_clock, std::chrono::duration<long, std::ratio<1l, 1000000000l> > >) :513 12x 96.6% 81.0% boost::corosio::detail::timer_service::insert_waiter(boost::corosio::detail::timer_service::implementation&, boost::corosio::detail::waiter_node*) :563 14979x 100.0% 82.0% boost::corosio::detail::timer_service::cancel_timer(boost::corosio::detail::timer_service::implementation&) :583 15991x 87.5% 78.0% boost::corosio::detail::timer_service::cancel_waiter(boost::corosio::detail::waiter_node*) :623 54x 92.9% 80.0% boost::corosio::detail::timer_service::cancel_one_waiter(boost::corosio::detail::timer_service::implementation&) :646 3x 76.5% 70.0% boost::corosio::detail::timer_service::process_expired() :673 450966x 100.0% 91.0% boost::corosio::detail::timer_service::remove_timer_impl(boost::corosio::detail::timer_service::implementation&) :708 14934x 84.6% 65.0% boost::corosio::detail::timer_service::up_heap(unsigned long) :735 14955x 100.0% 100.0% boost::corosio::detail::timer_service::down_heap(unsigned long) :748 13830x 100.0% 100.0% boost::corosio::detail::timer_service::swap_heap(unsigned long, unsigned long) :768 27618x 100.0% 100.0% boost::corosio::detail::waiter_node::canceller::operator()() const :780 54x 100.0% 100.0% boost::corosio::detail::waiter_node::completion_op::do_complete(void*, boost::corosio::detail::scheduler_op*, unsigned int, unsigned int) :786 0 0.0% 0.0% boost::corosio::detail::waiter_node::completion_op::operator()() :800 14955x 100.0% 100.0% boost::corosio::detail::waiter_node::completion_op::destroy() :825 12x 100.0% 100.0% boost::corosio::detail::timer_service::implementation::wait(std::__n4861::coroutine_handle<void>, boost::capy::executor_ref, std::stop_token, std::error_code*, boost::capy::continuation*) :852 14993x 100.0% 97.0% boost::corosio::detail::timer_service_update_expiry(boost::corosio::io_timer::implementation&) :895 12x 100.0% 100.0% boost::corosio::detail::timer_service_cancel(boost::corosio::io_timer::implementation&) :902 719x 100.0% 100.0% boost::corosio::detail::timer_service_cancel_one(boost::corosio::io_timer::implementation&) :909 3x 100.0% 100.0% boost::corosio::detail::get_timer_service(boost::capy::execution_context&, boost::corosio::detail::scheduler&) :916 1875x 100.0% 100.0%
Line TLA Hits Source Code
1 //
2 // Copyright (c) 2025 Vinnie Falco ([email protected])
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/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 <optional>
31 #include <stop_token>
32 #include <utility>
33 #include <vector>
34
35 namespace boost::corosio::detail {
36
37 struct scheduler;
38
39 /*
40 Timer Service
41 =============
42
43 Data Structures
44 ---------------
45 waiter_node holds per-waiter state: coroutine handle, executor,
46 error output, stop_token, embedded completion_op. Each concurrent
47 co_await t.wait() allocates one waiter_node.
48
49 timer_service::implementation holds per-timer state: expiry,
50 heap index, and an intrusive_list of waiter_nodes. Multiple
51 coroutines can wait on the same timer simultaneously.
52
53 timer_service owns a min-heap of active timers, a free list
54 of recycled impls, and a free list of recycled waiter_nodes. The
55 heap is ordered by expiry time; the scheduler queries
56 nearest_expiry() to set the epoll/timerfd timeout.
57
58 Optimization Strategy
59 ---------------------
60 1. Deferred heap insertion — expires_after() stores the expiry
61 but does not insert into the heap. Insertion happens in wait().
62 2. Thread-local impl cache — single-slot per-thread cache.
63 3. Embedded completion_op — eliminates heap allocation per fire/cancel.
64 4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().
65 5. might_have_pending_waits_ flag — skips lock when no wait issued.
66 6. Thread-local waiter cache — single-slot per-thread cache.
67
68 Concurrency
69 -----------
70 stop_token callbacks can fire from any thread. The impl_
71 pointer on waiter_node is used as a "still in list" marker.
72 */
73
74 struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node;
75
76 inline void timer_service_invalidate_cache() noexcept;
77
78 // timer_service class body — member function definitions are
79 // out-of-class (after implementation and waiter_node are complete)
80 class BOOST_COROSIO_DECL timer_service final
81 : public capy::execution_context::service
82 , public io_object::io_service
83 {
84 public:
85 using clock_type = std::chrono::steady_clock;
86 using time_point = clock_type::time_point;
87
88 /// Type-erased callback for earliest-expiry-changed notifications.
89 class callback
90 {
91 void* ctx_ = nullptr;
92 void (*fn_)(void*) = nullptr;
93
94 public:
95 /// Construct an empty callback.
96 1875x callback() = default;
97
98 /// Construct a callback with the given context and function.
99 1875x callback(void* ctx, void (*fn)(void*)) noexcept : ctx_(ctx), fn_(fn) {}
100
101 /// Return true if the callback is non-empty.
102 explicit operator bool() const noexcept
103 {
104 return fn_ != nullptr;
105 }
106
107 /// Invoke the callback.
108 14895x void operator()() const
109 {
110 14895x if (fn_)
111 14895x fn_(ctx_);
112 14895x }
113 };
114
115 struct implementation;
116
117 private:
118 struct heap_entry
119 {
120 time_point time_;
121 implementation* timer_;
122 };
123
124 scheduler* sched_ = nullptr;
125 mutable std::mutex mutex_;
126 std::vector<heap_entry> heap_;
127 implementation* free_list_ = nullptr;
128 waiter_node* waiter_free_list_ = nullptr;
129 callback on_earliest_changed_;
130 bool shutting_down_ = false;
131 // Avoids mutex in nearest_expiry() and empty()
132 mutable std::atomic<std::int64_t> cached_nearest_ns_{
133 (std::numeric_limits<std::int64_t>::max)()};
134
135 public:
136 /// Construct the timer service bound to a scheduler.
137 1875x inline timer_service(capy::execution_context&, scheduler& sched)
138 1875x : sched_(&sched)
139 {
140 1875x }
141
142 /// Return the associated scheduler.
143 29946x inline scheduler& get_scheduler() noexcept
144 {
145 29946x return *sched_;
146 }
147
148 /// Destroy the timer service.
149 3750x ~timer_service() override = default;
150
151 timer_service(timer_service const&) = delete;
152 timer_service& operator=(timer_service const&) = delete;
153
154 /// Register a callback invoked when the earliest expiry changes.
155 1875x inline void set_on_earliest_changed(callback cb)
156 {
157 1875x on_earliest_changed_ = cb;
158 1875x }
159
160 /// Return true if no timers are in the heap.
161 47932x inline bool empty() const noexcept
162 {
163 47932x return cached_nearest_ns_.load(std::memory_order_acquire) ==
164 47932x (std::numeric_limits<std::int64_t>::max)();
165 }
166
167 /// Return the nearest timer expiry without acquiring the mutex.
168 387273x inline time_point nearest_expiry() const noexcept
169 {
170 387273x auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
171 387273x return time_point(time_point::duration(ns));
172 }
173
174 /// Cancel all pending timers and free cached resources.
175 inline void shutdown() override;
176
177 /// Construct a new timer implementation.
178 inline io_object::implementation* construct() override;
179
180 /// Destroy a timer implementation, cancelling pending waiters.
181 inline void destroy(io_object::implementation* p) override;
182
183 /// Cancel and recycle a timer implementation.
184 inline void destroy_impl(implementation& impl);
185
186 /// Create or recycle a waiter node.
187 inline waiter_node* create_waiter();
188
189 /// Return a waiter node to the cache or free list.
190 inline void destroy_waiter(waiter_node* w);
191
192 /// Update the timer expiry, cancelling existing waiters.
193 inline std::size_t update_timer(implementation& impl, time_point new_time);
194
195 /// Insert a waiter into the timer's waiter list and the heap.
196 inline void insert_waiter(implementation& impl, waiter_node* w);
197
198 /// Cancel all waiters on a timer.
199 inline std::size_t cancel_timer(implementation& impl);
200
201 /// Cancel a single waiter ( stop_token callback path ).
202 inline void cancel_waiter(waiter_node* w);
203
204 /// Cancel one waiter on a timer.
205 inline std::size_t cancel_one_waiter(implementation& impl);
206
207 /// Complete all waiters whose timers have expired.
208 inline std::size_t process_expired();
209
210 private:
211 466727x inline void refresh_cached_nearest() noexcept
212 {
213 466727x auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
214 464508x : heap_[0].time_.time_since_epoch().count();
215 466727x cached_nearest_ns_.store(ns, std::memory_order_release);
216 466727x }
217
218 inline void remove_timer_impl(implementation& impl);
219 inline void up_heap(std::size_t index);
220 inline void down_heap(std::size_t index);
221 inline void swap_heap(std::size_t i1, std::size_t i2);
222 };
223
224 struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node
225 : intrusive_list<waiter_node>::node
226 {
227 // Embedded completion op — avoids heap allocation per fire/cancel
228 struct completion_op final : scheduler_op
229 {
230 waiter_node* waiter_ = nullptr;
231
232 static void do_complete(
233 void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);
234
235 478x completion_op() noexcept : scheduler_op(&do_complete) {}
236
237 void operator()() override;
238 void destroy() override;
239 };
240
241 // Per-waiter stop_token cancellation
242 struct canceller
243 {
244 waiter_node* waiter_;
245 void operator()() const;
246 };
247
248 // nullptr once removed from timer's waiter list (concurrency marker)
249 timer_service::implementation* impl_ = nullptr;
250 timer_service* svc_ = nullptr;
251 std::coroutine_handle<> h_;
252 capy::continuation* cont_ = nullptr;
253 capy::executor_ref d_;
254 std::error_code* ec_out_ = nullptr;
255 std::stop_token token_;
256 std::optional<std::stop_callback<canceller>> stop_cb_;
257 completion_op op_;
258 std::error_code ec_value_;
259 waiter_node* next_free_ = nullptr;
260
261 478x waiter_node() noexcept
262 478x {
263 478x op_.waiter_ = this;
264 478x }
265 };
266
267 struct timer_service::implementation final : timer::implementation
268 {
269 using clock_type = std::chrono::steady_clock;
270 using time_point = clock_type::time_point;
271 using duration = clock_type::duration;
272
273 timer_service* svc_ = nullptr;
274 intrusive_list<waiter_node> waiters_;
275
276 // Free list linkage (reused when impl is on free_list)
277 implementation* next_free_ = nullptr;
278
279 inline explicit implementation(timer_service& svc) noexcept;
280
281 inline std::coroutine_handle<> wait(
282 std::coroutine_handle<>,
283 capy::executor_ref,
284 std::stop_token,
285 std::error_code*,
286 capy::continuation*) override;
287 };
288
289 // Thread-local caches avoid hot-path mutex acquisitions:
290 // 1. Impl cache — single-slot, validated by comparing svc_
291 // 2. Waiter cache — single-slot, no service affinity
292 // All caches are cleared by timer_service_invalidate_cache() during shutdown.
293
294 inline thread_local_ptr<timer_service::implementation> tl_cached_impl;
295 inline thread_local_ptr<waiter_node> tl_cached_waiter;
296
297 inline timer_service::implementation*
298 15284x try_pop_tl_cache(timer_service* svc) noexcept
299 {
300 15284x auto* impl = tl_cached_impl.get();
301 15284x if (impl)
302 {
303 14686x tl_cached_impl.set(nullptr);
304 14686x if (impl->svc_ == svc)
305 14686x return impl;
306 // Stale impl from a destroyed service
307 delete impl;
308 }
309 598x return nullptr;
310 }
311
312 inline bool
313 15272x try_push_tl_cache(timer_service::implementation* impl) noexcept
314 {
315 15272x if (!tl_cached_impl.get())
316 {
317 15105x tl_cached_impl.set(impl);
318 15105x return true;
319 }
320 167x return false;
321 }
322
323 inline waiter_node*
324 14979x try_pop_waiter_tl_cache() noexcept
325 {
326 14979x auto* w = tl_cached_waiter.get();
327 14979x if (w)
328 {
329 14498x tl_cached_waiter.set(nullptr);
330 14498x return w;
331 }
332 481x return nullptr;
333 }
334
335 inline bool
336 14955x try_push_waiter_tl_cache(waiter_node* w) noexcept
337 {
338 14955x if (!tl_cached_waiter.get())
339 {
340 14826x tl_cached_waiter.set(w);
341 14826x return true;
342 }
343 129x return false;
344 }
345
346 inline void
347 1875x timer_service_invalidate_cache() noexcept
348 {
349 1875x delete tl_cached_impl.get();
350 1875x tl_cached_impl.set(nullptr);
351
352 1875x delete tl_cached_waiter.get();
353 1875x tl_cached_waiter.set(nullptr);
354 1875x }
355
356 // timer_service out-of-class member function definitions
357
358 598x inline timer_service::implementation::implementation(
359 598x timer_service& svc) noexcept
360 598x : svc_(&svc)
361 {
362 598x }
363
364 inline void
365 1875x timer_service::shutdown()
366 {
367 1875x timer_service_invalidate_cache();
368 1875x shutting_down_ = true;
369
370 // Snapshot impls and detach them from the heap so that
371 // coroutine-owned timer destructors (triggered by h.destroy()
372 // below) cannot re-enter remove_timer_impl() and mutate the
373 // vector during iteration.
374 1875x std::vector<implementation*> impls;
375 1875x impls.reserve(heap_.size());
376 1887x for (auto& entry : heap_)
377 {
378 12x entry.timer_->heap_index_ = (std::numeric_limits<std::size_t>::max)();
379 12x impls.push_back(entry.timer_);
380 }
381 1875x heap_.clear();
382 1875x cached_nearest_ns_.store(
383 (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);
384
385 // Cancel waiting timers. Each waiter called work_started()
386 // in implementation::wait(). On IOCP the scheduler shutdown
387 // loop exits when outstanding_work_ reaches zero, so we must
388 // call work_finished() here to balance it. On other backends
389 // this is harmless.
390 1887x for (auto* impl : impls)
391 {
392 24x while (auto* w = impl->waiters_.pop_front())
393 {
394 12x w->stop_cb_.reset();
395 12x auto h = std::exchange(w->h_, {});
396 12x sched_->work_finished();
397 12x if (h)
398 12x h.destroy();
399 12x delete w;
400 12x }
401 12x delete impl;
402 }
403
404 // Delete free-listed impls
405 2042x while (free_list_)
406 {
407 167x auto* next = free_list_->next_free_;
408 167x delete free_list_;
409 167x free_list_ = next;
410 }
411
412 // Delete free-listed waiters
413 2001x while (waiter_free_list_)
414 {
415 126x auto* next = waiter_free_list_->next_free_;
416 126x delete waiter_free_list_;
417 126x waiter_free_list_ = next;
418 }
419 1875x }
420
421 inline io_object::implementation*
422 15284x timer_service::construct()
423 {
424 15284x implementation* impl = try_pop_tl_cache(this);
425 15284x if (impl)
426 {
427 14686x impl->svc_ = this;
428 14686x impl->heap_index_ = (std::numeric_limits<std::size_t>::max)();
429 14686x impl->might_have_pending_waits_ = false;
430 14686x return impl;
431 }
432
433 598x std::lock_guard lock(mutex_);
434 598x if (free_list_)
435 {
436 impl = free_list_;
437 free_list_ = impl->next_free_;
438 impl->next_free_ = nullptr;
439 impl->svc_ = this;
440 impl->heap_index_ = (std::numeric_limits<std::size_t>::max)();
441 impl->might_have_pending_waits_ = false;
442 }
443 else
444 {
445 598x impl = new implementation(*this);
446 }
447 598x return impl;
448 598x }
449
450 inline void
451 15281x timer_service::destroy(io_object::implementation* p)
452 {
453 15281x destroy_impl(static_cast<implementation&>(*p));
454 15281x }
455
456 inline void
457 15281x timer_service::destroy_impl(implementation& impl)
458 {
459 // During shutdown the impl is owned by the shutdown loop.
460 // Re-entering here (from a coroutine-owned timer destructor
461 // triggered by h.destroy()) must not modify the heap or
462 // recycle the impl — shutdown deletes it directly.
463 15281x if (shutting_down_)
464 15114x return;
465
466 15272x cancel_timer(impl);
467
468 15272x if (impl.heap_index_ != (std::numeric_limits<std::size_t>::max)())
469 {
470 std::lock_guard lock(mutex_);
471 remove_timer_impl(impl);
472 refresh_cached_nearest();
473 }
474
475 15272x if (try_push_tl_cache(&impl))
476 15105x return;
477
478 167x std::lock_guard lock(mutex_);
479 167x impl.next_free_ = free_list_;
480 167x free_list_ = &impl;
481 167x }
482
483 inline waiter_node*
484 14979x timer_service::create_waiter()
485 {
486 14979x if (auto* w = try_pop_waiter_tl_cache())
487 14498x return w;
488
489 481x std::lock_guard lock(mutex_);
490 481x if (waiter_free_list_)
491 {
492 3x auto* w = waiter_free_list_;
493 3x waiter_free_list_ = w->next_free_;
494 3x w->next_free_ = nullptr;
495 3x return w;
496 }
497
498 478x return new waiter_node();
499 481x }
500
501 inline void
502 14955x timer_service::destroy_waiter(waiter_node* w)
503 {
504 14955x if (try_push_waiter_tl_cache(w))
505 14826x return;
506
507 129x std::lock_guard lock(mutex_);
508 129x w->next_free_ = waiter_free_list_;
509 129x waiter_free_list_ = w;
510 129x }
511
512 inline std::size_t
513 12x timer_service::update_timer(implementation& impl, time_point new_time)
514 {
515 bool in_heap =
516 12x (impl.heap_index_ != (std::numeric_limits<std::size_t>::max)());
517 12x if (!in_heap && impl.waiters_.empty())
518 return 0;
519
520 12x bool notify = false;
521 12x intrusive_list<waiter_node> canceled;
522
523 {
524 12x std::lock_guard lock(mutex_);
525
526 30x while (auto* w = impl.waiters_.pop_front())
527 {
528 18x w->impl_ = nullptr;
529 18x canceled.push_back(w);
530 18x }
531
532 12x if (impl.heap_index_ < heap_.size())
533 {
534 12x time_point old_time = heap_[impl.heap_index_].time_;
535 12x heap_[impl.heap_index_].time_ = new_time;
536
537 12x if (new_time < old_time)
538 9x up_heap(impl.heap_index_);
539 else
540 3x down_heap(impl.heap_index_);
541
542 12x notify = (impl.heap_index_ == 0);
543 }
544
545 12x refresh_cached_nearest();
546 12x }
547
548 12x std::size_t count = 0;
549 30x while (auto* w = canceled.pop_front())
550 {
551 18x w->ec_value_ = make_error_code(capy::error::canceled);
552 18x sched_->post(&w->op_);
553 18x ++count;
554 18x }
555
556 12x if (notify)
557 9x on_earliest_changed_();
558
559 12x return count;
560 }
561
562 inline void
563 14979x timer_service::insert_waiter(implementation& impl, waiter_node* w)
564 {
565 14979x bool notify = false;
566 {
567 14979x std::lock_guard lock(mutex_);
568 14979x if (impl.heap_index_ == (std::numeric_limits<std::size_t>::max)())
569 {
570 14946x impl.heap_index_ = heap_.size();
571 14946x heap_.push_back({impl.expiry_, &impl});
572 14946x up_heap(heap_.size() - 1);
573 14946x notify = (impl.heap_index_ == 0);
574 14946x refresh_cached_nearest();
575 }
576 14979x impl.waiters_.push_back(w);
577 14979x }
578 14979x if (notify)
579 14886x on_earliest_changed_();
580 14979x }
581
582 inline std::size_t
583 15991x timer_service::cancel_timer(implementation& impl)
584 {
585 15991x if (!impl.might_have_pending_waits_)
586 15245x return 0;
587
588 // Not in heap and no waiters — just clear the flag
589 746x if (impl.heap_index_ == (std::numeric_limits<std::size_t>::max)() &&
590 impl.waiters_.empty())
591 {
592 impl.might_have_pending_waits_ = false;
593 return 0;
594 }
595
596 746x intrusive_list<waiter_node> canceled;
597
598 {
599 746x std::lock_guard lock(mutex_);
600 746x remove_timer_impl(impl);
601 1495x while (auto* w = impl.waiters_.pop_front())
602 {
603 749x w->impl_ = nullptr;
604 749x canceled.push_back(w);
605 749x }
606 746x refresh_cached_nearest();
607 746x }
608
609 746x impl.might_have_pending_waits_ = false;
610
611 746x std::size_t count = 0;
612 1495x while (auto* w = canceled.pop_front())
613 {
614 749x w->ec_value_ = make_error_code(capy::error::canceled);
615 749x sched_->post(&w->op_);
616 749x ++count;
617 749x }
618
619 746x return count;
620 }
621
622 inline void
623 54x timer_service::cancel_waiter(waiter_node* w)
624 {
625 {
626 54x std::lock_guard lock(mutex_);
627 // Already removed by cancel_timer or process_expired
628 54x if (!w->impl_)
629 return;
630 54x auto* impl = w->impl_;
631 54x w->impl_ = nullptr;
632 54x impl->waiters_.remove(w);
633 54x if (impl->waiters_.empty())
634 {
635 51x remove_timer_impl(*impl);
636 51x impl->might_have_pending_waits_ = false;
637 }
638 54x refresh_cached_nearest();
639 54x }
640
641 54x w->ec_value_ = make_error_code(capy::error::canceled);
642 54x sched_->post(&w->op_);
643 }
644
645 inline std::size_t
646 3x timer_service::cancel_one_waiter(implementation& impl)
647 {
648 3x if (!impl.might_have_pending_waits_)
649 return 0;
650
651 3x waiter_node* w = nullptr;
652
653 {
654 3x std::lock_guard lock(mutex_);
655 3x w = impl.waiters_.pop_front();
656 3x if (!w)
657 return 0;
658 3x w->impl_ = nullptr;
659 3x if (impl.waiters_.empty())
660 {
661 remove_timer_impl(impl);
662 impl.might_have_pending_waits_ = false;
663 }
664 3x refresh_cached_nearest();
665 3x }
666
667 3x w->ec_value_ = make_error_code(capy::error::canceled);
668 3x sched_->post(&w->op_);
669 3x return 1;
670 }
671
672 inline std::size_t
673 450966x timer_service::process_expired()
674 {
675 450966x intrusive_list<waiter_node> expired;
676
677 {
678 450966x std::lock_guard lock(mutex_);
679 450966x auto now = clock_type::now();
680
681 465103x while (!heap_.empty() && heap_[0].time_ <= now)
682 {
683 14137x implementation* t = heap_[0].timer_;
684 14137x remove_timer_impl(*t);
685 28280x while (auto* w = t->waiters_.pop_front())
686 {
687 14143x w->impl_ = nullptr;
688 14143x w->ec_value_ = {};
689 14143x expired.push_back(w);
690 14143x }
691 14137x t->might_have_pending_waits_ = false;
692 }
693
694 450966x refresh_cached_nearest();
695 450966x }
696
697 450966x std::size_t count = 0;
698 465109x while (auto* w = expired.pop_front())
699 {
700 14143x sched_->post(&w->op_);
701 14143x ++count;
702 14143x }
703
704 450966x return count;
705 }
706
707 inline void
708 14934x timer_service::remove_timer_impl(implementation& impl)
709 {
710 14934x std::size_t index = impl.heap_index_;
711 14934x if (index >= heap_.size())
712 return; // Not in heap
713
714 14934x if (index == heap_.size() - 1)
715 {
716 // Last element, just pop
717 1107x impl.heap_index_ = (std::numeric_limits<std::size_t>::max)();
718 1107x heap_.pop_back();
719 }
720 else
721 {
722 // Swap with last and reheapify
723 13827x swap_heap(index, heap_.size() - 1);
724 13827x impl.heap_index_ = (std::numeric_limits<std::size_t>::max)();
725 13827x heap_.pop_back();
726
727 13827x if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
728 up_heap(index);
729 else
730 13827x down_heap(index);
731 }
732 }
733
734 inline void
735 14955x timer_service::up_heap(std::size_t index)
736 {
737 28740x while (index > 0)
738 {
739 13845x std::size_t parent = (index - 1) / 2;
740 13845x if (!(heap_[index].time_ < heap_[parent].time_))
741 60x break;
742 13785x swap_heap(index, parent);
743 13785x index = parent;
744 }
745 14955x }
746
747 inline void
748 13830x timer_service::down_heap(std::size_t index)
749 {
750 13830x std::size_t child = index * 2 + 1;
751 13836x while (child < heap_.size())
752 {
753 15x std::size_t min_child = (child + 1 == heap_.size() ||
754 3x heap_[child].time_ < heap_[child + 1].time_)
755 18x ? child
756 15x : child + 1;
757
758 15x if (heap_[index].time_ < heap_[min_child].time_)
759 9x break;
760
761 6x swap_heap(index, min_child);
762 6x index = min_child;
763 6x child = index * 2 + 1;
764 }
765 13830x }
766
767 inline void
768 27618x timer_service::swap_heap(std::size_t i1, std::size_t i2)
769 {
770 27618x heap_entry tmp = heap_[i1];
771 27618x heap_[i1] = heap_[i2];
772 27618x heap_[i2] = tmp;
773 27618x heap_[i1].timer_->heap_index_ = i1;
774 27618x heap_[i2].timer_->heap_index_ = i2;
775 27618x }
776
777 // waiter_node out-of-class member function definitions
778
779 inline void
780 54x waiter_node::canceller::operator()() const
781 {
782 54x waiter_->svc_->cancel_waiter(waiter_);
783 54x }
784
785 inline void
786 waiter_node::completion_op::do_complete(
787 [[maybe_unused]] void* owner,
788 scheduler_op* base,
789 std::uint32_t,
790 std::uint32_t)
791 {
792 // owner is always non-null here. The destroy path (owner == nullptr)
793 // is unreachable because completion_op overrides destroy() directly,
794 // bypassing scheduler_op::destroy() which would call func_(nullptr, ...).
795 BOOST_COROSIO_ASSERT(owner);
796 static_cast<completion_op*>(base)->operator()();
797 }
798
799 inline void
800 14955x waiter_node::completion_op::operator()()
801 {
802 14955x auto* w = waiter_;
803 14955x w->stop_cb_.reset();
804 14955x if (w->ec_out_)
805 14955x *w->ec_out_ = w->ec_value_;
806
807 14955x auto* cont = w->cont_;
808 14955x auto d = w->d_;
809 14955x auto* svc = w->svc_;
810 14955x auto& sched = svc->get_scheduler();
811
812 14955x svc->destroy_waiter(w);
813
814 14955x d.post(*cont);
815 14955x sched.work_finished();
816 14955x }
817
818 // GCC 14 false-positive: inlining ~optional<stop_callback> through
819 // delete loses track that stop_cb_ was already .reset() above.
820 #if defined(__GNUC__) && !defined(__clang__)
821 #pragma GCC diagnostic push
822 #pragma GCC diagnostic ignored "-Wmaybe-uninitialized"
823 #endif
824 inline void
825 12x waiter_node::completion_op::destroy()
826 {
827 // Called during scheduler shutdown drain when this completion_op is
828 // in the scheduler's ready queue (posted by cancel_timer() or
829 // process_expired()). Balances the work_started() from
830 // implementation::wait(). The scheduler drain loop separately
831 // balances the work_started() from post(). On IOCP both decrements
832 // are required for outstanding_work_ to reach zero; on other
833 // backends this is harmless.
834 //
835 // This override also prevents scheduler_op::destroy() from calling
836 // do_complete(nullptr, ...). See also: timer_service::shutdown()
837 // which drains waiters still in the timer heap (the other path).
838 12x auto* w = waiter_;
839 12x w->stop_cb_.reset();
840 12x auto h = std::exchange(w->h_, {});
841 12x auto& sched = w->svc_->get_scheduler();
842 12x delete w;
843 12x sched.work_finished();
844 12x if (h)
845 12x h.destroy();
846 12x }
847 #if defined(__GNUC__) && !defined(__clang__)
848 #pragma GCC diagnostic pop
849 #endif
850
851 inline std::coroutine_handle<>
852 14993x timer_service::implementation::wait(
853 std::coroutine_handle<> h,
854 capy::executor_ref d,
855 std::stop_token token,
856 std::error_code* ec,
857 capy::continuation* cont)
858 {
859 // Already-expired fast path — no waiter_node, no mutex.
860 // Post instead of dispatch so the coroutine yields to the
861 // scheduler, allowing other queued work to run.
862 14993x if (heap_index_ == (std::numeric_limits<std::size_t>::max)())
863 {
864 14960x if (expiry_ == (time_point::min)() || expiry_ <= clock_type::now())
865 {
866 14x if (ec)
867 14x *ec = {};
868 14x d.post(*cont);
869 14x return std::noop_coroutine();
870 }
871 }
872
873 14979x auto* w = svc_->create_waiter();
874 14979x w->impl_ = this;
875 14979x w->svc_ = svc_;
876 14979x w->h_ = h;
877 14979x w->cont_ = cont;
878 14979x w->d_ = d;
879 14979x w->token_ = std::move(token);
880 14979x w->ec_out_ = ec;
881
882 14979x svc_->insert_waiter(*this, w);
883 14979x might_have_pending_waits_ = true;
884 14979x svc_->get_scheduler().work_started();
885
886 14979x if (w->token_.stop_possible())
887 90x w->stop_cb_.emplace(w->token_, waiter_node::canceller{w});
888
889 14979x return std::noop_coroutine();
890 }
891
892 // Free functions
893
894 inline std::size_t
895 12x timer_service_update_expiry(timer::implementation& base)
896 {
897 12x auto& impl = static_cast<timer_service::implementation&>(base);
898 12x return impl.svc_->update_timer(impl, impl.expiry_);
899 }
900
901 inline std::size_t
902 719x timer_service_cancel(timer::implementation& base) noexcept
903 {
904 719x auto& impl = static_cast<timer_service::implementation&>(base);
905 719x return impl.svc_->cancel_timer(impl);
906 }
907
908 inline std::size_t
909 3x timer_service_cancel_one(timer::implementation& base) noexcept
910 {
911 3x auto& impl = static_cast<timer_service::implementation&>(base);
912 3x return impl.svc_->cancel_one_waiter(impl);
913 }
914
915 inline timer_service&
916 1875x get_timer_service(capy::execution_context& ctx, scheduler& sched)
917 {
918 1875x return ctx.make_service<timer_service>(sched);
919 }
920
921 } // namespace boost::corosio::detail
922
923 #endif
924