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

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