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

99.2% Lines (123/0/124) 100.0% List of functions (17/0/17)

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

io_uring_stream_file.hpp

f(x) Functions (17)

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_STREAM_FILE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_STREAM_FILE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_IO_URING
16
17		#include <boost/corosio/detail/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/stream_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_stream_file_service;
41
42		/** Native io_uring stream-file implementation.
43
44		Async `read_some` / `write_some` submit `IORING_OP_READV` /
45		`IORING_OP_WRITEV` with `offset == -1` (kernel f_pos). All
46		metadata operations (open, size, resize, sync, seek, close)
47		are synchronous syscalls.
48
49		@par Thread Safety
50		Concurrent `read_some` / `write_some` calls on the same file
51		interleave at the kernel level (matches POSIX `read(2)` /
52		`write(2)` semantics on a shared positional fd).
53
54		@note On `O_APPEND` open this backend relies on the kernel's
55		`f_pos` rather than tracking the offset in user space. Writes
56		still go to EOF atomically per `O_APPEND` semantics, but
57		`seek(0, seek_cur)` immediately after an append-mode open
58		returns `0` (the current f_pos), not the file size — observably
59		different from the POSIX backend, which seeds an internal offset
60		to size-at-open. Both behaviours are valid; documented for
61		cross-backend symmetry.
62		*/
63		class BOOST_COROSIO_DECL io_uring_stream_file final
64		: public stream_file::implementation
65		, public std::enable_shared_from_this<io_uring_stream_file>
66		, public intrusive_list<io_uring_stream_file>::node
67		{
68		friend class io_uring_stream_file_service;
69
70		int fd_ = -1;
71		io_uring_scheduler* sched_ = nullptr;
72
73		// Per-fd op slots — embedded to eliminate per-call heap allocation.
74		// Single-pending invariant per slot.
75		uring_file_read_op rd_;
76		uring_file_write_op wr_;
77
78		public:
79	43x	explicit io_uring_stream_file(io_uring_scheduler& sched) noexcept
80	43x	: sched_(&sched)
81	43x	{}
82
83	43x	~io_uring_stream_file() override
84	43x	{
85	43x	close_file();
86	43x	}
87
88		// -- io_stream::implementation --
89
90		std::coroutine_handle<> read_some(
91		std::coroutine_handle<>,
92		capy::executor_ref,
93		buffer_param,
94		std::stop_token,
95		std::error_code*,
96		std::size_t*) override;
97
98		std::coroutine_handle<> write_some(
99		std::coroutine_handle<>,
100		capy::executor_ref,
101		buffer_param,
102		std::stop_token,
103		std::error_code*,
104		std::size_t*) override;
105
106		// -- stream_file::implementation --
107
108	136x	native_handle_type native_handle() const noexcept override
109		{
110	136x	return fd_;
111		}
112
113	1x	void cancel() noexcept override
114		{
115	1x	if (fd_ >= 0)
116	1x	sched_->submit_cancel_by_fd(fd_);
117	1x	}
118
119	5x	std::uint64_t size() const override
120		{
121		struct stat st;
122	5x	if (::fstat(fd_, &st) < 0)
123	✗	throw_system_error(make_err(errno), "stream_file::size");
124	5x	return static_cast<std::uint64_t>(st.st_size);
125		}
126
127	3x	std::error_code resize(std::uint64_t new_size) noexcept override
128		{
129	3x	if (new_size > static_cast<std::uint64_t>(
130	3x	(std::numeric_limits<off_t>::max)()))
131	1x	return make_err(EOVERFLOW);
132	2x	if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
133	1x	return make_err(errno);
134	1x	return {};
135		}
136
137	2x	std::error_code sync_data() noexcept override
138		{
139		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
140	2x	if (::fdatasync(fd_) < 0)
141		#else
142		if (::fsync(fd_) < 0)
143		#endif
144	1x	return make_err(errno);
145	1x	return {};
146		}
147
148	2x	std::error_code sync_all() noexcept override
149		{
150	2x	if (::fsync(fd_) < 0)
151	1x	return make_err(errno);
152	1x	return {};
153		}
154
155	1x	native_handle_type release() override
156		{
157	1x	int fd = fd_;
158	1x	fd_ = -1;
159	1x	return fd;
160		}
161
162	3x	std::error_code assign(native_handle_type handle) noexcept override
163		{
164	3x	close_file();
165	3x	fd_ = handle;
166	3x	return {};
167		}
168
169	8x	capy::io_result<std::uint64_t> seek(
170		std::int64_t offset, file_base::seek_basis origin) noexcept override
171		{
172	8x	int whence = SEEK_SET;
173	8x	if (origin == file_base::seek_cur) whence = SEEK_CUR;
174	6x	else if (origin == file_base::seek_end) whence = SEEK_END;
175
176	8x	off_t r = ::lseek(fd_, static_cast<off_t>(offset), whence);
177	8x	if (r == static_cast<off_t>(-1))
178	3x	return {make_err(errno), 0};
179	5x	return {std::error_code{}, static_cast<std::uint64_t>(r)};
180		}
181
182		// -- Internal --
183
184		/// Open the file. Synchronous; sets `fd_`. Caller is the service.
185	33x	std::error_code open_file(
186		std::filesystem::path const& path, file_base::flags mode)
187		{
188	33x	close_file();
189
190	33x	int oflags = 0;
191	33x	unsigned access = static_cast<unsigned>(mode) & 3u;
192	33x	if (access == static_cast<unsigned>(file_base::read_write))
193	2x	oflags \|= O_RDWR;
194	31x	else if (access == static_cast<unsigned>(file_base::write_only))
195	11x	oflags \|= O_WRONLY;
196		else
197	20x	oflags \|= O_RDONLY;
198
199	33x	if ((mode & file_base::create) != file_base::flags(0))
200	9x	oflags \|= O_CREAT;
201	33x	if ((mode & file_base::exclusive) != file_base::flags(0))
202	1x	oflags \|= O_EXCL;
203	33x	if ((mode & file_base::truncate) != file_base::flags(0))
204	8x	oflags \|= O_TRUNC;
205	33x	if ((mode & file_base::append) != file_base::flags(0))
206	1x	oflags \|= O_APPEND;
207	33x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
208	1x	oflags \|= O_SYNC;
209
210	33x	oflags \|= O_CLOEXEC;
211
212	33x	int fd = ::open(path.c_str(), oflags, 0666);
213	33x	if (fd < 0)
214	2x	return make_err(errno);
215
216	31x	fd_ = fd;
217
218		#ifdef POSIX_FADV_SEQUENTIAL
219		// Hint the page cache about the access pattern; matches the
220		// POSIX backend.
221	31x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
222		#endif
223
224	31x	return {};
225		}
226
227		/// Cancel any in-flight ops and close the fd. Idempotent.
228	198x	void close_file() noexcept
229		{
230	198x	if (fd_ >= 0)
231		{
232	33x	sched_->cancel_and_flush(fd_);
233	33x	::close(fd_);
234	33x	fd_ = -1;
235		}
236	198x	}
237		};
238
239		inline std::coroutine_handle<>
240	13x	io_uring_stream_file::read_some(
241		std::coroutine_handle<> h,
242		capy::executor_ref ex,
243		buffer_param buffers,
244		std::stop_token token,
245		std::error_code* ec,
246		std::size_t* bytes)
247		{
248	13x	rd_.prepare(h, ex, ec, bytes, fd_, /file_offset=/-1, sched_,
249	26x	shared_from_this(), buffers, token);
250	13x	sched_->work_started();
251
252		// Closed-object contract outranks the zero-length no-op.
253	13x	if (fd_ < 0)
254		{
255	3x	rd_.empty_buffer = false;
256	3x	rd_.res = -EBADF;
257	3x	io_uring_scheduler::lock_type lock(sched_->dispatch_mutex());
258	3x	sched_->push_completed_locked(&rd_);
259	3x	return std::noop_coroutine();
260	3x	}
261
262	19x	if (rd_.empty_buffer \|\|
263	9x	rd_.cancelled.load(std::memory_order_acquire))
264		{
265	2x	io_uring_scheduler::lock_type lock(sched_->dispatch_mutex());
266	2x	sched_->push_completed_locked(&rd_);
267	2x	return std::noop_coroutine();
268	2x	}
269
270	8x	io_uring_submit_op(*sched_, &rd_);
271	8x	return std::noop_coroutine();
272		}
273
274		inline std::coroutine_handle<>
275	13x	io_uring_stream_file::write_some(
276		std::coroutine_handle<> h,
277		capy::executor_ref ex,
278		buffer_param buffers,
279		std::stop_token token,
280		std::error_code* ec,
281		std::size_t* bytes)
282		{
283	13x	wr_.prepare(h, ex, ec, bytes, fd_, /file_offset=/-1, sched_,
284	26x	shared_from_this(), buffers, token);
285	13x	sched_->work_started();
286
287		// Closed-object contract outranks the zero-length no-op.
288	13x	if (fd_ < 0)
289		{
290	3x	wr_.empty_buffer = false;
291	3x	wr_.res = -EBADF;
292	3x	io_uring_scheduler::lock_type lock(sched_->dispatch_mutex());
293	3x	sched_->push_completed_locked(&wr_);
294	3x	return std::noop_coroutine();
295	3x	}
296
297	19x	if (wr_.empty_buffer \|\|
298	9x	wr_.cancelled.load(std::memory_order_acquire))
299		{
300	1x	io_uring_scheduler::lock_type lock(sched_->dispatch_mutex());
301	1x	sched_->push_completed_locked(&wr_);
302	1x	return std::noop_coroutine();
303	1x	}
304
305	9x	io_uring_submit_op(*sched_, &wr_);
306	9x	return std::noop_coroutine();
307		}
308
309		/** Native io_uring stream-file service.
310
311		Owns all `io_uring_stream_file` impls. Replaces
312		`posix_stream_file_service` for the io_uring backend; registered
313		under the abstract `file_service` key by `io_uring_t::construct`.
314		*/
315		class BOOST_COROSIO_DECL io_uring_stream_file_service final
316		: public io_uring_file_service_base<
317		io_uring_stream_file_service, file_service, io_uring_stream_file>
318		{
319		using base_service = io_uring_file_service_base<
320		io_uring_stream_file_service, file_service, io_uring_stream_file>;
321
322		public:
323	655x	explicit io_uring_stream_file_service(
324		capy::execution_context& /ctx/, io_uring_scheduler& sched)
325	655x	: base_service(sched)
326	655x	{}
327
328		// construct / destroy / close / shutdown / scheduler() are inherited
329		// from io_uring_file_service_base.
330
331	33x	std::error_code open_file(
332		stream_file::implementation& impl,
333		std::filesystem::path const& path,
334		file_base::flags mode) override
335		{
336	33x	return static_cast<io_uring_stream_file&>(impl).open_file(
337	33x	path, mode);
338		}
339		};
340
341		} // namespace boost::corosio::detail
342
343		#endif // BOOST_COROSIO_HAS_IO_URING
344
345		#endif // BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_STREAM_FILE_HPP
346