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

74.8% Lines (101 / 135) 88.9% Functions (16 / 18)

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

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