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

88.6% Lines (405 / 457) 96.3% Functions (78 / 81)

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

uring_multishot_acceptor.hpp

f(x) Functions (81)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
3		//
4		// Distributed under the Boost Software License, Version 1.0. (See accompanying
5		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6		//
7		// Official repository: https://github.com/cppalliance/corosio
8		//
9
10		#ifndef BOOST_COROSIO_NATIVE_DETAIL_URING_URING_MULTISHOT_ACCEPTOR_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_MULTISHOT_ACCEPTOR_HPP
12
13		#include <boost/corosio/native/detail/endpoint_convert.hpp>
14		#include <boost/corosio/detail/config.hpp>
15		#include <boost/corosio/detail/platform.hpp>
16
17		#if BOOST_COROSIO_HAS_URING
18
19		#include <liburing.h>
20
21		#include <boost/corosio/detail/intrusive.hpp>
22		#include <boost/corosio/native/detail/uring/uring_acceptor_ops.hpp>
23		#include <boost/corosio/native/detail/uring/uring_buffer.hpp>
24		#include <boost/corosio/native/detail/uring/uring_op.hpp>
25		#include <boost/corosio/native/detail/uring/uring_scheduler.hpp>
26		#include <boost/corosio/native/detail/uring/uring_socket_ops.hpp>
27		#include <boost/corosio/native/detail/make_err.hpp>
28		#include <boost/corosio/detail/native_handle.hpp>
29		#include <boost/corosio/io/io_object.hpp>
30
31		#include <atomic>
32		#include <cstdint>
33		#include <coroutine>
34		#include <mutex>
35		#include <stop_token>
36		#include <system_error>
37		#include <utility>
38
39		#include <netinet/in.h>
40		#include <sys/socket.h>
41		#include <unistd.h>
42
43		namespace boost::corosio::detail {
44
45		/** Check whether a descriptor is in the listening state.
46
47		Multishot accept fails immediately on a non-listening socket and
48		the re-arm path would spin on that failure, so an adopted fd is
49		only armed when the kernel reports a listener. A query the
50		platform refuses is treated as listening so adoption still works.
51
52		@param fd The descriptor to probe.
53		@return True unless the kernel positively reports a non-listener.
54		*/
55		inline bool
56	8x	fd_is_listening(int fd) noexcept
57		{
58	8x	int accepting = 0;
59	8x	socklen_t alen = sizeof(accepting);
60	8x	if (::getsockopt(fd, SOL_SOCKET, SO_ACCEPTCONN, &accepting, &alen) != 0)
61	1x	return true;
62	7x	return accepting != 0;
63		}
64
65		template<
66		class Derived,
67		class ImplBase,
68		class Endpoint,
69		class PeerService,
70		class AcceptorService>
71		class uring_multishot_acceptor_base
72		: public ImplBase
73		, public intrusive_list<Derived>::node
74		{
75		protected:
76		struct ready_fd_node : intrusive_list<ready_fd_node>::node
77		{
78		int fd = -1;
79		sockaddr_storage peer{};
80		socklen_t peer_len = 0;
81		};
82
83		/** Pooled accept node: parked waiter and posted completion in one.
84
85		A parked accept used to be a `waiter_node` whose fields were
86		copied into a freshly allocated `uring_accept_op` at every
87		completion boundary. The two had strictly sequential lifetimes
88		(a waiter becomes a completion), so this node is both at
89		once — the intrusive hook parks it in `waiters_`/`free_nodes_`
90		and the inherited op posts it to the scheduler. Nodes recycle
91		through the owning acceptor's `free_nodes_` via `dispose`, so
92		steady-state accepts allocate nothing.
93
94		The inherited `cancelled` flag is the claim token: delivery,
95		arming-failure, and cancellation each claim the node with one
96		`exchange`, and the loser leaves completion to the winner.
97		Because the flag also decides how `do_handler` reports the
98		completion, every claiming path stores the final status after
99		`stop_cb.reset()` (which synchronizes with any in-flight
100		canceller) and before posting.
101
102		While in flight — parked or posted — the node holds an
103		`object_ref` on its acceptor (the inherited `object_ref_`
104		slot), so the impl cannot retire out from under `dispose`.
105		*/
106		struct accept_node
107		: uring_accept_op
108		, intrusive_list<accept_node>::node
109		{
110		Derived* owner = nullptr;
111		/// True once linked into `waiters_` (guarded by `mutex_`).
112		/// The stop callback is armed before the node is queued, so
113		/// cancel_waiter must not unlink a node it never queued.
114		bool queued = false;
115		/// A readiness wait rather than an accept: completion
116		/// observes a pending connection without consuming it.
117		bool peek = false;
118
119	728x	accept_node() noexcept
120	728x	{
121	728x	this->dispose = &recycle_thunk;
122	728x	}
123
124		/// Claim the node for cancellation (the stop token fired);
125		/// losing the exchange means a delivery already owns it.
126	221x	void on_cancel() noexcept override
127		{
128	221x	if (this->cancelled.exchange(true, std::memory_order_acq_rel))
129	✗	return;
130	221x	owner->cancel_waiter(this);
131		}
132
133	4066x	static void recycle_thunk(uring_accept_op* op) noexcept
134		{
135	4066x	auto* n = static_cast<accept_node*>(op);
136	4066x	n->owner->release_node(n);
137	4066x	}
138		};
139
140		int fd_ = -1;
141		uring_scheduler* sched_;
142		PeerService* peer_service_;
143		/// Owning acceptor service; exposes the recycling pool to
144		/// retire(). Not used for anything else — the service already
145		/// drives shutdown/destroy from its own side.
146		AcceptorService* acceptor_svc_;
147		Endpoint local_endpoint_{};
148		mutable std::mutex mutex_;
149		intrusive_list<ready_fd_node> ready_fds_;
150		intrusive_list<accept_node> waiters_;
151		/// Recycled accept nodes and ready-fd nodes (guarded by `mutex_`).
152		/// Both survive impl recycling — the storage belongs to this impl
153		/// and only the destructor frees it — so steady-state accepts
154		/// allocate nothing.
155		intrusive_list<accept_node> free_nodes_;
156		intrusive_list<ready_fd_node> free_ready_;
157		/// Single parked readiness wait (guarded by `mutex_`). Multishot
158		/// accepting drains the kernel queue instantly, so a listener's
159		/// readiness lives in `ready_fds_`, not in `poll()`; the wait is
160		/// completed by the next delivery instead of a kernel poll.
161		accept_node* read_wait_ = nullptr;
162		/// Ended armings, kept for the next one (guarded by `mutex_`).
163		/// Survives impl recycling like `free_nodes_`; entries hold no
164		/// reference.
165		intrusive_list<uring_multi_accept_op> free_ops_;
166		/// The arming whose deliveries belong to this acceptor (guarded by
167		/// `mutex_`). Identity only: the kernel owns the op and its
168		/// reference. Never set while `closing_` is.
169		uring_multi_accept_op* armed_ = nullptr;
170		bool closing_ = false;
171		/// Non-zero once an arming failed to reach the kernel (guarded by
172		/// `mutex_`). Nothing will ever deliver a connection through an
173		/// SQE the ring never took, so an accept reports this instead of
174		/// parking on a delivery that cannot come. Cleared by the next
175		/// arming that does reach the kernel.
176		int arm_err_ = 0;
177
178		private:
179		// CRTP ctor private + Derived friended so the base cannot be
180		// constructed except as a CRTP base of Derived
181		// (clang-tidy bugprone-crtp-constructor-accessibility).
182		friend Derived;
183	873x	uring_multishot_acceptor_base(
184		AcceptorService& acceptor_svc,
185		uring_scheduler& sched,
186		PeerService& peer_svc) noexcept
187	873x	: sched_(&sched)
188	873x	, peer_service_(&peer_svc)
189	873x	, acceptor_svc_(&acceptor_svc)
190		{
191	873x	}
192
193		protected:
194		/** Pop a recycled accept node, or allocate on the cold path.
195
196		Resets every per-use field and arms the `object_ref` keepalive
197		on this acceptor; the caller fills the request fields and
198		either parks or posts the node. OOM on the cold path
199		terminates, matching every other noexcept initiation path.
200		*/
201	4066x	accept_node* acquire_node() noexcept
202		{
203		accept_node* n;
204		{
205	4066x	std::lock_guard lk(mutex_);
206	4066x	n = free_nodes_.pop_front();
207	4066x	}
208	4066x	if (!n)
209		{
210		// NOLINTNEXTLINE(bugprone-unhandled-exception-at-new) — noexcept initiation path: OOM => std::terminate is the intended behavior
211	728x	n = new accept_node();
212		}
213	4066x	n->owner = static_cast<Derived*>(this);
214	4066x	n->queued = false;
215	4066x	n->peek = false;
216	4066x	n->err = 0;
217	4066x	n->accepted_fd = -1;
218	4066x	n->peer_len = 0;
219	4066x	n->ec_out = nullptr;
220	4066x	n->impl_out = nullptr;
221	4066x	n->peer_endpoint_out = nullptr;
222	4066x	n->peer_service = nullptr;
223	4066x	n->adopt_fn = nullptr;
224	4066x	n->cancelled.store(false, std::memory_order_relaxed);
225	4066x	n->object_ref_ = detail::object_ref(this);
226	4066x	return n;
227		}
228
229		/** Return a consumed node to the free list; `dispose` target.
230
231		The keepalive is moved out first and dropped after the lock,
232		so a final release (which may reenter this impl's retire())
233		never runs under `mutex_`.
234		*/
235	4066x	void release_node(accept_node* n) noexcept
236		{
237	4066x	auto keep = std::move(n->object_ref_);
238		{
239	4066x	std::lock_guard lk(mutex_);
240	4066x	free_nodes_.push_front(n);
241	4066x	}
242	4066x	}
243
244		/// Pop a recycled ready-fd node, or allocate on the cold path.
245		/// @pre `mutex_` is held.
246	746x	ready_fd_node* acquire_ready_locked() noexcept
247		{
248	746x	if (auto* r = free_ready_.pop_front())
249		{
250	205x	r->fd = -1;
251	205x	r->peer_len = 0;
252	205x	return r;
253		}
254		// NOLINTNEXTLINE(bugprone-unhandled-exception-at-new) — CQE handler: noexcept, OOM => std::terminate is the intended behavior
255	541x	return new ready_fd_node{};
256		}
257
258		public:
259		/** Close and free every fd parked in `ready_fds_`.
260
261		A multishot CQE can deliver a connection before any `accept()`
262		is outstanding to claim it; that connection's fd (and tracking
263		node) sits in `ready_fds_` until a later `accept()` consumes it
264		or the acceptor goes away. Idempotent: safe to call from both
265		`retire()` (recycling) and the destructor (pool-force-sweep
266		backstop) — the second call always sees an empty list.
267		*/
268	2248x	void drain_ready_fds() noexcept
269		{
270	2248x	intrusive_list<ready_fd_node> drained;
271		{
272	2248x	std::lock_guard lk(mutex_);
273	2658x	while (auto* r = ready_fds_.pop_front())
274	410x	drained.push_back(r);
275	2248x	}
276	2248x	discard_ready(drained);
277	2248x	}
278
279		/// Close each parked connection in @p stale and return its node
280		/// to `free_ready_`. @pre `mutex_` is not held.
281	3504x	void discard_ready(intrusive_list<ready_fd_node>& stale) noexcept
282		{
283	3504x	if (stale.empty())
284	3454x	return;
285	461x	stale.for_each([](ready_fd_node* r) { ::close(r->fd); });
286	50x	std::lock_guard lk(mutex_);
287	461x	while (auto* r = stale.pop_front())
288	411x	free_ready_.push_front(r);
289	50x	}
290
291		/** Recycle into the owning service's pool at zero references.
292
293		No arming can be live here: an armed op holds a reference.
294		*/
295	1375x	void retire() noexcept override
296		{
297	1375x	drain_ready_fds();
298	1375x	acceptor_svc_->pool_.recycle(static_cast<Derived*>(this));
299	1375x	}
300
301	873x	~uring_multishot_acceptor_base() override
302		{
303		uring_multi_accept_op* arming;
304		{
305	873x	std::lock_guard lk(mutex_);
306	873x	closing_ = true;
307	873x	arming = std::exchange(armed_, nullptr);
308	873x	}
309	873x	if (fd_ >= 0)
310		{
311	✗	sched_->flush_after_cancel(fd_, [&]() noexcept {
312		if (arming)
313		sched_->cancel_op_locked(arming);
314		});
315	✗	::close(fd_);
316	✗	fd_ = -1;
317		}
318
319		// Backstop: retire() already drains this on the normal
320		// recycle path. This call only does real work when this impl
321		// is reached by the pool's unconditional force-sweep at
322		// shutdown without ever going through retire() first.
323	873x	drain_ready_fds();
324
325		// The recycled-node storage belongs to this impl and is only
326		// freed here; in-flight nodes hold an object_ref on this impl,
327		// so none can still be parked or posted once the count reached
328		// zero (or the pool's force-sweep ran after scheduler drain).
329		// An arming the kernel never returned is not in `free_ops_`
330		// and is deliberately leaked: the ring may still write to it.
331	1601x	while (auto* n = free_nodes_.pop_front())
332	728x	delete n;
333	1414x	while (auto* r = free_ready_.pop_front())
334	541x	delete r;
335	1682x	while (auto* op = free_ops_.pop_front())
336	809x	delete op;
337	1746x	}
338
339	783x	Endpoint local_endpoint() const noexcept override
340		{
341	783x	return local_endpoint_;
342		}
343
344	11463x	bool is_open() const noexcept override
345		{
346	11463x	return fd_ >= 0;
347		}
348
349	24x	native_handle_type native_handle() const noexcept override
350		{
351	24x	return fd_;
352		}
353
354	601x	corosio::family family() const noexcept override
355		{
356	601x	return to_family(socket_family(fd_));
357		}
358
359	9x	native_handle_type release_socket() noexcept override
360		{
361		// Mirror the service close() path, except that the caller now
362		// owns the fd.
363	9x	if (fd_ >= 0)
364	9x	flush_listener();
365	9x	int fd = fd_;
366	9x	fd_ = -1;
367	9x	local_endpoint_ = Endpoint{};
368	9x	return fd;
369		}
370
371		/// Abort every operation and stop accepting, for service teardown.
372	601x	void abort_all() noexcept
373		{
374	601x	auto* arming = drain_waiters_only();
375	601x	static_cast<Derived*>(this)->cancel_wait_op();
376	601x	if (arming)
377	406x	sched_->cancel_ops({arming});
378	601x	}
379
380		/** Stop accepting and flush every SQE naming the listener.
381
382		Disowns the arming and cancels it with the readiness wait, so
383		the caller may then close or hand off the descriptor.
384
385		@pre The descriptor is open.
386		*/
387	1300x	void flush_listener() noexcept
388		{
389	1300x	auto* arming = drain_waiters_only();
390	1300x	static_cast<Derived*>(this)->cancel_wait_op();
391	1300x	sched_->flush_after_cancel(fd_, [&]() noexcept {
392	1295x	if (arming)
393	843x	sched_->cancel_op_locked(arming);
394		});
395	1300x	}
396
397		/** Cancel the pending accepts and waits.
398
399		The arming is left alone: the acceptor keeps listening, and a
400		connection that arrives meanwhile waits for the next accept.
401		*/
402	8x	void cancel() noexcept override
403		{
404	8x	intrusive_list<accept_node> claimed;
405		{
406	8x	std::lock_guard lk(mutex_);
407		// A waiter the stop callback already claimed belongs to it;
408		// see fail_arm().
409	8x	intrusive_list<accept_node> keep;
410	11x	while (auto* w = waiters_.pop_front())
411		{
412	3x	if (!w->cancelled.exchange(true, std::memory_order_acq_rel))
413	3x	claimed.push_back(w);
414		else
415	✗	keep.push_back(w);
416		}
417	8x	while (auto* w = keep.pop_front())
418	✗	waiters_.push_back(w);
419	9x	if (read_wait_ &&
420	1x	!read_wait_->cancelled.exchange(
421		true, std::memory_order_acq_rel))
422		{
423	1x	claimed.push_back(read_wait_);
424	1x	read_wait_ = nullptr;
425		}
426	8x	}
427	12x	while (auto* w = claimed.pop_front())
428		{
429		// The claim is also the completion status.
430	4x	w->stop_cb.reset();
431	4x	sched_->post(w);
432	4x	sched_->work_finished();
433		}
434	8x	static_cast<Derived*>(this)->cancel_wait_op();
435	8x	}
436
437		/** Reset state for recycling.
438
439		`close()`/`release_socket()` already drove fd_, local_endpoint_,
440		and the waiter/ready-fd queues to empty before the refcount
441		reached zero, so those are asserted rather than re-cleared.
442		`closing_` and `arm_err_` are the two fields that are NOT reset
443		by that close path (closing_ is only ever cleared by a re-listen
444		on a still-live acceptor, not by close) and must be explicitly
445		reset here or the next session would see this impl as still
446		shutting down.
447
448		@pre refs_ == 0, fd closed, no op or waiter in flight.
449		*/
450	502x	void reuse() noexcept
451		{
452	502x	BOOST_COROSIO_ASSERT(fd_ == -1);
453	502x	BOOST_COROSIO_ASSERT(local_endpoint_ == Endpoint{});
454	502x	BOOST_COROSIO_ASSERT(ready_fds_.empty());
455	502x	BOOST_COROSIO_ASSERT(waiters_.empty());
456	502x	BOOST_COROSIO_ASSERT(read_wait_ == nullptr);
457	502x	BOOST_COROSIO_ASSERT(armed_ == nullptr);
458	502x	closing_ = false;
459	502x	arm_err_ = 0;
460	502x	}
461
462		/** Disown the arming and drain queued waiters with
463		operation_aborted, without submitting a kernel cancel.
464
465		Runs before the caller cancels the arming it returns: an arming
466		admitted after this sees `closing_` and is refused, and the one
467		admitted before it is already in the submission queue ahead of
468		the cancel.
469
470		@return The disowned arming, or null if none was live.
471		*/
472	1901x	uring_multi_accept_op* drain_waiters_only() noexcept
473		{
474	1901x	intrusive_list<accept_node> drained;
475		uring_multi_accept_op* arming;
476		{
477	1901x	std::lock_guard lk(mutex_);
478	1901x	closing_ = true;
479	1901x	arming = std::exchange(armed_, nullptr);
480		// Drain under the lock — the kernel cancel may not produce
481		// a !more CQE before the fd is closed, so we can't rely on
482		// on_accept_cqe_impl to surface operation_aborted.
483	2138x	while (auto* w = waiters_.pop_front())
484	237x	drained.push_back(w);
485	1901x	if (read_wait_)
486		{
487	1x	drained.push_back(read_wait_);
488	1x	read_wait_ = nullptr;
489		}
490	1901x	}
491
492	2139x	while (auto* w = drained.pop_front())
493		{
494		// reset() synchronizes with an in-flight canceller; after
495		// it the stored status is ours to set.
496	238x	w->stop_cb.reset();
497	238x	w->cancelled.store(true, std::memory_order_release);
498	238x	sched_->post(w);
499	238x	sched_->work_finished();
500		}
501	1901x	return arming;
502		}
503
504		/** Park a readiness wait, or complete it if a connection is
505		already queued.
506
507		Multishot accepting consumes the kernel queue as connections
508		arrive, so a poll on the listener never reports it readable;
509		readiness is the impl's ready queue plus future deliveries.
510		*/
511	11x	void park_read_wait(
512		capy::continuation& cont,
513		capy::executor_ref ex,
514		std::stop_token const& token,
515		std::error_code* ec) noexcept
516		{
517	11x	auto* w = acquire_node();
518	11x	w->cont = &cont;
519	11x	w->ex = ex;
520	11x	w->ec_out = ec;
521	11x	w->peek = true;
522
523	11x	bool ready = false;
524	11x	bool aborted = false;
525		{
526	11x	std::lock_guard lk(mutex_);
527	11x	if (closing_)
528		{
529	✗	aborted = true;
530		}
531	11x	else if (!ready_fds_.empty())
532		{
533	4x	ready = true;
534		}
535	11x	}
536	11x	if (ready \|\| aborted)
537		{
538	4x	if (aborted)
539	✗	w->cancelled.store(true, std::memory_order_release);
540	4x	sched_->post(w);
541	4x	return;
542		}
543
544		// Same protocol as accept parking: arm the callback before
545		// the node is visible and outside `mutex_` (a pre-stopped
546		// token invokes the canceller synchronously, and the
547		// canceller takes `mutex_`).
548	7x	w->start(token);
549
550	7x	bool was_cancelled = false;
551	7x	int arm_err = 0;
552		{
553	7x	std::lock_guard lk(mutex_);
554	7x	if (w->cancelled.load(std::memory_order_acquire) \|\| closing_)
555		{
556	✗	was_cancelled = true;
557		}
558	7x	else if (ready_fds_.empty())
559		{
560		// Readiness here means a future delivery, which a failed
561		// arming has already ruled out: report it rather than
562		// wait on one.
563	7x	arm_err = arm_err_;
564	7x	if (arm_err == 0)
565		{
566	6x	w->queued = true;
567	6x	sched_->work_started();
568	6x	read_wait_ = w;
569	6x	return;
570		}
571		}
572		// else: a connection arrived while the callback was armed;
573		// complete as ready below.
574	7x	}
575
576	1x	w->stop_cb.reset();
577	1x	w->err = arm_err;
578	1x	w->cancelled.store(was_cancelled, std::memory_order_release);
579	1x	sched_->post(w);
580		}
581
582	595x	std::error_code set_option(
583		int level,
584		int optname,
585		void const* data,
586		std::size_t size) noexcept override
587		{
588	595x	if (fd_ < 0)
589	✗	return make_err(EBADF);
590	595x	if (::setsockopt(
591		fd_, level, optname, reinterpret_cast<char const*>(data),
592	595x	static_cast<socklen_t>(size)) < 0)
593	6x	return make_err(errno);
594	589x	return {};
595		}
596
597		std::error_code
598	6x	get_option(int level, int optname, void* data, std::size_t* size)
599		const noexcept override
600		{
601	6x	if (fd_ < 0)
602	✗	return make_err(EBADF);
603	6x	socklen_t len = static_cast<socklen_t>(*size);
604	12x	if (::getsockopt(
605	6x	fd_, level, optname, reinterpret_cast<char*>(data), &len) < 0)
606	2x	return make_err(errno);
607	4x	*size = static_cast<std::size_t>(len);
608	4x	return {};
609		}
610
611		/** Take over an already-listening descriptor.
612
613		Clears the shutdown latch a previous release left behind and
614		discards connections parked from the replaced descriptor:
615		those belong to the socket the caller is handing away.
616
617		@pre No arming is live (`drain_waiters_only` has run if the
618		acceptor had a descriptor).
619
620		@param fd The adopted descriptor.
621		*/
622	8x	void adopt_listening_fd(int fd) noexcept
623		{
624	8x	intrusive_list<ready_fd_node> stale;
625		{
626	8x	std::lock_guard lk(mutex_);
627	8x	BOOST_COROSIO_ASSERT(armed_ == nullptr);
628	8x	fd_ = fd;
629	8x	closing_ = false;
630	8x	while (auto* r = ready_fds_.pop_front())
631	✗	stale.push_back(r);
632	8x	}
633	8x	discard_ready(stale);
634	8x	}
635
636		/** Ready the acceptor for a listen-time arming.
637
638		Returns `false` when a live arming already covers the
639		descriptor: a re-listen only changes the backlog. Otherwise
640		clears the shutdown latch a released descriptor left behind and
641		discards the connections it delivered, which belong to the
642		socket the caller took away.
643		*/
644	1250x	bool prepare_listen_arm() noexcept
645		{
646	1250x	intrusive_list<ready_fd_node> stale;
647		{
648	1250x	std::lock_guard lk(mutex_);
649	1250x	if (armed_)
650	2x	return false;
651	1248x	closing_ = false;
652	1249x	while (auto* r = ready_fds_.pop_front())
653	1x	stale.push_back(r);
654	1250x	}
655	1248x	discard_ready(stale);
656	1248x	return true;
657		}
658
659		/** Arm a multishot accept on the current descriptor.
660
661		The op is handed to the kernel with a reference on this
662		acceptor; the caller must hold one. Refused when closing or
663		when an arming is already live.
664		*/
665	1256x	void start_multishot() noexcept
666		{
667		uring_multi_accept_op* op;
668		{
669	1256x	std::lock_guard lk(mutex_);
670	1256x	op = free_ops_.pop_front();
671	1256x	}
672	1256x	if (!op)
673		{
674		// NOLINTNEXTLINE(bugprone-unhandled-exception-at-new) — noexcept initiation path: OOM => std::terminate is the intended behavior
675	809x	op = new uring_multi_accept_op();
676	809x	op->sched_ = sched_;
677	809x	op->acceptor_impl = this;
678	809x	op->on_cqe = &uring_multishot_acceptor_base::on_accept_cqe;
679	809x	op->on_end = &uring_multishot_acceptor_base::on_arming_end;
680		}
681	1256x	op->object_ref_ = detail::object_ref(this);
682	1256x	submit_arming(op, [this, op]() noexcept {
683	1251x	if (closing_ \|\| armed_)
684	✗	return false;
685	1251x	armed_ = op;
686	1251x	op->listen_fd = fd_;
687	1251x	return true;
688		});
689	1256x	}
690
691		/** Report an arming that never reached the kernel.
692
693		No CQE can arrive for an SQE the ring never took, so an accept
694		parked on this acceptor would park for good. The error is
695		remembered for the accepts still to come and delivered now to
696		whoever is already parked, which is the same EAGAIN every other
697		op reports when the SQ stays full.
698
699		@param err The code to report to the accepts this arming can no
700		longer serve.
701		*/
702	2x	void fail_arm(int err) noexcept
703		{
704	2x	intrusive_list<accept_node> claimed;
705		{
706	2x	std::lock_guard lk(mutex_);
707	2x	arm_err_ = err;
708		// Claim each waiter the way a delivery does. One the
709		// canceller already claimed belongs to it: cancel_waiter
710		// is waiting on this mutex to unlink the node itself, so
711		// it has to still be in the list when it gets in.
712	2x	intrusive_list<accept_node> keep;
713	2x	while (auto* w = waiters_.pop_front())
714		{
715	✗	if (!w->cancelled.exchange(true, std::memory_order_acq_rel))
716	✗	claimed.push_back(w);
717		else
718	✗	keep.push_back(w);
719		}
720	2x	while (auto* w = keep.pop_front())
721	✗	waiters_.push_back(w);
722	2x	if (read_wait_ &&
723	✗	!read_wait_->cancelled.exchange(
724		true, std::memory_order_acq_rel))
725		{
726	✗	claimed.push_back(read_wait_);
727	✗	read_wait_ = nullptr;
728		}
729	2x	}
730
731	2x	while (auto* w = claimed.pop_front())
732		{
733		// The exchange above was a claim, not the completion
734		// status: after reset() quiesces the canceller, restore
735		// the flag so the handler reports `err`, not canceled.
736	✗	w->stop_cb.reset();
737	✗	w->err = err;
738	✗	w->cancelled.store(false, std::memory_order_release);
739	✗	sched_->post(w);
740	✗	sched_->work_finished(); // balance the waiter's work_started
741		}
742	2x	}
743
744		/// Pull a parked fd or queue a waiter — used by Derived::accept().
745		/// Either case ends with the calling coroutine suspending; the
746		/// caller returns `std::noop_coroutine()` unconditionally.
747	4051x	void dispatch_or_queue(
748		capy::continuation& cont,
749		capy::executor_ref ex,
750		std::stop_token const& token,
751		std::error_code* ec,
752		io_object::implementation** impl_out)
753		{
754	4051x	auto* w = acquire_node();
755	4051x	w->cont = &cont;
756	4051x	w->ex = ex;
757	4051x	w->ec_out = ec;
758	4051x	w->impl_out = impl_out;
759
760		// A live arming takes each connection as it lands, so a direct
761		// accept would only find the backlog empty.
762	4051x	bool armed = false;
763		{
764	4051x	std::lock_guard lk(mutex_);
765	4051x	armed = armed_ != nullptr;
766	4051x	}
767	4051x	if (!armed)
768		{
769	3x	sockaddr_storage peer_storage{};
770	3x	socklen_t peer_len = sizeof(peer_storage);
771	3x	int accepted_fd = ::accept4(
772		fd_, reinterpret_cast<sockaddr*>(&peer_storage), &peer_len,
773		SOCK_NONBLOCK \| SOCK_CLOEXEC);
774	3x	if (accepted_fd >= 0)
775		{
776	✗	w->peer_service = peer_service_;
777	✗	w->adopt_fn = &Derived::adopt_thunk;
778	✗	w->accepted_fd = accepted_fd;
779	✗	w->peer_storage = peer_storage;
780	✗	w->peer_len = peer_len;
781	✗	sched_->post(w);
782	2x	return;
783		}
784		// accept4 returned <0 — only EAGAIN/EWOULDBLOCK should fall
785		// through to the parked/waiter path. Other errors (EBADF,
786		// etc.) surface through the existing scheduler-completion
787		// path so the user sees them via the node's ec_out: `err`
788		// set means do_handler delivers make_err(err).
789	3x	if (errno != EAGAIN && errno != EWOULDBLOCK)
790		{
791	2x	w->err = errno;
792	2x	sched_->post(w);
793	2x	return;
794		}
795		}
796
797	4049x	bool have_ready = false;
798		{
799	4049x	std::lock_guard lk(mutex_);
800	4049x	if (auto* r = ready_fds_.pop_front())
801		{
802	335x	w->peer_service = peer_service_;
803	335x	w->adopt_fn = &Derived::adopt_thunk;
804	335x	w->accepted_fd = r->fd;
805	335x	w->peer_storage = r->peer;
806	335x	w->peer_len = r->peer_len;
807	335x	free_ready_.push_front(r);
808	335x	have_ready = true;
809		}
810	4049x	}
811	4049x	if (have_ready)
812		{
813		// Post outside the lock — acceptor mutex_ must never be
814		// held while dispatch_mutex_ is acquired by sched_->post().
815	335x	sched_->post(w);
816	335x	return;
817		}
818
819		// Arm the stop callback before the node is visible in
820		// `waiters_` and outside `mutex_`: an already-stopped token
821		// invokes the canceller synchronously from emplace, and
822		// cancel_waiter takes `mutex_` (self-deadlock if held).
823		// Arming pre-queue also keeps the CQE handler from claiming
824		// a node whose callback is not yet constructed.
825	3714x	w->start(token);
826
827	3714x	bool was_cancelled = false;
828		{
829	3714x	std::lock_guard lk(mutex_);
830	3714x	if (w->cancelled.load(std::memory_order_acquire))
831		{
832		// Canceller already fired (pre-stopped token); it saw
833		// queued == false and left completion to us.
834	✗	was_cancelled = true;
835		}
836	3714x	else if (auto* r = ready_fds_.pop_front())
837		{
838		// A connection arrived while the callback was armed;
839		// prefer it over parking the waiter behind it.
840	✗	w->peer_service = peer_service_;
841	✗	w->adopt_fn = &Derived::adopt_thunk;
842	✗	w->accepted_fd = r->fd;
843	✗	w->peer_storage = r->peer;
844	✗	w->peer_len = r->peer_len;
845	✗	free_ready_.push_front(r);
846		}
847	3714x	else if (arm_err_ != 0)
848		{
849		// No arming reached the kernel, so no CQE will deliver
850		// a connection: parking here would park for good.
851	1x	w->err = arm_err_;
852		}
853		else
854		{
855	3713x	w->queued = true;
856	3713x	sched_->work_started();
857	3713x	waiters_.push_back(w);
858	3713x	return;
859		}
860	3714x	}
861
862	1x	w->stop_cb.reset();
863	1x	w->cancelled.store(was_cancelled, std::memory_order_release);
864	1x	sched_->post(w);
865		}
866
867	221x	void cancel_waiter(accept_node* w) noexcept
868		{
869		{
870	221x	std::lock_guard lk(mutex_);
871	221x	if (closing_)
872	7x	return; // drain_waiters_only will complete with closing_ set
873	214x	if (!w->queued)
874	✗	return; // not queued yet; the parking path observes
875		// `cancelled` and completes the node
876	214x	if (w->peek)
877		{
878	1x	if (read_wait_ != w)
879	✗	return; // already claimed by a delivery
880	1x	read_wait_ = nullptr;
881		}
882		else
883		{
884	213x	waiters_.remove(w);
885		}
886	214x	sched_->cancel_posting_begin();
887	221x	}
888		// `cancelled` is already true (the claim that routed here);
889		// that is also the completion status, so post as-is. The node's
890		// stop_cb is still engaged — do_handler resets it, and running
891		// that reset from the handler thread while this callback
892		// returns is the stop_callback destructor's documented
893		// wait-or-self case.
894	214x	sched_->post(w);
895	214x	sched_->work_finished(); // balance the work_started() from accept()
896	214x	sched_->cancel_posting_end();
897		}
898
899		private:
900		/** Submit @p op as the live arming, or recycle it.
901
902		@param admit Runs under `mutex_`, itself under the ring mutex;
903		claims `armed_` for @p op and returns true, or returns
904		false to refuse.
905		*/
906		template<class Admit>
907	1259x	void submit_arming(uring_multi_accept_op* op, Admit admit) noexcept
908		{
909	1259x	op->peer_storage = sockaddr_storage{};
910	1259x	op->peer_len = sizeof(sockaddr_storage);
911	1259x	op->res = 0;
912		// Deliberately no work_started(): the arming is an internal
913		// mechanism; user-visible work is tracked per accept.
914		// Once submitted, a re-arm's only reference to this acceptor
915		// is the op's, which the kernel may already have ended, so
916		// the error is cleared while admission still holds the lock.
917	1259x	auto r = uring_try_submit_op_if(*sched_, op, [&]() noexcept {
918		std::lock_guard lk(mutex_);
919		if (!admit())
920		return false;
921		arm_err_ = 0;
922		return true;
923		});
924	1259x	if (r == uring_guarded_submit::submitted)
925	1257x	return;
926	2x	if (r == uring_guarded_submit::sq_full)
927		{
928		{
929	2x	std::lock_guard lk(mutex_);
930	2x	if (armed_ == op)
931	2x	armed_ = nullptr;
932	2x	}
933	2x	fail_arm(EAGAIN);
934		}
935	2x	recycle_arming(op);
936		}
937
938		/// Return an ended arming to `free_ops_`, then drop its reference,
939		/// which may retire this impl, after the lock.
940	1256x	void recycle_arming(uring_multi_accept_op* op) noexcept
941		{
942	1256x	auto keep = std::move(op->object_ref_);
943	1256x	std::lock_guard lk(mutex_);
944	1256x	free_ops_.push_front(op);
945	1256x	}
946
947	5566x	static void on_accept_cqe(
948		void* self_ptr,
949		uring_multi_accept_op* op,
950		int new_fd,
951		int err,
952		bool more) noexcept
953		{
954	5566x	static_cast<Derived*>(self_ptr)->on_accept_cqe_impl(
955		op, new_fd, err, more);
956	5566x	}
957
958		static void
959	1257x	on_arming_end(void* self_ptr, uring_multi_accept_op* op, bool live) noexcept
960		{
961	1257x	auto* self = static_cast<Derived*>(self_ptr);
962	1257x	if (live)
963		{
964		bool rearm;
965		{
966	655x	std::lock_guard lk(self->mutex_);
967	655x	rearm = self->armed_ == op && !self->closing_;
968	655x	}
969	655x	if (rearm)
970		{
971		// The kernel ended a live arming on its own; keep
972		// accepting with the same op and reference.
973	3x	self->submit_arming(op, [self, op]() noexcept {
974	3x	return self->armed_ == op && !self->closing_;
975		});
976	3x	return;
977		}
978		}
979		{
980	1254x	std::lock_guard lk(self->mutex_);
981	1254x	if (self->armed_ == op)
982	✗	self->armed_ = nullptr;
983	1254x	}
984	1254x	self->recycle_arming(op);
985		}
986
987		protected:
988	5566x	void on_accept_cqe_impl(
989		uring_multi_accept_op* op, int new_fd, int err, bool more) noexcept
990		{
991	5566x	accept_node* matched = nullptr;
992	5566x	accept_node* claimed_peek = nullptr;
993	5566x	intrusive_list<accept_node> closing_waiters;
994		{
995	5566x	std::lock_guard lk(mutex_);
996		// A disowned arming's deliveries belong to a descriptor
997		// this acceptor no longer has.
998	5566x	bool const stale = (op != armed_);
999	5569x	if (!stale && new_fd >= 0 && read_wait_ &&
1000	3x	!read_wait_->cancelled.exchange(
1001		true, std::memory_order_acq_rel))
1002		{
1003		// A parked readiness wait observes the delivery
1004		// without consuming it; the connection still flows
1005		// to a waiter or the ready queue below.
1006	3x	claimed_peek = read_wait_;
1007	3x	read_wait_ = nullptr;
1008		}
1009	5566x	if (stale)
1010		{
1011	1557x	if (new_fd >= 0)
1012	303x	::close(new_fd);
1013	1557x	if (!more && closing_)
1014		{
1015		// Collect waiters to drain after the lock is released.
1016	1248x	while (auto* w = waiters_.pop_front())
1017	✗	closing_waiters.push_back(w);
1018		}
1019		}
1020	4009x	else if (!waiters_.empty())
1021		{
1022		// Claim the head waiter atomically. If the canceller
1023		// already won the race (cancelled was already true),
1024		// leave the waiter in the list for cancel_waiter to
1025		// remove and dispatch with operation_aborted; park the
1026		// new_fd so the next waiter consumes it.
1027	3260x	auto* head_w = waiters_.front();
1028	3260x	if (!head_w->cancelled.exchange(
1029		true, std::memory_order_acq_rel))
1030		{
1031	3260x	waiters_.pop_front();
1032	3260x	matched = head_w;
1033		}
1034	✗	else if (new_fd >= 0)
1035		{
1036	✗	auto* node = acquire_ready_locked();
1037	✗	node->fd = new_fd;
1038	✗	node->peer = op->peer_storage;
1039	✗	node->peer_len = op->peer_len;
1040	✗	ready_fds_.push_back(node);
1041		}
1042		}
1043	749x	else if (new_fd >= 0)
1044		{
1045	746x	auto* node = acquire_ready_locked();
1046	746x	node->fd = new_fd;
1047	746x	node->peer = op->peer_storage;
1048	746x	node->peer_len = op->peer_len;
1049	746x	ready_fds_.push_back(node);
1050		}
1051	5566x	}
1052
1053		// Each claim's exchange set `cancelled`; after reset() has
1054		// quiesced any in-flight canceller, restore the flag to the
1055		// real completion status before posting.
1056	5566x	if (claimed_peek)
1057		{
1058	3x	claimed_peek->stop_cb.reset();
1059	3x	claimed_peek->cancelled.store(false, std::memory_order_release);
1060	3x	sched_->post(claimed_peek);
1061	3x	sched_->work_finished(); // balance the parking work_started
1062		}
1063
1064	5566x	if (matched)
1065		{
1066	3260x	matched->stop_cb.reset();
1067	3260x	matched->peer_service = peer_service_;
1068	3260x	matched->adopt_fn = &Derived::adopt_thunk;
1069	3260x	if (err)
1070		{
1071	1x	matched->err = err;
1072		}
1073	3259x	else if (new_fd >= 0)
1074		{
1075	3259x	matched->accepted_fd = new_fd;
1076	3259x	matched->peer_storage = op->peer_storage;
1077	3259x	matched->peer_len = op->peer_len;
1078		}
1079	3260x	matched->cancelled.store(false, std::memory_order_release);
1080	3260x	sched_->post(matched);
1081	3260x	sched_->work_finished(); // balance waiter's work_started
1082		}
1083
1084	5566x	while (auto* w = closing_waiters.pop_front())
1085		{
1086	✗	w->stop_cb.reset();
1087	✗	w->cancelled.store(true, std::memory_order_release);
1088	✗	sched_->post(w);
1089	✗	sched_->work_finished(); // balance waiter's work_started
1090		}
1091	5566x	}
1092		};
1093
1094		} // namespace boost::corosio::detail
1095
1096		#endif // BOOST_COROSIO_HAS_URING
1097
1098		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_MULTISHOT_ACCEPTOR_HPP
1099