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

100.0% Lines (137 / 137) 100.0% Functions (19 / 19)

[ ] prev/next file

uring_stream_file.hpp

f(x) Functions (19)

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_STREAM_FILE_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_STREAM_FILE_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_URING
17
18		#include <boost/corosio/detail/file_service.hpp>
19		#include <boost/corosio/detail/config.hpp>
20		#include <boost/corosio/detail/intrusive.hpp>
21		#include <boost/corosio/native/detail/uring/uring_file_ops.hpp>
22		#include <boost/corosio/native/detail/uring/uring_file_service_base.hpp>
23		#include <boost/corosio/native/detail/uring/uring_scheduler.hpp>
24		#include <boost/corosio/native/detail/make_err.hpp>
25		#include <boost/corosio/native/detail/posix/large_file.hpp>
26		#include <boost/corosio/native/detail/validate_fd.hpp>
27		#include <boost/corosio/stream_file.hpp>
28
29		#include <cstdint>
30		#include <filesystem>
31		#include <limits>
32		#include <memory>
33		#include <mutex>
34		#include <system_error>
35
36		#include <fcntl.h>
37		#include <sys/stat.h>
38		#include <sys/types.h>
39		#include <unistd.h>
40
41		namespace boost::corosio::detail {
42
43		class uring_stream_file_service;
44
45		/** Native io_uring stream-file implementation.
46
47		Async `read_some` / `write_some` submit `IORING_OP_READV` /
48		`IORING_OP_WRITEV` with `offset == -1` (kernel f_pos). All
49		metadata operations (open, size, resize, sync, seek, close)
50		are synchronous syscalls.
51
52		@par Thread Safety
53		Concurrent `read_some` / `write_some` calls on the same file
54		interleave at the kernel level (matches POSIX `read(2)` /
55		`write(2)` semantics on a shared positional fd).
56
57		@note On `O_APPEND` open this backend relies on the kernel's
58		`f_pos` rather than tracking the offset in user space. Writes
59		still go to EOF atomically per `O_APPEND` semantics, but
60		`seek(0, seek_cur)` immediately after an append-mode open
61		returns `0` (the current f_pos), not the file size — observably
62		different from the POSIX backend, which seeds an internal offset
63		to size-at-open. Both behaviours are valid; documented for
64		cross-backend symmetry.
65		*/
66		class BOOST_COROSIO_DECL uring_stream_file final
67		: public stream_file::implementation
68		, public intrusive_list<uring_stream_file>::node
69		{
70		friend class uring_stream_file_service;
71
72		int fd_ = -1;
73		uring_scheduler* sched_ = nullptr;
74		uring_stream_file_service* svc_ = nullptr;
75
76		// Per-fd op slots — embedded to eliminate per-call heap allocation.
77		// Single-pending invariant per slot.
78		uring_file_read_op rd_;
79		uring_file_write_op wr_;
80
81		public:
82	104x	explicit uring_stream_file(
83		uring_stream_file_service& svc, uring_scheduler& sched) noexcept
84	208x	: sched_(&sched)
85	104x	, svc_(&svc)
86		{
87	104x	}
88
89	208x	~uring_stream_file() override
90	104x	{
91	104x	close_file();
92	208x	}
93
94		/// Recycle into the owning service's pool. Defined out-of-line
95		/// after uring_stream_file_service for its complete type.
96		void retire() noexcept override;
97
98		/** Reset op slots for recycling.
99
100		`close_file()` already drove fd_ to its closed value before the
101		refcount reached zero; each op's own `prepare()` overwrites its
102		state before the next use, so only the `stop_cb` invariant is
103		worth asserting here.
104
105		@pre refs_ == 0, fd closed, no op in flight.
106		*/
107	32x	void reuse() noexcept
108		{
109	32x	BOOST_COROSIO_ASSERT(fd_ == -1);
110	32x	BOOST_COROSIO_ASSERT(!rd_.stop_cb);
111	32x	BOOST_COROSIO_ASSERT(!wr_.stop_cb);
112	32x	}
113
114		// -- io_stream::implementation --
115
116		std::coroutine_handle<> read_some(
117		capy::continuation&,
118		capy::executor_ref,
119		buffer_param,
120		std::stop_token,
121		std::error_code*,
122		std::size_t*) override;
123
124		std::coroutine_handle<> write_some(
125		capy::continuation&,
126		capy::executor_ref,
127		buffer_param,
128		std::stop_token,
129		std::error_code*,
130		std::size_t*) override;
131
132		// -- stream_file::implementation --
133
134	392x	native_handle_type native_handle() const noexcept override
135		{
136	392x	return fd_;
137		}
138
139	1x	void cancel() noexcept override
140		{
141	1x	if (fd_ >= 0)
142	1x	sched_->cancel_ops({&rd_, &wr_});
143	1x	}
144
145	6x	capy::results<std::uint64_t> size() const noexcept override
146		{
147		file_stat_t st;
148	6x	if (file_fstat(fd_, &st) < 0)
149	2x	return {make_err(errno), 0};
150	4x	return {std::error_code{}, static_cast<std::uint64_t>(st.st_size)};
151		}
152
153	5x	std::error_code resize(std::uint64_t new_size) noexcept override
154		{
155	5x	if (new_size > static_cast<std::uint64_t>(
156	5x	(std::numeric_limits<file_off_t>::max)()))
157	1x	return make_err(EOVERFLOW);
158	4x	if (file_ftruncate(fd_, static_cast<file_off_t>(new_size)) < 0)
159	3x	return make_err(errno);
160	1x	return {};
161		}
162
163	3x	std::error_code sync_data() noexcept override
164		{
165		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
166	3x	if (::fdatasync(fd_) < 0)
167		#else
168		if (::fsync(fd_) < 0)
169		#endif
170	2x	return make_err(errno);
171	1x	return {};
172		}
173
174	3x	std::error_code sync_all() noexcept override
175		{
176	3x	if (::fsync(fd_) < 0)
177	2x	return make_err(errno);
178	1x	return {};
179		}
180
181	1x	capy::results<native_handle_type> release() noexcept override
182		{
183		// Flush while the fd is still open, so no queued SQE can name
184		// a number the caller closes and recycles.
185	1x	if (fd_ >= 0)
186	1x	sched_->flush_after_cancel(fd_, {&rd_, &wr_});
187	1x	int fd = fd_;
188	1x	fd_ = -1;
189	1x	return {std::error_code{}, fd};
190		}
191
192	3x	std::error_code assign(native_handle_type handle) noexcept override
193		{
194		// The public assign() guarantees the object is closed.
195	3x	if (auto ec = validate_file_fd(handle))
196	2x	return ec;
197
198	1x	fd_ = handle;
199	1x	return {};
200		}
201
202		capy::results<std::uint64_t>
203	11x	seek(std::int64_t offset, file_base::seek_basis origin) noexcept override
204		{
205	11x	int whence = SEEK_SET;
206	11x	if (origin == file_base::seek_cur)
207	2x	whence = SEEK_CUR;
208	9x	else if (origin == file_base::seek_end)
209	4x	whence = SEEK_END;
210
211	11x	file_off_t r = file_lseek(fd_, static_cast<file_off_t>(offset), whence);
212	11x	if (r == static_cast<file_off_t>(-1))
213	5x	return {make_err(errno), 0};
214	6x	return {std::error_code{}, static_cast<std::uint64_t>(r)};
215		}
216
217		// -- Internal --
218
219		/// Open the file. Synchronous; sets `fd_`. Caller is the service.
220		std::error_code
221	124x	open_file(std::filesystem::path const& path, file_base::flags mode)
222		{
223	124x	close_file();
224
225	124x	int oflags = 0;
226	124x	unsigned access = static_cast<unsigned>(mode) & 3u;
227	124x	if (access == static_cast<unsigned>(file_base::read_write))
228	40x	oflags \|= O_RDWR;
229	84x	else if (access == static_cast<unsigned>(file_base::write_only))
230	37x	oflags \|= O_WRONLY;
231		else
232	47x	oflags \|= O_RDONLY;
233
234	124x	if ((mode & file_base::create) != file_base::flags(0))
235	47x	oflags \|= O_CREAT;
236	124x	if ((mode & file_base::exclusive) != file_base::flags(0))
237	1x	oflags \|= O_EXCL;
238	124x	if ((mode & file_base::truncate) != file_base::flags(0))
239	41x	oflags \|= O_TRUNC;
240	124x	if ((mode & file_base::append) != file_base::flags(0))
241	1x	oflags \|= O_APPEND;
242	124x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
243	1x	oflags \|= O_SYNC;
244
245	124x	oflags \|= O_CLOEXEC;
246
247	124x	int fd = ::open(path.c_str(), oflags \| large_file_open_flag, 0666);
248	124x	if (fd < 0)
249	4x	return make_err(errno);
250
251	120x	fd_ = fd;
252
253		#ifdef POSIX_FADV_SEQUENTIAL
254		// Hint the page cache about the access pattern; matches the
255		// POSIX backend.
256	120x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
257		#endif
258
259	120x	return {};
260		}
261
262		/// Cancel any in-flight ops and close the fd. Idempotent.
263	620x	void close_file() noexcept
264		{
265	620x	if (fd_ >= 0)
266		{
267		// The kernel may run a queued pipe write as task work at
268		// either kernel entry below; with the reader already gone
269		// that raises SIGPIPE.
270	120x	scoped_sigpipe_block no_sigpipe;
271	120x	sched_->flush_after_cancel(fd_, {&rd_, &wr_});
272	120x	::close(fd_);
273	120x	fd_ = -1;
274	120x	}
275	620x	}
276		};
277
278		inline std::coroutine_handle<>
279	38x	uring_stream_file::read_some(
280		capy::continuation& cont,
281		capy::executor_ref ex,
282		buffer_param buffers,
283		std::stop_token token,
284		std::error_code* ec,
285		std::size_t* bytes)
286		{
287	38x	rd_.prepare(
288		cont, ex, ec, bytes, fd_, /file_offset=/-1, sched_,
289	76x	detail::object_ref(this), buffers, token);
290	38x	sched_->work_started();
291
292		// Closed-object contract outranks the zero-length no-op.
293	38x	if (fd_ < 0)
294		{
295	3x	rd_.empty_buffer = false;
296	3x	rd_.res = -EBADF;
297	3x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
298	3x	sched_->push_completed_locked(&rd_);
299	3x	return std::noop_coroutine();
300	3x	}
301
302	35x	if (rd_.empty_buffer \|\| rd_.cancelled.load(std::memory_order_acquire))
303		{
304	1x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
305	1x	sched_->push_completed_locked(&rd_);
306	1x	return std::noop_coroutine();
307	1x	}
308
309	34x	uring_submit_op(*sched_, &rd_);
310	34x	return std::noop_coroutine();
311		}
312
313		inline std::coroutine_handle<>
314	41x	uring_stream_file::write_some(
315		capy::continuation& cont,
316		capy::executor_ref ex,
317		buffer_param buffers,
318		std::stop_token token,
319		std::error_code* ec,
320		std::size_t* bytes)
321		{
322	41x	wr_.prepare(
323		cont, ex, ec, bytes, fd_, /file_offset=/-1, sched_,
324	82x	detail::object_ref(this), buffers, token);
325	41x	sched_->work_started();
326
327		// Closed-object contract outranks the zero-length no-op.
328	41x	if (fd_ < 0)
329		{
330	3x	wr_.empty_buffer = false;
331	3x	wr_.res = -EBADF;
332	3x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
333	3x	sched_->push_completed_locked(&wr_);
334	3x	return std::noop_coroutine();
335	3x	}
336
337	38x	if (wr_.empty_buffer \|\| wr_.cancelled.load(std::memory_order_acquire))
338		{
339	1x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
340	1x	sched_->push_completed_locked(&wr_);
341	1x	return std::noop_coroutine();
342	1x	}
343
344	37x	uring_submit_op(*sched_, &wr_);
345	37x	return std::noop_coroutine();
346		}
347
348		/** Native io_uring stream-file service.
349
350		Owns all `uring_stream_file` impls. Replaces
351		`posix_stream_file_service` for the io_uring backend; registered
352		under the abstract `file_service` key by `uring_t::construct`.
353		*/
354		class BOOST_COROSIO_DECL uring_stream_file_service final
355		: public uring_file_service_base<
356		uring_stream_file_service,
357		file_service,
358		uring_stream_file>
359		{
360		using base_service = uring_file_service_base<
361		uring_stream_file_service,
362		file_service,
363		uring_stream_file>;
364
365		public:
366	76x	explicit uring_stream_file_service(capy::execution_context& ctx)
367	76x	: base_service(ctx.use_service<uring_scheduler>())
368		{
369	76x	}
370
371		// construct / destroy / close / shutdown / scheduler() are inherited
372		// from uring_file_service_base.
373
374	124x	std::error_code open_file(
375		stream_file::implementation& impl,
376		std::filesystem::path const& path,
377		file_base::flags mode) override
378		{
379	124x	return static_cast<uring_stream_file&>(impl).open_file(path, mode);
380		}
381		};
382
383		inline void
384	136x	uring_stream_file::retire() noexcept
385		{
386	136x	svc_->pool_.recycle(this);
387	136x	}
388
389		} // namespace boost::corosio::detail
390
391		#endif // BOOST_COROSIO_HAS_URING
392
393		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_STREAM_FILE_HPP
394