include/boost/corosio/detail/timer_service.hpp

96.4% Lines (269 / 279) 73.0% Branches (111 / 152) 100.0% Functions (30 / 30)

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

timer_service.hpp

f(x) Functions (30)

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