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 <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::implementation holds per-timer state: expiry, heap
50 : 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 HIT 1225 : callback() = default;
97 :
98 : /// Construct a callback with the given context and function.
99 1225 : 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 7457 : void operator()() const
109 : {
110 7457 : if (fn_)
111 7457 : fn_(ctx_);
112 7457 : }
113 : };
114 :
115 : private:
116 : struct heap_entry
117 : {
118 : time_point time_;
119 : timer::implementation* timer_;
120 : };
121 :
122 : scheduler* sched_ = nullptr;
123 : BOOST_COROSIO_MSVC_WARNING_PUSH
124 : BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std:: members, dll-interface
125 : mutable std::mutex mutex_;
126 : std::vector<heap_entry> heap_;
127 : timer::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 : BOOST_COROSIO_MSVC_WARNING_POP
135 :
136 : public:
137 : /// Construct the timer service bound to a scheduler.
138 1225 : inline timer_service(capy::execution_context&, scheduler& sched)
139 1225 : : sched_(&sched)
140 : {
141 1225 : }
142 :
143 : /// Return the associated scheduler.
144 15176 : inline scheduler& get_scheduler() noexcept
145 : {
146 15176 : return *sched_;
147 : }
148 :
149 : /// Destroy the timer service.
150 2450 : ~timer_service() override = default;
151 :
152 : timer_service(timer_service const&) = delete;
153 : timer_service& operator=(timer_service const&) = delete;
154 :
155 : /// Register a callback invoked when the earliest expiry changes.
156 1225 : inline void set_on_earliest_changed(callback cb)
157 : {
158 1225 : on_earliest_changed_ = cb;
159 1225 : }
160 :
161 : /// Return true if no timers are in the heap.
162 : inline bool empty() const noexcept
163 : {
164 : return cached_nearest_ns_.load(std::memory_order_acquire) ==
165 : (std::numeric_limits<std::int64_t>::max)();
166 : }
167 :
168 : /// Return the nearest timer expiry without acquiring the mutex.
169 225055 : inline time_point nearest_expiry() const noexcept
170 : {
171 225055 : auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
172 225055 : return time_point(time_point::duration(ns));
173 : }
174 :
175 : /// Cancel all pending timers and free cached resources.
176 : inline void shutdown() override;
177 :
178 : /// Construct a new timer implementation.
179 : inline io_object::implementation* construct() override;
180 :
181 : /// Destroy a timer implementation, cancelling pending waiters.
182 : inline void destroy(io_object::implementation* p) override;
183 :
184 : /// Cancel and recycle a timer implementation.
185 : inline void destroy_impl(timer::implementation& impl);
186 :
187 : /// Create or recycle a waiter node.
188 : inline waiter_node* create_waiter();
189 :
190 : /// Return a waiter node to the cache or free list.
191 : inline void destroy_waiter(waiter_node* w);
192 :
193 : /// Update the timer expiry, cancelling existing waiters.
194 : inline std::size_t update_timer(
195 : timer::implementation& impl, time_point new_time);
196 :
197 : /// Insert a waiter into the timer's waiter list and the heap.
198 : inline void insert_waiter(timer::implementation& impl, waiter_node* w);
199 :
200 : /// Cancel all waiters on a timer.
201 : inline std::size_t cancel_timer(timer::implementation& impl);
202 :
203 : /// Cancel a single waiter ( stop_token callback path ).
204 : inline void cancel_waiter(waiter_node* w);
205 :
206 : /// Cancel one waiter on a timer.
207 : inline std::size_t cancel_one_waiter(timer::implementation& impl);
208 :
209 : /// Complete all waiters whose timers have expired.
210 : inline std::size_t process_expired();
211 :
212 : private:
213 250093 : inline void refresh_cached_nearest() noexcept
214 : {
215 250093 : auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
216 246962 : : heap_[0].time_.time_since_epoch().count();
217 250093 : cached_nearest_ns_.store(ns, std::memory_order_release);
218 250093 : }
219 :
220 : inline void remove_timer_impl(timer::implementation& impl);
221 : inline void up_heap(std::size_t index);
222 : inline void down_heap(std::size_t index);
223 : inline void swap_heap(std::size_t i1, std::size_t i2);
224 : };
225 :
226 : struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node
227 : : intrusive_list<waiter_node>::node
228 : {
229 : // Embedded completion op — avoids heap allocation per fire/cancel
230 : struct completion_op final : scheduler_op
231 : {
232 : waiter_node* waiter_ = nullptr;
233 :
234 : static void do_complete(
235 : void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);
236 :
237 1606 : completion_op() noexcept : scheduler_op(&do_complete) {}
238 :
239 : void operator()() override;
240 : void destroy() override;
241 : };
242 :
243 : // Per-waiter stop_token cancellation
244 : struct canceller
245 : {
246 : waiter_node* waiter_;
247 : void operator()() const;
248 : };
249 :
250 : // nullptr once removed from timer's waiter list (concurrency marker)
251 : timer::implementation* impl_ = nullptr;
252 : timer_service* svc_ = nullptr;
253 : std::coroutine_handle<> h_;
254 : capy::continuation* cont_ = nullptr;
255 : capy::executor_ref d_;
256 : std::error_code* ec_out_ = nullptr;
257 : std::stop_token token_;
258 : std::optional<std::stop_callback<canceller>> stop_cb_;
259 : completion_op op_;
260 : std::error_code ec_value_;
261 : waiter_node* next_free_ = nullptr;
262 :
263 1606 : waiter_node() noexcept
264 1606 : {
265 1606 : op_.waiter_ = this;
266 1606 : }
267 : };
268 :
269 : // Thread-local caches avoid hot-path mutex acquisitions:
270 : // 1. Impl cache — single-slot, validated by comparing svc_
271 : // 2. Waiter cache — single-slot, no service affinity
272 : // All caches are cleared by timer_service_invalidate_cache() during shutdown.
273 :
274 : inline thread_local_ptr<timer::implementation> tl_cached_impl;
275 : inline thread_local_ptr<waiter_node> tl_cached_waiter;
276 :
277 : // The POD TLS slots above never run destructors, so a short-lived
278 : // run() thread would leak its cached impl and waiter. Each push
279 : // arms this owner, whose destructor frees the slots at thread exit.
280 : // Cached entries are quiescent heap objects (stop callbacks reset,
281 : // nothing in the heap or free lists) and deletion touches no
282 : // service state, so it is safe after the owning service is gone
283 : // (the stale-entry path in try_pop_tl_cache deletes the same way).
284 : struct tl_cache_owner
285 : {
286 40 : ~tl_cache_owner()
287 : {
288 40 : delete tl_cached_impl.get();
289 40 : tl_cached_impl.set(nullptr);
290 :
291 40 : delete tl_cached_waiter.get();
292 40 : tl_cached_waiter.set(nullptr);
293 40 : }
294 : };
295 :
296 : inline void
297 14602 : arm_tl_cache_cleanup() noexcept
298 : {
299 14602 : thread_local tl_cache_owner owner;
300 : (void)owner;
301 14602 : }
302 :
303 : inline timer::implementation*
304 8457 : try_pop_tl_cache(timer_service* svc) noexcept
305 : {
306 8457 : auto* impl = tl_cached_impl.get();
307 8457 : if (impl)
308 : {
309 8164 : tl_cached_impl.set(nullptr);
310 8164 : if (impl->svc_ == svc)
311 8164 : return impl;
312 : // Stale impl from a destroyed service
313 MIS 0 : delete impl;
314 : }
315 HIT 293 : return nullptr;
316 : }
317 :
318 : inline bool
319 8431 : try_push_tl_cache(timer::implementation* impl) noexcept
320 : {
321 8431 : if (!tl_cached_impl.get())
322 : {
323 8387 : arm_tl_cache_cleanup();
324 8387 : tl_cached_impl.set(impl);
325 8387 : return true;
326 : }
327 44 : return false;
328 : }
329 :
330 : inline waiter_node*
331 7601 : try_pop_waiter_tl_cache() noexcept
332 : {
333 7601 : auto* w = tl_cached_waiter.get();
334 7601 : if (w)
335 : {
336 5994 : tl_cached_waiter.set(nullptr);
337 5994 : return w;
338 : }
339 1607 : return nullptr;
340 : }
341 :
342 : inline bool
343 7575 : try_push_waiter_tl_cache(waiter_node* w) noexcept
344 : {
345 7575 : if (!tl_cached_waiter.get())
346 : {
347 6215 : arm_tl_cache_cleanup();
348 6215 : tl_cached_waiter.set(w);
349 6215 : return true;
350 : }
351 1360 : return false;
352 : }
353 :
354 : inline void
355 1225 : timer_service_invalidate_cache() noexcept
356 : {
357 1225 : delete tl_cached_impl.get();
358 1225 : tl_cached_impl.set(nullptr);
359 :
360 1225 : delete tl_cached_waiter.get();
361 1225 : tl_cached_waiter.set(nullptr);
362 1225 : }
363 :
364 : // timer_service out-of-class member function definitions
365 :
366 : inline void
367 1225 : timer_service::shutdown()
368 : {
369 1225 : timer_service_invalidate_cache();
370 1225 : shutting_down_ = true;
371 :
372 : // Snapshot impls and detach them from the heap so that
373 : // coroutine-owned timer destructors (triggered by h.destroy()
374 : // below) cannot re-enter remove_timer_impl() and mutate the
375 : // vector during iteration.
376 1225 : std::vector<timer::implementation*> impls;
377 1225 : impls.reserve(heap_.size());
378 1251 : for (auto& entry : heap_)
379 : {
380 26 : entry.timer_->heap_index_.store(
381 : (std::numeric_limits<std::size_t>::max)(),
382 : std::memory_order_relaxed);
383 26 : impls.push_back(entry.timer_);
384 : }
385 1225 : heap_.clear();
386 1225 : cached_nearest_ns_.store(
387 : (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);
388 :
389 : // Cancel waiting timers. Each waiter called work_started()
390 : // in implementation::wait(). On IOCP the scheduler shutdown
391 : // loop exits when outstanding_work_ reaches zero, so we must
392 : // call work_finished() here to balance it. On other backends
393 : // this is harmless.
394 1251 : for (auto* impl : impls)
395 : {
396 52 : while (auto* w = impl->waiters_.pop_front())
397 : {
398 26 : w->stop_cb_.reset();
399 26 : auto h = std::exchange(w->h_, {});
400 26 : sched_->work_finished();
401 26 : if (h)
402 26 : h.destroy();
403 26 : delete w;
404 26 : }
405 26 : delete impl;
406 : }
407 :
408 : // Delete free-listed impls
409 1269 : while (free_list_)
410 : {
411 44 : auto* next = free_list_->next_free_;
412 44 : delete free_list_;
413 44 : free_list_ = next;
414 : }
415 :
416 : // Delete free-listed waiters
417 2584 : while (waiter_free_list_)
418 : {
419 1359 : auto* next = waiter_free_list_->next_free_;
420 1359 : delete waiter_free_list_;
421 1359 : waiter_free_list_ = next;
422 : }
423 1225 : }
424 :
425 : inline io_object::implementation*
426 8457 : timer_service::construct()
427 : {
428 8457 : timer::implementation* impl = try_pop_tl_cache(this);
429 8457 : if (impl)
430 : {
431 8164 : impl->svc_ = this;
432 8164 : impl->heap_index_.store(
433 : (std::numeric_limits<std::size_t>::max)(),
434 : std::memory_order_relaxed);
435 8164 : impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
436 8164 : return impl;
437 : }
438 :
439 293 : std::lock_guard lock(mutex_);
440 293 : if (free_list_)
441 : {
442 MIS 0 : impl = free_list_;
443 0 : free_list_ = impl->next_free_;
444 0 : impl->next_free_ = nullptr;
445 0 : impl->svc_ = this;
446 0 : impl->heap_index_.store(
447 : (std::numeric_limits<std::size_t>::max)(),
448 : std::memory_order_relaxed);
449 0 : impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
450 : }
451 : else
452 : {
453 HIT 293 : impl = new timer::implementation(*this);
454 : }
455 293 : return impl;
456 293 : }
457 :
458 : inline void
459 8457 : timer_service::destroy(io_object::implementation* p)
460 : {
461 : // During shutdown the drain loop owns every impl and deletes
462 : // them directly. A frame destroyed by that loop can unwind a
463 : // handle whose impl was freed in an earlier iteration (a
464 : // timeout's parent frame owns the timeout timer while
465 : // suspended on the inner delay's timer), so bail out before
466 : // even downcasting the pointer.
467 8457 : if (shutting_down_)
468 26 : return;
469 8431 : destroy_impl(static_cast<timer::implementation&>(*p));
470 : }
471 :
472 : inline void
473 8431 : timer_service::destroy_impl(timer::implementation& impl)
474 : {
475 : // During shutdown the impl is owned by the shutdown loop.
476 : // Re-entering here (from a coroutine-owned timer destructor
477 : // triggered by h.destroy()) must not modify the heap or
478 : // recycle the impl — shutdown deletes it directly.
479 8431 : if (shutting_down_)
480 8387 : return;
481 :
482 8431 : cancel_timer(impl);
483 :
484 16862 : if (impl.heap_index_.load(std::memory_order_relaxed) !=
485 8431 : (std::numeric_limits<std::size_t>::max)())
486 : {
487 MIS 0 : std::lock_guard lock(mutex_);
488 0 : remove_timer_impl(impl);
489 0 : refresh_cached_nearest();
490 0 : }
491 :
492 HIT 8431 : if (try_push_tl_cache(&impl))
493 8387 : return;
494 :
495 44 : std::lock_guard lock(mutex_);
496 44 : impl.next_free_ = free_list_;
497 44 : free_list_ = &impl;
498 44 : }
499 :
500 : inline waiter_node*
501 7601 : timer_service::create_waiter()
502 : {
503 7601 : if (auto* w = try_pop_waiter_tl_cache())
504 5994 : return w;
505 :
506 1607 : std::lock_guard lock(mutex_);
507 1607 : if (waiter_free_list_)
508 : {
509 1 : auto* w = waiter_free_list_;
510 1 : waiter_free_list_ = w->next_free_;
511 1 : w->next_free_ = nullptr;
512 1 : return w;
513 : }
514 :
515 1606 : return new waiter_node();
516 1607 : }
517 :
518 : inline void
519 7575 : timer_service::destroy_waiter(waiter_node* w)
520 : {
521 7575 : if (try_push_waiter_tl_cache(w))
522 6215 : return;
523 :
524 1360 : std::lock_guard lock(mutex_);
525 1360 : w->next_free_ = waiter_free_list_;
526 1360 : waiter_free_list_ = w;
527 1360 : }
528 :
529 : inline std::size_t
530 MIS 0 : timer_service::update_timer(timer::implementation& impl, time_point new_time)
531 : {
532 : // Gate on the flag, not waiters_: reading the non-atomic list
533 : // here would race a concurrent drain. A false flag is safe to
534 : // trust pre-lock because every draining critical section stores
535 : // it false as its final touch of the impl.
536 : bool in_heap =
537 0 : (impl.heap_index_.load(std::memory_order_relaxed) !=
538 0 : (std::numeric_limits<std::size_t>::max)());
539 0 : if (!in_heap &&
540 0 : !impl.might_have_pending_waits_.load(std::memory_order_relaxed))
541 0 : return 0;
542 :
543 0 : bool notify = false;
544 0 : intrusive_list<waiter_node> canceled;
545 :
546 : {
547 0 : std::lock_guard lock(mutex_);
548 :
549 0 : while (auto* w = impl.waiters_.pop_front())
550 : {
551 0 : w->impl_ = nullptr;
552 0 : canceled.push_back(w);
553 0 : }
554 :
555 0 : std::size_t idx = impl.heap_index_.load(std::memory_order_relaxed);
556 0 : if (idx < heap_.size())
557 : {
558 0 : time_point old_time = heap_[idx].time_;
559 0 : heap_[idx].time_ = new_time;
560 :
561 0 : if (new_time < old_time)
562 0 : up_heap(idx);
563 : else
564 0 : down_heap(idx);
565 :
566 0 : notify =
567 0 : (impl.heap_index_.load(std::memory_order_relaxed) == 0);
568 : }
569 :
570 0 : refresh_cached_nearest();
571 0 : }
572 :
573 0 : std::size_t count = 0;
574 0 : while (auto* w = canceled.pop_front())
575 : {
576 0 : w->ec_value_ = make_error_code(capy::error::canceled);
577 0 : sched_->post(&w->op_);
578 0 : ++count;
579 0 : }
580 :
581 0 : if (notify)
582 0 : on_earliest_changed_();
583 :
584 0 : return count;
585 : }
586 :
587 : inline void
588 HIT 7601 : timer_service::insert_waiter(timer::implementation& impl, waiter_node* w)
589 : {
590 7601 : bool notify = false;
591 7601 : bool lost_cancel = false;
592 : {
593 7601 : std::lock_guard lock(mutex_);
594 : // Publish: from here the waiter is visible to the fire path and
595 : // to its own stop callback (impl_ non-null enables cancel_waiter).
596 7601 : w->impl_ = &impl;
597 15202 : if (impl.heap_index_.load(std::memory_order_relaxed) ==
598 7601 : (std::numeric_limits<std::size_t>::max)())
599 : {
600 7601 : impl.heap_index_.store(heap_.size(), std::memory_order_relaxed);
601 7601 : heap_.push_back({impl.expiry_, &impl});
602 7601 : up_heap(heap_.size() - 1);
603 7601 : notify =
604 7601 : (impl.heap_index_.load(std::memory_order_relaxed) == 0);
605 7601 : refresh_cached_nearest();
606 : }
607 7601 : impl.waiters_.push_back(w);
608 :
609 : // Lost-cancel re-check: a stop requested after the canceller was
610 : // armed in wait() but before this publication found impl_ null
611 : // and returned a no-op. Observe it now and undo the insertion.
612 7601 : if (w->token_.stop_requested())
613 : {
614 3 : w->impl_ = nullptr;
615 3 : impl.waiters_.remove(w);
616 3 : if (impl.waiters_.empty())
617 : {
618 3 : remove_timer_impl(impl);
619 3 : impl.might_have_pending_waits_.store(
620 : false, std::memory_order_relaxed);
621 : }
622 3 : refresh_cached_nearest();
623 3 : lost_cancel = true;
624 3 : notify = false; // insertion undone; nearest unchanged
625 : }
626 7601 : }
627 7601 : if (notify)
628 7457 : on_earliest_changed_();
629 7601 : if (lost_cancel)
630 : {
631 3 : w->ec_value_ = make_error_code(capy::error::canceled);
632 3 : sched_->post(&w->op_);
633 : }
634 7601 : }
635 :
636 : inline std::size_t
637 8431 : timer_service::cancel_timer(timer::implementation& impl)
638 : {
639 8431 : if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
640 8429 : return 0;
641 :
642 : // No unlocked already-done fast-out here: it would need the
643 : // non-atomic waiters_ (a race with concurrent drains), and an
644 : // index-only check is lifetime-unsafe because npos is stored
645 : // before the drain finishes touching the impl. A stale-true
646 : // flag is rare with the stateless API; the locked path below
647 : // re-validates.
648 :
649 2 : intrusive_list<waiter_node> canceled;
650 :
651 : {
652 2 : std::lock_guard lock(mutex_);
653 2 : remove_timer_impl(impl);
654 4 : while (auto* w = impl.waiters_.pop_front())
655 : {
656 2 : w->impl_ = nullptr;
657 2 : canceled.push_back(w);
658 2 : }
659 : // Store false as the final touch of the impl under the lock so
660 : // update_timer's pre-lock false-flag trust holds unqualified.
661 2 : impl.might_have_pending_waits_.store(false, std::memory_order_relaxed);
662 2 : refresh_cached_nearest();
663 2 : }
664 :
665 2 : std::size_t count = 0;
666 4 : while (auto* w = canceled.pop_front())
667 : {
668 2 : w->ec_value_ = make_error_code(capy::error::canceled);
669 2 : sched_->post(&w->op_);
670 2 : ++count;
671 2 : }
672 :
673 2 : return count;
674 : }
675 :
676 : inline void
677 1387 : timer_service::cancel_waiter(waiter_node* w)
678 : {
679 : {
680 1387 : std::lock_guard lock(mutex_);
681 : // Already removed by cancel_timer or process_expired
682 1387 : if (!w->impl_)
683 4 : return;
684 1383 : auto* impl = w->impl_;
685 1383 : w->impl_ = nullptr;
686 1383 : impl->waiters_.remove(w);
687 1383 : if (impl->waiters_.empty())
688 : {
689 1383 : remove_timer_impl(*impl);
690 1383 : impl->might_have_pending_waits_.store(
691 : false, std::memory_order_relaxed);
692 : }
693 1383 : refresh_cached_nearest();
694 1387 : }
695 :
696 1383 : w->ec_value_ = make_error_code(capy::error::canceled);
697 1383 : sched_->post(&w->op_);
698 : }
699 :
700 : inline std::size_t
701 MIS 0 : timer_service::cancel_one_waiter(timer::implementation& impl)
702 : {
703 0 : if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
704 0 : return 0;
705 :
706 0 : waiter_node* w = nullptr;
707 :
708 : {
709 0 : std::lock_guard lock(mutex_);
710 0 : w = impl.waiters_.pop_front();
711 0 : if (!w)
712 0 : return 0;
713 0 : w->impl_ = nullptr;
714 0 : if (impl.waiters_.empty())
715 : {
716 0 : remove_timer_impl(impl);
717 0 : impl.might_have_pending_waits_.store(
718 : false, std::memory_order_relaxed);
719 : }
720 0 : refresh_cached_nearest();
721 0 : }
722 :
723 0 : w->ec_value_ = make_error_code(capy::error::canceled);
724 0 : sched_->post(&w->op_);
725 0 : return 1;
726 : }
727 :
728 : inline std::size_t
729 HIT 241104 : timer_service::process_expired()
730 : {
731 241104 : intrusive_list<waiter_node> expired;
732 :
733 : {
734 241104 : std::lock_guard lock(mutex_);
735 241104 : auto now = clock_type::now();
736 :
737 247291 : while (!heap_.empty() && heap_[0].time_ <= now)
738 : {
739 6187 : timer::implementation* t = heap_[0].timer_;
740 6187 : remove_timer_impl(*t);
741 12374 : while (auto* w = t->waiters_.pop_front())
742 : {
743 6187 : w->impl_ = nullptr;
744 6187 : w->ec_value_ = {};
745 6187 : expired.push_back(w);
746 6187 : }
747 6187 : t->might_have_pending_waits_.store(
748 : false, std::memory_order_relaxed);
749 : }
750 :
751 241104 : refresh_cached_nearest();
752 241104 : }
753 :
754 241104 : std::size_t count = 0;
755 247291 : while (auto* w = expired.pop_front())
756 : {
757 6187 : sched_->post(&w->op_);
758 6187 : ++count;
759 6187 : }
760 :
761 241104 : return count;
762 : }
763 :
764 : inline void
765 7575 : timer_service::remove_timer_impl(timer::implementation& impl)
766 : {
767 7575 : std::size_t index = impl.heap_index_.load(std::memory_order_relaxed);
768 7575 : if (index >= heap_.size())
769 MIS 0 : return; // Not in heap
770 :
771 HIT 7575 : if (index == heap_.size() - 1)
772 : {
773 : // Last element, just pop
774 1609 : impl.heap_index_.store(
775 : (std::numeric_limits<std::size_t>::max)(),
776 : std::memory_order_relaxed);
777 1609 : heap_.pop_back();
778 : }
779 : else
780 : {
781 : // Swap with last and reheapify
782 5966 : swap_heap(index, heap_.size() - 1);
783 5966 : impl.heap_index_.store(
784 : (std::numeric_limits<std::size_t>::max)(),
785 : std::memory_order_relaxed);
786 5966 : heap_.pop_back();
787 :
788 5966 : if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
789 MIS 0 : up_heap(index);
790 : else
791 HIT 5966 : down_heap(index);
792 : }
793 : }
794 :
795 : inline void
796 7601 : timer_service::up_heap(std::size_t index)
797 : {
798 13544 : while (index > 0)
799 : {
800 6084 : std::size_t parent = (index - 1) / 2;
801 6084 : if (!(heap_[index].time_ < heap_[parent].time_))
802 141 : break;
803 5943 : swap_heap(index, parent);
804 5943 : index = parent;
805 : }
806 7601 : }
807 :
808 : inline void
809 5966 : timer_service::down_heap(std::size_t index)
810 : {
811 5966 : std::size_t child = index * 2 + 1;
812 5968 : while (child < heap_.size())
813 : {
814 4 : std::size_t min_child = (child + 1 == heap_.size() ||
815 MIS 0 : heap_[child].time_ < heap_[child + 1].time_)
816 HIT 4 : ? child
817 4 : : child + 1;
818 :
819 4 : if (heap_[index].time_ < heap_[min_child].time_)
820 2 : break;
821 :
822 2 : swap_heap(index, min_child);
823 2 : index = min_child;
824 2 : child = index * 2 + 1;
825 : }
826 5966 : }
827 :
828 : inline void
829 11911 : timer_service::swap_heap(std::size_t i1, std::size_t i2)
830 : {
831 11911 : heap_entry tmp = heap_[i1];
832 11911 : heap_[i1] = heap_[i2];
833 11911 : heap_[i2] = tmp;
834 11911 : heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed);
835 11911 : heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed);
836 11911 : }
837 :
838 : // waiter_node out-of-class member function definitions
839 :
840 : inline void
841 1387 : waiter_node::canceller::operator()() const
842 : {
843 1387 : waiter_->svc_->cancel_waiter(waiter_);
844 1387 : }
845 :
846 : inline void
847 MIS 0 : waiter_node::completion_op::do_complete(
848 : [[maybe_unused]] void* owner,
849 : scheduler_op* base,
850 : std::uint32_t,
851 : std::uint32_t)
852 : {
853 : // owner is always non-null here. The destroy path (owner == nullptr)
854 : // is unreachable because completion_op overrides destroy() directly,
855 : // bypassing scheduler_op::destroy() which would call func_(nullptr, ...).
856 0 : BOOST_COROSIO_ASSERT(owner);
857 0 : static_cast<completion_op*>(base)->operator()();
858 0 : }
859 :
860 : inline void
861 HIT 7575 : waiter_node::completion_op::operator()()
862 : {
863 7575 : auto* w = waiter_;
864 7575 : w->stop_cb_.reset();
865 7575 : if (w->ec_out_)
866 7575 : *w->ec_out_ = w->ec_value_;
867 :
868 7575 : auto* cont = w->cont_;
869 7575 : auto d = w->d_;
870 7575 : auto* svc = w->svc_;
871 7575 : auto& sched = svc->get_scheduler();
872 :
873 7575 : svc->destroy_waiter(w);
874 :
875 7575 : d.post(*cont);
876 7575 : sched.work_finished();
877 7575 : }
878 :
879 : // GCC 14 false-positive: inlining ~optional<stop_callback> through
880 : // delete loses track that stop_cb_ was already .reset() above.
881 : #if defined(__GNUC__) && !defined(__clang__)
882 : #pragma GCC diagnostic push
883 : #pragma GCC diagnostic ignored "-Wmaybe-uninitialized"
884 : #endif
885 : inline void
886 MIS 0 : waiter_node::completion_op::destroy()
887 : {
888 : // Called during scheduler shutdown drain when this completion_op is
889 : // in the scheduler's ready queue (posted by cancel_timer() or
890 : // process_expired()). Balances the work_started() from
891 : // implementation::wait(). The scheduler drain loop separately
892 : // balances the work_started() from post(). On IOCP both decrements
893 : // are required for outstanding_work_ to reach zero; on other
894 : // backends this is harmless.
895 : //
896 : // This override also prevents scheduler_op::destroy() from calling
897 : // do_complete(nullptr, ...). See also: timer_service::shutdown()
898 : // which drains waiters still in the timer heap (the other path).
899 0 : auto* w = waiter_;
900 0 : w->stop_cb_.reset();
901 0 : auto h = std::exchange(w->h_, {});
902 0 : auto& sched = w->svc_->get_scheduler();
903 0 : delete w;
904 0 : sched.work_finished();
905 0 : if (h)
906 0 : h.destroy();
907 0 : }
908 : #if defined(__GNUC__) && !defined(__clang__)
909 : #pragma GCC diagnostic pop
910 : #endif
911 :
912 : // timer::implementation::wait() is defined in timer.cpp, not here.
913 : // It must be a non-inline definition in a translation unit that is
914 : // always pulled into the link whenever detail::timer is used (every
915 : // consumer needs timer's constructors from that same object file).
916 : // An inline definition in this header would only be emitted in
917 : // translation units that happen to also include this header, which
918 : // is not guaranteed for every caller of wait_awaitable::await_suspend
919 : // in timer.hpp (e.g. code that only reaches timer.hpp through
920 : // delay.hpp, without transitively including a scheduler header).
921 :
922 : // Free functions
923 :
924 : inline std::size_t
925 0 : timer_service_update_expiry(timer::implementation& impl)
926 : {
927 0 : return impl.svc_->update_timer(impl, impl.expiry_);
928 : }
929 :
930 : inline std::size_t
931 0 : timer_service_cancel(timer::implementation& impl) noexcept
932 : {
933 0 : return impl.svc_->cancel_timer(impl);
934 : }
935 :
936 : inline std::size_t
937 0 : timer_service_cancel_one(timer::implementation& impl) noexcept
938 : {
939 0 : return impl.svc_->cancel_one_waiter(impl);
940 : }
941 :
942 : inline timer_service&
943 HIT 1225 : get_timer_service(capy::execution_context& ctx, scheduler& sched)
944 : {
945 1225 : return ctx.make_service<timer_service>(sched);
946 : }
947 :
948 : } // namespace boost::corosio::detail
949 :
950 : #endif
|
