include/boost/corosio/detail/timer_service.hpp

72.3% Lines (214/296) 84.4% List of functions (27/32) 61.5% Branches (91/148)

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

timer_service.hpp

f(x) Functions (32)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2025 Vinnie Falco ([email protected])
3			// Copyright (c) 2026 Steve Gerbino
4			//
5			// Distributed under the Boost Software License, Version 1.0. (See accompanying
6			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7			//
8			// Official repository: https://github.com/cppalliance/corosio
9			//
10
11			#ifndef BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP
12			#define BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP
13
14			#include <boost/corosio/detail/timer.hpp>
15			#include <boost/corosio/detail/scheduler.hpp>
16			#include <boost/corosio/detail/scheduler_op.hpp>
17			#include <boost/corosio/detail/intrusive.hpp>
18			#include <boost/corosio/detail/thread_local_ptr.hpp>
19			#include <boost/capy/error.hpp>
20			#include <boost/capy/ex/execution_context.hpp>
21			#include <boost/capy/ex/executor_ref.hpp>
22			#include <system_error>
23
24			#include <atomic>
25			#include <chrono>
26			#include <coroutine>
27			#include <cstddef>
28			#include <limits>
29			#include <mutex>
30			#include <stop_token>
31			#include <utility>
32			#include <vector>
33
34			namespace boost::corosio::detail {
35
36			struct scheduler;
37
38			/*
39			Timer Service
40			=============
41
42			Data Structures
43			---------------
44			waiter_node (defined in timer.hpp) holds per-waiter state:
45			coroutine handle, executor, error output, embedded
46			completion_op. Each concurrent co_await t.wait() embeds one
47			waiter_node in the awaitable on the suspended coroutine's
48			frame — waits perform no allocation.
49
50			timer::implementation holds per-timer state: expiry, heap
51			index, and an intrusive_list of waiter_nodes. Multiple
52			coroutines can wait on the same timer simultaneously.
53
54			timer_service owns a min-heap of active timers and a free list
55			of recycled impls. The heap is ordered by expiry time; the
56			scheduler queries nearest_expiry() to set the epoll/timerfd
57			timeout.
58
59			Optimization Strategy
60			---------------------
61			1. Deferred heap insertion — expires_after() stores the expiry
62			but does not insert into the heap. Insertion happens in wait().
63			2. Thread-local impl cache — single-slot per-thread cache.
64			3. Frame-resident waiter_node with embedded completion_op —
65			eliminates heap allocation per wait/fire/cancel.
66			4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().
67			5. might_have_pending_waits_ flag — skips lock when no wait issued.
68
69			Concurrency
70			-----------
71			stop_token callbacks can fire from any thread. The impl_
72			pointer on waiter_node is used as a "still in list" marker.
73			A waiter_node's storage is the suspended coroutine's frame:
74			every completion path must finish touching the node before
75			posting the continuation or destroying the handle.
76			*/
77
78			inline void timer_service_invalidate_cache() noexcept;
79
80			// timer_service class body — member function definitions are
81			// out-of-class (after implementation and waiter_node are complete)
82			class BOOST_COROSIO_DECL timer_service final
83			: public capy::execution_context::service
84			, public io_object::io_service
85			{
86			public:
87			using clock_type = std::chrono::steady_clock;
88			using time_point = clock_type::time_point;
89
90			/// Type-erased callback for earliest-expiry-changed notifications.
91			class callback
92			{
93			void* ctx_ = nullptr;
94			void (fn_)(void) = nullptr;
95
96			public:
97			/// Construct an empty callback.
98		615x	callback() = default;
99
100			/// Construct a callback with the given context and function.
101		615x	callback(void* ctx, void (fn)(void)) noexcept : ctx_(ctx), fn_(fn) {}
102
103			/// Return true if the callback is non-empty.
104			explicit operator bool() const noexcept
105			{
106			return fn_ != nullptr;
107			}
108
109			/// Invoke the callback.
110		1808x	void operator()() const
111			{
112	1/2 ✓ Branch 2 → 3 taken 1808 times. ✗ Branch 2 → 4 not taken.	1808x	if (fn_)
113		1808x	fn_(ctx_);
114		1808x	}
115			};
116
117			private:
118			struct heap_entry
119			{
120			time_point time_;
121			timer::implementation* timer_;
122			};
123
124			scheduler* sched_ = nullptr;
125			BOOST_COROSIO_MSVC_WARNING_PUSH
126			BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std:: members, dll-interface
127			mutable std::mutex mutex_;
128			std::vector<heap_entry> heap_;
129			timer::implementation* free_list_ = nullptr;
130			callback on_earliest_changed_;
131			bool shutting_down_ = false;
132			// Avoids mutex in nearest_expiry() and empty()
133			mutable std::atomic<std::int64_t> cached_nearest_ns_{
134			(std::numeric_limits<std::int64_t>::max)()};
135			BOOST_COROSIO_MSVC_WARNING_POP
136
137			public:
138			/// Construct the timer service bound to a scheduler.
139		615x	inline timer_service(capy::execution_context&, scheduler& sched)
140		615x	: sched_(&sched)
141			{
142		615x	}
143
144			/// Return the associated scheduler.
145		3725x	inline scheduler& get_scheduler() noexcept
146			{
147		3725x	return *sched_;
148			}
149
150			/// Destroy the timer service.
151		1230x	~timer_service() override = default;
152
153			timer_service(timer_service const&) = delete;
154			timer_service& operator=(timer_service const&) = delete;
155
156			/// Register a callback invoked when the earliest expiry changes.
157		615x	inline void set_on_earliest_changed(callback cb)
158			{
159		615x	on_earliest_changed_ = cb;
160		615x	}
161
162			/// Return true if no timers are in the heap.
163			inline bool empty() const noexcept
164			{
165			return cached_nearest_ns_.load(std::memory_order_acquire) ==
166			(std::numeric_limits<std::int64_t>::max)();
167			}
168
169			/// Return the nearest timer expiry without acquiring the mutex.
170		2254x	inline time_point nearest_expiry() const noexcept
171			{
172		2254x	auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
173		2254x	return time_point(time_point::duration(ns));
174			}
175
176			/// Cancel all pending timers and free cached resources.
177			inline void shutdown() override;
178
179			/// Construct a new timer implementation.
180			inline io_object::implementation* construct() override;
181
182			/// Destroy a timer implementation, cancelling pending waiters.
183			inline void destroy(io_object::implementation* p) override;
184
185			/// Cancel and recycle a timer implementation.
186			inline void destroy_impl(timer::implementation& impl);
187
188			/// Update the timer expiry, cancelling existing waiters.
189			inline std::size_t update_timer(
190			timer::implementation& impl, time_point new_time);
191
192			/// Insert a waiter into the timer's waiter list and the heap.
193			inline void insert_waiter(timer::implementation& impl, waiter_node* w);
194
195			/// Cancel all waiters on a timer.
196			inline std::size_t cancel_timer(timer::implementation& impl);
197
198			/// Cancel one specific waiter ( stop_token callback path ).
199			inline void cancel_waiter(waiter_node* w);
200
201			/// Cancel the oldest pending waiter on a timer ( FIFO ).
202			inline std::size_t cancel_one_waiter(timer::implementation& impl);
203
204			/// Complete all waiters whose timers have expired.
205			inline std::size_t process_expired();
206
207			private:
208		3732x	inline void refresh_cached_nearest() noexcept
209			{
210	2/2 ✓ Branch 3 → 4 taken 1455 times. ✓ Branch 3 → 5 taken 2277 times.	3732x	auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
211		2277x	: heap_[0].time_.time_since_epoch().count();
212		3732x	cached_nearest_ns_.store(ns, std::memory_order_release);
213		3732x	}
214
215			inline void remove_timer_impl(timer::implementation& impl);
216			inline void up_heap(std::size_t index);
217			inline void down_heap(std::size_t index);
218			inline void swap_heap(std::size_t i1, std::size_t i2);
219			};
220
221			// Thread-local cache avoids hot-path mutex acquisitions:
222			// single-slot impl cache, validated by comparing svc_. Cleared by
223			// timer_service_invalidate_cache() during shutdown.
224
225			inline thread_local_ptr<timer::implementation> tl_cached_impl;
226
227			// The POD TLS slot above never runs destructors, so a short-lived
228			// run() thread would leak its cached impl. Each push arms this
229			// owner, whose destructor frees the slot at thread exit. A cached
230			// entry is a quiescent heap object (nothing in the heap or free
231			// list) and deletion touches no service state, so it is safe after
232			// the owning service is gone (the stale-entry path in
233			// try_pop_tl_cache deletes the same way).
234			struct tl_cache_owner
235			{
236		30x	~tl_cache_owner()
237			{
238	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 5 taken 30 times.	30x	delete tl_cached_impl.get();
239		30x	tl_cached_impl.set(nullptr);
240		30x	}
241			};
242
243			inline void
244		3144x	arm_tl_cache_cleanup() noexcept
245			{
246	2/2 ✓ Branch 2 → 3 taken 30 times. ✓ Branch 2 → 4 taken 3114 times.	3144x	thread_local tl_cache_owner owner;
247			(void)owner;
248		3144x	}
249
250			inline timer::implementation*
251		3182x	try_pop_tl_cache(timer_service* svc) noexcept
252			{
253		3182x	auto* impl = tl_cached_impl.get();
254	2/2 ✓ Branch 3 → 4 taken 3028 times. ✓ Branch 3 → 9 taken 154 times.	3182x	if (impl)
255			{
256		3028x	tl_cached_impl.set(nullptr);
257	1/2 ✓ Branch 5 → 6 taken 3028 times. ✗ Branch 5 → 7 not taken.	3028x	if (impl->svc_ == svc)
258		3028x	return impl;
259			// Stale impl from a destroyed service
260		✗	delete impl;
261			}
262		154x	return nullptr;
263			}
264
265			inline bool
266		3168x	try_push_tl_cache(timer::implementation* impl) noexcept
267			{
268	2/2 ✓ Branch 3 → 4 taken 3144 times. ✓ Branch 3 → 7 taken 24 times.	3168x	if (!tl_cached_impl.get())
269			{
270		3144x	arm_tl_cache_cleanup();
271		3144x	tl_cached_impl.set(impl);
272		3144x	return true;
273			}
274		24x	return false;
275			}
276
277			inline void
278		1230x	timer_service_invalidate_cache() noexcept
279			{
280	2/2 ✓ Branch 3 → 4 taken 106 times. ✓ Branch 3 → 5 taken 1124 times.	1230x	delete tl_cached_impl.get();
281		1230x	tl_cached_impl.set(nullptr);
282		1230x	}
283
284			// timer_service out-of-class member function definitions
285
286			inline void
287		1230x	timer_service::shutdown()
288			{
289		1230x	timer_service_invalidate_cache();
290		1230x	shutting_down_ = true;
291
292			// Snapshot impls and detach them from the heap so that
293			// coroutine-owned timer destructors (triggered by h.destroy()
294			// below) cannot re-enter remove_timer_impl() and mutate the
295			// vector during iteration.
296		1230x	std::vector<timer::implementation*> impls;
297	1/1 ✓ Branch 4 → 5 taken 1230 times.	1230x	impls.reserve(heap_.size());
298	2/2 ✓ Branch 32 → 7 taken 13 times. ✓ Branch 32 → 33 taken 1230 times.	1243x	for (auto& entry : heap_)
299			{
300		13x	entry.timer_->heap_index_.store(
301			(std::numeric_limits<std::size_t>::max)(),
302			std::memory_order_relaxed);
303	1/1 ✓ Branch 29 → 30 taken 13 times.	13x	impls.push_back(entry.timer_);
304			}
305		1230x	heap_.clear();
306		1230x	cached_nearest_ns_.store(
307			(std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);
308
309			// Cancel waiting timers. Each waiter called work_started()
310			// in implementation::wait(). On IOCP the scheduler shutdown
311			// loop exits when outstanding_work_ reaches zero, so we must
312			// call work_finished() here to balance it. On other backends
313			// this is harmless.
314	2/2 ✓ Branch 73 → 57 taken 13 times. ✓ Branch 73 → 74 taken 1230 times.	1243x	for (auto* impl : impls)
315			{
316	2/2 ✓ Branch 60 → 61 taken 13 times. ✓ Branch 60 → 69 taken 13 times.	26x	while (auto* w = impl->waiters_.pop_front())
317			{
318		13x	w->reset_stop_cb();
319		13x	auto h = std::exchange(w->h_, {});
320		13x	sched_->work_finished();
321			// Destroying the frame also ends the node's storage
322	1/2 ✓ Branch 66 → 67 taken 13 times. ✗ Branch 66 → 68 not taken.	13x	if (h)
323	1/1 ✓ Branch 67 → 68 taken 13 times.	13x	h.destroy();
324		13x	}
325	1/2 ✓ Branch 69 → 70 taken 13 times. ✗ Branch 69 → 71 not taken.	13x	delete impl;
326			}
327
328			// Delete free-listed impls
329	2/2 ✓ Branch 78 → 75 taken 24 times. ✓ Branch 78 → 79 taken 1230 times.	1254x	while (free_list_)
330			{
331		24x	auto* next = free_list_->next_free_;
332	1/2 ✓ Branch 75 → 76 taken 24 times. ✗ Branch 75 → 77 not taken.	24x	delete free_list_;
333		24x	free_list_ = next;
334			}
335		1230x	}
336
337			inline io_object::implementation*
338		3182x	timer_service::construct()
339			{
340		3182x	timer::implementation* impl = try_pop_tl_cache(this);
341	2/2 ✓ Branch 3 → 4 taken 3028 times. ✓ Branch 3 → 28 taken 154 times.	3182x	if (impl)
342			{
343		3028x	impl->svc_ = this;
344			// Reset expiry_ too: a recycled impl must behave like a fresh
345			// one, whose default expiry reads as already elapsed
346		3028x	impl->expiry_ = {};
347		3028x	impl->heap_index_.store(
348			(std::numeric_limits<std::size_t>::max)(),
349			std::memory_order_relaxed);
350		3028x	impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
351		3028x	return impl;
352			}
353
354	1/1 ✓ Branch 28 → 29 taken 154 times.	154x	std::lock_guard lock(mutex_);
355	1/2 ✗ Branch 29 → 30 not taken. ✓ Branch 29 → 53 taken 154 times.	154x	if (free_list_)
356			{
357		✗	impl = free_list_;
358		✗	free_list_ = impl->next_free_;
359		✗	impl->next_free_ = nullptr;
360		✗	impl->svc_ = this;
361		✗	impl->expiry_ = {};
362		✗	impl->heap_index_.store(
363			(std::numeric_limits<std::size_t>::max)(),
364			std::memory_order_relaxed);
365		✗	impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
366			}
367			else
368			{
369	1/1 ✓ Branch 53 → 54 taken 154 times.	154x	impl = new timer::implementation(*this);
370			}
371		154x	return impl;
372		154x	}
373
374			inline void
375		3182x	timer_service::destroy(io_object::implementation* p)
376			{
377			// During shutdown the drain loop owns every impl and deletes
378			// them directly. A frame destroyed by that loop can unwind a
379			// handle whose impl was freed in an earlier iteration (a
380			// timeout's parent frame owns the timeout timer while
381			// suspended on the inner delay's timer), so bail out before
382			// even downcasting the pointer.
383	2/2 ✓ Branch 2 → 3 taken 14 times. ✓ Branch 2 → 4 taken 3168 times.	3182x	if (shutting_down_)
384		14x	return;
385		3168x	destroy_impl(static_cast<timer::implementation&>(*p));
386			}
387
388			inline void
389		3168x	timer_service::destroy_impl(timer::implementation& impl)
390			{
391			// During shutdown the impl is owned by the shutdown loop.
392			// Re-entering here (from a coroutine-owned timer destructor
393			// triggered by h.destroy()) must not modify the heap or
394			// recycle the impl — shutdown deletes it directly.
395	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 4 taken 3168 times.	3168x	if (shutting_down_)
396		3144x	return;
397
398	1/1 ✓ Branch 4 → 5 taken 3168 times.	3168x	cancel_timer(impl);
399
400	1/2 ✗ Branch 21 → 22 not taken. ✓ Branch 21 → 27 taken 3168 times.	6336x	if (impl.heap_index_.load(std::memory_order_relaxed) !=
401		3168x	(std::numeric_limits<std::size_t>::max)())
402			{
403		✗	std::lock_guard lock(mutex_);
404		✗	remove_timer_impl(impl);
405		✗	refresh_cached_nearest();
406		✗	}
407
408	2/2 ✓ Branch 28 → 29 taken 3144 times. ✓ Branch 28 → 30 taken 24 times.	3168x	if (try_push_tl_cache(&impl))
409		3144x	return;
410
411	1/1 ✓ Branch 30 → 31 taken 24 times.	24x	std::lock_guard lock(mutex_);
412		24x	impl.next_free_ = free_list_;
413		24x	free_list_ = &impl;
414		24x	}
415
416			inline std::size_t
417		✗	timer_service::update_timer(timer::implementation& impl, time_point new_time)
418			{
419			// Gate on the flag, not waiters_: reading the non-atomic list
420			// here would race a concurrent drain. A false flag is safe to
421			// trust pre-lock: wait() stores it true before publishing, and
422			// it is cleared only under the mutex when the waiter list is
423			// empty, so false implies no published waiters.
424			bool in_heap =
425		✗	(impl.heap_index_.load(std::memory_order_relaxed) !=
426		✗	(std::numeric_limits<std::size_t>::max)());
427		✗	if (!in_heap &&
428		✗	!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
429		✗	return 0;
430
431		✗	bool notify = false;
432		✗	intrusive_list<waiter_node> canceled;
433
434			{
435		✗	std::lock_guard lock(mutex_);
436
437		✗	while (auto* w = impl.waiters_.pop_front())
438			{
439		✗	w->impl_ = nullptr;
440		✗	canceled.push_back(w);
441		✗	}
442
443		✗	std::size_t idx = impl.heap_index_.load(std::memory_order_relaxed);
444		✗	if (idx < heap_.size())
445			{
446		✗	time_point old_time = heap_[idx].time_;
447		✗	heap_[idx].time_ = new_time;
448
449		✗	if (new_time < old_time)
450		✗	up_heap(idx);
451			else
452		✗	down_heap(idx);
453
454		✗	notify =
455		✗	(impl.heap_index_.load(std::memory_order_relaxed) == 0);
456			}
457
458		✗	refresh_cached_nearest();
459		✗	}
460
461		✗	std::size_t count = 0;
462		✗	while (auto* w = canceled.pop_front())
463			{
464		✗	w->ec_ = make_error_code(capy::error::canceled);
465		✗	sched_->post(&w->op_);
466		✗	++count;
467		✗	}
468
469		✗	if (notify)
470		✗	on_earliest_changed_();
471
472		✗	return count;
473			}
474
475			inline void
476		1869x	timer_service::insert_waiter(timer::implementation& impl, waiter_node* w)
477			{
478		1869x	bool notify = false;
479		1869x	bool lost_cancel = false;
480			{
481	1/1 ✓ Branch 2 → 3 taken 1869 times.	1869x	std::lock_guard lock(mutex_);
482			// Publish: from here the waiter is visible to the fire path and
483			// to its own stop callback (impl_ non-null enables cancel_waiter).
484		1869x	w->impl_ = &impl;
485	1/2 ✓ Branch 19 → 20 taken 1869 times. ✗ Branch 19 → 60 not taken.	3738x	if (impl.heap_index_.load(std::memory_order_relaxed) ==
486		1869x	(std::numeric_limits<std::size_t>::max)())
487			{
488		1869x	impl.heap_index_.store(heap_.size(), std::memory_order_relaxed);
489	1/1 ✓ Branch 41 → 42 taken 1869 times.	1869x	heap_.push_back({impl.expiry_, &impl});
490	1/1 ✓ Branch 43 → 44 taken 1869 times.	1869x	up_heap(heap_.size() - 1);
491		1869x	notify =
492		1869x	(impl.heap_index_.load(std::memory_order_relaxed) == 0);
493		1869x	refresh_cached_nearest();
494			}
495		1869x	impl.waiters_.push_back(w);
496
497			// Lost-cancel re-check: a stop requested after the canceller was
498			// armed in wait() but before this publication found impl_ null
499			// and returned a no-op. Observe it now and undo the insertion.
500	1/2 ✗ Branch 62 → 63 not taken. ✓ Branch 62 → 70 taken 1869 times.	1869x	if (w->token_->stop_requested())
501			{
502		✗	w->impl_ = nullptr;
503		✗	impl.waiters_.remove(w);
504		✗	if (impl.waiters_.empty())
505			{
506		✗	remove_timer_impl(impl);
507		✗	impl.might_have_pending_waits_.store(
508			false, std::memory_order_relaxed);
509			}
510		✗	refresh_cached_nearest();
511		✗	lost_cancel = true;
512		✗	notify = false; // insertion undone; nearest unchanged
513			}
514		1869x	}
515	2/2 ✓ Branch 71 → 72 taken 1808 times. ✓ Branch 71 → 73 taken 61 times.	1869x	if (notify)
516		1808x	on_earliest_changed_();
517	1/2 ✗ Branch 73 → 74 not taken. ✓ Branch 73 → 76 taken 1869 times.	1869x	if (lost_cancel)
518			{
519		✗	w->ec_ = make_error_code(capy::error::canceled);
520		✗	sched_->post(&w->op_);
521			}
522		1869x	}
523
524			inline std::size_t
525		3168x	timer_service::cancel_timer(timer::implementation& impl)
526			{
527	2/2 ✓ Branch 3 → 4 taken 3167 times. ✓ Branch 3 → 5 taken 1 time.	3168x	if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
528		3167x	return 0;
529
530			// No unlocked already-done fast-out here: it would need the
531			// non-atomic waiters_ (a race with concurrent drains), and an
532			// index-only check is lifetime-unsafe because npos is stored
533			// before the drain finishes touching the impl. A stale-true
534			// flag is rare with the stateless API; the locked path below
535			// re-validates.
536
537		1x	intrusive_list<waiter_node> canceled;
538
539			{
540	1/1 ✓ Branch 5 → 6 taken 1 time.	1x	std::lock_guard lock(mutex_);
541	1/1 ✓ Branch 6 → 7 taken 1 time.	1x	remove_timer_impl(impl);
542	2/2 ✓ Branch 8 → 9 taken 1 time. ✓ Branch 8 → 11 taken 1 time.	2x	while (auto* w = impl.waiters_.pop_front())
543			{
544		1x	w->impl_ = nullptr;
545		1x	canceled.push_back(w);
546		1x	}
547			// Store false as the final touch of the impl under the lock so
548			// update_timer's pre-lock false-flag trust holds unqualified.
549		1x	impl.might_have_pending_waits_.store(false, std::memory_order_relaxed);
550		1x	refresh_cached_nearest();
551		1x	}
552
553		1x	std::size_t count = 0;
554	2/2 ✓ Branch 16 → 17 taken 1 time. ✓ Branch 16 → 20 taken 1 time.	2x	while (auto* w = canceled.pop_front())
555			{
556		1x	w->ec_ = make_error_code(capy::error::canceled);
557	1/1 ✓ Branch 18 → 19 taken 1 time.	1x	sched_->post(&w->op_);
558		1x	++count;
559		1x	}
560
561		1x	return count;
562			}
563
564			inline void
565		1416x	timer_service::cancel_waiter(waiter_node* w)
566			{
567			{
568	1/1 ✓ Branch 2 → 3 taken 1416 times.	1416x	std::lock_guard lock(mutex_);
569			// Already removed by another drain: cancel_timer,
570			// cancel_one_waiter, update_timer, process_expired, or
571			// insert_waiter's lost-cancel recheck
572	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 5 taken 1416 times.	1416x	if (!w->impl_)
573		✗	return;
574		1416x	auto* impl = w->impl_;
575		1416x	w->impl_ = nullptr;
576		1416x	impl->waiters_.remove(w);
577	1/2 ✓ Branch 7 → 8 taken 1416 times. ✗ Branch 7 → 10 not taken.	1416x	if (impl->waiters_.empty())
578			{
579	1/1 ✓ Branch 8 → 9 taken 1416 times.	1416x	remove_timer_impl(*impl);
580		1416x	impl->might_have_pending_waits_.store(
581			false, std::memory_order_relaxed);
582			}
583		1416x	refresh_cached_nearest();
584		1416x	}
585
586		1416x	w->ec_ = make_error_code(capy::error::canceled);
587		1416x	sched_->post(&w->op_);
588			}
589
590			inline std::size_t
591		✗	timer_service::cancel_one_waiter(timer::implementation& impl)
592			{
593		✗	if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
594		✗	return 0;
595
596		✗	waiter_node* w = nullptr;
597
598			{
599		✗	std::lock_guard lock(mutex_);
600		✗	w = impl.waiters_.pop_front();
601		✗	if (!w)
602		✗	return 0;
603		✗	w->impl_ = nullptr;
604		✗	if (impl.waiters_.empty())
605			{
606		✗	remove_timer_impl(impl);
607		✗	impl.might_have_pending_waits_.store(
608			false, std::memory_order_relaxed);
609			}
610		✗	refresh_cached_nearest();
611		✗	}
612
613		✗	w->ec_ = make_error_code(capy::error::canceled);
614		✗	sched_->post(&w->op_);
615		✗	return 1;
616			}
617
618			inline std::size_t
619		446x	timer_service::process_expired()
620			{
621		446x	intrusive_list<waiter_node> expired;
622
623			{
624	1/1 ✓ Branch 2 → 3 taken 446 times.	446x	std::lock_guard lock(mutex_);
625		446x	auto now = clock_type::now();
626
627	7/7 ✓ Branch 14 → 15 taken 793 times. ✓ Branch 14 → 20 taken 92 times. ✓ Branch 16 → 17 taken 793 times. ✓ Branch 18 → 19 taken 439 times. ✓ Branch 18 → 20 taken 354 times. ✓ Branch 21 → 5 taken 439 times. ✓ Branch 21 → 22 taken 446 times.	885x	while (!heap_.empty() && heap_[0].time_ <= now)
628			{
629		439x	timer::implementation* t = heap_[0].timer_;
630	1/1 ✓ Branch 6 → 7 taken 439 times.	439x	remove_timer_impl(*t);
631	2/2 ✓ Branch 8 → 9 taken 439 times. ✓ Branch 8 → 12 taken 439 times.	878x	while (auto* w = t->waiters_.pop_front())
632			{
633		439x	w->impl_ = nullptr;
634		439x	w->ec_ = {};
635		439x	expired.push_back(w);
636		439x	}
637		439x	t->might_have_pending_waits_.store(
638			false, std::memory_order_relaxed);
639			}
640
641		446x	refresh_cached_nearest();
642		446x	}
643
644		446x	std::size_t count = 0;
645	2/2 ✓ Branch 26 → 27 taken 439 times. ✓ Branch 26 → 29 taken 446 times.	885x	while (auto* w = expired.pop_front())
646			{
647	1/1 ✓ Branch 27 → 28 taken 439 times.	439x	sched_->post(&w->op_);
648		439x	++count;
649		439x	}
650
651		446x	return count;
652			}
653
654			inline void
655		1856x	timer_service::remove_timer_impl(timer::implementation& impl)
656			{
657		1856x	std::size_t index = impl.heap_index_.load(std::memory_order_relaxed);
658	1/2 ✗ Branch 18 → 19 not taken. ✓ Branch 18 → 20 taken 1856 times.	1856x	if (index >= heap_.size())
659		✗	return; // Not in heap
660
661	2/2 ✓ Branch 21 → 22 taken 1484 times. ✓ Branch 21 → 44 taken 372 times.	1856x	if (index == heap_.size() - 1)
662			{
663			// Last element, just pop
664		1484x	impl.heap_index_.store(
665			(std::numeric_limits<std::size_t>::max)(),
666			std::memory_order_relaxed);
667		1484x	heap_.pop_back();
668			}
669			else
670			{
671			// Swap with last and reheapify
672		372x	swap_heap(index, heap_.size() - 1);
673		372x	impl.heap_index_.store(
674			(std::numeric_limits<std::size_t>::max)(),
675			std::memory_order_relaxed);
676		372x	heap_.pop_back();
677
678	4/6 ✓ Branch 68 → 69 taken 1 time. ✓ Branch 68 → 75 taken 371 times. ✗ Branch 73 → 74 not taken. ✓ Branch 73 → 75 taken 1 time. ✗ Branch 76 → 77 not taken. ✓ Branch 76 → 78 taken 372 times.	372x	if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
679		✗	up_heap(index);
680			else
681		372x	down_heap(index);
682			}
683			}
684
685			inline void
686		1869x	timer_service::up_heap(std::size_t index)
687			{
688	2/2 ✓ Branch 11 → 3 taken 410 times. ✓ Branch 11 → 12 taken 1808 times.	2218x	while (index > 0)
689			{
690		410x	std::size_t parent = (index - 1) / 2;
691	2/2 ✓ Branch 7 → 8 taken 61 times. ✓ Branch 7 → 9 taken 349 times.	410x	if (!(heap_[index].time_ < heap_[parent].time_))
692		61x	break;
693		349x	swap_heap(index, parent);
694		349x	index = parent;
695			}
696		1869x	}
697
698			inline void
699		372x	timer_service::down_heap(std::size_t index)
700			{
701		372x	std::size_t child = index * 2 + 1;
702	2/2 ✓ Branch 21 → 3 taken 2 times. ✓ Branch 21 → 22 taken 371 times.	373x	while (child < heap_.size())
703			{
704		2x	std::size_t min_child = (child + 1 == heap_.size() \|\|
705		✗	heap_[child].time_ < heap_[child + 1].time_)
706	1/2 ✗ Branch 4 → 5 not taken. ✓ Branch 4 → 11 taken 2 times.	2x	? child
707		2x	: child + 1;
708
709	2/2 ✓ Branch 16 → 17 taken 1 time. ✓ Branch 16 → 18 taken 1 time.	2x	if (heap_[index].time_ < heap_[min_child].time_)
710		1x	break;
711
712		1x	swap_heap(index, min_child);
713		1x	index = min_child;
714		1x	child = index * 2 + 1;
715			}
716		372x	}
717
718			inline void
719		722x	timer_service::swap_heap(std::size_t i1, std::size_t i2)
720			{
721		722x	heap_entry tmp = heap_[i1];
722		722x	heap_[i1] = heap_[i2];
723		722x	heap_[i2] = tmp;
724		722x	heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed);
725		722x	heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed);
726		722x	}
727
728			// waiter_node's completion_op and canceller members are defined in
729			// timer.cpp alongside implementation::wait(), for the same reason
730			// wait() lives there (see below).
731
732			// timer::implementation::wait() is defined in timer.cpp, not here.
733			// It must be a non-inline definition in a translation unit that is
734			// always pulled into the link whenever detail::timer is used (every
735			// consumer needs timer's constructors from that same object file).
736			// An inline definition in this header would only be emitted in
737			// translation units that happen to also include this header, which
738			// is not guaranteed for every caller of wait_awaitable::await_suspend
739			// in timer.hpp (e.g. code that only reaches timer.hpp through
740			// delay.hpp, without transitively including a scheduler header).
741
742			// Free functions
743
744			inline std::size_t
745		✗	timer_service_update_expiry(timer::implementation& impl)
746			{
747		✗	return impl.svc_->update_timer(impl, impl.expiry_);
748			}
749
750			inline std::size_t
751		✗	timer_service_cancel(timer::implementation& impl) noexcept
752			{
753		✗	return impl.svc_->cancel_timer(impl);
754			}
755
756			inline std::size_t
757		✗	timer_service_cancel_one(timer::implementation& impl) noexcept
758			{
759		✗	return impl.svc_->cancel_one_waiter(impl);
760			}
761
762			inline timer_service&
763		615x	get_timer_service(capy::execution_context& ctx, scheduler& sched)
764			{
765		615x	return ctx.make_service<timer_service>(sched);
766			}
767
768			} // namespace boost::corosio::detail
769
770			#endif
771