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

100.0% Lines (113/0/113) 100.0% List of functions (10/0/10)

[ ] prev/next file

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