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

92.1% Lines (116/126) 100.0% List of functions (18/18)

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

io_uring_file_ops.hpp

f(x) Functions (18)

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_FILE_OPS_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_FILE_OPS_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_IO_URING
16
17		#include <boost/corosio/native/detail/io_uring/io_uring_op.hpp>
18		#include <boost/corosio/native/detail/io_uring/io_uring_socket_ops.hpp>
19		#include <boost/corosio/detail/dispatch_coro.hpp>
20
21		#include <cstdint>
22		#include <sys/uio.h>
23
24		namespace boost::corosio::detail {
25
26		/** Scatter-gather file read via `IORING_OP_READV`.
27
28		Stream files pass `offset == -1` so the kernel uses (and updates)
29		the fd's `f_pos`, matching POSIX `read(2)` semantics. Random-
30		access files pass an explicit caller-supplied offset.
31
32		@par Handler dispatch
33		`do_cqe` captures `res`/`cqe_flags` and queues self into `local`;
34		`do_handler` runs from the scheduler queue and resumes the
35		coroutine.
36		*/
37		/// Shared state and submission logic for file read ops. Concrete
38		/// subclasses pick a `do_handler` that matches their storage model:
39		/// `uring_file_read_op` for embedded slots (stream_file), and
40		/// `uring_random_access_read_op` for heap-allocated per-call ops
41		/// (random_access_file, where concurrent reads at different offsets
42		/// are legitimate).
43		struct uring_file_read_op_base : io_uring_op
44		{
45		iovec iovecs[io_uring_max_iov];
46		int iovec_count = 0;
47		int fd = -1;
48		std::int64_t offset = -1; // -1 means kernel f_pos
49
50		protected:
51	7x	explicit uring_file_read_op_base(func_type handler) noexcept
52	7x	: io_uring_op(handler, &do_cqe, &do_prep)
53		{
54	7x	is_read = true;
55	7x	}
56
57		public:
58		/** Reset and initialize for a new submission.
59
60		@param file_offset -1 selects the kernel's `f_pos` (POSIX
61		`read(2)` semantics for stream files); otherwise the explicit
62		offset for random-access files.
63		*/
64	4x	void prepare(
65		std::coroutine_handle<> handle,
66		capy::executor_ref executor,
67		std::error_code* ec,
68		std::size_t* bytes,
69		int file_descriptor,
70		std::int64_t file_offset,
71		io_uring_scheduler* scheduler,
72		std::shared_ptr<void> impl,
73		buffer_param buffers,
74		std::stop_token const& token) noexcept
75		{
76	4x	h = handle;
77	4x	ex = executor;
78	4x	ec_out = ec;
79	4x	bytes_out = bytes;
80	4x	fd = file_descriptor;
81	4x	offset = file_offset;
82	4x	sched_ = scheduler;
83	4x	impl_ptr = std::move(impl);
84	4x	res = 0;
85	4x	cqe_flags = 0;
86	4x	iovec_count = static_cast<int>(
87	4x	buffers.copy_to(
88	4x	reinterpret_cast<capy::mutable_buffer*>(iovecs),
89		io_uring_max_iov));
90	4x	empty_buffer = (iovec_count == 0);
91	4x	start(token);
92	4x	}
93
94	4x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
95		{
96	4x	auto* self = static_cast<uring_file_read_op_base*>(base);
97	4x	::io_uring_prep_readv(
98	4x	sqe, self->fd, self->iovecs, self->iovec_count,
99	4x	static_cast<__u64>(self->offset));
100	4x	}
101
102	4x	static void do_cqe(
103		io_uring_op* base, int res, unsigned flags,
104		op_queue& local) noexcept
105		{
106	4x	auto* self = static_cast<uring_file_read_op_base*>(base);
107	4x	self->res = res;
108	4x	self->cqe_flags = flags;
109	4x	local.push(self);
110	4x	}
111
112		/// Common post-completion work used by both handlers: fill ec_out
113		/// and bytes_out, then return the coroutine to resume.
114		static std::coroutine_handle<>
115	4x	finish(uring_file_read_op_base* self) noexcept
116		{
117	4x	uring_set_result(self, /is_read=/true, self->empty_buffer);
118	4x	if (self->bytes_out)
119	4x	*self->bytes_out =
120	4x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
121	4x	self->cont_op.cont.h = self->h;
122	4x	return dispatch_coro(self->ex, self->cont_op.cont);
123		}
124		};
125
126		/// Scatter-gather file read embedded as a member of stream_file
127		/// (single-pending per fd). Handler uses the suicide-move pattern;
128		/// the impl owns this slot.
129		struct uring_file_read_op : uring_file_read_op_base
130		{
131	5x	uring_file_read_op() noexcept
132	5x	: uring_file_read_op_base(&do_handler) {}
133
134	2x	static void do_handler(
135		void* owner, scheduler_op* base,
136		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
137		{
138	2x	auto* self = static_cast<uring_file_read_op*>(base);
139	2x	self->stop_cb.reset();
140
141	2x	if (owner == nullptr)
142		{
143	✗	auto suicide = std::move(self->impl_ptr);
144	✗	return;
145	✗	}
146
147	2x	auto next = finish(self);
148	2x	auto suicide = std::move(self->impl_ptr);
149	2x	next.resume();
150	2x	}
151		};
152
153		/// Heap-allocated scatter-gather file read for random_access_file —
154		/// each `read_some_at` call allocates a fresh op so multiple reads
155		/// at different offsets on the same fd can be in flight concurrently.
156		struct uring_random_access_read_op : uring_file_read_op_base
157		{
158	2x	uring_random_access_read_op() noexcept
159	2x	: uring_file_read_op_base(&do_handler) {}
160
161	2x	static void do_handler(
162		void* owner, scheduler_op* base,
163		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
164		{
165	2x	auto* self = static_cast<uring_random_access_read_op*>(base);
166	2x	self->stop_cb.reset();
167
168	2x	if (owner == nullptr)
169		{
170	✗	delete self;
171	✗	return;
172		}
173
174	2x	auto next = finish(self);
175	2x	delete self;
176	2x	next.resume();
177		}
178		};
179
180		/** Scatter-gather file write via `IORING_OP_WRITEV`.
181
182		Stream files pass `offset == -1` (kernel f_pos); random-access
183		files pass an explicit caller-supplied offset. Unlike socket
184		writes, no `MSG_NOSIGNAL` is needed — files don't generate
185		SIGPIPE on closed peers.
186		*/
187		/// Shared state and submission logic for file write ops. Concrete
188		/// subclasses pick a `do_handler` matching their storage model.
189		struct uring_file_write_op_base : io_uring_op
190		{
191		iovec iovecs[io_uring_max_iov];
192		int iovec_count = 0;
193		int fd = -1;
194		std::int64_t offset = -1;
195
196		protected:
197	6x	explicit uring_file_write_op_base(func_type handler) noexcept
198	6x	: io_uring_op(handler, &do_cqe, &do_prep) {}
199
200		public:
201		/** Reset and initialize for a new submission.
202
203		See uring_file_read_op_base::prepare for the offset convention.
204		*/
205	2x	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		std::int64_t file_offset,
212		io_uring_scheduler* scheduler,
213		std::shared_ptr<void> impl,
214		buffer_param buffers,
215		std::stop_token const& token) noexcept
216		{
217	2x	h = handle;
218	2x	ex = executor;
219	2x	ec_out = ec;
220	2x	bytes_out = bytes;
221	2x	fd = file_descriptor;
222	2x	offset = file_offset;
223	2x	sched_ = scheduler;
224	2x	impl_ptr = std::move(impl);
225	2x	res = 0;
226	2x	cqe_flags = 0;
227	2x	iovec_count = static_cast<int>(
228	2x	buffers.copy_to(
229	2x	reinterpret_cast<capy::mutable_buffer*>(iovecs),
230		io_uring_max_iov));
231	2x	empty_buffer = (iovec_count == 0);
232	2x	start(token);
233	2x	}
234
235	2x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
236		{
237	2x	auto* self = static_cast<uring_file_write_op_base*>(base);
238	2x	::io_uring_prep_writev(
239	2x	sqe, self->fd, self->iovecs, self->iovec_count,
240	2x	static_cast<__u64>(self->offset));
241	2x	}
242
243	2x	static void do_cqe(
244		io_uring_op* base, int res, unsigned flags,
245		op_queue& local) noexcept
246		{
247	2x	auto* self = static_cast<uring_file_write_op_base*>(base);
248	2x	self->res = res;
249	2x	self->cqe_flags = flags;
250	2x	local.push(self);
251	2x	}
252
253		static std::coroutine_handle<>
254	2x	finish(uring_file_write_op_base* self) noexcept
255		{
256	2x	uring_set_result(self, /is_read=/false, self->empty_buffer);
257	2x	if (self->bytes_out)
258	2x	*self->bytes_out =
259	2x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
260	2x	self->cont_op.cont.h = self->h;
261	2x	return dispatch_coro(self->ex, self->cont_op.cont);
262		}
263		};
264
265		/// Embedded file write op for stream_file.
266		struct uring_file_write_op : uring_file_write_op_base
267		{
268	5x	uring_file_write_op() noexcept
269	5x	: uring_file_write_op_base(&do_handler) {}
270
271	1x	static void do_handler(
272		void* owner, scheduler_op* base,
273		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
274		{
275	1x	auto* self = static_cast<uring_file_write_op*>(base);
276	1x	self->stop_cb.reset();
277
278	1x	if (owner == nullptr)
279		{
280	✗	auto suicide = std::move(self->impl_ptr);
281	✗	return;
282	✗	}
283
284	1x	auto next = finish(self);
285	1x	auto suicide = std::move(self->impl_ptr);
286	1x	next.resume();
287	1x	}
288		};
289
290		/// Heap-allocated file write op for random_access_file.
291		struct uring_random_access_write_op : uring_file_write_op_base
292		{
293	1x	uring_random_access_write_op() noexcept
294	1x	: uring_file_write_op_base(&do_handler) {}
295
296	1x	static void do_handler(
297		void* owner, scheduler_op* base,
298		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
299		{
300	1x	auto* self = static_cast<uring_random_access_write_op*>(base);
301	1x	self->stop_cb.reset();
302
303	1x	if (owner == nullptr)
304		{
305	✗	delete self;
306	✗	return;
307		}
308
309	1x	auto next = finish(self);
310	1x	delete self;
311	1x	next.resume();
312		}
313		};
314
315		} // namespace boost::corosio::detail
316
317		#endif // BOOST_COROSIO_HAS_IO_URING
318
319		#endif // BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_FILE_OPS_HPP
320