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

94.4% Lines (51 / 54, 3 excl) 85.7% Functions (6 / 7, 1 excl)

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

uring_acceptor_ops.hpp

f(x) Functions (8)

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_ACCEPTOR_OPS_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_ACCEPTOR_OPS_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_URING
16
17		#include <liburing.h>
18
19		#include <boost/capy/error.hpp>
20		#include <boost/corosio/detail/dispatch_coro.hpp>
21		#include <boost/corosio/detail/intrusive.hpp>
22		#include <boost/corosio/io/io_object.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/make_err.hpp>
27
28		#include <netinet/in.h>
29		#include <sys/socket.h>
30		#include <unistd.h>
31
32		namespace boost::corosio::detail {
33
34		/** Multishot accept op: one kernel arming of an acceptor's listener.
35
36		The kernel produces a CQE for each accepted connection, carrying
37		the new fd in `res` (>= 0) or a negative errno. Every CQE but the
38		last sets `IORING_CQE_F_MORE`; the last ends the arming.
39
40		While armed, the op is owned by the kernel: it holds a reference on
41		its acceptor (`object_ref_`) and the acceptor only remembers it by
42		address. Deliveries are routed inline by `on_cqe` under the ring
43		mutex. The terminal CQE queues the op for dispatch, and its handler
44		(`on_end`, off every lock) either re-arms it or returns it to the
45		acceptor's free list and drops the reference.
46		*/
47		struct uring_multi_accept_op
48		: uring_op
49		, intrusive_list<uring_multi_accept_op>::node
50		{
51		/// Filled by the kernel for each accept. Address of this struct
52		/// is registered with the SQE; kernel writes peer address here.
53		sockaddr_storage peer_storage{};
54		socklen_t peer_len = sizeof(peer_storage);
55		int listen_fd = -1;
56
57		/// The acceptor that armed this op.
58		void* acceptor_impl = nullptr;
59
60		/** Route one accept CQE; runs under the ring mutex.
61
62		@param acceptor The acceptor that armed @p op.
63		@param op The arming the CQE belongs to.
64		@param new_fd Accepted fd on success, -1 on error.
65		@param err errno value on failure, 0 on success.
66		@param more True unless this CQE ends the arming.
67		*/
68		void (*on_cqe)(
69		void* acceptor,
70		uring_multi_accept_op* op,
71		int new_fd,
72		int err,
73		bool more) noexcept = nullptr;
74
75		/** Handle the end of an arming; runs from dispatch, off every lock.
76
77		@param acceptor The acceptor that armed @p op.
78		@param op The ended arming.
79		@param live False when the scheduler is shutting down.
80		*/
81		void (*on_end)(
82		void* acceptor,
83		uring_multi_accept_op* op,
84		bool live) noexcept = nullptr;
85
86	809x	uring_multi_accept_op() noexcept : uring_op(&do_handler, &do_cqe, &do_prep)
87		{
88	809x	}
89
90	1257x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
91		{
92	1257x	auto* self = static_cast<uring_multi_accept_op*>(base);
93	1257x	::io_uring_prep_multishot_accept(
94		sqe, self->listen_fd,
95	1257x	reinterpret_cast<sockaddr*>(&self->peer_storage), &self->peer_len,
96		SOCK_NONBLOCK \| SOCK_CLOEXEC);
97	1257x	}
98
99		static void
100	5566x	do_cqe(uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
101		{
102	5566x	auto* self = static_cast<uring_multi_accept_op*>(base);
103	5566x	bool more = (flags & IORING_CQE_F_MORE) != 0;
104	5566x	int err = (res < 0) ? -res : 0;
105	5566x	int new_fd = (res >= 0) ? res : -1;
106	5566x	self->on_cqe(self->acceptor_impl, self, new_fd, err, more);
107	5566x	if (!more)
108		{
109		// Balance the work_finished() do_one runs after dispatching.
110	1257x	self->sched_->work_started();
111	1257x	local.push(self);
112		}
113	5566x	}
114
115	1257x	static void do_handler(
116		void* owner,
117		scheduler_op* base,
118		std::uint32_t /bytes/,
119		std::uint32_t /error/) noexcept
120		{
121	1257x	auto* self = static_cast<uring_multi_accept_op*>(base);
122	1257x	self->on_end(self->acceptor_impl, self, owner != nullptr);
123	1257x	}
124		};
125
126		/** Synthesized accept op — manufactured by the acceptor for parked fds.
127
128		When `async_accept` arrives and a ready fd is already parked, the
129		acceptor builds one of these, fills `accepted_fd` and peer storage
130		from the parked node, and posts it to the scheduler. This op never
131		interacts with the ring directly — it goes straight to handler
132		dispatch via `(*op)()`.
133
134		`do_cqe` is unused (this op never receives a kernel CQE).
135		*/
136		struct uring_accept_op : uring_op
137		{
138		int accepted_fd = -1;
139		int err = 0;
140		sockaddr_storage peer_storage{};
141		socklen_t peer_len = 0;
142
143		/// Set by the acceptor's `async_accept` entry point; filled by
144		/// `do_handler` with the new socket impl.
145		io_object::implementation** impl_out = nullptr;
146
147		/// Optional output for the peer endpoint.
148		endpoint* peer_endpoint_out = nullptr;
149
150		/// The peer service used to wrap the accepted fd.
151		void* peer_service = nullptr;
152
153		/// Acceptor-supplied wrapper: adopts `fd` into the right impl type.
154		io_object::implementation* (*adopt_fn)(
155		void* peer_service,
156		int fd,
157		sockaddr_storage const& peer,
158		socklen_t peer_len) noexcept = nullptr;
159
160		/** Disposal hook, run exactly once when the op is consumed.
161
162		Heap-allocated ops delete themselves (the default); the
163		multishot acceptor's pooled nodes override this to return to
164		their acceptor's free list instead. Factoring disposal out of
165		`do_handler` is what lets one completion routine serve both
166		ownership models.
167		*/
168		void (dispose)(uring_accept_op) noexcept = &dispose_delete;
169
170	728x	uring_accept_op() noexcept : uring_op(&do_handler, &do_cqe) {}
171
172	✗	static void dispose_delete(uring_accept_op* op) noexcept
173		{
174	✗	delete op;
175	✗	}
176
177		// LCOV_EXCL_START: never receives a CQE; present for vtable
178		// completeness.
179	−	static void do_cqe(uring_op*, int, unsigned, ready_queue&) noexcept {}
180		// LCOV_EXCL_STOP
181
182	4066x	static void do_handler(
183		void* owner,
184		scheduler_op* base,
185		std::uint32_t /bytes/,
186		std::uint32_t /error/) noexcept
187		{
188	4066x	auto* self = static_cast<uring_accept_op*>(base);
189	4066x	self->stop_cb.reset();
190
191	4066x	if (owner == nullptr)
192		{
193		// A delivered connection that nobody will adopt.
194	408x	if (self->accepted_fd >= 0)
195	1x	::close(self->accepted_fd);
196		// Outlives dispose(), which may hand the node back to an
197		// acceptor that the unwound frame would otherwise release.
198	408x	auto keep = std::move(self->object_ref_);
199	408x	self->destroy_awaiting();
200	408x	self->dispose(self);
201	408x	return;
202	408x	}
203
204	3658x	bool was_cancelled = self->cancelled.load(std::memory_order_acquire);
205
206	3658x	if (was_cancelled \|\| self->err)
207		{
208	59x	if (self->ec_out)
209	59x	*self->ec_out = was_cancelled
210	68x	? std::error_code(capy::error::canceled)
211	9x	: make_err(self->err);
212	59x	auto next = dispatch_coro(self->ex, *self->cont);
213	59x	self->dispose(self);
214	59x	next.resume();
215	59x	return;
216		}
217
218	3599x	if (self->adopt_fn && self->impl_out)
219	3593x	*self->impl_out = self->adopt_fn(
220	3593x	self->peer_service, self->accepted_fd, self->peer_storage,
221		self->peer_len);
222
223		// LCOV_EXCL_START: no public accept overload reports the peer
224		// endpoint on this backend yet.
225	−	if (self->peer_endpoint_out)
226	−	*self->peer_endpoint_out = sockaddr_to_endpoint(self->peer_storage);
227		// LCOV_EXCL_STOP
228
229	3599x	if (self->ec_out)
230	3599x	*self->ec_out = {};
231
232	3599x	auto next = dispatch_coro(self->ex, *self->cont);
233	3599x	self->dispose(self);
234	3599x	next.resume();
235		}
236		};
237
238		} // namespace boost::corosio::detail
239
240		#endif // BOOST_COROSIO_HAS_URING
241
242		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_ACCEPTOR_OPS_HPP
243