include/boost/corosio/detail/timer_service.hpp

91.8% Lines (345/376) 97.8% List of functions (44/45)

[ ] prev/next file n p prev/next uncovered line

timer_service.hpp

f(x) Functions (45)

Function Calls Lines Blocks

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