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

95.1% Lines (116/0/122) 100.0% List of functions (18/0/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	183x	explicit uring_file_read_op_base(func_type handler) noexcept
54	183x	: io_uring_op(handler, &do_cqe, &do_prep)
55		{
56	183x	is_read = true;
57	183x	}
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	153x	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	153x	h = handle;
79	153x	ex = executor;
80	153x	ec_out = ec;
81	153x	bytes_out = bytes;
82	153x	fd = file_descriptor;
83	153x	offset = file_offset;
84	153x	sched_ = scheduler;
85	153x	impl_ptr = std::move(impl);
86	153x	res = 0;
87	153x	cqe_flags = 0;
88	153x	iovec_count = copy_to_iovec(buffers, iovecs);
89	153x	empty_buffer = (iovec_count == 0);
90	153x	start(token);
91	153x	}
92
93	144x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
94		{
95	144x	auto* self = static_cast<uring_file_read_op_base*>(base);
96	144x	::io_uring_prep_readv(
97	144x	sqe, self->fd, self->iovecs, self->iovec_count,
98	144x	static_cast<__u64>(self->offset));
99	144x	}
100
101	144x	static void do_cqe(
102		io_uring_op* base, int res, unsigned flags,
103		ready_queue& local) noexcept
104		{
105	144x	auto* self = static_cast<uring_file_read_op_base*>(base);
106	144x	self->res = res;
107	144x	self->cqe_flags = flags;
108	144x	local.push(self);
109	144x	}
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	140x	finish(uring_file_read_op_base* self) noexcept
115		{
116	140x	uring_set_result(self, /is_read=/true, self->empty_buffer);
117	140x	if (self->bytes_out)
118	140x	*self->bytes_out =
119	140x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
120	140x	self->cont.h = self->h;
121	140x	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	43x	uring_file_read_op() noexcept
131	43x	: uring_file_read_op_base(&do_handler) {}
132
133	13x	static void do_handler(
134		void* owner, scheduler_op* base,
135		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
136		{
137	13x	auto* self = static_cast<uring_file_read_op*>(base);
138	13x	if (coro_drain_if_shutdown(owner, self))
139	✗	return;
140
141	13x	if (self->sched_)
142	13x	self->sched_->reset_inline_budget();
143
144	13x	uring_set_result(self, /is_read=/true, self->empty_buffer);
145	13x	if (self->bytes_out)
146	13x	*self->bytes_out =
147	13x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
148	13x	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	140x	uring_random_access_read_op() noexcept
158	140x	: uring_file_read_op_base(&do_handler) {}
159
160	140x	static void do_handler(
161		void* owner, scheduler_op* base,
162		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
163		{
164	140x	auto* self = static_cast<uring_random_access_read_op*>(base);
165	140x	self->stop_cb.reset();
166
167	140x	if (owner == nullptr)
168		{
169	✗	delete self;
170	✗	return;
171		}
172
173	140x	auto next = finish(self);
174	140x	delete self;
175	140x	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	58x	explicit uring_file_write_op_base(func_type handler) noexcept
197	58x	: 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	28x	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	28x	h = handle;
217	28x	ex = executor;
218	28x	ec_out = ec;
219	28x	bytes_out = bytes;
220	28x	fd = file_descriptor;
221	28x	offset = file_offset;
222	28x	sched_ = scheduler;
223	28x	impl_ptr = std::move(impl);
224	28x	res = 0;
225	28x	cqe_flags = 0;
226	28x	iovec_count = copy_to_iovec(buffers, iovecs);
227	28x	empty_buffer = (iovec_count == 0);
228	28x	start(token);
229	28x	}
230
231	22x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
232		{
233	22x	auto* self = static_cast<uring_file_write_op_base*>(base);
234	22x	::io_uring_prep_writev(
235	22x	sqe, self->fd, self->iovecs, self->iovec_count,
236	22x	static_cast<__u64>(self->offset));
237	22x	}
238
239	22x	static void do_cqe(
240		io_uring_op* base, int res, unsigned flags,
241		ready_queue& local) noexcept
242		{
243	22x	auto* self = static_cast<uring_file_write_op_base*>(base);
244	22x	self->res = res;
245	22x	self->cqe_flags = flags;
246	22x	local.push(self);
247	22x	}
248
249		static std::coroutine_handle<>
250	15x	finish(uring_file_write_op_base* self) noexcept
251		{
252	15x	uring_set_result(self, /is_read=/false, self->empty_buffer);
253	15x	if (self->bytes_out)
254	15x	*self->bytes_out =
255	15x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
256	15x	self->cont.h = self->h;
257	15x	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	43x	uring_file_write_op() noexcept
265	43x	: uring_file_write_op_base(&do_handler) {}
266
267	13x	static void do_handler(
268		void* owner, scheduler_op* base,
269		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
270		{
271	13x	auto* self = static_cast<uring_file_write_op*>(base);
272	13x	if (coro_drain_if_shutdown(owner, self))
273	✗	return;
274
275	13x	if (self->sched_)
276	13x	self->sched_->reset_inline_budget();
277
278	13x	uring_set_result(self, /is_read=/false, self->empty_buffer);
279	13x	if (self->bytes_out)
280	13x	*self->bytes_out =
281	13x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
282	13x	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	15x	uring_random_access_write_op() noexcept
290	15x	: uring_file_write_op_base(&do_handler) {}
291
292	15x	static void do_handler(
293		void* owner, scheduler_op* base,
294		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
295		{
296	15x	auto* self = static_cast<uring_random_access_write_op*>(base);
297	15x	self->stop_cb.reset();
298
299	15x	if (owner == nullptr)
300		{
301	✗	delete self;
302	✗	return;
303		}
304
305	15x	auto next = finish(self);
306	15x	delete self;
307	15x	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