include/boost/corosio/native/detail/io_uring/io_uring_acceptor_ops.hpp

85.7% Lines (48/0/56) 75.0% List of functions (6/0/8)

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

io_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_IO_URING_IO_URING_ACCEPTOR_OPS_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_ACCEPTOR_OPS_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_IO_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/io_uring/io_uring_buffer.hpp>
23		#include <boost/corosio/native/detail/io_uring/io_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 : io_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,
64		bool more) noexcept = nullptr;
65
66	165x	uring_multi_accept_op() noexcept
67	165x	: io_uring_op(&do_handler, &do_cqe, &do_prep)
68	165x	{}
69
70	165x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
71		{
72	165x	auto* self = static_cast<uring_multi_accept_op*>(base);
73	165x	::io_uring_prep_multishot_accept(
74		sqe, self->listen_fd,
75	165x	reinterpret_cast<sockaddr*>(&self->peer_storage),
76		&self->peer_len,
77		SOCK_NONBLOCK \| SOCK_CLOEXEC);
78	165x	}
79
80		/** Dispose of a connection the kernel accepted for a retired
81		arming.
82
83		The acceptor that armed this op has moved to another
84		descriptor, so no waiter will ever take delivery. The fd is
85		already installed in the process table — dropping the CQE
86		without closing it leaks it for the life of the process.
87		*/
88	6x	static void do_retired_cqe(
89		io_uring_op* /base/, int res, unsigned /flags/) noexcept
90		{
91	6x	if (res >= 0)
92	✗	::close(res);
93	6x	}
94
95	2221x	static void do_cqe(io_uring_op* base, int res, unsigned flags,
96		ready_queue& /local/) noexcept
97		{
98	2221x	auto* self = static_cast<uring_multi_accept_op*>(base);
99	2221x	bool more = (flags & IORING_CQE_F_MORE) != 0;
100	2221x	int err = (res < 0) ? -res : 0;
101	2221x	int new_fd = (res >= 0) ? res : -1;
102	2221x	if (self->on_cqe)
103	2221x	self->on_cqe(self->acceptor_impl, new_fd, err, more);
104		// Intentionally NOT pushed into local: the acceptor decides
105		// whether to surface the fd via a waiter or park it.
106	2221x	}
107
108		/// Never invoked: the multishot op is owned by the acceptor and
109		/// never queued for handler dispatch. Provided so the vtable is
110		/// complete.
111	✗	static void do_handler(
112		void* /owner/, scheduler_op* /base/,
113		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
114		{
115		// No-op. The acceptor's per-accept callback handles everything.
116	✗	}
117		};
118
119		/** Synthesized accept op — manufactured by the acceptor for parked fds.
120
121		When `async_accept` arrives and a ready fd is already parked, the
122		acceptor builds one of these, fills `accepted_fd` and peer storage
123		from the parked node, and posts it to the scheduler. This op never
124		interacts with the ring directly — it goes straight to handler
125		dispatch via `(*op)()`.
126
127		`do_cqe` is unused (this op never receives a kernel CQE).
128		*/
129		struct uring_accept_op : io_uring_op
130		{
131		int accepted_fd = -1;
132		int err = 0;
133		sockaddr_storage peer_storage{};
134		socklen_t peer_len = 0;
135
136		/// Set by the acceptor's `async_accept` entry point; filled by
137		/// `do_handler` with the new socket impl.
138		io_object::implementation** impl_out = nullptr;
139
140		/// Optional output for the peer endpoint.
141		endpoint* peer_endpoint_out = nullptr;
142
143		/// The peer service used to wrap the accepted fd.
144		void* peer_service = nullptr;
145
146		/// Acceptor-supplied wrapper: adopts `fd` into the right impl type.
147		io_object::implementation*
148		(adopt_fn)(void peer_service, int fd,
149		sockaddr_storage const& peer,
150		socklen_t peer_len) noexcept = nullptr;
151
152	2237x	uring_accept_op() noexcept
153	2237x	: io_uring_op(&do_handler, &do_cqe)
154	2237x	{}
155
156	✗	static void do_cqe(io_uring_op*, int, unsigned,
157		ready_queue&) noexcept
158		{
159		// Unreachable: this op never receives a CQE.
160	✗	}
161
162	2237x	static void do_handler(
163		void* owner, scheduler_op* base,
164		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
165		{
166	2237x	auto* self = static_cast<uring_accept_op*>(base);
167	2237x	self->stop_cb.reset();
168
169	2237x	if (owner == nullptr)
170		{
171	✗	delete self;
172	24x	return;
173		}
174
175		bool was_cancelled =
176	2237x	self->cancelled.load(std::memory_order_acquire);
177
178	2237x	if (was_cancelled \|\| self->err)
179		{
180	24x	if (self->ec_out)
181	24x	*self->ec_out = was_cancelled
182	30x	? std::error_code(capy::error::canceled)
183	6x	: make_err(self->err);
184	24x	self->cont.h = self->h;
185	24x	auto next = dispatch_coro(self->ex, self->cont);
186	24x	delete self;
187	24x	next.resume();
188	24x	return;
189		}
190
191	2213x	if (self->adopt_fn && self->impl_out)
192	2208x	*self->impl_out = self->adopt_fn(
193		self->peer_service, self->accepted_fd,
194	2208x	self->peer_storage, self->peer_len);
195
196	2213x	if (self->peer_endpoint_out)
197	✗	*self->peer_endpoint_out =
198	✗	sockaddr_to_endpoint(self->peer_storage);
199
200	2213x	if (self->ec_out)
201	2213x	*self->ec_out = {};
202
203	2213x	self->cont.h = self->h;
204	2213x	auto next = dispatch_coro(self->ex, self->cont);
205	2213x	delete self;
206	2213x	next.resume();
207		}
208		};
209
210		} // namespace boost::corosio::detail
211
212		#endif // BOOST_COROSIO_HAS_IO_URING
213
214		#endif // BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_ACCEPTOR_OPS_HPP
215