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/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 : posting 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 HIT 4413 : callback() = default;
126 :
127 : /// Construct a callback with the given context and function.
128 4413 : 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 8070 : void operator()() const
138 : {
139 8070 : if (fn_)
140 8070 : fn_(ctx_);
141 8070 : }
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 :
160 : // Cancellers that claimed a waiter and have yet to post it;
161 : // shutdown waits them out so the scheduler's drain finds the post.
162 : std::size_t cancels_posting_ = 0; // guarded by mutex_
163 : // Avoids mutex in nearest_expiry() and empty()
164 : mutable std::atomic<std::int64_t> cached_nearest_ns_{
165 : (std::numeric_limits<std::int64_t>::max)()};
166 : BOOST_COROSIO_MSVC_WARNING_POP
167 :
168 : public:
169 : /// Construct the timer service bound to a scheduler.
170 4413 : inline timer_service(capy::execution_context&, scheduler& sched)
171 4413 : : sched_(&sched)
172 : {
173 4413 : }
174 :
175 : /// Return the associated scheduler.
176 35708 : inline scheduler& get_scheduler() noexcept
177 : {
178 35708 : return *sched_;
179 : }
180 :
181 : /// Destroy the timer service.
182 8826 : ~timer_service() override = default;
183 :
184 : timer_service(timer_service const&) = delete;
185 : timer_service& operator=(timer_service const&) = delete;
186 :
187 : /// Register a callback invoked when the earliest expiry changes.
188 4413 : inline void set_on_earliest_changed(callback cb)
189 : {
190 4413 : on_earliest_changed_ = cb;
191 4413 : }
192 :
193 : /// Return true if no timers are in the heap.
194 : inline bool empty() const noexcept
195 : {
196 : return cached_nearest_ns_.load(std::memory_order_acquire) ==
197 : (std::numeric_limits<std::int64_t>::max)();
198 : }
199 :
200 : /// Return the nearest timer expiry without acquiring the mutex.
201 93928 : inline time_point nearest_expiry() const noexcept
202 : {
203 93928 : auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
204 93928 : return time_point(time_point::duration(ns));
205 : }
206 :
207 : /// Cancel all pending timers and free cached resources.
208 : inline void shutdown() override;
209 :
210 : /// Construct a new timer implementation.
211 : inline io_object::implementation* construct() override;
212 :
213 : /// Destroy a timer implementation, cancelling pending waiters.
214 : inline void destroy(io_object::implementation* p) override;
215 :
216 : /// Cancel and recycle a timer implementation.
217 : inline void destroy_impl(timer::implementation& impl);
218 :
219 : /// Publish the timer's waiter and insert the timer into the heap.
220 : inline void insert_waiter(timer::implementation& impl, waiter_node* w);
221 :
222 : /// Cancel the timer's published waiter, if any.
223 : inline void cancel_timer(timer::implementation& impl);
224 :
225 : /// Cancel one specific waiter ( stop_token callback path ).
226 : inline void cancel_waiter(waiter_node* w);
227 :
228 : /// Complete all waiters whose timers have expired.
229 : inline std::size_t process_expired();
230 :
231 : private:
232 245265 : inline void refresh_cached_nearest() noexcept
233 : {
234 245265 : auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
235 61763 : : heap_[0].time_.time_since_epoch().count();
236 245265 : cached_nearest_ns_.store(ns, std::memory_order_release);
237 245265 : }
238 :
239 : inline void remove_timer_impl(timer::implementation& impl);
240 : inline void up_heap(std::size_t index);
241 : inline void down_heap(std::size_t index);
242 : inline void swap_heap(std::size_t i1, std::size_t i2);
243 : };
244 :
245 : // Thread-local single-slot impl cache, validated by comparing svc_.
246 : // Keeps a hot impl out of the shared free list; both sides still
247 : // take the pool lock to move it in and out of live tracking.
248 : // Cleared by timer_service_invalidate_cache() during shutdown.
249 :
250 : inline thread_local_ptr<timer::implementation> tl_cached_impl;
251 :
252 : // The POD TLS slot above never runs destructors, so a short-lived
253 : // run() thread would leak its cached impl. Each push arms this
254 : // owner, whose destructor frees the slot at thread exit. A cached
255 : // entry is a quiescent heap object (nothing in the heap or free
256 : // list) and deletion touches no service state, so it is safe after
257 : // the owning service is gone (the stale-entry path in
258 : // try_pop_tl_cache deletes the same way).
259 : struct tl_cache_owner
260 : {
261 61 : ~tl_cache_owner()
262 : {
263 61 : delete tl_cached_impl.get();
264 61 : tl_cached_impl.set(nullptr);
265 61 : }
266 : };
267 :
268 : inline void
269 15248 : arm_tl_cache_cleanup() noexcept
270 : {
271 15248 : [[maybe_unused]] thread_local tl_cache_owner owner;
272 15248 : }
273 :
274 : inline timer::implementation*
275 20401 : try_pop_tl_cache(timer_service* svc) noexcept
276 : {
277 20401 : auto* impl = tl_cached_impl.get();
278 20401 : if (impl)
279 : {
280 14744 : tl_cached_impl.set(nullptr);
281 14744 : if (impl->svc_ == svc)
282 14744 : return impl;
283 : // Stale impl from a destroyed service
284 MIS 0 : delete impl;
285 : }
286 HIT 5657 : return nullptr;
287 : }
288 :
289 : inline bool
290 17154 : try_push_tl_cache(timer::implementation* impl) noexcept
291 : {
292 17154 : if (!tl_cached_impl.get())
293 : {
294 15248 : arm_tl_cache_cleanup();
295 15248 : tl_cached_impl.set(impl);
296 15248 : return true;
297 : }
298 1906 : return false;
299 : }
300 :
301 : inline void
302 8826 : timer_service_invalidate_cache() noexcept
303 : {
304 8826 : delete tl_cached_impl.get();
305 8826 : tl_cached_impl.set(nullptr);
306 8826 : }
307 :
308 : // timer_service out-of-class member function definitions
309 :
310 : inline void
311 8826 : timer_service::shutdown()
312 : {
313 8826 : timer_service_invalidate_cache();
314 8826 : shutting_down_ = true;
315 : // Flag only; the real drain walks the expiry heap below, not
316 : // object_pool's live_ list, so the visit callback is a no-op.
317 8826 : pool_.shutdown([](timer::implementation*) {});
318 :
319 : // Snapshot impls and detach them from the heap so that
320 : // coroutine-owned timer destructors (run by the unwinds the
321 : // abandons below start) cannot re-enter remove_timer_impl() and
322 : // mutate the vector during iteration. A stop requested on another thread
323 : // reaches cancel_waiter() under the same lock: whichever side
324 : // claims a waiter first owns it, and a waiter its canceller
325 : // claimed is posted, so the scheduler's drain abandons it.
326 8826 : std::vector<timer::implementation*> impls;
327 8826 : std::vector<waiter_node*> waiters;
328 : {
329 8826 : std::lock_guard lock(mutex_);
330 8826 : impls.reserve(heap_.size());
331 8826 : waiters.reserve(heap_.size());
332 12068 : for (auto& entry : heap_)
333 : {
334 3242 : entry.timer_->heap_index_.store(
335 : (std::numeric_limits<std::size_t>::max)(),
336 : std::memory_order_relaxed);
337 3242 : impls.push_back(entry.timer_);
338 3242 : if (auto* w = std::exchange(entry.timer_->waiter_, nullptr))
339 : {
340 3242 : w->impl_ = nullptr;
341 3242 : waiters.push_back(w);
342 : }
343 : }
344 8826 : heap_.clear();
345 8826 : cached_nearest_ns_.store(
346 : (std::numeric_limits<std::int64_t>::max)(),
347 : std::memory_order_release);
348 8826 : }
349 :
350 : // A stop on another thread may have claimed its waiter before the
351 : // pass above and still be posting it; the scheduler drains next,
352 : // and must find that post.
353 : for (;;)
354 : {
355 : {
356 8826 : std::lock_guard lock(mutex_);
357 8826 : if (cancels_posting_ == 0)
358 8826 : break;
359 8826 : }
360 MIS 0 : std::this_thread::yield();
361 0 : }
362 :
363 : // Cancel waiting timers. Each waiter called work_started()
364 : // in implementation::wait(); a waiter given up here never
365 : // completes, so its work is finished here to keep the count
366 : // balanced. Resetting the stop callback outside the lock
367 : // waits out a canceller already running, which finds the waiter
368 : // claimed.
369 HIT 12068 : for (auto* w : waiters)
370 : {
371 3242 : w->reset_stop_cb();
372 3242 : auto c = std::exchange(w->cont_, {});
373 3242 : sched_->work_finished();
374 : // Unwinding the frame also ends the node's storage
375 3242 : if (c.h)
376 3242 : capy::abandon(c);
377 : }
378 :
379 12068 : for (auto* impl : impls)
380 : {
381 : // Unlink from the pool's live_ list before the direct delete
382 : // below, or ~object_pool()'s unconditional sweep would delete
383 : // this impl a second time.
384 3242 : [[maybe_unused]] bool const removed = pool_.remove(impl);
385 3242 : BOOST_COROSIO_ASSERT(removed);
386 3242 : delete impl;
387 : }
388 :
389 : // Anything still parked in pool_ (free-listed or live with no
390 : // pending wait at shutdown) is freed by object_pool's own destructor.
391 8826 : }
392 :
393 : inline io_object::implementation*
394 20401 : timer_service::construct()
395 : {
396 20401 : timer::implementation* impl = try_pop_tl_cache(this);
397 20401 : if (impl)
398 : {
399 14744 : impl->svc_ = this;
400 14744 : impl->reuse();
401 14744 : impl->refs_.store(1, std::memory_order_relaxed);
402 : // The TL push that parked impl here also removed it from
403 : // live_ (see destroy_impl) — re-adopt it now that it is live
404 : // again.
405 14744 : pool_.adopt(impl);
406 14744 : return impl;
407 : }
408 :
409 5657 : return pool_.acquire(*this);
410 : }
411 :
412 : inline void
413 20401 : timer_service::destroy(io_object::implementation* p)
414 : {
415 : // During shutdown the drain loop owns every impl and deletes
416 : // them directly. A frame destroyed by that loop can unwind a
417 : // handle whose impl was freed in an earlier iteration (a
418 : // timeout's parent frame owns the timeout timer while
419 : // suspended on the inner delay's timer), so bail out before
420 : // even downcasting the pointer.
421 20401 : if (shutting_down_)
422 3247 : return;
423 17154 : destroy_impl(static_cast<timer::implementation&>(*p));
424 : }
425 :
426 : inline void
427 17154 : timer_service::destroy_impl(timer::implementation& impl)
428 : {
429 : // During shutdown the impl is owned by the shutdown loop.
430 : // Re-entering here (from a coroutine-owned timer destructor
431 : // triggered by h.destroy()) must not modify the heap or
432 : // recycle the impl — shutdown deletes it directly.
433 17154 : if (shutting_down_)
434 MIS 0 : return;
435 :
436 HIT 17154 : cancel_timer(impl);
437 :
438 34308 : if (impl.heap_index_.load(std::memory_order_relaxed) !=
439 17154 : (std::numeric_limits<std::size_t>::max)())
440 : {
441 MIS 0 : std::lock_guard lock(mutex_);
442 0 : remove_timer_impl(impl);
443 0 : refresh_cached_nearest();
444 0 : }
445 :
446 HIT 17154 : if (try_push_tl_cache(&impl))
447 : {
448 : // The TL slot takes sole ownership: it is deleted directly by
449 : // tl_cache_owner / the stale-service path, never through this
450 : // pool, so it must leave live_ now or ~object_pool() would see
451 : // it as abandoned (and, at the wrong moment, double-delete it
452 : // after a later construct() re-adopts it).
453 15248 : [[maybe_unused]] bool const removed = pool_.remove(&impl);
454 15248 : BOOST_COROSIO_ASSERT(removed);
455 15248 : return;
456 : }
457 :
458 : // No op keepalives on a timer impl: this is the only reference,
459 : // so release() always recycles immediately via retire().
460 1906 : release(&impl);
461 : }
462 :
463 : inline void
464 24215 : timer_service::insert_waiter(timer::implementation& impl, waiter_node* w)
465 : {
466 24215 : bool notify = false;
467 24215 : bool lost_cancel = false;
468 : {
469 24215 : std::lock_guard lock(mutex_);
470 : // Grow before publishing anything, so the push_back below
471 : // cannot throw: a failure here leaves the waiter untouched,
472 : // the strong guarantee rearm_wait's recovery relies on.
473 24215 : if (impl.heap_index_.load(std::memory_order_relaxed) ==
474 48430 : (std::numeric_limits<std::size_t>::max)() &&
475 24215 : heap_.size() == heap_.capacity())
476 943 : heap_.reserve(heap_.capacity() == 0 ? 16 : 2 * heap_.capacity());
477 : // Publish: from here the waiter is visible to the fire path and
478 : // to its own stop callback (impl_ non-null enables cancel_waiter).
479 24215 : w->impl_ = &impl;
480 48430 : if (impl.heap_index_.load(std::memory_order_relaxed) ==
481 24215 : (std::numeric_limits<std::size_t>::max)())
482 : {
483 24215 : impl.heap_index_.store(heap_.size(), std::memory_order_relaxed);
484 24215 : heap_.push_back({impl.expiry_, &impl});
485 24215 : up_heap(heap_.size() - 1);
486 24215 : notify = (impl.heap_index_.load(std::memory_order_relaxed) == 0);
487 24215 : refresh_cached_nearest();
488 : }
489 24215 : BOOST_COROSIO_ASSERT(impl.waiter_ == nullptr);
490 24215 : impl.waiter_ = w;
491 :
492 : // Lost-cancel re-check: a stop requested after the canceller was
493 : // armed in wait() but before this publication found impl_ null
494 : // and returned a no-op. Observe it now and undo the insertion.
495 24215 : if (w->token_->stop_requested())
496 : {
497 239 : w->impl_ = nullptr;
498 239 : impl.waiter_ = nullptr;
499 239 : remove_timer_impl(impl);
500 239 : impl.might_have_pending_waits_.store(
501 : false, std::memory_order_release);
502 239 : refresh_cached_nearest();
503 239 : lost_cancel = true;
504 239 : notify = false; // insertion undone; nearest unchanged
505 : }
506 24215 : }
507 24215 : if (notify)
508 8070 : on_earliest_changed_();
509 24215 : if (lost_cancel)
510 : {
511 239 : w->ec_ = make_error_code(capy::error::canceled);
512 239 : sched_->post(&w->op_);
513 : }
514 24215 : }
515 :
516 : inline void
517 17154 : timer_service::cancel_timer(timer::implementation& impl)
518 : {
519 : // Acquire pairs with the drains' release: a caller that skips the
520 : // lock may recycle the impl next, so the drain's writes to it must
521 : // happen before.
522 17154 : if (!impl.might_have_pending_waits_.load(std::memory_order_acquire))
523 17152 : return;
524 :
525 : // No unlocked already-done fast-out here: it would need the
526 : // non-atomic waiter_ (a race with concurrent drains), and an
527 : // index-only check is lifetime-unsafe because npos is stored
528 : // before the drain finishes touching the impl. A stale-true
529 : // flag is rare with the stateless API; the locked path below
530 : // re-validates.
531 :
532 2 : waiter_node* canceled = nullptr;
533 :
534 : {
535 2 : std::lock_guard lock(mutex_);
536 2 : remove_timer_impl(impl);
537 2 : canceled = std::exchange(impl.waiter_, nullptr);
538 2 : if (canceled)
539 2 : canceled->impl_ = nullptr;
540 : // Store false as the final touch of the impl under the lock so
541 : // a pre-lock false-flag check trusts it unqualified.
542 2 : impl.might_have_pending_waits_.store(false, std::memory_order_release);
543 2 : refresh_cached_nearest();
544 2 : }
545 :
546 2 : if (canceled)
547 : {
548 2 : canceled->ec_ = make_error_code(capy::error::canceled);
549 2 : sched_->post(&canceled->op_);
550 : }
551 : }
552 :
553 : inline void
554 2474 : timer_service::cancel_waiter(waiter_node* w)
555 : {
556 : {
557 2474 : std::lock_guard lock(mutex_);
558 : // Already removed by another drain: cancel_timer,
559 : // process_expired, or insert_waiter's lost-cancel recheck
560 2474 : if (!w->impl_)
561 415 : return;
562 2059 : auto* impl = w->impl_;
563 2059 : w->impl_ = nullptr;
564 2059 : impl->waiter_ = nullptr;
565 2059 : remove_timer_impl(*impl);
566 2059 : impl->might_have_pending_waits_.store(false, std::memory_order_release);
567 2059 : refresh_cached_nearest();
568 2059 : ++cancels_posting_;
569 2474 : }
570 :
571 2059 : w->ec_ = make_error_code(capy::error::canceled);
572 2059 : sched_->post(&w->op_);
573 :
574 2059 : std::lock_guard lock(mutex_);
575 2059 : --cancels_posting_;
576 2059 : }
577 :
578 : inline std::size_t
579 218750 : timer_service::process_expired()
580 : {
581 218750 : intrusive_list<waiter_node> expired;
582 :
583 : {
584 218750 : std::lock_guard lock(mutex_);
585 218750 : auto now = clock_type::now();
586 :
587 237423 : while (!heap_.empty() && heap_[0].time_ <= now)
588 : {
589 18673 : timer::implementation* t = heap_[0].timer_;
590 18673 : remove_timer_impl(*t);
591 18673 : if (auto* w = std::exchange(t->waiter_, nullptr))
592 : {
593 18673 : w->impl_ = nullptr;
594 18673 : w->ec_ = {};
595 18673 : expired.push_back(w);
596 : }
597 18673 : t->might_have_pending_waits_.store(
598 : false, std::memory_order_release);
599 : }
600 :
601 218750 : refresh_cached_nearest();
602 218750 : }
603 :
604 218750 : std::size_t count = 0;
605 237423 : while (auto* w = expired.pop_front())
606 : {
607 18673 : sched_->post(&w->op_);
608 18673 : ++count;
609 18673 : }
610 :
611 218750 : return count;
612 : }
613 :
614 : inline void
615 20973 : timer_service::remove_timer_impl(timer::implementation& impl)
616 : {
617 20973 : std::size_t index = impl.heap_index_.load(std::memory_order_relaxed);
618 20973 : if (index >= heap_.size())
619 MIS 0 : return; // Not in heap
620 :
621 HIT 20973 : if (index == heap_.size() - 1)
622 : {
623 : // Last element, just pop
624 3023 : impl.heap_index_.store(
625 : (std::numeric_limits<std::size_t>::max)(),
626 : std::memory_order_relaxed);
627 3023 : heap_.pop_back();
628 : }
629 : else
630 : {
631 : // Swap with last and reheapify
632 17950 : swap_heap(index, heap_.size() - 1);
633 17950 : impl.heap_index_.store(
634 : (std::numeric_limits<std::size_t>::max)(),
635 : std::memory_order_relaxed);
636 17950 : heap_.pop_back();
637 :
638 17950 : if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
639 3 : up_heap(index);
640 : else
641 17947 : down_heap(index);
642 : }
643 : }
644 :
645 : inline void
646 24218 : timer_service::up_heap(std::size_t index)
647 : {
648 32155 : while (index > 0)
649 : {
650 23876 : std::size_t parent = (index - 1) / 2;
651 23876 : if (!(heap_[index].time_ < heap_[parent].time_))
652 15939 : break;
653 7937 : swap_heap(index, parent);
654 7937 : index = parent;
655 : }
656 24218 : }
657 :
658 : inline void
659 17947 : timer_service::down_heap(std::size_t index)
660 : {
661 17947 : std::size_t child = index * 2 + 1;
662 39466 : while (child < heap_.size())
663 : {
664 23776 : std::size_t min_child = (child + 1 == heap_.size() ||
665 19592 : heap_[child].time_ < heap_[child + 1].time_)
666 43368 : ? child
667 23776 : : child + 1;
668 :
669 23776 : if (heap_[index].time_ < heap_[min_child].time_)
670 2257 : break;
671 :
672 21519 : swap_heap(index, min_child);
673 21519 : index = min_child;
674 21519 : child = index * 2 + 1;
675 : }
676 17947 : }
677 :
678 : inline void
679 47406 : timer_service::swap_heap(std::size_t i1, std::size_t i2)
680 : {
681 47406 : heap_entry tmp = heap_[i1];
682 47406 : heap_[i1] = heap_[i2];
683 47406 : heap_[i2] = tmp;
684 47406 : heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed);
685 47406 : heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed);
686 47406 : }
687 :
688 : // waiter_node's completion_op and canceller members are defined in
689 : // timer.cpp alongside implementation::wait(), for the same reason
690 : // wait() lives there (see below).
691 :
692 : // timer::implementation::wait() is defined in timer.cpp, not here.
693 : // It must be a non-inline definition in a translation unit that is
694 : // always pulled into the link whenever detail::timer is used (every
695 : // consumer needs timer's constructors from that same object file).
696 : // An inline definition in this header would only be emitted in
697 : // translation units that happen to also include this header, which
698 : // is not guaranteed for every caller of wait_awaitable::await_suspend
699 : // in timer.hpp (e.g. code that only reaches timer.hpp through
700 : // delay.hpp, without transitively including a scheduler header).
701 :
702 : // Free functions
703 :
704 : inline timer_service&
705 4413 : get_timer_service(capy::execution_context& ctx, scheduler& sched)
706 : {
707 4413 : return ctx.make_service<timer_service>(sched);
708 : }
709 :
710 : } // namespace boost::corosio::detail
711 :
712 : #endif
|
