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

98.3% Lines (116/0/118) 100.0% List of functions (10/0/10)

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

io_uring_dgram_ops.hpp

f(x) Functions (10)

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_IO_URING_IO_URING_DGRAM_OPS_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_DGRAM_OPS_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_IO_URING
17
18		#include <liburing.h>
19
20		#include <boost/corosio/detail/dispatch_coro.hpp>
21		#include <boost/corosio/native/detail/io_uring/io_uring_op.hpp>
22		#include <boost/corosio/native/detail/coro_op_complete.hpp>
23		#include <boost/corosio/native/detail/speculative_state.hpp>
24		#include <boost/corosio/native/detail/io_uring/io_uring_socket_ops.hpp>
25		#include <boost/corosio/native/detail/make_err.hpp>
26		#include <boost/capy/error.hpp>
27
28		#include <cstddef>
29		#include <cstdint>
30
31		#include <netinet/in.h>
32		#include <sys/socket.h>
33		#include <sys/uio.h>
34
35		namespace boost::corosio::detail {
36
37		/** Datagram send op — connected and unconnected.
38
39		Always uses `IORING_OP_SENDMSG`. In connected mode, `dest_len == 0`
40		and `msg.msg_name == nullptr`. In unconnected mode, `dest_storage`
41		holds the destination and `msg.msg_name` points at it.
42
43		`iovec[io_uring_max_iov]` for scatter/gather: a single datagram
44		can be assembled from N user buffers via `msg.msg_iov`.
45		*/
46		struct uring_dgram_send_op : io_uring_op
47		{
48		iovec iovecs[io_uring_max_iov];
49		int iovec_count = 0;
50		msghdr msg{};
51		sockaddr_storage dest_storage{};
52		socklen_t dest_len = 0;
53		int fd = -1;
54		int msg_flags = 0;
55		detail::speculative_state* spec_state = nullptr;
56
57	193x	uring_dgram_send_op() noexcept
58	193x	: io_uring_op(&do_handler, &do_cqe, &do_prep) {}
59
60		/** Reset and initialize for a new submission.
61
62		Pass `dest_addr_len == 0` for connected-mode datagram sockets
63		(the kernel uses the connected peer); otherwise fill
64		`dest_addr_storage` with the destination address.
65		*/
66	20x	void prepare(
67		std::coroutine_handle<> handle,
68		capy::executor_ref executor,
69		std::error_code* ec,
70		std::size_t* bytes,
71		int file_descriptor,
72		io_uring_scheduler* scheduler,
73		std::shared_ptr<void> impl,
74		detail::speculative_state* spec,
75		buffer_param buffers,
76		socklen_t dest_addr_len,
77		sockaddr_storage const& dest_addr_storage,
78		int flags,
79		std::stop_token const& token) noexcept
80		{
81	20x	h = handle;
82	20x	ex = executor;
83	20x	ec_out = ec;
84	20x	bytes_out = bytes;
85	20x	fd = file_descriptor;
86	20x	sched_ = scheduler;
87	20x	impl_ptr = std::move(impl);
88	20x	spec_state = spec;
89	20x	res = 0;
90	20x	cqe_flags = 0;
91	20x	msg_flags = flags;
92
93	20x	iovec_count = copy_to_iovec(buffers, iovecs);
94
95	20x	msg = {};
96	20x	msg.msg_iov = iovecs;
97	20x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(iovec_count);
98	20x	if (dest_addr_len > 0)
99		{
100	10x	dest_storage = dest_addr_storage;
101	10x	dest_len = dest_addr_len;
102	10x	msg.msg_name = &dest_storage;
103	10x	msg.msg_namelen = dest_addr_len;
104		}
105		else
106		{
107	10x	dest_len = 0;
108		}
109	20x	start(token);
110	20x	}
111
112	20x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
113		{
114	20x	auto* self = static_cast<uring_dgram_send_op*>(base);
115	20x	::io_uring_prep_sendmsg(
116	20x	sqe, self->fd, &self->msg,
117	20x	self->msg_flags \| MSG_NOSIGNAL);
118	20x	}
119
120	20x	static void do_cqe(
121		io_uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
122		{
123	20x	auto* self = static_cast<uring_dgram_send_op*>(base);
124	20x	self->res = res;
125	20x	self->cqe_flags = flags;
126	20x	local.push(self);
127	20x	}
128
129	20x	static void do_handler(
130		void* owner, scheduler_op* base,
131		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
132		{
133	20x	auto* self = static_cast<uring_dgram_send_op*>(base);
134	20x	if (coro_drain_if_shutdown(owner, self))
135	✗	return;
136
137	20x	if (self->sched_)
138	20x	self->sched_->reset_inline_budget();
139
140		// Datagram send: no EOF (a 0-byte send is success).
141	38x	decode_io_result(
142		self->ec_out,
143	20x	self->cancelled.load(std::memory_order_acquire),
144	20x	self->res < 0 ? make_err(-self->res) : std::error_code{},
145		/is_read=/false, /bytes=/0, /empty_buffer=/false);
146	20x	if (self->bytes_out)
147	20x	*self->bytes_out = (self->res >= 0)
148	20x	? static_cast<std::size_t>(self->res) : 0;
149
150	20x	if (self->res > 0 && self->spec_state)
151		{
152		// Kernel signalled readiness — restore speculation.
153	18x	self->spec_state->on_async_write_ready();
154		}
155
156	20x	coro_resume(self);
157		}
158		};
159
160		/** Datagram receive op — connected and unconnected.
161
162		Always uses `IORING_OP_RECVMSG`. In connected mode `msg.msg_name`
163		is null. In unconnected mode `msg.msg_name` points at
164		`source_storage` and the kernel writes the source address there.
165
166		`res == 0` is success (zero-byte datagrams are valid), NOT EOF.
167
168		The `source_writer` callback lets the concrete socket type
169		translate `sockaddr_storage` into `endpoint` or `local_endpoint`
170		without the op needing to know which family it is.
171		*/
172		struct uring_dgram_recv_op : io_uring_op
173		{
174		iovec iovecs[io_uring_max_iov];
175		int iovec_count = 0;
176		msghdr msg{};
177		sockaddr_storage source_storage{};
178		socklen_t source_len = 0;
179		int fd = -1;
180		int msg_flags = 0;
181		detail::speculative_state* spec_state = nullptr;
182
183		/// Type-erased translator: writes source_storage into the user's
184		/// endpoint output via concrete-class-specific conversion.
185		void* source_writer_ctx = nullptr;
186		void (*source_writer)(
187		void*, sockaddr_storage const&, socklen_t) noexcept = nullptr;
188
189	193x	uring_dgram_recv_op() noexcept
190	193x	: io_uring_op(&do_handler, &do_cqe, &do_prep) {}
191
192		/** Reset and initialize for a new submission.
193
194		When `source_fn` is non-null, the kernel writes the peer
195		address into `source_storage` and `source_fn(source_ctx, ...)`
196		is invoked from the handler on success to translate it to
197		the user's endpoint output. Connected-mode receivers should
198		pass `source_fn = nullptr`.
199
200		A zero-iovec `buffers` argument yields `iovec_count == 0`;
201		the caller should push the slot onto `completed_ops_`
202		directly (bypassing the kernel) since `recvmsg` would
203		otherwise block forever.
204		*/
205	59x	void prepare(
206		std::coroutine_handle<> handle,
207		capy::executor_ref executor,
208		std::error_code* ec,
209		std::size_t* bytes,
210		int file_descriptor,
211		io_uring_scheduler* scheduler,
212		std::shared_ptr<void> impl,
213		detail::speculative_state* spec,
214		buffer_param buffers,
215		void* source_ctx,
216		void (source_fn)(void, sockaddr_storage const&, socklen_t) noexcept,
217		int flags,
218		std::stop_token const& token) noexcept
219		{
220	59x	h = handle;
221	59x	ex = executor;
222	59x	ec_out = ec;
223	59x	bytes_out = bytes;
224	59x	fd = file_descriptor;
225	59x	sched_ = scheduler;
226	59x	impl_ptr = std::move(impl);
227	59x	spec_state = spec;
228	59x	res = 0;
229	59x	cqe_flags = 0;
230	59x	msg_flags = flags;
231
232	59x	iovec_count = copy_to_iovec(buffers, iovecs);
233
234	59x	msg = {};
235		// For the zero-iovec bypass path the caller pushes the slot
236		// straight onto completed_ops_; source_writer must NOT run in
237		// that case (no recvmsg ever happens, source_storage is empty
238		// and would clobber the user's endpoint). Arm the writer only
239		// when there's a real buffer AND the caller asked for it.
240	59x	if (iovec_count > 0 && source_fn)
241		{
242	32x	msg.msg_iov = iovecs;
243	32x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(
244	32x	iovec_count);
245	32x	source_storage = {};
246	32x	source_len = sizeof(source_storage);
247	32x	msg.msg_name = &source_storage;
248	32x	msg.msg_namelen = source_len;
249	32x	source_writer_ctx = source_ctx;
250	32x	source_writer = source_fn;
251		}
252		else
253		{
254	27x	if (iovec_count > 0)
255		{
256	27x	msg.msg_iov = iovecs;
257	27x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(
258	27x	iovec_count);
259		}
260	27x	source_len = 0;
261	27x	source_writer_ctx = nullptr;
262	27x	source_writer = nullptr;
263		}
264	59x	start(token);
265	59x	}
266
267	59x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
268		{
269	59x	auto* self = static_cast<uring_dgram_recv_op*>(base);
270	59x	::io_uring_prep_recvmsg(
271	59x	sqe, self->fd, &self->msg, self->msg_flags);
272	59x	}
273
274	59x	static void do_cqe(
275		io_uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
276		{
277	59x	auto* self = static_cast<uring_dgram_recv_op*>(base);
278	59x	self->res = res;
279	59x	self->cqe_flags = flags;
280		// recvmsg writes the actual source addrlen back into msg.msg_namelen.
281	59x	self->source_len = self->msg.msg_namelen;
282	59x	local.push(self);
283	59x	}
284
285	59x	static void do_handler(
286		void* owner, scheduler_op* base,
287		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
288		{
289	59x	auto* self = static_cast<uring_dgram_recv_op*>(base);
290	59x	if (coro_drain_if_shutdown(owner, self))
291	✗	return;
292
293	59x	if (self->sched_)
294	59x	self->sched_->reset_inline_budget();
295
296		// Datagram recv: a 0-byte datagram is success, not EOF — is_read
297		// stays false so the shared decode never maps it to end_of_file.
298	106x	decode_io_result(
299		self->ec_out,
300	59x	self->cancelled.load(std::memory_order_acquire),
301	59x	self->res < 0 ? make_err(-self->res) : std::error_code{},
302		/is_read=/false, /bytes=/0, /empty_buffer=/false);
303	59x	if (self->bytes_out)
304	59x	*self->bytes_out = (self->res >= 0)
305	59x	? static_cast<std::size_t>(self->res) : 0;
306
307	59x	if (self->res > 0 && self->spec_state)
308		{
309		// Kernel signalled readiness — restore speculation.
310	47x	self->spec_state->on_async_read_ready();
311		}
312
313		// Translate source storage into user's endpoint output (only on
314		// success and only when the concrete socket type asked for it).
315	59x	if (self->source_writer && self->res >= 0)
316	24x	self->source_writer(self->source_writer_ctx,
317	24x	self->source_storage, self->source_len);
318
319	59x	coro_resume(self);
320		}
321		};
322
323		} // namespace boost::corosio::detail
324
325		#endif // BOOST_COROSIO_HAS_IO_URING
326
327		#endif // BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_DGRAM_OPS_HPP
328