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

100.0% Lines (44/7/51) 100.0% List of functions (6/2/8)

[ ] 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/io/io_object.hpp>
22		#include <boost/corosio/native/detail/uring/uring_buffer.hpp>
23		#include <boost/corosio/native/detail/uring/uring_op.hpp>
24		#include <boost/corosio/native/detail/make_err.hpp>
25
26		#include <netinet/in.h>
27		#include <sys/socket.h>
28		#include <unistd.h>
29
30		namespace boost::corosio::detail {
31
32		/** Multishot accept op — one submitted per acceptor lifetime.
33
34		The kernel produces a CQE for each accepted connection. Each CQE
35		carries the new fd in `res` (>= 0) or a negative errno on failure.
36		The `IORING_CQE_F_MORE` flag is set on every CQE except the last,
37		indicating whether the multishot armament is still active.
38
39		`do_cqe` does NOT push self into `local` — the owning acceptor's
40		`on_cqe` handler decides whether to dispatch immediately (waiter
41		present) or park the fd (no waiter). The multishot op persists
42		across CQEs; only `acceptor_impl` owns its lifetime.
43		*/
44		struct uring_multi_accept_op : uring_op
45		{
46		/// Filled by the kernel for each accept. Address of this struct
47		/// is registered with the SQE; kernel writes peer address here.
48		sockaddr_storage peer_storage{};
49		socklen_t peer_len = sizeof(peer_storage);
50		int listen_fd = -1;
51
52		/// Owning acceptor; raw because the op IS owned by the acceptor.
53		void* acceptor_impl = nullptr;
54
55		/** Callback into the acceptor for each accept CQE.
56
57		@param acceptor The owning acceptor_impl pointer.
58		@param new_fd Accepted fd on success, -1 on error.
59		@param err errno value on failure, 0 on success.
60		@param more True unless this is the terminating CQE
61		(e.g. kernel dropped multishot on -ENOMEM).
62		*/
63		void (on_cqe)(void acceptor, int new_fd, int err, bool more) noexcept =
64		nullptr;
65
66	218x	uring_multi_accept_op() noexcept : uring_op(&do_handler, &do_cqe, &do_prep)
67		{
68	218x	}
69
70	217x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
71		{
72	217x	auto* self = static_cast<uring_multi_accept_op*>(base);
73	217x	::io_uring_prep_multishot_accept(
74		sqe, self->listen_fd,
75	217x	reinterpret_cast<sockaddr*>(&self->peer_storage), &self->peer_len,
76		SOCK_NONBLOCK \| SOCK_CLOEXEC);
77	217x	}
78
79		/** Dispose of a connection the kernel accepted for a retired
80		arming.
81
82		The acceptor that armed this op has moved to another
83		descriptor, so no waiter will ever take delivery. The fd is
84		already installed in the process table — dropping the CQE
85		without closing it leaks it for the life of the process.
86		*/
87		static void
88	6x	do_retired_cqe(uring_op* /base/, int res, unsigned /flags/) noexcept
89		{
90	−	if (res >= 0) // LCOV_EXCL_LINE adopt-over-armed race leak guard
91	−	::close(res); // LCOV_EXCL_LINE adopt-over-armed race leak guard
92	6x	}
93
94	3273x	static void do_cqe(
95		uring_op* base,
96		int res,
97		unsigned flags,
98		ready_queue& /local/) noexcept
99		{
100	3273x	auto* self = static_cast<uring_multi_accept_op*>(base);
101	3273x	bool more = (flags & IORING_CQE_F_MORE) != 0;
102	3273x	int err = (res < 0) ? -res : 0;
103	3273x	int new_fd = (res >= 0) ? res : -1;
104	3273x	if (self->on_cqe)
105	3273x	self->on_cqe(self->acceptor_impl, new_fd, err, more);
106		// Intentionally NOT pushed into local: the acceptor decides
107		// whether to surface the fd via a waiter or park it.
108	3273x	}
109
110		// LCOV_EXCL_START: never invoked; the multishot op is owned by
111		// the acceptor and never queued for handler dispatch. Provided so
112		// the vtable is complete.
113	−	static void do_handler(
114		void* /owner/,
115		scheduler_op* /base/,
116		std::uint32_t /bytes/,
117		std::uint32_t /error/) noexcept
118		{
119	−	}
120		// LCOV_EXCL_STOP
121		};
122
123		/** Synthesized accept op — manufactured by the acceptor for parked fds.
124
125		When `async_accept` arrives and a ready fd is already parked, the
126		acceptor builds one of these, fills `accepted_fd` and peer storage
127		from the parked node, and posts it to the scheduler. This op never
128		interacts with the ring directly — it goes straight to handler
129		dispatch via `(*op)()`.
130
131		`do_cqe` is unused (this op never receives a kernel CQE).
132		*/
133		struct uring_accept_op : uring_op
134		{
135		int accepted_fd = -1;
136		int err = 0;
137		sockaddr_storage peer_storage{};
138		socklen_t peer_len = 0;
139
140		/// Set by the acceptor's `async_accept` entry point; filled by
141		/// `do_handler` with the new socket impl.
142		io_object::implementation** impl_out = nullptr;
143
144		/// Optional output for the peer endpoint.
145		endpoint* peer_endpoint_out = nullptr;
146
147		/// The peer service used to wrap the accepted fd.
148		void* peer_service = nullptr;
149
150		/// Acceptor-supplied wrapper: adopts `fd` into the right impl type.
151		io_object::implementation* (*adopt_fn)(
152		void* peer_service,
153		int fd,
154		sockaddr_storage const& peer,
155		socklen_t peer_len) noexcept = nullptr;
156
157	3303x	uring_accept_op() noexcept : uring_op(&do_handler, &do_cqe) {}
158
159		// LCOV_EXCL_START: never receives a CQE; present for vtable
160		// completeness.
161	−	static void do_cqe(uring_op*, int, unsigned, ready_queue&) noexcept {}
162		// LCOV_EXCL_STOP
163
164	3303x	static void do_handler(
165		void* owner,
166		scheduler_op* base,
167		std::uint32_t /bytes/,
168		std::uint32_t /error/) noexcept
169		{
170	3303x	auto* self = static_cast<uring_accept_op*>(base);
171	3303x	self->stop_cb.reset();
172
173	3303x	if (owner == nullptr)
174		{
175	5x	delete self;
176	52x	return;
177		}
178
179	3298x	bool was_cancelled = self->cancelled.load(std::memory_order_acquire);
180
181	3298x	if (was_cancelled \|\| self->err)
182		{
183	47x	if (self->ec_out)
184	47x	*self->ec_out = was_cancelled
185	59x	? std::error_code(capy::error::canceled)
186	12x	: make_err(self->err);
187	47x	self->cont.h = self->h;
188	47x	auto next = dispatch_coro(self->ex, self->cont);
189	47x	delete self;
190	47x	next.resume();
191	47x	return;
192		}
193
194	3251x	if (self->adopt_fn && self->impl_out)
195	3245x	*self->impl_out = self->adopt_fn(
196	3245x	self->peer_service, self->accepted_fd, self->peer_storage,
197		self->peer_len);
198
199		// LCOV_EXCL_START: no public accept overload reports the peer
200		// endpoint on this backend yet.
201	−	if (self->peer_endpoint_out)
202	−	*self->peer_endpoint_out = sockaddr_to_endpoint(self->peer_storage);
203		// LCOV_EXCL_STOP
204
205	3251x	if (self->ec_out)
206	3251x	*self->ec_out = {};
207
208	3251x	self->cont.h = self->h;
209	3251x	auto next = dispatch_coro(self->ex, self->cont);
210	3251x	delete self;
211	3251x	next.resume();
212		}
213		};
214
215		} // namespace boost::corosio::detail
216
217		#endif // BOOST_COROSIO_HAS_URING
218
219		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_ACCEPTOR_OPS_HPP
220