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

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

[ ] prev/next file

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_URING_URING_STREAM_FILE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_STREAM_FILE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_URING
16
17		#include <boost/corosio/detail/file_service.hpp>
18		#include <boost/corosio/detail/intrusive.hpp>
19		#include <boost/corosio/native/detail/uring/uring_file_ops.hpp>
20		#include <boost/corosio/native/detail/uring/uring_file_service_base.hpp>
21		#include <boost/corosio/native/detail/uring/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 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 uring_stream_file final
64		: public stream_file::implementation
65		, public std::enable_shared_from_this<uring_stream_file>
66		, public intrusive_list<uring_stream_file>::node
67		{
68		friend class uring_stream_file_service;
69
70		int fd_ = -1;
71		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	57x	explicit uring_stream_file(uring_scheduler& sched) noexcept : sched_(&sched)
80		{
81	57x	}
82
83	54x	~uring_stream_file() override
84	54x	{
85	54x	close_file();
86	54x	}
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	177x	native_handle_type native_handle() const noexcept override
109		{
110	177x	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	7x	std::uint64_t size() const override
120		{
121		struct stat st;
122	7x	if (::fstat(fd_, &st) < 0)
123	2x	throw_system_error(make_err(errno), "stream_file::size");
124	5x	return static_cast<std::uint64_t>(st.st_size);
125		}
126
127	5x	std::error_code resize(std::uint64_t new_size) noexcept override
128		{
129	5x	if (new_size >
130	5x	static_cast<std::uint64_t>((std::numeric_limits<off_t>::max)()))
131	1x	return make_err(EOVERFLOW);
132	4x	if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
133	3x	return make_err(errno);
134	1x	return {};
135		}
136
137	4x	std::error_code sync_data() noexcept override
138		{
139		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
140	4x	if (::fdatasync(fd_) < 0)
141		#else
142		if (::fsync(fd_) < 0)
143		#endif
144	3x	return make_err(errno);
145	1x	return {};
146		}
147
148	4x	std::error_code sync_all() noexcept override
149		{
150	4x	if (::fsync(fd_) < 0)
151	3x	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	9x	std::error_code assign(native_handle_type handle) noexcept override
163		{
164	9x	close_file();
165	9x	fd_ = handle;
166	9x	return {};
167		}
168
169		capy::io_result<std::uint64_t>
170	11x	seek(std::int64_t offset, file_base::seek_basis origin) noexcept override
171		{
172	11x	int whence = SEEK_SET;
173	11x	if (origin == file_base::seek_cur)
174	2x	whence = SEEK_CUR;
175	9x	else if (origin == file_base::seek_end)
176	4x	whence = SEEK_END;
177
178	11x	off_t r = ::lseek(fd_, static_cast<off_t>(offset), whence);
179	11x	if (r == static_cast<off_t>(-1))
180	5x	return {make_err(errno), 0};
181	6x	return {std::error_code{}, static_cast<std::uint64_t>(r)};
182		}
183
184		// -- Internal --
185
186		/// Open the file. Synchronous; sets `fd_`. Caller is the service.
187		std::error_code
188	41x	open_file(std::filesystem::path const& path, file_base::flags mode)
189		{
190	41x	close_file();
191
192	41x	int oflags = 0;
193	41x	unsigned access = static_cast<unsigned>(mode) & 3u;
194	41x	if (access == static_cast<unsigned>(file_base::read_write))
195	7x	oflags \|= O_RDWR;
196	34x	else if (access == static_cast<unsigned>(file_base::write_only))
197	12x	oflags \|= O_WRONLY;
198		else
199	22x	oflags \|= O_RDONLY;
200
201	41x	if ((mode & file_base::create) != file_base::flags(0))
202	14x	oflags \|= O_CREAT;
203	41x	if ((mode & file_base::exclusive) != file_base::flags(0))
204	1x	oflags \|= O_EXCL;
205	41x	if ((mode & file_base::truncate) != file_base::flags(0))
206	8x	oflags \|= O_TRUNC;
207	41x	if ((mode & file_base::append) != file_base::flags(0))
208	1x	oflags \|= O_APPEND;
209	41x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
210	1x	oflags \|= O_SYNC;
211
212	41x	oflags \|= O_CLOEXEC;
213
214	41x	int fd = ::open(path.c_str(), oflags, 0666);
215	41x	if (fd < 0)
216	4x	return make_err(errno);
217
218	37x	fd_ = fd;
219
220		#ifdef POSIX_FADV_SEQUENTIAL
221		// Hint the page cache about the access pattern; matches the
222		// POSIX backend.
223	37x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
224		#endif
225
226	37x	return {};
227		}
228
229		/// Cancel any in-flight ops and close the fd. Idempotent.
230	249x	void close_file() noexcept
231		{
232	249x	if (fd_ >= 0)
233		{
234		// The kernel may run a queued pipe write as task work at
235		// either kernel entry below; with the reader already gone
236		// that raises SIGPIPE.
237	45x	scoped_sigpipe_block no_sigpipe;
238	45x	sched_->cancel_and_flush(fd_);
239	45x	::close(fd_);
240	45x	fd_ = -1;
241	45x	}
242	249x	}
243		};
244
245		inline std::coroutine_handle<>
246	18x	uring_stream_file::read_some(
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	18x	rd_.prepare(
255	36x	h, ex, ec, bytes, fd_, /file_offset=/-1, sched_, shared_from_this(),
256		buffers, token);
257	18x	sched_->work_started();
258
259		// Closed-object contract outranks the zero-length no-op.
260	18x	if (fd_ < 0)
261		{
262	3x	rd_.empty_buffer = false;
263	3x	rd_.res = -EBADF;
264	3x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
265	3x	sched_->push_completed_locked(&rd_);
266	3x	return std::noop_coroutine();
267	3x	}
268
269	15x	if (rd_.empty_buffer \|\| rd_.cancelled.load(std::memory_order_acquire))
270		{
271	3x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
272	3x	sched_->push_completed_locked(&rd_);
273	3x	return std::noop_coroutine();
274	3x	}
275
276	12x	uring_submit_op(*sched_, &rd_);
277	12x	return std::noop_coroutine();
278		}
279
280		inline std::coroutine_handle<>
281	19x	uring_stream_file::write_some(
282		std::coroutine_handle<> h,
283		capy::executor_ref ex,
284		buffer_param buffers,
285		std::stop_token token,
286		std::error_code* ec,
287		std::size_t* bytes)
288		{
289	19x	wr_.prepare(
290	38x	h, ex, ec, bytes, fd_, /file_offset=/-1, sched_, shared_from_this(),
291		buffers, token);
292	19x	sched_->work_started();
293
294		// Closed-object contract outranks the zero-length no-op.
295	19x	if (fd_ < 0)
296		{
297	3x	wr_.empty_buffer = false;
298	3x	wr_.res = -EBADF;
299	3x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
300	3x	sched_->push_completed_locked(&wr_);
301	3x	return std::noop_coroutine();
302	3x	}
303
304	16x	if (wr_.empty_buffer \|\| wr_.cancelled.load(std::memory_order_acquire))
305		{
306	2x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
307	2x	sched_->push_completed_locked(&wr_);
308	2x	return std::noop_coroutine();
309	2x	}
310
311	14x	uring_submit_op(*sched_, &wr_);
312	14x	return std::noop_coroutine();
313		}
314
315		/** Native io_uring stream-file service.
316
317		Owns all `uring_stream_file` impls. Replaces
318		`posix_stream_file_service` for the io_uring backend; registered
319		under the abstract `file_service` key by `uring_t::construct`.
320		*/
321		class BOOST_COROSIO_DECL uring_stream_file_service final
322		: public uring_file_service_base<
323		uring_stream_file_service,
324		file_service,
325		uring_stream_file>
326		{
327		using base_service = uring_file_service_base<
328		uring_stream_file_service,
329		file_service,
330		uring_stream_file>;
331
332		public:
333	799x	explicit uring_stream_file_service(
334		capy::execution_context& /ctx/, uring_scheduler& sched)
335	799x	: base_service(sched)
336		{
337	799x	}
338
339		// construct / destroy / close / shutdown / scheduler() are inherited
340		// from uring_file_service_base.
341
342	41x	std::error_code open_file(
343		stream_file::implementation& impl,
344		std::filesystem::path const& path,
345		file_base::flags mode) override
346		{
347	41x	return static_cast<uring_stream_file&>(impl).open_file(path, mode);
348		}
349		};
350
351		} // namespace boost::corosio::detail
352
353		#endif // BOOST_COROSIO_HAS_URING
354
355		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_STREAM_FILE_HPP
356