include/boost/corosio/detail/timer_service.hpp

77.2% Lines (284/368) 83.3% List of functions (35/42)

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

timer_service.hpp

f(x) Functions (42)

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