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_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 HIT 1219 : callback() = default;
99 :
100 : /// Construct a callback with the given context and function.
101 1219 : 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 8422 : void operator()() const
111 : {
112 8422 : if (fn_)
113 8422 : fn_(ctx_);
114 8422 : }
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 1219 : inline timer_service(capy::execution_context&, scheduler& sched)
140 1219 : : sched_(&sched)
141 : {
142 1219 : }
143 :
144 : /// Return the associated scheduler.
145 16984 : inline scheduler& get_scheduler() noexcept
146 : {
147 16984 : return *sched_;
148 : }
149 :
150 : /// Destroy the timer service.
151 2438 : ~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 1219 : inline void set_on_earliest_changed(callback cb)
158 : {
159 1219 : on_earliest_changed_ = cb;
160 1219 : }
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 220457 : inline time_point nearest_expiry() const noexcept
171 : {
172 220457 : auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
173 220457 : 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 250840 : inline void refresh_cached_nearest() noexcept
209 : {
210 250840 : auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
211 247655 : : heap_[0].time_.time_since_epoch().count();
212 250840 : cached_nearest_ns_.store(ns, std::memory_order_release);
213 250840 : }
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 35 : ~tl_cache_owner()
237 : {
238 35 : delete tl_cached_impl.get();
239 35 : tl_cached_impl.set(nullptr);
240 35 : }
241 : };
242 :
243 : inline void
244 9290 : arm_tl_cache_cleanup() noexcept
245 : {
246 9290 : thread_local tl_cache_owner owner;
247 : (void)owner;
248 9290 : }
249 :
250 : inline timer::implementation*
251 9360 : try_pop_tl_cache(timer_service* svc) noexcept
252 : {
253 9360 : auto* impl = tl_cached_impl.get();
254 9360 : if (impl)
255 : {
256 9070 : tl_cached_impl.set(nullptr);
257 9070 : if (impl->svc_ == svc)
258 9070 : return impl;
259 : // Stale impl from a destroyed service
260 MIS 0 : delete impl;
261 : }
262 HIT 290 : return nullptr;
263 : }
264 :
265 : inline bool
266 9334 : try_push_tl_cache(timer::implementation* impl) noexcept
267 : {
268 9334 : if (!tl_cached_impl.get())
269 : {
270 9290 : arm_tl_cache_cleanup();
271 9290 : tl_cached_impl.set(impl);
272 9290 : return true;
273 : }
274 44 : return false;
275 : }
276 :
277 : inline void
278 1219 : timer_service_invalidate_cache() noexcept
279 : {
280 1219 : delete tl_cached_impl.get();
281 1219 : tl_cached_impl.set(nullptr);
282 1219 : }
283 :
284 : // timer_service out-of-class member function definitions
285 :
286 : inline void
287 1219 : timer_service::shutdown()
288 : {
289 1219 : timer_service_invalidate_cache();
290 1219 : 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 1219 : std::vector<timer::implementation*> impls;
297 1219 : impls.reserve(heap_.size());
298 1245 : for (auto& entry : heap_)
299 : {
300 26 : entry.timer_->heap_index_.store(
301 : (std::numeric_limits<std::size_t>::max)(),
302 : std::memory_order_relaxed);
303 26 : impls.push_back(entry.timer_);
304 : }
305 1219 : heap_.clear();
306 1219 : 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 1245 : for (auto* impl : impls)
315 : {
316 52 : while (auto* w = impl->waiters_.pop_front())
317 : {
318 26 : w->reset_stop_cb();
319 26 : auto h = std::exchange(w->h_, {});
320 26 : sched_->work_finished();
321 : // Destroying the frame also ends the node's storage
322 26 : if (h)
323 26 : h.destroy();
324 26 : }
325 26 : delete impl;
326 : }
327 :
328 : // Delete free-listed impls
329 1263 : while (free_list_)
330 : {
331 44 : auto* next = free_list_->next_free_;
332 44 : delete free_list_;
333 44 : free_list_ = next;
334 : }
335 1219 : }
336 :
337 : inline io_object::implementation*
338 9360 : timer_service::construct()
339 : {
340 9360 : timer::implementation* impl = try_pop_tl_cache(this);
341 9360 : if (impl)
342 : {
343 9070 : 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 9070 : impl->expiry_ = {};
347 9070 : impl->heap_index_.store(
348 : (std::numeric_limits<std::size_t>::max)(),
349 : std::memory_order_relaxed);
350 9070 : impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
351 9070 : return impl;
352 : }
353 :
354 290 : std::lock_guard lock(mutex_);
355 290 : if (free_list_)
356 : {
357 MIS 0 : impl = free_list_;
358 0 : free_list_ = impl->next_free_;
359 0 : impl->next_free_ = nullptr;
360 0 : impl->svc_ = this;
361 0 : impl->expiry_ = {};
362 0 : impl->heap_index_.store(
363 : (std::numeric_limits<std::size_t>::max)(),
364 : std::memory_order_relaxed);
365 0 : impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
366 : }
367 : else
368 : {
369 HIT 290 : impl = new timer::implementation(*this);
370 : }
371 290 : return impl;
372 290 : }
373 :
374 : inline void
375 9360 : 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 9360 : if (shutting_down_)
384 26 : return;
385 9334 : destroy_impl(static_cast<timer::implementation&>(*p));
386 : }
387 :
388 : inline void
389 9334 : 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 9334 : if (shutting_down_)
396 9290 : return;
397 :
398 9334 : cancel_timer(impl);
399 :
400 18668 : if (impl.heap_index_.load(std::memory_order_relaxed) !=
401 9334 : (std::numeric_limits<std::size_t>::max)())
402 : {
403 MIS 0 : std::lock_guard lock(mutex_);
404 0 : remove_timer_impl(impl);
405 0 : refresh_cached_nearest();
406 0 : }
407 :
408 HIT 9334 : if (try_push_tl_cache(&impl))
409 9290 : return;
410 :
411 44 : std::lock_guard lock(mutex_);
412 44 : impl.next_free_ = free_list_;
413 44 : free_list_ = &impl;
414 44 : }
415 :
416 : inline std::size_t
417 MIS 0 : 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 0 : (impl.heap_index_.load(std::memory_order_relaxed) !=
426 0 : (std::numeric_limits<std::size_t>::max)());
427 0 : if (!in_heap &&
428 0 : !impl.might_have_pending_waits_.load(std::memory_order_relaxed))
429 0 : return 0;
430 :
431 0 : bool notify = false;
432 0 : intrusive_list<waiter_node> canceled;
433 :
434 : {
435 0 : std::lock_guard lock(mutex_);
436 :
437 0 : while (auto* w = impl.waiters_.pop_front())
438 : {
439 0 : w->impl_ = nullptr;
440 0 : canceled.push_back(w);
441 0 : }
442 :
443 0 : std::size_t idx = impl.heap_index_.load(std::memory_order_relaxed);
444 0 : if (idx < heap_.size())
445 : {
446 0 : time_point old_time = heap_[idx].time_;
447 0 : heap_[idx].time_ = new_time;
448 :
449 0 : if (new_time < old_time)
450 0 : up_heap(idx);
451 : else
452 0 : down_heap(idx);
453 :
454 0 : notify =
455 0 : (impl.heap_index_.load(std::memory_order_relaxed) == 0);
456 : }
457 :
458 0 : refresh_cached_nearest();
459 0 : }
460 :
461 0 : std::size_t count = 0;
462 0 : while (auto* w = canceled.pop_front())
463 : {
464 0 : w->ec_ = make_error_code(capy::error::canceled);
465 0 : sched_->post(&w->op_);
466 0 : ++count;
467 0 : }
468 :
469 0 : if (notify)
470 0 : on_earliest_changed_();
471 :
472 0 : return count;
473 : }
474 :
475 : inline void
476 HIT 8505 : timer_service::insert_waiter(timer::implementation& impl, waiter_node* w)
477 : {
478 8505 : bool notify = false;
479 8505 : bool lost_cancel = false;
480 : {
481 8505 : 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 8505 : w->impl_ = &impl;
485 17010 : if (impl.heap_index_.load(std::memory_order_relaxed) ==
486 8505 : (std::numeric_limits<std::size_t>::max)())
487 : {
488 8505 : impl.heap_index_.store(heap_.size(), std::memory_order_relaxed);
489 8505 : heap_.push_back({impl.expiry_, &impl});
490 8505 : up_heap(heap_.size() - 1);
491 8505 : notify =
492 8505 : (impl.heap_index_.load(std::memory_order_relaxed) == 0);
493 8505 : refresh_cached_nearest();
494 : }
495 8505 : 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 8505 : if (w->token_->stop_requested())
501 : {
502 MIS 0 : w->impl_ = nullptr;
503 0 : impl.waiters_.remove(w);
504 0 : if (impl.waiters_.empty())
505 : {
506 0 : remove_timer_impl(impl);
507 0 : impl.might_have_pending_waits_.store(
508 : false, std::memory_order_relaxed);
509 : }
510 0 : refresh_cached_nearest();
511 0 : lost_cancel = true;
512 0 : notify = false; // insertion undone; nearest unchanged
513 : }
514 HIT 8505 : }
515 8505 : if (notify)
516 8422 : on_earliest_changed_();
517 8505 : if (lost_cancel)
518 : {
519 MIS 0 : w->ec_ = make_error_code(capy::error::canceled);
520 0 : sched_->post(&w->op_);
521 : }
522 HIT 8505 : }
523 :
524 : inline std::size_t
525 9334 : timer_service::cancel_timer(timer::implementation& impl)
526 : {
527 9334 : if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
528 9332 : 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 2 : intrusive_list<waiter_node> canceled;
538 :
539 : {
540 2 : std::lock_guard lock(mutex_);
541 2 : remove_timer_impl(impl);
542 4 : while (auto* w = impl.waiters_.pop_front())
543 : {
544 2 : w->impl_ = nullptr;
545 2 : canceled.push_back(w);
546 2 : }
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 2 : impl.might_have_pending_waits_.store(false, std::memory_order_relaxed);
550 2 : refresh_cached_nearest();
551 2 : }
552 :
553 2 : std::size_t count = 0;
554 4 : while (auto* w = canceled.pop_front())
555 : {
556 2 : w->ec_ = make_error_code(capy::error::canceled);
557 2 : sched_->post(&w->op_);
558 2 : ++count;
559 2 : }
560 :
561 2 : return count;
562 : }
563 :
564 : inline void
565 1386 : timer_service::cancel_waiter(waiter_node* w)
566 : {
567 : {
568 1386 : 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 1386 : if (!w->impl_)
573 MIS 0 : return;
574 HIT 1386 : auto* impl = w->impl_;
575 1386 : w->impl_ = nullptr;
576 1386 : impl->waiters_.remove(w);
577 1386 : if (impl->waiters_.empty())
578 : {
579 1386 : remove_timer_impl(*impl);
580 1386 : impl->might_have_pending_waits_.store(
581 : false, std::memory_order_relaxed);
582 : }
583 1386 : refresh_cached_nearest();
584 1386 : }
585 :
586 1386 : w->ec_ = make_error_code(capy::error::canceled);
587 1386 : sched_->post(&w->op_);
588 : }
589 :
590 : inline std::size_t
591 MIS 0 : timer_service::cancel_one_waiter(timer::implementation& impl)
592 : {
593 0 : if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
594 0 : return 0;
595 :
596 0 : waiter_node* w = nullptr;
597 :
598 : {
599 0 : std::lock_guard lock(mutex_);
600 0 : w = impl.waiters_.pop_front();
601 0 : if (!w)
602 0 : return 0;
603 0 : w->impl_ = nullptr;
604 0 : if (impl.waiters_.empty())
605 : {
606 0 : remove_timer_impl(impl);
607 0 : impl.might_have_pending_waits_.store(
608 : false, std::memory_order_relaxed);
609 : }
610 0 : refresh_cached_nearest();
611 0 : }
612 :
613 0 : w->ec_ = make_error_code(capy::error::canceled);
614 0 : sched_->post(&w->op_);
615 0 : return 1;
616 : }
617 :
618 : inline std::size_t
619 HIT 240947 : timer_service::process_expired()
620 : {
621 240947 : intrusive_list<waiter_node> expired;
622 :
623 : {
624 240947 : std::lock_guard lock(mutex_);
625 240947 : auto now = clock_type::now();
626 :
627 248038 : while (!heap_.empty() && heap_[0].time_ <= now)
628 : {
629 7091 : timer::implementation* t = heap_[0].timer_;
630 7091 : remove_timer_impl(*t);
631 14182 : while (auto* w = t->waiters_.pop_front())
632 : {
633 7091 : w->impl_ = nullptr;
634 7091 : w->ec_ = {};
635 7091 : expired.push_back(w);
636 7091 : }
637 7091 : t->might_have_pending_waits_.store(
638 : false, std::memory_order_relaxed);
639 : }
640 :
641 240947 : refresh_cached_nearest();
642 240947 : }
643 :
644 240947 : std::size_t count = 0;
645 248038 : while (auto* w = expired.pop_front())
646 : {
647 7091 : sched_->post(&w->op_);
648 7091 : ++count;
649 7091 : }
650 :
651 240947 : return count;
652 : }
653 :
654 : inline void
655 8479 : timer_service::remove_timer_impl(timer::implementation& impl)
656 : {
657 8479 : std::size_t index = impl.heap_index_.load(std::memory_order_relaxed);
658 8479 : if (index >= heap_.size())
659 MIS 0 : return; // Not in heap
660 :
661 HIT 8479 : if (index == heap_.size() - 1)
662 : {
663 : // Last element, just pop
664 1622 : impl.heap_index_.store(
665 : (std::numeric_limits<std::size_t>::max)(),
666 : std::memory_order_relaxed);
667 1622 : heap_.pop_back();
668 : }
669 : else
670 : {
671 : // Swap with last and reheapify
672 6857 : swap_heap(index, heap_.size() - 1);
673 6857 : impl.heap_index_.store(
674 : (std::numeric_limits<std::size_t>::max)(),
675 : std::memory_order_relaxed);
676 6857 : heap_.pop_back();
677 :
678 6857 : if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
679 MIS 0 : up_heap(index);
680 : else
681 HIT 6857 : down_heap(index);
682 : }
683 : }
684 :
685 : inline void
686 8505 : timer_service::up_heap(std::size_t index)
687 : {
688 15338 : while (index > 0)
689 : {
690 6916 : std::size_t parent = (index - 1) / 2;
691 6916 : if (!(heap_[index].time_ < heap_[parent].time_))
692 83 : break;
693 6833 : swap_heap(index, parent);
694 6833 : index = parent;
695 : }
696 8505 : }
697 :
698 : inline void
699 6857 : timer_service::down_heap(std::size_t index)
700 : {
701 6857 : std::size_t child = index * 2 + 1;
702 6859 : while (child < heap_.size())
703 : {
704 4 : std::size_t min_child = (child + 1 == heap_.size() ||
705 MIS 0 : heap_[child].time_ < heap_[child + 1].time_)
706 HIT 4 : ? child
707 4 : : child + 1;
708 :
709 4 : if (heap_[index].time_ < heap_[min_child].time_)
710 2 : break;
711 :
712 2 : swap_heap(index, min_child);
713 2 : index = min_child;
714 2 : child = index * 2 + 1;
715 : }
716 6857 : }
717 :
718 : inline void
719 13692 : timer_service::swap_heap(std::size_t i1, std::size_t i2)
720 : {
721 13692 : heap_entry tmp = heap_[i1];
722 13692 : heap_[i1] = heap_[i2];
723 13692 : heap_[i2] = tmp;
724 13692 : heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed);
725 13692 : heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed);
726 13692 : }
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 MIS 0 : timer_service_update_expiry(timer::implementation& impl)
746 : {
747 0 : return impl.svc_->update_timer(impl, impl.expiry_);
748 : }
749 :
750 : inline std::size_t
751 0 : timer_service_cancel(timer::implementation& impl) noexcept
752 : {
753 0 : return impl.svc_->cancel_timer(impl);
754 : }
755 :
756 : inline std::size_t
757 0 : timer_service_cancel_one(timer::implementation& impl) noexcept
758 : {
759 0 : return impl.svc_->cancel_one_waiter(impl);
760 : }
761 :
762 : inline timer_service&
763 HIT 1219 : get_timer_service(capy::execution_context& ctx, scheduler& sched)
764 : {
765 1219 : return ctx.make_service<timer_service>(sched);
766 : }
767 :
768 : } // namespace boost::corosio::detail
769 :
770 : #endif
|
