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

64.3% Lines (90/140) 68.2% List of functions (15/22)

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

io_uring_random_access_file.hpp

f(x) Functions (22)

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