include/boost/corosio/detail/timer_service.hpp

71.0% Lines (230/324) 84.4% List of functions (27/33) 53.2% Branches (83/156)

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

timer_service.hpp

f(x) Functions (33)

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		1331x	void* ctx_ = nullptr;
94		1331x	void (fn_)(void) = nullptr;
95
96			public:
97			/// Construct an empty callback.
98		3993x	callback() = default;
99
100			/// Construct a callback with the given context and function.
101		2662x	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		8203x	void operator()() const
111			{
112	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 8203 times.	8203x	if (fn_)
113		8203x	fn_(ctx_);
114		8203x	}
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		1331x	timer::implementation* free_list_ = nullptr;
130			callback on_earliest_changed_;
131		1331x	bool shutting_down_ = false;
132			// Avoids mutex in nearest_expiry() and empty()
133		2662x	mutable std::atomic<std::int64_t> cached_nearest_ns_{
134		1331x	(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		6655x	inline timer_service(capy::execution_context&, scheduler& sched)
140		1331x	: sched_(&sched)
141		3993x	{
142		2662x	}
143
144			/// Return the associated scheduler.
145		16583x	inline scheduler& get_scheduler() noexcept
146			{
147		16583x	return *sched_;
148			}
149
150			/// Destroy the timer service.
151		3993x	~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		1331x	inline void set_on_earliest_changed(callback cb)
158			{
159		1331x	on_earliest_changed_ = cb;
160		1331x	}
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		1074301x	inline time_point nearest_expiry() const noexcept
171			{
172		1074301x	auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
173	2/4 ✓ Branch 0 taken 1074301 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 1074301 times. ✗ Branch 3 not taken.	1074301x	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		1138266x	inline void refresh_cached_nearest() noexcept
209			{
210	2/2 ✓ Branch 0 taken 13410 times. ✓ Branch 1 taken 1124856 times.	1138266x	auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
211		1124856x	: heap_[0].time_.time_since_epoch().count();
212		1138266x	cached_nearest_ns_.store(ns, std::memory_order_release);
213		1138266x	}
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		126x	~tl_cache_owner()
237		63x	{
238	2/2 ✓ Branch 0 taken 43 times. ✓ Branch 1 taken 20 times.	63x	delete tl_cached_impl.get();
239		63x	tl_cached_impl.set(nullptr);
240		126x	}
241			};
242
243			inline void
244		8915x	arm_tl_cache_cleanup() noexcept
245			{
246	2/2 ✓ Branch 0 taken 8852 times. ✓ Branch 1 taken 63 times.	8915x	thread_local tl_cache_owner owner;
247		8915x	(void)owner;
248		8915x	}
249
250			inline timer::implementation*
251		8987x	try_pop_tl_cache(timer_service* svc) noexcept
252			{
253		8987x	auto* impl = tl_cached_impl.get();
254	2/2 ✓ Branch 0 taken 329 times. ✓ Branch 1 taken 8658 times.	8987x	if (impl)
255			{
256		8658x	tl_cached_impl.set(nullptr);
257	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 8658 times.	8658x	if (impl->svc_ == svc)
258		8658x	return impl;
259			// Stale impl from a destroyed service
260		✗	delete impl;
261		✗	}
262		329x	return nullptr;
263		8987x	}
264
265			inline bool
266		8960x	try_push_tl_cache(timer::implementation* impl) noexcept
267			{
268	2/2 ✓ Branch 0 taken 46 times. ✓ Branch 1 taken 8914 times.	8960x	if (!tl_cached_impl.get())
269			{
270		8914x	arm_tl_cache_cleanup();
271		8914x	tl_cached_impl.set(impl);
272		8914x	return true;
273			}
274		46x	return false;
275		8960x	}
276
277			inline void
278		1331x	timer_service_invalidate_cache() noexcept
279			{
280	2/2 ✓ Branch 0 taken 1094 times. ✓ Branch 1 taken 237 times.	1331x	delete tl_cached_impl.get();
281		1331x	tl_cached_impl.set(nullptr);
282		1331x	}
283
284			// timer_service out-of-class member function definitions
285
286			inline void
287		1331x	timer_service::shutdown()
288			{
289		1331x	timer_service_invalidate_cache();
290		1331x	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		1331x	std::vector<timer::implementation*> impls;
297	1/2 ✓ Branch 0 taken 1331 times. ✗ Branch 1 not taken.	1331x	impls.reserve(heap_.size());
298	2/2 ✓ Branch 0 taken 1331 times. ✓ Branch 1 taken 26 times.	1357x	for (auto& entry : heap_)
299			{
300		52x	entry.timer_->heap_index_.store(
301		26x	(std::numeric_limits<std::size_t>::max)(),
302			std::memory_order_relaxed);
303	1/2 ✓ Branch 0 taken 26 times. ✗ Branch 1 not taken.	26x	impls.push_back(entry.timer_);
304			}
305		1331x	heap_.clear();
306		2662x	cached_nearest_ns_.store(
307		1331x	(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 0 taken 1331 times. ✓ Branch 1 taken 26 times.	1357x	for (auto* impl : impls)
315			{
316	2/2 ✓ Branch 0 taken 26 times. ✓ Branch 1 taken 26 times.	52x	while (auto* w = impl->waiters_.pop_front())
317			{
318		26x	w->reset_stop_cb();
319		26x	auto h = std::exchange(w->h_, {});
320		26x	sched_->work_finished();
321			// Destroying the frame also ends the node's storage
322	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 26 times.	26x	if (h)
323	1/2 ✓ Branch 0 taken 26 times. ✗ Branch 1 not taken.	26x	h.destroy();
324			}
325	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 26 times.	26x	delete impl;
326			}
327
328			// Delete free-listed impls
329	2/2 ✓ Branch 0 taken 46 times. ✓ Branch 1 taken 1331 times.	1377x	while (free_list_)
330			{
331		46x	auto* next = free_list_->next_free_;
332	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 46 times.	46x	delete free_list_;
333		46x	free_list_ = next;
334			}
335		1331x	}
336
337			inline io_object::implementation*
338		8987x	timer_service::construct()
339			{
340		8987x	timer::implementation* impl = try_pop_tl_cache(this);
341	2/2 ✓ Branch 0 taken 8658 times. ✓ Branch 1 taken 329 times.	8987x	if (impl)
342			{
343		8658x	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		8658x	impl->expiry_ = {};
347		17316x	impl->heap_index_.store(
348		8658x	(std::numeric_limits<std::size_t>::max)(),
349			std::memory_order_relaxed);
350		8658x	impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
351		8658x	return impl;
352			}
353
354		329x	std::lock_guard lock(mutex_);
355	1/2 ✓ Branch 0 taken 329 times. ✗ Branch 1 not taken.	329x	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/2 ✓ Branch 0 taken 329 times. ✗ Branch 1 not taken.	329x	impl = new timer::implementation(*this);
370			}
371		329x	return impl;
372		8987x	}
373
374			inline void
375		8987x	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 0 taken 26 times. ✓ Branch 1 taken 8961 times.	8987x	if (shutting_down_)
384		26x	return;
385		8961x	destroy_impl(static_cast<timer::implementation&>(*p));
386		8987x	}
387
388			inline void
389		8959x	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 0 not taken. ✓ Branch 1 taken 8959 times.	8959x	if (shutting_down_)
396		✗	return;
397
398		8959x	cancel_timer(impl);
399
400	2/4 ✓ Branch 0 taken 8959 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 8959 times. ✗ Branch 3 not taken.	17918x	if (impl.heap_index_.load(std::memory_order_relaxed) !=
401		8959x	(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 0 taken 8913 times. ✓ Branch 1 taken 46 times.	8959x	if (try_push_tl_cache(&impl))
409		8913x	return;
410
411		46x	std::lock_guard lock(mutex_);
412		46x	impl.next_free_ = free_list_;
413		46x	free_list_ = &impl;
414		8959x	}
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		8305x	timer_service::insert_waiter(timer::implementation& impl, waiter_node* w)
477			{
478		8305x	bool notify = false;
479		8305x	bool lost_cancel = false;
480			{
481		8305x	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		8305x	w->impl_ = &impl;
485	2/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 8305 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 8305 times.	16610x	if (impl.heap_index_.load(std::memory_order_relaxed) ==
486		8305x	(std::numeric_limits<std::size_t>::max)())
487			{
488		8305x	impl.heap_index_.store(heap_.size(), std::memory_order_relaxed);
489	1/2 ✓ Branch 0 taken 8305 times. ✗ Branch 1 not taken.	8305x	heap_.push_back({impl.expiry_, &impl});
490	1/2 ✓ Branch 0 taken 8305 times. ✗ Branch 1 not taken.	8305x	up_heap(heap_.size() - 1);
491		8305x	notify =
492		8305x	(impl.heap_index_.load(std::memory_order_relaxed) == 0);
493		8305x	refresh_cached_nearest();
494		8305x	}
495		8305x	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 0 taken 8305 times. ✗ Branch 1 not taken.	8305x	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		8305x	}
515	2/2 ✓ Branch 0 taken 8203 times. ✓ Branch 1 taken 102 times.	8305x	if (notify)
516		8203x	on_earliest_changed_();
517	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 8305 times.	8305x	if (lost_cancel)
518			{
519		✗	w->ec_ = make_error_code(capy::error::canceled);
520		✗	sched_->post(&w->op_);
521		✗	}
522		8305x	}
523
524			inline std::size_t
525		8961x	timer_service::cancel_timer(timer::implementation& impl)
526			{
527	2/2 ✓ Branch 0 taken 8959 times. ✓ Branch 1 taken 2 times.	8961x	if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
528		8959x	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		2x	intrusive_list<waiter_node> canceled;
538
539			{
540		2x	std::lock_guard lock(mutex_);
541	1/2 ✓ Branch 0 taken 2 times. ✗ Branch 1 not taken.	2x	remove_timer_impl(impl);
542	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 2 times.	4x	while (auto* w = impl.waiters_.pop_front())
543			{
544		2x	w->impl_ = nullptr;
545		2x	canceled.push_back(w);
546			}
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		2x	impl.might_have_pending_waits_.store(false, std::memory_order_relaxed);
550		2x	refresh_cached_nearest();
551		2x	}
552
553		2x	std::size_t count = 0;
554	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 2 times.	4x	while (auto* w = canceled.pop_front())
555			{
556		2x	w->ec_ = make_error_code(capy::error::canceled);
557		2x	sched_->post(&w->op_);
558		2x	++count;
559			}
560
561		2x	return count;
562		8961x	}
563
564			inline void
565		1975x	timer_service::cancel_waiter(waiter_node* w)
566			{
567			{
568		1975x	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 0 taken 1975 times. ✗ Branch 1 not taken.	1975x	if (!w->impl_)
573		✗	return;
574		1975x	auto* impl = w->impl_;
575		1975x	w->impl_ = nullptr;
576		1975x	impl->waiters_.remove(w);
577	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 1975 times.	1975x	if (impl->waiters_.empty())
578			{
579	1/2 ✓ Branch 0 taken 1975 times. ✗ Branch 1 not taken.	1975x	remove_timer_impl(*impl);
580		1975x	impl->might_have_pending_waits_.store(
581			false, std::memory_order_relaxed);
582		1975x	}
583		1975x	refresh_cached_nearest();
584		1975x	}
585
586		1975x	w->ec_ = make_error_code(capy::error::canceled);
587		1975x	sched_->post(&w->op_);
588		1975x	}
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		1127984x	timer_service::process_expired()
620			{
621		1127984x	intrusive_list<waiter_node> expired;
622
623			{
624		1127984x	std::lock_guard lock(mutex_);
625		1127984x	auto now = clock_type::now();
626
627	4/4 ✓ Branch 0 taken 11508 times. ✓ Branch 1 taken 1122778 times. ✓ Branch 2 taken 1127984 times. ✓ Branch 3 taken 6302 times.	1134286x	while (!heap_.empty() && heap_[0].time_ <= now)
628			{
629		6302x	timer::implementation* t = heap_[0].timer_;
630	1/2 ✓ Branch 0 taken 6302 times. ✗ Branch 1 not taken.	6302x	remove_timer_impl(*t);
631	2/2 ✓ Branch 0 taken 6302 times. ✓ Branch 1 taken 6302 times.	12604x	while (auto* w = t->waiters_.pop_front())
632			{
633		6302x	w->impl_ = nullptr;
634		6302x	w->ec_ = {};
635		6302x	expired.push_back(w);
636			}
637		6302x	t->might_have_pending_waits_.store(
638			false, std::memory_order_relaxed);
639			}
640
641		1127984x	refresh_cached_nearest();
642		1127984x	}
643
644		1127984x	std::size_t count = 0;
645	2/2 ✓ Branch 0 taken 6302 times. ✓ Branch 1 taken 1127984 times.	1134286x	while (auto* w = expired.pop_front())
646			{
647		6302x	sched_->post(&w->op_);
648		6302x	++count;
649			}
650
651		1127984x	return count;
652		✗	}
653
654			inline void
655		8279x	timer_service::remove_timer_impl(timer::implementation& impl)
656			{
657		8279x	std::size_t index = impl.heap_index_.load(std::memory_order_relaxed);
658	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 8279 times.	8279x	if (index >= heap_.size())
659		✗	return; // Not in heap
660
661	2/2 ✓ Branch 0 taken 2188 times. ✓ Branch 1 taken 6091 times.	8279x	if (index == heap_.size() - 1)
662			{
663			// Last element, just pop
664		4376x	impl.heap_index_.store(
665		2188x	(std::numeric_limits<std::size_t>::max)(),
666			std::memory_order_relaxed);
667		2188x	heap_.pop_back();
668		2188x	}
669			else
670			{
671			// Swap with last and reheapify
672		6091x	swap_heap(index, heap_.size() - 1);
673		12182x	impl.heap_index_.store(
674		6091x	(std::numeric_limits<std::size_t>::max)(),
675			std::memory_order_relaxed);
676		6091x	heap_.pop_back();
677
678	3/4 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 6089 times. ✓ Branch 2 taken 2 times. ✗ Branch 3 not taken.	6091x	if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
679		✗	up_heap(index);
680			else
681		6091x	down_heap(index);
682			}
683		8279x	}
684
685			inline void
686		8305x	timer_service::up_heap(std::size_t index)
687			{
688	2/2 ✓ Branch 0 taken 8203 times. ✓ Branch 1 taken 6152 times.	14355x	while (index > 0)
689			{
690		6152x	std::size_t parent = (index - 1) / 2;
691	2/2 ✓ Branch 0 taken 102 times. ✓ Branch 1 taken 6050 times.	6152x	if (!(heap_[index].time_ < heap_[parent].time_))
692		102x	break;
693		6050x	swap_heap(index, parent);
694		6050x	index = parent;
695			}
696		8305x	}
697
698			inline void
699		6091x	timer_service::down_heap(std::size_t index)
700			{
701		6091x	std::size_t child = index * 2 + 1;
702	2/2 ✓ Branch 0 taken 6089 times. ✓ Branch 1 taken 4 times.	6093x	while (child < heap_.size())
703			{
704	1/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 4 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken.	4x	std::size_t min_child = (child + 1 == heap_.size() \|\|
705		✗	heap_[child].time_ < heap_[child + 1].time_)
706		4x	? child
707		✗	: child + 1;
708
709	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 2 times.	4x	if (heap_[index].time_ < heap_[min_child].time_)
710		2x	break;
711
712		2x	swap_heap(index, min_child);
713		2x	index = min_child;
714		2x	child = index * 2 + 1;
715			}
716		6091x	}
717
718			inline void
719		12143x	timer_service::swap_heap(std::size_t i1, std::size_t i2)
720			{
721		12143x	heap_entry tmp = heap_[i1];
722		12143x	heap_[i1] = heap_[i2];
723		12143x	heap_[i2] = tmp;
724		12143x	heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed);
725		12143x	heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed);
726		12143x	}
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		1331x	get_timer_service(capy::execution_context& ctx, scheduler& sched)
764			{
765		1331x	return ctx.make_service<timer_service>(sched);
766			}
767
768			} // namespace boost::corosio::detail
769
770			#endif
771