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

100.0% Lines (123/0/123) 100.0% List of functions (18/0/18)

[ ] prev/next file

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_URING_URING_FILE_OPS_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_FILE_OPS_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_URING
17
18		#include <boost/corosio/native/detail/uring/uring_op.hpp>
19		#include <boost/corosio/native/detail/uring/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 : uring_op
46		{
47		iovec iovecs[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	202x	explicit uring_file_read_op_base(func_type handler) noexcept
54	202x	: uring_op(handler, &do_cqe, &do_prep)
55		{
56	202x	is_read = true;
57	202x	}
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	163x	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		uring_scheduler* scheduler,
74		std::shared_ptr<void> impl,
75		buffer_param buffers,
76		std::stop_token const& token) noexcept
77		{
78	163x	h = handle;
79	163x	ex = executor;
80	163x	ec_out = ec;
81	163x	bytes_out = bytes;
82	163x	fd = file_descriptor;
83	163x	offset = file_offset;
84	163x	sched_ = scheduler;
85	163x	impl_ptr = std::move(impl);
86	163x	res = 0;
87	163x	cqe_flags = 0;
88	163x	iovec_count = copy_to_iovec(buffers, iovecs);
89	163x	empty_buffer = (iovec_count == 0);
90	163x	start(token);
91	163x	}
92
93	151x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
94		{
95	151x	auto* self = static_cast<uring_file_read_op_base*>(base);
96	151x	::io_uring_prep_readv(
97	151x	sqe, self->fd, self->iovecs, self->iovec_count,
98	151x	static_cast<__u64>(self->offset));
99	151x	}
100
101		static void
102	150x	do_cqe(uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
103		{
104	150x	auto* self = static_cast<uring_file_read_op_base*>(base);
105	150x	self->res = res;
106	150x	self->cqe_flags = flags;
107	150x	local.push(self);
108	150x	}
109
110		/// Common post-completion work used by both handlers: fill ec_out
111		/// and bytes_out, then return the coroutine to resume.
112		static std::coroutine_handle<>
113	143x	finish(uring_file_read_op_base* self) noexcept
114		{
115	143x	uring_set_result(self, /is_read=/true, self->empty_buffer);
116	143x	if (self->bytes_out)
117	143x	*self->bytes_out =
118	143x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
119	143x	self->cont.h = self->h;
120	143x	return dispatch_coro(self->ex, self->cont);
121		}
122		};
123
124		/// Scatter-gather file read embedded as a member of stream_file
125		/// (single-pending per fd). Handler uses the suicide-move pattern;
126		/// the impl owns this slot.
127		struct uring_file_read_op : uring_file_read_op_base
128		{
129	57x	uring_file_read_op() noexcept : uring_file_read_op_base(&do_handler) {}
130
131	17x	static void do_handler(
132		void* owner,
133		scheduler_op* base,
134		std::uint32_t /bytes/,
135		std::uint32_t /error/) noexcept
136		{
137	17x	auto* self = static_cast<uring_file_read_op*>(base);
138	17x	if (coro_drain_if_shutdown(owner, self))
139	2x	return;
140
141	15x	if (self->sched_)
142	15x	self->sched_->reset_inline_budget();
143
144	15x	uring_set_result(self, /is_read=/true, self->empty_buffer);
145	15x	if (self->bytes_out)
146	15x	*self->bytes_out =
147	15x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
148	15x	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	145x	uring_random_access_read_op() noexcept
158	145x	: uring_file_read_op_base(&do_handler)
159		{
160	145x	}
161
162	145x	static void do_handler(
163		void* owner,
164		scheduler_op* base,
165		std::uint32_t /bytes/,
166		std::uint32_t /error/) noexcept
167		{
168	145x	auto* self = static_cast<uring_random_access_read_op*>(base);
169	145x	self->stop_cb.reset();
170
171	145x	if (owner == nullptr)
172		{
173	2x	delete self;
174	2x	return;
175		}
176
177	143x	auto next = finish(self);
178	143x	delete self;
179	143x	next.resume();
180		}
181		};
182
183		/** Scatter-gather file write via `IORING_OP_WRITEV`.
184
185		Stream files pass `offset == -1` (kernel f_pos); random-access
186		files pass an explicit caller-supplied offset. Unlike socket
187		writes there is no `MSG_NOSIGNAL` equivalent: a write to a pipe
188		or FIFO whose reader has closed raises SIGPIPE when the kernel
189		executes it, so the teardown paths that flush queued writes hold
190		the signal blocked (see `scoped_sigpipe_block`).
191		*/
192		/// Shared state and submission logic for file write ops. Concrete
193		/// subclasses pick a `do_handler` matching their storage model.
194		struct uring_file_write_op_base : uring_op
195		{
196		iovec iovecs[uring_max_iov];
197		int iovec_count = 0;
198		int fd = -1;
199		std::int64_t offset = -1;
200
201		protected:
202	78x	explicit uring_file_write_op_base(func_type handler) noexcept
203	78x	: uring_op(handler, &do_cqe, &do_prep)
204		{
205	78x	}
206
207		public:
208		/** Reset and initialize for a new submission.
209
210		See uring_file_read_op_base::prepare for the offset convention.
211		*/
212	40x	void prepare(
213		std::coroutine_handle<> handle,
214		capy::executor_ref executor,
215		std::error_code* ec,
216		std::size_t* bytes,
217		int file_descriptor,
218		std::int64_t file_offset,
219		uring_scheduler* scheduler,
220		std::shared_ptr<void> impl,
221		buffer_param buffers,
222		std::stop_token const& token) noexcept
223		{
224	40x	h = handle;
225	40x	ex = executor;
226	40x	ec_out = ec;
227	40x	bytes_out = bytes;
228	40x	fd = file_descriptor;
229	40x	offset = file_offset;
230	40x	sched_ = scheduler;
231	40x	impl_ptr = std::move(impl);
232	40x	res = 0;
233	40x	cqe_flags = 0;
234	40x	iovec_count = copy_to_iovec(buffers, iovecs);
235	40x	empty_buffer = (iovec_count == 0);
236	40x	start(token);
237	40x	}
238
239	31x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
240		{
241	31x	auto* self = static_cast<uring_file_write_op_base*>(base);
242	31x	::io_uring_prep_writev(
243	31x	sqe, self->fd, self->iovecs, self->iovec_count,
244	31x	static_cast<__u64>(self->offset));
245	31x	}
246
247		static void
248	28x	do_cqe(uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
249		{
250	28x	auto* self = static_cast<uring_file_write_op_base*>(base);
251	28x	self->res = res;
252	28x	self->cqe_flags = flags;
253	28x	local.push(self);
254	28x	}
255
256		static std::coroutine_handle<>
257	19x	finish(uring_file_write_op_base* self) noexcept
258		{
259	19x	uring_set_result(self, /is_read=/false, self->empty_buffer);
260	19x	if (self->bytes_out)
261	19x	*self->bytes_out =
262	19x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
263	19x	self->cont.h = self->h;
264	19x	return dispatch_coro(self->ex, self->cont);
265		}
266		};
267
268		/// Embedded file write op for stream_file.
269		struct uring_file_write_op : uring_file_write_op_base
270		{
271	57x	uring_file_write_op() noexcept : uring_file_write_op_base(&do_handler) {}
272
273	17x	static void do_handler(
274		void* owner,
275		scheduler_op* base,
276		std::uint32_t /bytes/,
277		std::uint32_t /error/) noexcept
278		{
279	17x	auto* self = static_cast<uring_file_write_op*>(base);
280	17x	if (coro_drain_if_shutdown(owner, self))
281	1x	return;
282
283	16x	if (self->sched_)
284	16x	self->sched_->reset_inline_budget();
285
286	16x	uring_set_result(self, /is_read=/false, self->empty_buffer);
287	16x	if (self->bytes_out)
288	16x	*self->bytes_out =
289	16x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
290	16x	coro_resume(self);
291		}
292		};
293
294		/// Heap-allocated file write op for random_access_file.
295		struct uring_random_access_write_op : uring_file_write_op_base
296		{
297	21x	uring_random_access_write_op() noexcept
298	21x	: uring_file_write_op_base(&do_handler)
299		{
300	21x	}
301
302	20x	static void do_handler(
303		void* owner,
304		scheduler_op* base,
305		std::uint32_t /bytes/,
306		std::uint32_t /error/) noexcept
307		{
308	20x	auto* self = static_cast<uring_random_access_write_op*>(base);
309	20x	self->stop_cb.reset();
310
311	20x	if (owner == nullptr)
312		{
313	1x	delete self;
314	1x	return;
315		}
316
317	19x	auto next = finish(self);
318	19x	delete self;
319	19x	next.resume();
320		}
321		};
322
323		} // namespace boost::corosio::detail
324
325		#endif // BOOST_COROSIO_HAS_URING
326
327		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_FILE_OPS_HPP
328