include/boost/corosio/native/detail/uring/uring_scheduler.hpp

97.9% Lines (619 / 632, 10 excl) 97.5% Functions (79 / 81, 2 excl)

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

uring_scheduler.hpp

f(x) Functions (83)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
3		// Copyright (c) 2026 Michael Vandeberg
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_NATIVE_DETAIL_URING_URING_SCHEDULER_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_SCHEDULER_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_URING
17
18		// Include before any project headers open a namespace — prevents the
19		// boost::corosio::uring tag variable from shadowing struct ::io_uring.
20		#include <liburing.h>
21
22		#include <boost/corosio/detail/conditionally_enabled_event.hpp>
23		#include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
24		#include <boost/corosio/detail/config.hpp>
25		#include <boost/corosio/detail/except.hpp>
26		#include <boost/corosio/detail/ready_queue.hpp>
27		#include <boost/corosio/detail/scheduler.hpp>
28		#include <boost/corosio/detail/scheduler_op.hpp>
29		#include <boost/corosio/detail/timer_service.hpp>
30		#include <boost/corosio/native/detail/uring/uring_op.hpp>
31		#include <boost/corosio/native/detail/make_err.hpp>
32		#include <boost/corosio/native/detail/posix/posix_signal_service.hpp>
33		#include <boost/capy/ex/execution_context.hpp>
34		#include <boost/capy/ex/abandon.hpp>
35
36		#include <atomic>
37		#include <chrono>
38		#include <coroutine>
39		#include <cstddef>
40		#include <cstdint>
41		#include <limits>
42		#include <memory>
43		#include <mutex>
44		#include <thread>
45		#include <tuple>
46		#include <initializer_list>
47		#include <vector>
48
49		#include <errno.h>
50		#include <fcntl.h>
51		#include <poll.h>
52		#include <pthread.h>
53		#include <signal.h>
54		#include <sys/eventfd.h>
55		#include <time.h>
56		#include <unistd.h>
57
58		namespace boost::corosio::detail {
59
60		/** Block SIGPIPE on the calling thread for the current scope.
61
62		A queued pipe write can execute inline while this thread is in
63		the kernel submitting SQEs; if the pipe's reader is already gone
64		the kernel raises a thread-directed SIGPIPE, which kills any
65		process that has not ignored the signal. The write's CQE still
66		reports `EPIPE`, so the signal carries no information the
67		completion path does not already deliver. The destructor consumes
68		any SIGPIPE raised while blocked and restores the caller's mask.
69		*/
70		class scoped_sigpipe_block
71		{
72		sigset_t old_{};
73		bool restore_ = false;
74
75		public:
76		/// Consume any SIGPIPE raised in scope and restore the mask.
77	3525x	~scoped_sigpipe_block()
78	3525x	{
79	3525x	if (!restore_)
80	✗	return;
81	3525x	if (!sigismember(&old_, SIGPIPE))
82		{
83		// Only consume what this scope could have generated; a
84		// caller who blocked SIGPIPE keeps their pending state.
85		sigset_t set;
86	3418x	sigemptyset(&set);
87	3418x	sigaddset(&set, SIGPIPE);
88	3418x	timespec zero{};
89	3419x	while (::sigtimedwait(&set, nullptr, &zero) == SIGPIPE)
90		{
91		}
92		}
93	3525x	::pthread_sigmask(SIG_SETMASK, &old_, nullptr);
94	3525x	}
95
96		/// Construct and block SIGPIPE for the calling thread.
97	3525x	scoped_sigpipe_block() noexcept
98	3525x	{
99		sigset_t set;
100	3525x	sigemptyset(&set);
101	3525x	sigaddset(&set, SIGPIPE);
102	3525x	restore_ = ::pthread_sigmask(SIG_BLOCK, &set, &old_) == 0;
103	3525x	}
104
105		scoped_sigpipe_block(scoped_sigpipe_block const&) = delete;
106		scoped_sigpipe_block& operator=(scoped_sigpipe_block const&) = delete;
107		};
108
109		// Forward-declared so the out-of-line inline definitions below the class
110		// can reference the frame stack without a circular dependency.
111		struct uring_scheduler_frame;
112		extern thread_local uring_scheduler_frame* tl_running_scheduler_frame_;
113
114		/** io_uring scheduler — proactor model on Linux 6.x+.
115
116		Owns one io_uring per io_context. Lazy batched submit;
117		cross-thread post wakes a registered eventfd via multishot
118		POLL_ADD.
119
120		@par Thread Safety
121		All public member functions are thread-safe.
122		*/
123		class BOOST_COROSIO_DECL uring_scheduler final
124		: public scheduler
125		{
126		public:
127		using mutex_type = conditionally_enabled_mutex;
128		using lock_type = mutex_type::scoped_lock;
129		using event_type = conditionally_enabled_event;
130
131		uring_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
132		~uring_scheduler() override;
133		uring_scheduler(uring_scheduler const&) = delete;
134		uring_scheduler& operator=(uring_scheduler const&) = delete;
135
136		void shutdown() override;
137
138		void post(std::coroutine_handle<>) const override;
139		void post(scheduler_op*) const override;
140		void post(capy::continuation&) const override;
141
142		bool running_in_this_thread() const noexcept override;
143		void stop() override;
144		bool stopped() const noexcept override;
145		void restart() override;
146		std::size_t run() override;
147		std::size_t run_one() override;
148		std::size_t wait_one(long usec) override;
149		std::size_t poll() override;
150		std::size_t poll_one() override;
151		void work_started() noexcept override;
152		void work_finished() noexcept override;
153
154		/// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
155		/// Submits a multishot POLL on @p read_fd; on readiness the drain+deliver
156		/// runs in dispatch context via signal_drain_op_.
157		[[nodiscard]] std::error_code register_signal_reader(int read_fd) override;
158
159		/** Return the underlying liburing ring.
160
161		Triggers lazy ring initialisation on first call. Used by
162		socket op submission helpers (e.g. `uring_submit_op`) and
163		any other code path that needs a live ring pointer.
164		*/
165	4x	struct ::io_uring* ring() noexcept
166		{
167	4x	lazy_init_ring();
168	4x	return &ring_;
169		}
170
171		/// Return the dispatch mutex (protects completed_ops_ / cond_).
172	18596x	mutex_type& dispatch_mutex() const noexcept
173		{
174	18596x	return dispatch_mutex_;
175		}
176
177		/// Return the ring mutex (serialises userspace SQ/CQ access).
178	6212x	mutex_type& ring_mutex() const noexcept
179		{
180	6212x	return ring_mutex_;
181		}
182
183		/** Reset the calling thread's inline-budget for this scheduler.
184
185		Called at the top of each dispatched op in `do_one` so each
186		op handler gets a fresh budget for inline speculative
187		completions. Walks the frame stack; no-op if this scheduler
188		isn't on the stack (i.e. called from a non-run thread).
189		*/
190		void reset_inline_budget() const noexcept;
191
192		/** Consume one unit of inline budget if available.
193
194		@return `true` if budget was available and consumed; `false`
195		if the budget is exhausted or this scheduler is not on
196		the calling thread's run stack.
197		*/
198		bool try_consume_inline_budget() const noexcept;
199
200		/// Exchange the submit-batch posted flag. Returns the prior value.
201		/// Caller MUST hold ring_mutex_.
202	6207x	bool submit_op_posted_exchange(bool desired) const noexcept
203		{
204	6207x	return submit_op_posted_.exchange(desired, std::memory_order_relaxed);
205		}
206
207		/** Check whether a stream write should join the queued batch.
208
209		A write that would otherwise be issued directly can instead
210		ride the batch submit's kernel entry. Only a single running
211		thread joins: with more, the batch may run on another thread,
212		which would then own this write's completion. The answer is a
213		hint and may be stale.
214
215		@return `true` if a batch submit is queued and exactly one
216		thread is running the scheduler.
217		*/
218	158328x	bool write_may_join_batch() const noexcept
219		{
220	158577x	return submit_op_posted_.load(std::memory_order_relaxed) &&
221	158826x	runners_.load(std::memory_order_relaxed) == 1;
222		}
223
224		/// Return the number of threads running the scheduler.
225		int runners() const noexcept
226		{
227		return runners_.load(std::memory_order_relaxed);
228		}
229
230		/// Record a thread entering a run function.
231	4063x	void runner_entered() const noexcept
232		{
233	4063x	runners_.fetch_add(1, std::memory_order_relaxed);
234	4063x	}
235
236		/// Record a thread leaving a run function.
237	4063x	void runner_exited() const noexcept
238		{
239	4063x	runners_.fetch_sub(1, std::memory_order_relaxed);
240	4063x	}
241
242		/// Return a reference to the mutable embedded submit_sqes_op.
243	5074x	scheduler_op& submit_op_ref() const noexcept
244		{
245	5074x	return submit_op_;
246		}
247
248		/// Increment the io_uring in-flight counter. Callers prep an SQE
249		/// whose CQE must be reaped. Excluded: the wakeup-eventfd poll,
250		/// which stays armed for the ring's lifetime.
251	9737x	void inflight_inc() const noexcept
252		{
253	9737x	uring_inflight_.fetch_add(1, std::memory_order_release);
254	9737x	}
255
256		/** Return the current io_uring in-flight counter.
257
258		Test-only helper: `uring_inflight_` is an internal accounting
259		counter with no bearing on the public API. It is exposed solely
260		so tests can assert the counter stays balanced across op
261		submission and teardown. Not thread-safe with
262		respect to a concurrently running scheduler; call it from a quiesced
263		context.
264
265		@return The number of SQEs currently counted as in flight.
266		*/
267	11x	std::int64_t inflight() const noexcept
268		{
269	22x	return uring_inflight_.load(std::memory_order_acquire);
270		}
271
272		/// Initialize the io_uring ring on first access. Idempotent.
273		void lazy_init_ring() const;
274
275		/** Create the io_uring ring.
276
277		Called once the owning `io_context` has applied every option
278		that feeds the ring's setup flags, so that a kernel that
279		refuses the ring is reported from the `io_context`
280		constructor rather than from the first operation submitted
281		through it. Idempotent.
282
283		@par Exception Safety
284		Strong guarantee. A ring that cannot be created leaves no
285		descriptor behind.
286
287		@throws std::system_error If the ring, its wakeup eventfd, or
288		the poll watching that eventfd could not be set up.
289		*/
290	1721x	void init_ring() const
291		{
292	1721x	lazy_init_ring();
293	1715x	}
294
295		/// Wake the leader if it's blocked in `submit_and_wait_timeout`.
296		/// Best-effort: the wakeup is suppressed if the leader has already
297		/// been signalled and not yet acked.
298		void interrupt_reactor() const noexcept;
299
300		/** Wake a thread to run newly queued work.
301
302		Called under `dispatch_mutex_`: the caller may be on another
303		context's thread, and once the lock is released the work can
304		run, finish and let this scheduler be destroyed.
305		*/
306		void wake_one_locked() const noexcept;
307
308		/** Submit `IORING_OP_ASYNC_CANCEL` targeting an in-flight op by its
309		user_data pointer.
310
311		The kernel delivers `-ECANCELED` on the target's CQE if it was
312		still in flight; the op's completion handler then reports
313		`operation_aborted`. Best-effort: if the SQ is full after one
314		flush attempt the function returns without cancelling (the op
315		will complete normally on its own).
316
317		@param target The in-flight op to cancel.
318		*/
319		void submit_cancel_by_user_data(uring_op* target) noexcept;
320
321		/** Cancel @p op by its user_data if the kernel may hold it.
322
323		A cancel that cannot be queued yet is deferred, and queued
324		ahead of the next operation's SQE.
325
326		@pre `ring_mutex()` is held.
327
328		@param op The operation to cancel.
329		*/
330		void cancel_op_locked(uring_op* op) noexcept;
331
332		/** Cancel each of @p ops the kernel may hold.
333
334		Only these operations are cancelled, never another object's
335		operation on the same file.
336
337		@param ops The operations to cancel.
338		*/
339		void cancel_ops(std::initializer_list<uring_op*> ops) noexcept;
340
341		/** Cancel an object's operations.
342
343		@param cancel Called under the ring mutex; cancels the object's
344		operations with @ref cancel_op_locked.
345		*/
346		template<class Cancel>
347	923x	void cancel_ops(Cancel&& cancel) noexcept
348		{
349	923x	lazy_init_ring();
350		{
351	923x	lock_type lock(ring_mutex_);
352	923x	cancel();
353	923x	}
354		// After queuing: a leader woken first could re-enter its wait
355		// before the cancel SQE exists, leaving it unsubmitted.
356	923x	interrupt_reactor();
357	923x	}
358
359		/** Cancel an object's operations and flush every SQE naming `fd`.
360
361		Once this returns no queued SQE refers to `fd` by number, so
362		the caller may close it, or hand it to an owner who may. If the
363		flush fails, this object's SQEs still queued are turned into
364		no-ops and their operations complete as cancelled.
365
366		@param fd The object's descriptor.
367		@param cancel Called under the ring mutex; cancels the object's
368		operations with @ref cancel_op_locked.
369		*/
370		template<class Cancel>
371		void flush_after_cancel(int fd, Cancel&& cancel) noexcept;
372
373		/// @copydoc flush_after_cancel
374		void
375	7559x	flush_after_cancel(int fd, std::initializer_list<uring_op*> ops) noexcept
376		{
377		// An op clears sqe_set only once the kernel is done with its
378		// entry, so with none set nothing names `fd`. Closing an idle
379		// socket then skips the flush's kernel entries and the wake.
380	7559x	bool armed = false;
381	37649x	for (auto* op : ops)
382	30090x	armed = armed \|\| op->sqe_set.load(std::memory_order_acquire);
383	7559x	if (!armed)
384	7467x	return;
385
386	92x	flush_after_cancel(fd, [&]() noexcept {
387	466x	for (auto* op : ops)
388	374x	cancel_op_locked(op);
389	92x	});
390		}
391
392		/** Return a submission queue entry for a new operation.
393
394		Deferred cancels are queued first: they name operations by
395		address, and an operation armed later at the same address must
396		not be what they reach. Flushes once if the queue is full.
397
398		@pre `ring_mutex()` is held.
399
400		@return The entry, or null if the queue stays full or a
401		deferred cancel still cannot be queued.
402		*/
403		::io_uring_sqe* acquire_sqe() noexcept;
404
405		/** Queue an already-counted op while the caller holds dispatch_mutex_.
406
407		Does NOT increment `outstanding_work_`. Use for synchronous
408		completion paths (e.g. SQE backpressure) where the caller called
409		`work_started()` and already holds the dispatch lock.
410
411		@pre `dispatch_mutex_` must be locked by the calling thread.
412		*/
413	18596x	void push_completed_locked(scheduler_op* op) const noexcept
414		{
415	18596x	completed_ops_.push(op);
416		// A run thread finds the op on its next pass; any other caller
417		// must wake one, or it waits for the nearest timer.
418	18596x	if (!running_in_this_thread())
419	6x	wake_one_locked();
420	18596x	}
421
422	1721x	void configure_threading(threading_config cfg) noexcept override
423		{
424	1721x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
425		// reactor_io_locking off also drives the eventfd wake elision (see
426		// interrupt_reactor). one_thread is unused: the leader-follower wake
427		// model is not gated on it.
428	1721x	reactor_io_locking_ = cfg.reactor_io_locking;
429	1721x	dispatch_mutex_.set_enabled(cfg.scheduler_locking);
430	1721x	ring_mutex_.set_enabled(cfg.reactor_io_locking);
431	1721x	cond_.set_enabled(cfg.scheduler_locking);
432	1721x	}
433
434		/** Configure SQPOLL parameters.
435
436		Must be called before the ring is created — the values are
437		cached and read when it is. No-op if `enable` is false (the
438		default).
439
440		@param enable Set IORING_SETUP_SQPOLL on ring init.
441		@param idle_ms sq_thread_idle in milliseconds; 0 = kernel
442		default (1ms).
443		@param cpu Pin the polling thread to this CPU; -1 to
444		not pin.
445		*/
446	2x	void configure_sqpoll(bool enable, unsigned idle_ms, int cpu) noexcept
447		{
448	2x	enable_sqpoll_ = enable;
449	2x	sq_thread_idle_ms_ = idle_ms;
450	2x	sq_thread_cpu_ = cpu;
451	2x	}
452
453		/// Return true when scheduler locking is disabled (fully-lockless tier).
454	10x	bool scheduler_locking_disabled() const noexcept override
455		{
456	10x	return scheduler_locking_disabled_;
457		}
458
459		private:
460		void on_timer_service_attached(timer_service& svc) override;
461
462		/// Destroy every queued operation without running it.
463		void destroy_queued(lock_type& lock) noexcept;
464
465		// ring_ + wakeup_eventfd_ are mutable so lazy_init_ring() (called
466		// from const contexts like post()) can populate them on first use.
467		mutable struct ::io_uring ring_{};
468		mutable int wakeup_eventfd_ = -1;
469
470		// dispatch_mutex_ protects completed_ops_, cond_, task_running_.
471		// ring_mutex_ protects every userspace touch of ring_ (SQ tail,
472		// CQ head): get_sqe / submit / submit_and_wait_timeout /
473		// for_each_cqe / cq_advance.
474		//
475		// process_completions runs under ring_mutex_ and briefly takes
476		// dispatch_mutex_ to splice into completed_ops_. The locks are
477		// never held simultaneously for the full duration of any other
478		// path's critical section, so no deadlock.
479		mutable mutex_type dispatch_mutex_{true};
480		mutable mutex_type ring_mutex_{true};
481		mutable event_type cond_{true};
482		mutable ready_queue completed_ops_;
483		// outstanding_work_ and uring_inflight_ are both atomic
484		// counters updated at high frequency on different paths:
485		// - outstanding_work_ : every work_started / work_finished call,
486		// including timers, posts, and SQE submits.
487		// - uring_inflight_ : only SQE submit + non-F_MORE CQE consume.
488		// Under multi-thread workloads the threads tend to update these
489		// from different code paths; placing them on the same cache line
490		// would cause false sharing and unnecessary cache-line ping-pong.
491		// Hold each on its own line.
492		alignas(64) mutable std::atomic<std::int64_t> outstanding_work_{0};
493		// Count of io_uring SQEs in flight whose completion must be reaped.
494		// Excludes the wakeup-eventfd poll (registered in lazy_init_ring), and
495		// is updated by uring_submit_op and by process_completions on
496		// each non-F_MORE, non-eventfd CQE. shutdown() waits for it to drain.
497		alignas(64) mutable std::atomic<std::int64_t> uring_inflight_{0};
498		std::atomic<bool> stopped_{false};
499		// True while a thread holds task_op_ and runs a ring pass.
500		// Protected by dispatch_mutex_.
501		mutable bool task_running_ = false;
502		bool scheduler_locking_disabled_ = false;
503		bool reactor_io_locking_ = true;
504		bool enable_sqpoll_ = false;
505		unsigned sq_thread_idle_ms_ = 0;
506		int sq_thread_cpu_ = -1;
507
508		int cancel_sentinel_ = 0;
509
510		std::vector<uring_op*> deferred_cancels_; // guarded by ring_mutex_
511		std::atomic<bool> has_deferred_cancels_{false};
512
513		/// io_uring_submit, retried while a signal or a full completion
514		/// queue delays it. @pre ring_mutex_ is held.
515		int submit_retrying() noexcept;
516
517		// Submit as submit_retrying() does, and under SQPOLL also wait for
518		// the kernel thread to take every queued entry: an SQE names its
519		// file by number until it is issued, so the number must stay open
520		// until then.
521		int submit_issued() noexcept;
522
523		/// Queue a cancel of @p op by user_data; return false if the SQ
524		/// stays full. @pre ring_mutex_ is held.
525		bool queue_cancel_locked(uring_op* op) noexcept;
526
527		/// Queue the deferred cancels; return false if one still cannot
528		/// be. @pre ring_mutex_ is held.
529		bool queue_deferred_cancels() noexcept;
530
531		/// Turn every SQE on @p fd the kernel has not taken into a no-op,
532		/// completing its operation as cancelled. @pre ring_mutex_ is held.
533		void neutralize_pending_locked(int fd) noexcept;
534
535		// Signal self-pipe integration. The read end is watched via a multishot
536		// POLL SQE tagged with &signal_pipe_sentinel_ (distinct from nullptr =
537		// wakeup eventfd and &cancel_sentinel_). On its CQE we re-arm the poll if
538		// needed and enqueue signal_drain_op_ so the drain+deliver runs in
539		// dispatch context — never under ring_mutex_ — keeping deliver_signal's
540		// mutex locking off the ring critical section.
541		int signal_pipe_read_fd_ = -1;
542		int signal_pipe_sentinel_ = 0;
543
544		/// Dispatch-context op that drains the signal self-pipe and delivers each
545		/// pending signal. Enqueued (once at a time, guarded by queued_) from
546		/// process_completions when the poll CQE fires. Scheduler-owned; destroy()
547		/// is a no-op.
548		struct signal_drain_op final : scheduler_op
549		{
550		std::atomic<bool> queued_{false};
551
552	365x	void operator()() override
553		{
554		// Clear before draining so a signal that arrives mid-drain re-arms
555		// the op via a fresh CQE rather than being lost.
556	365x	queued_.store(false, std::memory_order_release);
557	365x	posix_signal_detail::drain_signal_pipe();
558	365x	}
559
560	274x	void destroy() override {}
561		};
562		mutable signal_drain_op signal_drain_op_;
563
564		/// Flushes the SQ ring and drains CQEs in one mutex-held pass.
565		/// One instance covers a whole batch; subsequent SQEs in the same
566		/// batch skip the post, amortising syscall cost across the batch.
567		/// Mirrors Asio's `submit_sqes_op` (`io_uring_service.ipp:730-742`).
568		struct submit_sqes_op final : scheduler_op
569		{
570		uring_scheduler* sched_ = nullptr;
571
572	1721x	submit_sqes_op() noexcept : scheduler_op(&do_handler) {}
573
574		// do_one dispatches through operator(); do_handler alone
575		// serves only the destroy path.
576	5035x	void operator()() override
577		{
578	5035x	do_handler(sched_, this, 0, 0);
579	5035x	}
580
581		static void do_handler(
582		void* owner,
583		scheduler_op* base,
584		std::uint32_t /bytes/,
585		std::uint32_t /error/) noexcept;
586		};
587
588		/// True between the first submitter of a batch posting `submit_op_`
589		/// and the dispatched op clearing the flag inside its handler.
590		/// Written only while holding `ring_mutex_`.
591		mutable std::atomic<bool> submit_op_posted_{false};
592
593		mutable std::atomic<int> runners_{0};
594
595		/// Single embedded `submit_sqes_op` instance, owned by the scheduler.
596		mutable submit_sqes_op submit_op_;
597
598		/// Sentinel op that triggers a ring pass when dequeued.
599		struct task_op final : scheduler_op
600		{
601		// LCOV_EXCL_START: the sentinel is intercepted by pointer
602		// identity; its virtuals exist for vtable completeness.
603	−	void operator()() override {}
604	−	void destroy() override {}
605		// LCOV_EXCL_STOP
606		};
607
608		// Exactly one instance circulates through completed_ops_, so the
609		// ring is entered once per pass over the queue rather than once
610		// per handler.
611		mutable task_op task_op_;
612
613		// ring_inited_ goes true once the ring exists. The init is deferred
614		// from the constructor so configure_threading() and configure_sqpoll()
615		// can take effect before io_uring_queue_init_params chooses flags.
616		mutable std::once_flag ring_init_once_;
617		mutable bool ring_inited_ = false;
618
619		std::size_t do_one(long timeout_us);
620
621		/** Reap the ring once, then expire due timers.
622
623		A zero timeout only peeks the completion queue, which needs no
624		kernel entry unless the kernel holds completions back.
625
626		@throws std::system_error If the kernel wait fails.
627		*/
628		void run_task(long timeout_us);
629
630		void process_completions();
631		void drain_wakeup_eventfd() const noexcept;
632		bool prep_multishot_poll(int fd, void* data) noexcept;
633		void lazy_init_ring_unlocked() const;
634		};
635
636	1721x	inline uring_scheduler::uring_scheduler(
637	1721x	capy::execution_context&, int /concurrency_hint/)
638		{
639		// sched_ cannot be set in the member initialiser — `this` is not
640		// available there.
641	1721x	submit_op_.sched_ = this;
642	1721x	completed_ops_.push(&task_op_);
643
644		// Ring init is deferred so the options the io_context applies
645		// after this constructor — the locking tier and SQPOLL — can feed
646		// the flags io_uring_queue_init_params is given. The io_context
647		// calls init_ring() once they are in place; the lazy_init_ring()
648		// calls scattered through the op paths only matter for a
649		// scheduler used without one.
650	1721x	}
651
652	3442x	inline uring_scheduler::~uring_scheduler()
653		{
654		// Services give up their work in shutdown(), but anything posted
655		// after the drain, such as from a destructor, is still destroyed.
656		{
657	1721x	lock_type lock(dispatch_mutex_);
658	1721x	destroy_queued(lock);
659	1721x	}
660	1721x	if (ring_inited_)
661		{
662		// Ring teardown can still run a doomed pipe write as task
663		// work; absorb the SIGPIPE it would raise.
664	1715x	scoped_sigpipe_block no_sigpipe;
665	1715x	if (wakeup_eventfd_ >= 0)
666	1715x	::close(wakeup_eventfd_);
667	1715x	::io_uring_queue_exit(&ring_);
668	1715x	}
669	3442x	}
670
671		inline void
672	146560x	uring_scheduler::lazy_init_ring() const
673		{
674	148281x	std::call_once(ring_init_once_, [this] { lazy_init_ring_unlocked(); });
675	146554x	}
676
677		inline void
678	1721x	uring_scheduler::lazy_init_ring_unlocked() const
679		{
680		// No flags bind the ring to one thread, even in the lockless tiers,
681		// so a context may be made, run and destroyed on different threads,
682		// and its teardown can always drain the ring.
683	1721x	io_uring_params params{};
684
685	1721x	if (enable_sqpoll_)
686		{
687		// SQPOLL forks a kernel thread that busy-polls the SQ ring;
688		// submission becomes a userspace-only memory store. Idle
689		// timeout 0 means kernel default (1ms); we only forward when
690		// explicitly set so the kernel default is preserved.
691	2x	params.flags \|= IORING_SETUP_SQPOLL;
692	2x	if (sq_thread_idle_ms_ != 0)
693	1x	params.sq_thread_idle = sq_thread_idle_ms_;
694	2x	if (sq_thread_cpu_ >= 0)
695		{
696	✗	params.flags \|= IORING_SETUP_SQ_AFF;
697	✗	params.sq_thread_cpu = static_cast<__u32>(sq_thread_cpu_);
698		}
699		}
700
701		// SUBMIT_ALL keeps a batch going past an entry the kernel rejects,
702		// so a cancel queued behind it is still issued.
703		//
704		// The kernel finishes a socket completion on the thread that
705		// submitted its entry, and a batch submit makes one thread the
706		// submitter for its peers' entries too. Without COOP_TASKRUN the
707		// kernel interrupts that thread for each completion while it runs
708		// handlers; with it, the work waits for the thread's next kernel
709		// entry, and a sleeping thread is still woken. TASKRUN_FLAG marks
710		// that waiting work in the ring, so a pass that only peeks still
711		// enters the kernel to finish it rather than missing completions
712		// while handlers keep the thread in user space. SQPOLL rings already
713		// work that way and refuse both flags.
714		//
715		// Kernels that predate a flag refuse it. The two TASKRUN flags are
716		// the newer, so they are dropped first.
717	1721x	io_uring_params const base = params;
718	3442x	unsigned const coop = enable_sqpoll_
719	1721x	? 0u
720		: IORING_SETUP_COOP_TASKRUN \| IORING_SETUP_TASKRUN_FLAG;
721	1721x	params.flags \|= IORING_SETUP_SUBMIT_ALL \| coop;
722	1721x	int rc = ::io_uring_queue_init_params(256, &ring_, &params);
723	1721x	if (rc == -EINVAL && coop != 0)
724		{
725	✗	params = base;
726	✗	params.flags \|= IORING_SETUP_SUBMIT_ALL;
727	✗	rc = ::io_uring_queue_init_params(256, &ring_, &params);
728		}
729	1721x	if (rc == -EINVAL)
730		{
731	✗	params = base;
732	✗	rc = ::io_uring_queue_init_params(256, &ring_, &params);
733		}
734	1721x	if (rc < 0)
735	1x	detail::throw_system_error(make_err(-rc), "io_uring_queue_init_params");
736
737	1720x	wakeup_eventfd_ = ::eventfd(0, EFD_NONBLOCK \| EFD_CLOEXEC);
738	1720x	if (wakeup_eventfd_ < 0)
739		{
740	1x	int errn = errno;
741	1x	::io_uring_queue_exit(&ring_);
742	1x	detail::throw_system_error(make_err(errn), "eventfd");
743		}
744
745		// Register a one-shot poll on the wake eventfd. user_data nullptr
746		// is the sentinel recognized by process_completions, which calls
747		// drain_wakeup_eventfd() to consume the eventfd byte AND re-arm
748		// the poll. Plan 5a switched away from IORING_POLL_MULTISHOT
749		// because multishot ops can silently terminate (e.g. under CQ
750		// pressure), and we don't observe the termination — leaving the
751		// wake mechanism dead and the leader stuck in kernel wait. One-
752		// shot rearm-on-fire is fail-fast: every wake event is paired
753		// with an explicit rearm, so a missed rearm would manifest
754		// immediately as the next wake being lost (test-visible).
755	1719x	::io_uring_sqe* sqe = ::io_uring_get_sqe(&ring_);
756	1719x	if (!sqe)
757		{
758	3x	::close(wakeup_eventfd_);
759	3x	wakeup_eventfd_ = -1;
760	3x	::io_uring_queue_exit(&ring_);
761	3x	detail::throw_system_error(
762	6x	make_err(ENOSPC), "io_uring_get_sqe (wakeup)");
763		}
764		// Multishot poll: fires a CQE on each eventfd POLLIN without
765		// consuming the SQE. Avoids the re-arm hazard of one-shot poll
766		// (where drain_wakeup_eventfd's get_sqe could return null on a
767		// full SQ, leaving no SQE to detect future wakes).
768	1716x	::io_uring_prep_poll_multishot(sqe, wakeup_eventfd_, POLLIN);
769	1716x	::io_uring_sqe_set_data(sqe, nullptr);
770	1716x	int submit_rc = ::io_uring_submit(&ring_);
771	1716x	if (submit_rc < 0)
772		{
773	1x	::close(wakeup_eventfd_);
774	1x	wakeup_eventfd_ = -1;
775	1x	::io_uring_queue_exit(&ring_);
776	1x	detail::throw_system_error(
777	2x	make_err(-submit_rc), "io_uring_submit (wakeup)");
778		}
779
780	1715x	ring_inited_ = true;
781	1715x	}
782
783		inline void
784	1721x	uring_scheduler::shutdown()
785		{
786	1721x	stopped_.store(true, std::memory_order_release);
787
788		// Cancel every request still in the kernel and wait for each to
789		// come back: the services free their impls next, and an op still
790		// armed would write into freed memory. One cancel-any covers ops
791		// whose own service-side cancel could not be queued. The kernel
792		// answers every cancel, so once it is submitted the wait ends; a
793		// ring that will not take it leaves nothing that could end a wait.
794		// A signal or a full completion queue only delays either step.
795	1721x	if (ring_inited_)
796		{
797	1715x	lock_type ring_lock(ring_mutex_);
798	1715x	io_uring_sqe* sqe = ::io_uring_get_sqe(&ring_);
799	1715x	if (!sqe)
800		{
801	3x	submit_issued();
802	3x	sqe = ::io_uring_get_sqe(&ring_);
803		}
804	1715x	if (sqe)
805		{
806	1715x	::io_uring_prep_cancel64(sqe, 0, IORING_ASYNC_CANCEL_ANY);
807	1715x	::io_uring_sqe_set_data(sqe, &cancel_sentinel_);
808	1715x	inflight_inc();
809		}
810	1715x	int const submitted = sqe ? submit_retrying() : -EINVAL;
811	6878x	while (submitted >= 0 &&
812	6878x	uring_inflight_.load(std::memory_order_acquire) > 0)
813		{
814	1724x	::io_uring_cqe* cqe = nullptr;
815	1724x	int const rc = ::io_uring_wait_cqe_timeout(&ring_, &cqe, nullptr);
816	1724x	if (rc == -EINTR \|\| rc == -EAGAIN \|\| rc == -ETIME)
817	2x	continue;
818	1722x	if (rc != 0)
819	✗	break;
820	1722x	process_completions();
821		}
822
823	1715x	deferred_cancels_.clear();
824	1715x	has_deferred_cancels_.store(false, std::memory_order_release);
825	1715x	}
826
827	1721x	wait_cancels_posting();
828
829		// Drain posted ops, including the completions reaped above,
830		// calling destroy() on each so embedded handles (coroutine frames,
831		// error_code outputs) get torn down rather than leaked. Mirrors
832		// reactor_scheduler::shutdown_drain.
833	1721x	lock_type lock(dispatch_mutex_);
834	1721x	destroy_queued(lock);
835	1721x	cond_.notify_all();
836	1721x	}
837
838		inline void
839	3442x	uring_scheduler::destroy_queued(lock_type& lock) noexcept
840		{
841	10661x	while (auto e = completed_ops_.pop())
842		{
843	7219x	if (ready_is_continuation(e))
844		{
845	3924x	lock.unlock();
846	3924x	if (auto* c = ready_as_cont(e); c->h)
847	3924x	capy::abandon(*c);
848	3924x	lock.lock();
849		}
850		else
851		{
852	3295x	auto* op = ready_as_op(e);
853	3295x	if (op == &task_op_)
854	1721x	continue;
855	1574x	lock.unlock();
856	1574x	op->destroy();
857	1574x	lock.lock();
858		}
859	7219x	}
860	3442x	}
861
862		inline void
863	2665x	uring_scheduler::stop()
864		{
865		// All under the lock: releasing the last work on another thread
866		// gets here, and a run thread that sees stopped_ may return and
867		// let this scheduler be destroyed.
868	2665x	lock_type lock(dispatch_mutex_);
869	2665x	stopped_.store(true, std::memory_order_release);
870	2665x	cond_.notify_all();
871		// Force-wake unconditionally — bypass interrupt_reactor's CAS
872		// coalescing. A dropped wake here leaves the leader blocked
873		// forever in submit_and_wait_timeout (no further CQE will
874		// arrive after stop()). With multishot poll on wakeup_eventfd_,
875		// this write reliably produces a CQE.
876	2665x	if (ring_inited_)
877		{
878	2665x	std::uint64_t v = 1;
879	2665x	[[maybe_unused]] auto r = ::write(wakeup_eventfd_, &v, sizeof(v));
880		}
881	2665x	}
882
883		inline bool
884	485x	uring_scheduler::stopped() const noexcept
885		{
886	485x	return stopped_.load(std::memory_order_acquire);
887		}
888
889		inline void
890	1863x	uring_scheduler::restart()
891		{
892	1863x	stopped_.store(false, std::memory_order_release);
893	1863x	}
894
895		inline void
896	47097x	uring_scheduler::work_started() noexcept
897		{
898	47097x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
899	47097x	}
900
901		inline void
902	171121x	uring_scheduler::work_finished() noexcept
903		{
904	342242x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
905	2097x	stop();
906	171121x	}
907
908		inline void
909	10864x	uring_scheduler::interrupt_reactor() const noexcept
910		{
911		// Skip if the ring hasn't been initialised yet — there's no leader
912		// to wake and no eventfd to write.
913	10864x	if (!ring_inited_)
914	179x	return;
915
916		// Only the unsafe tier forbids cross-thread post(); there the caller
917		// is the run thread itself, which is not in a kernel wait. Under
918		// unsafe_io a pool thread (the resolver) may post, and the eventfd
919		// write leaves the single ring submitter untouched.
920	10864x	if (scheduler_locking_disabled_)
921	✗	return;
922
923		// Under unsafe_io a single thread runs the ring; when that thread is
924		// the caller it is not in a kernel wait.
925	10864x	if (!reactor_io_locking_ && running_scheduler.get() == this)
926	179x	return;
927
928		// Multi-thread: write the eventfd unconditionally. CAS-coalescing
929		// is unsafe here because the leader's Phase 2 in do_one waits
930		// indefinitely for a CQE; a dropped wake leaves the leader
931		// blocked forever when there is no other CQE-producing activity.
932		// Multishot poll on wakeup_eventfd_ delivers a CQE for every
933		// write, so multiple writes in flight produce multiple CQEs
934		// (drained together by drain_wakeup_eventfd's single read of
935		// the eventfd counter).
936	10685x	std::uint64_t v = 1;
937	10685x	std::ignore = ::write(wakeup_eventfd_, &v, sizeof(v));
938		}
939
940		inline void
941	376x	uring_scheduler::on_timer_service_attached(timer_service& svc)
942		{
943	376x	svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) {
944	3959x	static_cast<uring_scheduler*>(p)->interrupt_reactor();
945	3959x	}));
946	376x	}
947
948		inline void
949	13899x	uring_scheduler::drain_wakeup_eventfd() const noexcept
950		{
951		std::uint64_t v;
952	13899x	std::ignore = ::read(wakeup_eventfd_, &v, sizeof(v));
953
954		// Multishot poll never needs re-arming. The poll-add was queued
955		// once at lazy_init_ring with IORING_POLL_ADD_MULTI; each eventfd
956		// POLLIN produces a CQE without consuming the SQE.
957	13899x	}
958
959		inline bool
960	2270x	uring_scheduler::prep_multishot_poll(int fd, void* data) noexcept
961		{
962		// Prepare a multishot POLLIN SQE on `fd` tagged with `data`. Caller holds
963		// ring_mutex_ and flushes separately (re-arm sites ride the batch submit;
964		// register/init submit explicitly). Returns false when no SQE could be
965		// had after one flush, so a caller that has someone to report to does
966		// not mistake the submit of an empty queue for an armed poll. Shared by
967		// the wakeup-eventfd and signal self-pipe multishot polls.
968	2270x	::io_uring_sqe* sqe = ::io_uring_get_sqe(&ring_);
969	2270x	if (!sqe)
970		{
971	2x	::io_uring_submit(&ring_);
972	2x	sqe = ::io_uring_get_sqe(&ring_);
973		}
974	2270x	if (!sqe)
975	1x	return false;
976	2269x	::io_uring_prep_poll_multishot(sqe, fd, POLLIN);
977	2269x	::io_uring_sqe_set_data(sqe, data);
978	2269x	return true;
979		}
980
981		inline std::error_code
982	277x	uring_scheduler::register_signal_reader(int read_fd)
983		{
984		// Called once per service from add_signal(), holding neither the
985		// signal_state mutex nor the service mutex (see the call site). Submit a
986		// multishot POLL on the pipe read end; its CQE (tagged with
987		// &signal_pipe_sentinel_) is recognised in process_completions. The actual
988		// drain+deliver is deferred to dispatch context, so the only lock taken
989		// here is ring_mutex_ with no outer lock held, hence no lock-order
990		// inversion with the reactor drain path.
991	277x	signal_pipe_read_fd_ = read_fd;
992	277x	lazy_init_ring();
993
994	277x	lock_type lock(ring_mutex_);
995		// EAGAIN, not ENOBUFS: an exhausted submission queue reports the
996		// same retryable code wherever it is hit, and the next add() is
997		// the retry.
998	277x	if (!prep_multishot_poll(read_fd, &signal_pipe_sentinel_))
999	1x	return make_err(EAGAIN);
1000	276x	int rc = ::io_uring_submit(&ring_);
1001	276x	if (rc < 0)
1002	1x	return make_err(-rc);
1003	275x	return {};
1004	277x	}
1005
1006		inline void
1007	1840x	uring_scheduler::post(std::coroutine_handle<> h) const
1008		{
1009		struct post_handler final : scheduler_op
1010		{
1011		std::coroutine_handle<> h_;
1012	1840x	explicit post_handler(std::coroutine_handle<> h) noexcept : h_(h) {}
1013
1014	1832x	void operator()() override
1015		{
1016	1832x	auto saved = h_;
1017	1832x	delete this;
1018	1832x	saved.resume();
1019	1832x	}
1020
1021	8x	void destroy() override
1022		{
1023		// A bare handle carries no owner to unwind from: it is
1024		// destroyed directly.
1025	8x	auto saved = h_;
1026	8x	delete this;
1027	8x	if (saved)
1028	8x	saved.destroy();
1029	8x	}
1030		};
1031
1032	1840x	auto* op = new post_handler(h);
1033	1840x	lazy_init_ring();
1034	1840x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
1035		{
1036	1840x	lock_type lock(dispatch_mutex_);
1037	1840x	completed_ops_.push(op);
1038	1840x	wake_one_locked();
1039	1840x	}
1040	1840x	}
1041
1042		inline void
1043	13693x	uring_scheduler::post(scheduler_op* op) const
1044		{
1045	13693x	lazy_init_ring();
1046	13693x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
1047		{
1048	13693x	lock_type lock(dispatch_mutex_);
1049	13693x	completed_ops_.push(op);
1050	13693x	wake_one_locked();
1051	13693x	}
1052	13693x	}
1053
1054		inline void
1055	115595x	uring_scheduler::post(capy::continuation& c) const
1056		{
1057	115595x	lazy_init_ring();
1058	115595x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
1059		{
1060	115595x	lock_type lock(dispatch_mutex_);
1061	115595x	completed_ops_.push(c);
1062	115595x	wake_one_locked();
1063	115595x	}
1064	115595x	}
1065
1066		inline void
1067	131134x	uring_scheduler::wake_one_locked() const noexcept
1068		{
1069	131134x	if (!task_running_)
1070	126842x	cond_.notify_one();
1071		else
1072	4292x	interrupt_reactor();
1073	131134x	}
1074
1075		// Thread-local stack of frames for io_uring schedulers being run on the
1076		// current thread. Holds the running-scheduler pointer (for
1077		// running_in_this_thread reporting) and the inline completion budget
1078		// used by the speculative non-blocking I/O path (plan 5j). Nesting
1079		// stacks frames via prev_ so each scheduler gets its own budget.
1080		struct uring_scheduler_frame
1081		{
1082		uring_scheduler const* sched;
1083		uring_scheduler_frame* prev;
1084		int inline_budget;
1085		int inline_budget_max;
1086		};
1087
1088		inline thread_local uring_scheduler_frame* tl_running_scheduler_frame_ =
1089		nullptr;
1090
1091		// Default inline budget. Matches reactor's initial budget (2). Adaptive
1092		// ramp-up to a max is intentionally NOT implemented yet — keep it simple
1093		// for plan 5j and revisit if benches show fairness issues.
1094		inline constexpr int uring_inline_budget_initial = 2;
1095		inline constexpr int uring_inline_budget_max = 16;
1096
1097		/// RAII guard: pushes a frame onto the thread's running-scheduler stack
1098		/// on construction, restores the previous on destruction. Used by
1099		/// run/run_one/wait_one/poll/poll_one to mark the running thread and
1100		/// hold a fresh inline budget for speculative completions.
1101		struct uring_run_guard
1102		{
1103		uring_scheduler_frame frame_;
1104		running_scheduler_guard running_;
1105
1106	4063x	explicit uring_run_guard(uring_scheduler const* self) noexcept
1107	4063x	: frame_{
1108		self, tl_running_scheduler_frame_, uring_inline_budget_initial,
1109		uring_inline_budget_max}
1110	4063x	, running_(self)
1111		{
1112	4063x	tl_running_scheduler_frame_ = &frame_;
1113	4063x	self->runner_entered();
1114	4063x	}
1115
1116	4063x	~uring_run_guard() noexcept
1117		{
1118	4063x	frame_.sched->runner_exited();
1119	4063x	tl_running_scheduler_frame_ = frame_.prev;
1120	4063x	}
1121		};
1122
1123		inline bool
1124	28227x	uring_scheduler::running_in_this_thread() const noexcept
1125		{
1126	28357x	for (auto* f = tl_running_scheduler_frame_; f != nullptr; f = f->prev)
1127		{
1128	18720x	if (f->sched == this)
1129	18590x	return true;
1130		}
1131	9637x	return false;
1132		}
1133
1134		inline void
1135	174129x	uring_scheduler::reset_inline_budget() const noexcept
1136		{
1137	174129x	for (auto* f = tl_running_scheduler_frame_; f != nullptr; f = f->prev)
1138		{
1139	174129x	if (f->sched == this)
1140		{
1141	174129x	f->inline_budget = f->inline_budget_max;
1142	174129x	return;
1143		}
1144		}
1145		}
1146
1147		inline bool
1148	316970x	uring_scheduler::try_consume_inline_budget() const noexcept
1149		{
1150	316976x	for (auto* f = tl_running_scheduler_frame_; f != nullptr; f = f->prev)
1151		{
1152	316968x	if (f->sched == this)
1153		{
1154	316962x	if (f->inline_budget > 0)
1155		{
1156	298393x	--f->inline_budget;
1157	298393x	return true;
1158		}
1159	18569x	return false;
1160		}
1161		}
1162	8x	return false;
1163		}
1164
1165		inline std::size_t
1166	1639x	uring_scheduler::run()
1167		{
1168	1639x	lazy_init_ring();
1169	3278x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
1170		{
1171	115x	stop();
1172	115x	return 0;
1173		}
1174
1175	1524x	uring_run_guard guard(this);
1176	1524x	std::size_t n = 0;
1177		for (;;)
1178		{
1179	146511x	std::size_t r = do_one(-1);
1180	146509x	if (r)
1181		{
1182	144987x	if (n != (std::numeric_limits<std::size_t>::max)())
1183	144987x	++n;
1184	144987x	continue;
1185		}
1186	3064x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
1187	20x	stopped_.load(std::memory_order_acquire))
1188	1522x	break;
1189		// do_one returned 0 but work still outstanding (e.g. timer
1190		// expiry dispatched async work). Continue.
1191	144987x	}
1192	1522x	return n;
1193	1524x	}
1194
1195		inline std::size_t
1196	79x	uring_scheduler::run_one()
1197		{
1198	79x	lazy_init_ring();
1199	158x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
1200		{
1201	1x	stop();
1202	1x	return 0;
1203		}
1204	78x	uring_run_guard guard(this);
1205	78x	return do_one(-1);
1206	78x	}
1207
1208		inline std::size_t
1209	2112x	uring_scheduler::wait_one(long usec)
1210		{
1211	2112x	lazy_init_ring();
1212	4224x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
1213		{
1214	418x	stop();
1215	418x	return 0;
1216		}
1217	1694x	uring_run_guard guard(this);
1218	1694x	return do_one(usec);
1219	1694x	}
1220
1221		inline std::size_t
1222	753x	uring_scheduler::poll()
1223		{
1224	753x	lazy_init_ring();
1225	1506x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
1226		{
1227	6x	stop();
1228	6x	return 0;
1229		}
1230	747x	uring_run_guard guard(this);
1231	747x	std::size_t n = 0;
1232	5071x	while (do_one(0))
1233		{
1234	4324x	if (n != (std::numeric_limits<std::size_t>::max)())
1235	4324x	++n;
1236		}
1237	747x	return n;
1238	747x	}
1239
1240		inline std::size_t
1241	22x	uring_scheduler::poll_one()
1242		{
1243	22x	lazy_init_ring();
1244	44x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
1245		{
1246	2x	stop();
1247	2x	return 0;
1248		}
1249	20x	uring_run_guard guard(this);
1250	20x	return do_one(0);
1251	20x	}
1252
1253		inline std::size_t
1254	153374x	uring_scheduler::do_one(long timeout_us)
1255		{
1256	153374x	if (stopped_.load(std::memory_order_acquire))
1257		{
1258		// stop() sets the flag under the lock and still writes the
1259		// eventfd; returning before it unlocks lets the caller destroy
1260		// this scheduler under it.
1261	1511x	lock_type lock(dispatch_mutex_);
1262	1511x	return 0;
1263	1511x	}
1264
1265	151863x	lock_type lock(dispatch_mutex_);
1266		for (;;)
1267		{
1268	278042x	if (stopped_.load(std::memory_order_acquire))
1269	131x	return 0;
1270
1271	277911x	auto const e = completed_ops_.pop();
1272
1273	277911x	if (e && !ready_is_continuation(e) && ready_as_op(e) == &task_op_)
1274		{
1275	126355x	bool const more_handlers = !completed_ops_.empty();
1276
1277	134375x	if (!more_handlers &&
1278	16040x	outstanding_work_.load(std::memory_order_acquire) == 0)
1279		{
1280	1x	completed_ops_.push(&task_op_);
1281	1x	return 0;
1282		}
1283
1284		// A peer takes the queued handlers while this thread is in
1285		// the ring.
1286	126354x	if (more_handlers)
1287	118335x	cond_.notify_one();
1288	126354x	task_running_ = true;
1289	126354x	lock.unlock();
1290
1291		try
1292		{
1293		// Queued handlers must not wait behind a kernel block.
1294	126354x	run_task(more_handlers ? 0 : timeout_us);
1295		}
1296	2x	catch (...)
1297		{
1298		// Restore state so followers don't wait for a leader
1299		// that never returns.
1300	2x	lock.lock();
1301	2x	task_running_ = false;
1302	2x	completed_ops_.push(&task_op_);
1303	2x	cond_.notify_all();
1304	2x	throw;
1305	2x	}
1306
1307	126352x	lock.lock();
1308	126352x	task_running_ = false;
1309	126352x	bool const reaped = !completed_ops_.empty();
1310	126352x	completed_ops_.push(&task_op_);
1311
1312		// poll() and wait_one() make one ring pass, handing the
1313		// sentinel to a waiting peer.
1314	126352x	if (timeout_us >= 0 && !reaped)
1315		{
1316	656x	cond_.notify_one();
1317	656x	return 0;
1318		}
1319	125696x	continue;
1320	125696x	}
1321
1322	151556x	if (e)
1323		{
1324		// Hand off any remaining queued work to a follower so we
1325		// dispatch in parallel.
1326	151072x	if (!completed_ops_.empty())
1327	150559x	cond_.notify_one();
1328	151072x	lock.unlock();
1329		// Speculative follow-ups in the handler share this budget.
1330	151072x	reset_inline_budget();
1331	151072x	if (ready_is_continuation(e))
1332	111671x	ready_as_cont(e)->h.resume();
1333		else
1334	39401x	(*ready_as_op(e))();
1335	151072x	work_finished();
1336	151072x	return 1;
1337		}
1338
1339	968x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
1340	✗	return 0;
1341
1342		// LCOV_EXCL_START: follower path — entered only when another
1343		// thread holds the sentinel at the instant this thread finds
1344		// the queue empty. That overlap is a race no deterministic
1345		// test reliably forces (see the io_uring mt_scheduler
1346		// regression test, which exercises MT run but does not pin
1347		// this window).
1348	−	if (timeout_us == 0)
1349	−	return 0;
1350	−	if (timeout_us < 0)
1351	−	cond_.wait(lock);
1352		else
1353		{
1354	−	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
1355		// wait_one honoured its timeout; if nothing arrived,
1356		// return rather than re-arm.
1357	−	if (completed_ops_.empty() &&
1358	−	!stopped_.load(std::memory_order_acquire))
1359	−	return 0;
1360		}
1361		// LCOV_EXCL_STOP
1362	126179x	}
1363	151863x	}
1364
1365		inline void
1366	126354x	uring_scheduler::run_task(long timeout_us)
1367		{
1368	126354x	auto* tsvc = timers();
1369
1370	126354x	if (timeout_us == 0)
1371		{
1372	118962x	lock_type ring_lock(ring_mutex_);
1373	118962x	if (has_deferred_cancels_.load(std::memory_order_acquire))
1374	5x	queue_deferred_cancels();
1375		// A queued submit_op_ flushes its batch; anything else in the
1376		// SQ (cancels, poll re-arms) would otherwise wait for the
1377		// queue to drain.
1378	232994x	if (!submit_op_posted_.load(std::memory_order_relaxed) &&
1379	114032x	::io_uring_sq_ready(&ring_) != 0)
1380	66x	::io_uring_submit_and_get_events(&ring_);
1381		else
1382		{
1383	118896x	::io_uring_cqe* cqe = nullptr;
1384	118896x	std::ignore = ::io_uring_peek_batch_cqe(&ring_, &cqe, 1);
1385		}
1386	118962x	process_completions();
1387	118962x	}
1388		else
1389		{
1390	7392x	__kernel_timespec ts{};
1391	7392x	__kernel_timespec* ts_ptr = nullptr;
1392		auto next_expiry =
1393	7392x	tsvc ? tsvc->nearest_expiry() : timer_service::time_point::max();
1394	7392x	auto now = std::chrono::steady_clock::now();
1395
1396	7392x	if (next_expiry != timer_service::time_point::max())
1397		{
1398		// Timers expire after the wait, so a past deadline can be as
1399		// early as time_point::min(), where the subtraction overflows.
1400	6106x	std::chrono::nanoseconds::rep delta_ns = 0;
1401	6106x	if (next_expiry > now)
1402	5995x	delta_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
1403	11990x	next_expiry - now)
1404	5995x	.count();
1405	6106x	if (timeout_us > 0 && timeout_us < delta_ns / 1000)
1406	4x	delta_ns = static_cast<decltype(delta_ns)>(timeout_us) * 1000;
1407	6106x	ts.tv_sec = delta_ns / 1'000'000'000;
1408	6106x	ts.tv_nsec = delta_ns % 1'000'000'000;
1409	6106x	ts_ptr = &ts;
1410		}
1411	1286x	else if (timeout_us > 0)
1412		{
1413	411x	ts.tv_sec = timeout_us / 1'000'000;
1414	411x	ts.tv_nsec = (timeout_us % 1'000'000) * 1000;
1415	411x	ts_ptr = &ts;
1416		}
1417		else
1418		{
1419		// run() with no pending timers: cap the kernel wait at 1s
1420		// so the leader periodically re-checks state. Defense in
1421		// depth against a lost wakeup (e.g. multishot poll on the
1422		// wakeup eventfd terminates and the re-arm SQE doesn't
1423		// reach the kernel in time). Worst case: one extra
1424		// wake-up per io_context per second when truly idle.
1425	875x	ts.tv_sec = 1;
1426	875x	ts.tv_nsec = 0;
1427	875x	ts_ptr = &ts;
1428		}
1429
1430		// Three-phase kernel wait, matching Boost.Asio's
1431		// io_uring_service::run pattern. ring_mutex_ is held briefly
1432		// to push pending SQEs and to drain CQEs, but NOT during
1433		// the blocking io_uring_wait_cqe_timeout. Cross-thread
1434		// submitters (uring_submit_op, cancel paths) can take
1435		// ring_mutex_ during the wait and prep new SQEs without
1436		// blocking on the leader; their wake eventfd write fires the
1437		// multishot poll and returns the leader from wait_cqe_timeout
1438		// promptly.
1439		//
1440		// Phase 1 — submit any pending SQEs to the kernel.
1441		{
1442	7392x	lock_type ring_lock(ring_mutex_);
1443	7392x	if (has_deferred_cancels_.load(std::memory_order_acquire))
1444	3x	queue_deferred_cancels();
1445	7392x	::io_uring_submit(&ring_);
1446	7392x	}
1447
1448		// Phase 2 — wait for at least one CQE without holding the
1449		// mutex. wait_cqe_timeout only peeks the CQ ring;
1450		// head advancement happens under the mutex in
1451		// process_completions below.
1452	7392x	::io_uring_cqe* cqe = nullptr;
1453	7392x	int rc = ::io_uring_wait_cqe_timeout(&ring_, &cqe, ts_ptr);
1454
1455		// Phase 3 — drain CQEs under the mutex.
1456		// EBUSY: the CQ is full, so the kernel cannot flush overflowed
1457		// completions into it. Reaping makes room, so drain and wait
1458		// again.
1459	7392x	bool const transient = rc == -ETIME \|\| rc == -EINTR \|\| rc == -EBUSY;
1460		{
1461	7392x	lock_type ring_lock(ring_mutex_);
1462	7392x	if (rc == 0 \|\| transient)
1463	7390x	process_completions();
1464	7392x	}
1465
1466	7392x	if (rc < 0 && !transient)
1467	2x	detail::throw_system_error(
1468	4x	make_err(-rc), "io_uring_wait_cqe_timeout");
1469		}
1470
1471	126352x	if (tsvc && !tsvc->empty())
1472	17588x	tsvc->process_expired();
1473	126352x	}
1474
1475		inline void
1476	133109x	uring_scheduler::process_completions()
1477		{
1478		unsigned head;
1479		::io_uring_cqe* cqe;
1480	133109x	unsigned consumed = 0;
1481
1482		// Collect completed I/O ops locally; splice into completed_ops_
1483		// after the loop so do_one dispatches them one at a time.
1484	133109x	ready_queue local_ops;
1485
1486	133109x	std::int64_t inflight_dec = 0;
1487	162093x	io_uring_for_each_cqe(&ring_, head, cqe)
1488		{
1489	28984x	void* ud = io_uring_cqe_get_data(cqe);
1490	28984x	if (ud == nullptr)
1491		{
1492		// Wakeup eventfd CQE: drain the eventfd byte. Not counted
1493		// by uring_inflight_; we never incremented for the
1494		// wakeup multishot SQE (its progress doesn't depend on
1495		// userspace getevents).
1496	13899x	drain_wakeup_eventfd();
1497		// If multishot terminated (kernel dropped under memory
1498		// pressure or similar), re-arm. Each CQE except the last
1499		// sets IORING_CQE_F_MORE.
1500		// Best-effort: nothing here has a caller to report to, and
1501		// prep_multishot_poll already flushed once to make room.
1502	13899x	if ((cqe->flags & IORING_CQE_F_MORE) == 0)
1503	1717x	std::ignore = prep_multishot_poll(wakeup_eventfd_, nullptr);
1504		}
1505	15085x	else if (ud == &cancel_sentinel_)
1506		{
1507		// CQE for an ASYNC_CANCEL op — ignore; the actual op's
1508		// CQE arrives separately and is dispatched via cqe_func.
1509		// Cancels are one-shot, no F_MORE, decrement inflight.
1510	3534x	++inflight_dec;
1511		}
1512	11551x	else if (ud == &signal_pipe_sentinel_)
1513		{
1514		// Signal self-pipe readiness. Re-arm the multishot poll if it
1515		// terminated (F_MORE cleared), then enqueue signal_drain_op_ to
1516		// drain + deliver in dispatch context. Not counted in
1517		// uring_inflight_ (like the wakeup eventfd poll): it stays
1518		// armed for the ring's lifetime.
1519	1039x	if ((cqe->flags & IORING_CQE_F_MORE) == 0)
1520	276x	std::ignore = prep_multishot_poll(
1521	276x	signal_pipe_read_fd_, &signal_pipe_sentinel_);
1522	1039x	bool expected = false;
1523	1039x	if (signal_drain_op_.queued_.compare_exchange_strong(
1524		expected, true, std::memory_order_acq_rel,
1525		std::memory_order_relaxed))
1526		{
1527		// Balance the work_finished() do_one runs after dispatching.
1528	639x	work_started();
1529	639x	local_ops.push(&signal_drain_op_);
1530		}
1531		}
1532		else
1533		{
1534	10512x	auto* iop = static_cast<uring_op*>(ud);
1535		// Nothing is left for a cancel to reach.
1536	10512x	if ((cqe->flags & IORING_CQE_F_MORE) == 0)
1537	6203x	iop->sqe_set.store(false, std::memory_order_release);
1538	10512x	(*iop->cqe_func)(iop, cqe->res, cqe->flags, local_ops);
1539		// Decrement inflight on the terminal CQE only — multishot
1540		// ops (acceptor) hold the SQE alive across F_MORE CQEs and
1541		// free it only when F_MORE is cleared.
1542	10512x	if ((cqe->flags & IORING_CQE_F_MORE) == 0)
1543	6203x	++inflight_dec;
1544		}
1545	28984x	++consumed;
1546		}
1547	133109x	if (inflight_dec)
1548	6519x	uring_inflight_.fetch_sub(inflight_dec, std::memory_order_acq_rel);
1549
1550	133109x	if (consumed)
1551	14700x	io_uring_cq_advance(&ring_, consumed);
1552
1553		// Caller holds ring_mutex_. Take dispatch_mutex_ briefly to
1554		// splice locally-collected ops onto the global queue (lock order
1555		// ring_mutex_ -> dispatch_mutex_).
1556	133109x	if (!local_ops.empty())
1557		{
1558	5710x	lock_type lock(dispatch_mutex_);
1559	5710x	completed_ops_.splice(local_ops);
1560		// Wake any follower waiting on cond_; it'll pop and dispatch.
1561	5710x	cond_.notify_one();
1562	5710x	}
1563	133109x	}
1564
1565		inline void
1566	5074x	uring_scheduler::submit_sqes_op::do_handler(
1567		void* owner,
1568		scheduler_op* base,
1569		std::uint32_t /bytes/,
1570		std::uint32_t /error/) noexcept
1571		{
1572	5074x	if (owner == nullptr)
1573	39x	return; // shutdown drain — nothing to do; SQE storage is
1574		// kernel-mapped and discarded by io_uring_queue_exit.
1575
1576	5035x	auto* self = static_cast<submit_sqes_op*>(base);
1577	5035x	auto* sched = self->sched_;
1578
1579	5035x	uring_scheduler::lock_type ring_lock(sched->ring_mutex_);
1580	5035x	sched->submit_op_posted_.store(false, std::memory_order_relaxed);
1581	5035x	::io_uring_submit_and_get_events(&sched->ring_);
1582	5035x	sched->process_completions();
1583	5035x	}
1584
1585		inline void
1586	195x	uring_scheduler::submit_cancel_by_user_data(uring_op* target) noexcept
1587		{
1588	195x	lazy_init_ring();
1589		{
1590	195x	lock_type lock(ring_mutex_);
1591	195x	cancel_op_locked(target);
1592	195x	}
1593		// After queuing, so the woken leader's submit carries the cancel.
1594	195x	interrupt_reactor();
1595	195x	}
1596
1597		inline bool
1598	1827x	uring_scheduler::queue_cancel_locked(uring_op* target) noexcept
1599		{
1600	1827x	io_uring_sqe* sqe = io_uring_get_sqe(&ring_);
1601	1827x	if (!sqe)
1602		{
1603		// A flush alone frees no space under SQPOLL until the kernel
1604		// thread consumes it.
1605	18x	submit_issued();
1606	18x	sqe = io_uring_get_sqe(&ring_);
1607		}
1608	1827x	if (!sqe)
1609	12x	return false;
1610
1611	1815x	io_uring_prep_cancel(sqe, target, 0);
1612	1815x	io_uring_sqe_set_data(sqe, &cancel_sentinel_);
1613	1815x	inflight_inc();
1614	1815x	return true;
1615		}
1616
1617		inline void
1618	3905x	uring_scheduler::cancel_op_locked(uring_op* op) noexcept
1619		{
1620	3905x	if (!op->sqe_set.load(std::memory_order_acquire))
1621	2087x	return;
1622	1818x	if (queue_cancel_locked(op))
1623	1809x	return;
1624	9x	deferred_cancels_.push_back(op);
1625	9x	has_deferred_cancels_.store(true, std::memory_order_release);
1626		}
1627
1628		inline void
1629	921x	uring_scheduler::cancel_ops(std::initializer_list<uring_op*> ops) noexcept
1630		{
1631	921x	cancel_ops([&]() noexcept {
1632	3404x	for (auto* op : ops)
1633	2483x	cancel_op_locked(op);
1634	921x	});
1635	921x	}
1636
1637		template<class Cancel>
1638		void
1639	1495x	uring_scheduler::flush_after_cancel(int fd, Cancel&& cancel) noexcept
1640		{
1641		// The flush can execute a queued write on `fd` inline; when the
1642		// fd is a pipe whose reader has already closed — service
1643		// shutdown closes impls one at a time, so teardown itself
1644		// creates that state — the kernel raises SIGPIPE.
1645	1495x	scoped_sigpipe_block no_sigpipe;
1646
1647	1495x	lazy_init_ring();
1648	1495x	interrupt_reactor();
1649	1495x	lock_type lock(ring_mutex_);
1650	1495x	cancel();
1651	1495x	if (submit_issued() < 0)
1652	8x	neutralize_pending_locked(fd);
1653	1495x	}
1654
1655		inline void
1656	2142x	uring_op::on_cancel() noexcept
1657		{
1658		// Paired with uring_link_sqe: with no SQE linked yet, the
1659		// initiator either sees the flag and completes without the kernel,
1660		// or links and queues the cancel itself.
1661	2142x	cancelled.store(true, std::memory_order_seq_cst);
1662	2142x	if (sched_ && sqe_set.load(std::memory_order_seq_cst))
1663	195x	sched_->submit_cancel_by_user_data(this);
1664	2142x	}
1665
1666		inline int
1667	3306x	uring_scheduler::submit_retrying() noexcept
1668		{
1669		int r;
1670		do
1671		{
1672	3307x	r = ::io_uring_submit(&ring_);
1673	3307x	if (r == -EBUSY)
1674	✗	process_completions();
1675		}
1676	3307x	while (r == -EINTR \|\| r == -EAGAIN \|\| r == -EBUSY);
1677	3306x	return r;
1678		}
1679
1680		inline int
1681	1516x	uring_scheduler::submit_issued() noexcept
1682		{
1683	1516x	bool const sqpoll = (ring_.flags & IORING_SETUP_SQPOLL) != 0;
1684	1516x	int r = submit_retrying();
1685		// The kernel can take fewer entries than were flushed, and under
1686		// SQPOLL takes them later on its own thread.
1687	1591x	while (r >= 0 && ::io_uring_sq_ready(&ring_) > 0)
1688		{
1689	94x	if (sqpoll)
1690	75x	std::this_thread::yield();
1691	19x	else if (r == 0)
1692	19x	return -EBUSY;
1693		// Under SQPOLL this also wakes the kernel thread if it went idle.
1694	75x	r = submit_retrying();
1695		}
1696	1497x	return r;
1697		}
1698
1699		inline bool
1700	9x	uring_scheduler::queue_deferred_cancels() noexcept
1701		{
1702	9x	std::size_t kept = 0;
1703	18x	for (auto* op : deferred_cancels_)
1704	9x	if (!queue_cancel_locked(op))
1705	3x	deferred_cancels_[kept++] = op;
1706	9x	deferred_cancels_.resize(kept);
1707	9x	has_deferred_cancels_.store(kept != 0, std::memory_order_release);
1708	9x	return kept == 0;
1709		}
1710
1711		inline ::io_uring_sqe*
1712	6212x	uring_scheduler::acquire_sqe() noexcept
1713		{
1714	6213x	if (has_deferred_cancels_.load(std::memory_order_relaxed) &&
1715	1x	!queue_deferred_cancels())
1716	✗	return nullptr;
1717	6212x	::io_uring_sqe* sqe = ::io_uring_get_sqe(&ring_);
1718	6212x	if (!sqe)
1719		{
1720	5x	::io_uring_submit(&ring_);
1721	5x	sqe = ::io_uring_get_sqe(&ring_);
1722		}
1723	6212x	return sqe;
1724		}
1725
1726		inline void
1727	8x	uring_scheduler::neutralize_pending_locked(int fd) noexcept
1728		{
1729		// Without SQPOLL the kernel reads the SQ only inside
1730		// io_uring_enter, which runs under ring_mutex_, so everything from
1731		// its head on is still ours. The SQPOLL thread reads published
1732		// entries on its own; only the unpublished ones are safe to touch.
1733	8x	auto& sq = ring_.sq;
1734	8x	bool const sqpoll = (ring_.flags & IORING_SETUP_SQPOLL) != 0;
1735		unsigned const from =
1736	8x	sqpoll ? sq.sqe_head : __atomic_load_n(sq.khead, __ATOMIC_ACQUIRE);
1737
1738	8x	ready_queue local_ops;
1739	1547x	for (unsigned i = from; i != sq.sqe_tail; ++i)
1740		{
1741	1539x	auto* sqe = &sq.sqes[i & sq.ring_mask];
1742		// NOLINTNEXTLINE(performance-no-int-to-ptr) — user_data holds the address io_uring_sqe_set_data stored
1743	1539x	auto* ud = reinterpret_cast<void*>(sqe->user_data);
1744	1539x	if (sqe->fd != fd \|\| ud == nullptr \|\| ud == &cancel_sentinel_ \|\|
1745	4x	ud == &signal_pipe_sentinel_)
1746	1535x	continue;
1747	4x	auto* op = static_cast<uring_op*>(ud);
1748	4x	::io_uring_prep_nop(sqe);
1749	4x	::io_uring_sqe_set_data(sqe, &cancel_sentinel_);
1750	4x	op->sqe_set.store(false, std::memory_order_release);
1751	4x	(*op->cqe_func)(op, -ECANCELED, 0, local_ops);
1752		}
1753	8x	if (!local_ops.empty())
1754		{
1755	4x	lock_type lock(dispatch_mutex_);
1756	4x	completed_ops_.splice(local_ops);
1757	4x	cond_.notify_one();
1758	4x	}
1759	8x	}
1760
1761		} // namespace boost::corosio::detail
1762
1763		#endif // BOOST_COROSIO_HAS_URING
1764
1765		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_SCHEDULER_HPP
1766