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