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

92.7% Lines (101/109) 100.0% List of functions (16/16)

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

io_uring_random_access_file.hpp

f(x) Functions (16)

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_RANDOM_ACCESS_FILE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_RANDOM_ACCESS_FILE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_IO_URING
16
17		#include <boost/corosio/detail/random_access_file_service.hpp>
18		#include <boost/corosio/detail/intrusive.hpp>
19		#include <boost/corosio/native/detail/io_uring/io_uring_file_ops.hpp>
20		#include <boost/corosio/native/detail/io_uring/io_uring_file_service_base.hpp>
21		#include <boost/corosio/native/detail/io_uring/io_uring_scheduler.hpp>
22		#include <boost/corosio/native/detail/make_err.hpp>
23		#include <boost/corosio/random_access_file.hpp>
24
25		#include <cstdint>
26		#include <filesystem>
27		#include <limits>
28		#include <memory>
29		#include <mutex>
30		#include <system_error>
31		#include <unordered_map>
32
33		#include <fcntl.h>
34		#include <sys/stat.h>
35		#include <sys/types.h>
36		#include <unistd.h>
37
38		namespace boost::corosio::detail {
39
40		class io_uring_random_access_file_service;
41
42		/** Native io_uring random-access-file implementation.
43
44		Async `read_some_at` / `write_some_at` submit `IORING_OP_READV`
45		/ `IORING_OP_WRITEV` with the caller-supplied offset. Metadata
46		operations (open, size, resize, sync, close) are synchronous
47		syscalls.
48
49		@par Thread Safety
50		Concurrent `read_some_at` / `write_some_at` calls on the same
51		file at distinct offsets are safe; ordering between two
52		submissions at the same offset is unspecified at the kernel
53		level (matches POSIX `pread(2)` / `pwrite(2)` semantics).
54		*/
55		class BOOST_COROSIO_DECL io_uring_random_access_file final
56		: public random_access_file::implementation
57		, public std::enable_shared_from_this<io_uring_random_access_file>
58		, public intrusive_list<io_uring_random_access_file>::node
59		{
60		friend class io_uring_random_access_file_service;
61
62		int fd_ = -1;
63		io_uring_scheduler* sched_ = nullptr;
64
65		// Random-access files legitimately support concurrent ops at
66		// different offsets on the same fd (e.g. parallel reads in
67		// testConcurrentReads). Embedding a single slot would smash
68		// state across calls; ops are heap-allocated per submission.
69
70		public:
71	38x	explicit io_uring_random_access_file(io_uring_scheduler& sched) noexcept
72	38x	: sched_(&sched)
73	38x	{}
74
75	38x	~io_uring_random_access_file() override
76	38x	{
77	38x	close_file();
78	38x	}
79
80		// -- random_access_file::implementation --
81
82		std::coroutine_handle<> read_some_at(
83		std::uint64_t,
84		std::coroutine_handle<>,
85		capy::executor_ref,
86		buffer_param,
87		std::stop_token,
88		std::error_code*,
89		std::size_t*) override;
90
91		std::coroutine_handle<> write_some_at(
92		std::uint64_t,
93		std::coroutine_handle<>,
94		capy::executor_ref,
95		buffer_param,
96		std::stop_token,
97		std::error_code*,
98		std::size_t*) override;
99
100	259x	native_handle_type native_handle() const noexcept override
101		{
102	259x	return fd_;
103		}
104
105	2x	void cancel() noexcept override
106		{
107	2x	if (fd_ >= 0)
108	2x	sched_->submit_cancel_by_fd(fd_);
109	2x	}
110
111	2x	std::uint64_t size() const override
112		{
113		struct stat st;
114	2x	if (::fstat(fd_, &st) < 0)
115	✗	throw_system_error(
116	✗	make_err(errno), "random_access_file::size");
117	2x	return static_cast<std::uint64_t>(st.st_size);
118		}
119
120	3x	void resize(std::uint64_t new_size) override
121		{
122	3x	if (new_size > static_cast<std::uint64_t>(
123	3x	(std::numeric_limits<off_t>::max)()))
124	1x	throw_system_error(
125	2x	make_err(EOVERFLOW), "random_access_file::resize");
126	2x	if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
127	✗	throw_system_error(
128	✗	make_err(errno), "random_access_file::resize");
129	2x	}
130
131	1x	void sync_data() override
132		{
133		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
134	1x	if (::fdatasync(fd_) < 0)
135		#else
136		if (::fsync(fd_) < 0)
137		#endif
138	✗	throw_system_error(
139	✗	make_err(errno), "random_access_file::sync_data");
140	1x	}
141
142	1x	void sync_all() override
143		{
144	1x	if (::fsync(fd_) < 0)
145	✗	throw_system_error(
146	✗	make_err(errno), "random_access_file::sync_all");
147	1x	}
148
149	1x	native_handle_type release() override
150		{
151	1x	int fd = fd_;
152	1x	fd_ = -1;
153	1x	return fd;
154		}
155
156	1x	void assign(native_handle_type handle) override
157		{
158	1x	close_file();
159	1x	fd_ = handle;
160	1x	}
161
162		// -- Internal --
163
164		/// Open the file. Synchronous; sets `fd_`. Caller is the service.
165	32x	std::error_code open_file(
166		std::filesystem::path const& path, file_base::flags mode)
167		{
168	32x	close_file();
169
170	32x	int oflags = 0;
171	32x	unsigned access = static_cast<unsigned>(mode) & 3u;
172	32x	if (access == static_cast<unsigned>(file_base::read_write))
173	7x	oflags \|= O_RDWR;
174	25x	else if (access == static_cast<unsigned>(file_base::write_only))
175	5x	oflags \|= O_WRONLY;
176		else
177	20x	oflags \|= O_RDONLY;
178
179	32x	if ((mode & file_base::create) != file_base::flags(0))
180	9x	oflags \|= O_CREAT;
181	32x	if ((mode & file_base::exclusive) != file_base::flags(0))
182	2x	oflags \|= O_EXCL;
183	32x	if ((mode & file_base::truncate) != file_base::flags(0))
184	7x	oflags \|= O_TRUNC;
185	32x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
186	1x	oflags \|= O_SYNC;
187
188	32x	oflags \|= O_CLOEXEC;
189
190	32x	int fd = ::open(path.c_str(), oflags, 0666);
191	32x	if (fd < 0)
192	2x	return make_err(errno);
193
194	30x	fd_ = fd;
195
196		#ifdef POSIX_FADV_RANDOM
197		// Hint the page cache that access will be random; matches
198		// the POSIX backend.
199	30x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
200		#endif
201
202	30x	return {};
203		}
204
205		/// Cancel any in-flight ops and close the fd. Idempotent.
206	177x	void close_file() noexcept
207		{
208	177x	if (fd_ >= 0)
209		{
210	30x	sched_->cancel_and_flush(fd_);
211	30x	::close(fd_);
212	30x	fd_ = -1;
213		}
214	177x	}
215		};
216
217		inline std::coroutine_handle<>
218	136x	io_uring_random_access_file::read_some_at(
219		std::uint64_t user_offset,
220		std::coroutine_handle<> h,
221		capy::executor_ref ex,
222		buffer_param buffers,
223		std::stop_token token,
224		std::error_code* ec,
225		std::size_t* bytes)
226		{
227	136x	auto op_guard = std::make_unique<uring_random_access_read_op>();
228	272x	op_guard->prepare(h, ex, ec, bytes, fd_,
229		static_cast<std::int64_t>(user_offset),
230	272x	sched_, shared_from_this(), buffers, token);
231	136x	sched_->work_started();
232
233	271x	if (op_guard->empty_buffer \|\|
234	135x	op_guard->cancelled.load(std::memory_order_acquire))
235		{
236	2x	io_uring_scheduler::lock_type lock(sched_->dispatch_mutex());
237	2x	sched_->push_completed_locked(op_guard.release());
238	2x	return std::noop_coroutine();
239	2x	}
240
241	134x	io_uring_submit_op(*sched_, op_guard.release());
242	134x	return std::noop_coroutine();
243	136x	}
244
245		inline std::coroutine_handle<>
246	13x	io_uring_random_access_file::write_some_at(
247		std::uint64_t user_offset,
248		std::coroutine_handle<> h,
249		capy::executor_ref ex,
250		buffer_param buffers,
251		std::stop_token token,
252		std::error_code* ec,
253		std::size_t* bytes)
254		{
255	13x	auto op_guard = std::make_unique<uring_random_access_write_op>();
256	26x	op_guard->prepare(h, ex, ec, bytes, fd_,
257		static_cast<std::int64_t>(user_offset),
258	26x	sched_, shared_from_this(), buffers, token);
259	13x	sched_->work_started();
260
261	25x	if (op_guard->empty_buffer \|\|
262	12x	op_guard->cancelled.load(std::memory_order_acquire))
263		{
264	1x	io_uring_scheduler::lock_type lock(sched_->dispatch_mutex());
265	1x	sched_->push_completed_locked(op_guard.release());
266	1x	return std::noop_coroutine();
267	1x	}
268
269	12x	io_uring_submit_op(*sched_, op_guard.release());
270	12x	return std::noop_coroutine();
271	13x	}
272
273		/** Native io_uring random-access-file service.
274
275		Owns all `io_uring_random_access_file` impls. Replaces
276		`posix_random_access_file_service` for the io_uring backend;
277		registered under the abstract `random_access_file_service` key
278		by `io_uring_t::construct`.
279		*/
280		class BOOST_COROSIO_DECL io_uring_random_access_file_service final
281		: public io_uring_file_service_base<
282		io_uring_random_access_file_service,
283		random_access_file_service,
284		io_uring_random_access_file>
285		{
286		using base_service = io_uring_file_service_base<
287		io_uring_random_access_file_service,
288		random_access_file_service,
289		io_uring_random_access_file>;
290
291		public:
292	551x	explicit io_uring_random_access_file_service(
293		capy::execution_context& /ctx/, io_uring_scheduler& sched)
294	551x	: base_service(sched)
295	551x	{}
296
297		// construct / destroy / close / shutdown / scheduler() are inherited
298		// from io_uring_file_service_base.
299
300	32x	std::error_code open_file(
301		random_access_file::implementation& impl,
302		std::filesystem::path const& path,
303		file_base::flags mode) override
304		{
305	32x	return static_cast<io_uring_random_access_file&>(impl).open_file(
306	32x	path, mode);
307		}
308		};
309
310		} // namespace boost::corosio::detail
311
312		#endif // BOOST_COROSIO_HAS_IO_URING
313
314		#endif // BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_RANDOM_ACCESS_FILE_HPP
315