include/boost/corosio/native/detail/posix/posix_random_access_file.hpp

100.0% Lines (95 / 95) 100.0% Functions (14 / 14)

[ ] prev/next file

posix_random_access_file.hpp

f(x) Functions (14)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Michael Vandeberg
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_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_POSIX
16
17		#include <boost/corosio/detail/config.hpp>
18		#include <boost/corosio/random_access_file.hpp>
19		#include <boost/corosio/file_base.hpp>
20		#include <boost/corosio/detail/intrusive.hpp>
21		#include <boost/corosio/detail/scheduler_op.hpp>
22		#include <boost/corosio/detail/thread_pool.hpp>
23		#include <boost/corosio/detail/scheduler.hpp>
24		#include <boost/corosio/detail/buffer_param.hpp>
25		#include <boost/corosio/native/detail/coro_op.hpp>
26		#include <boost/corosio/native/detail/coro_op_complete.hpp>
27		#include <boost/corosio/native/detail/make_err.hpp>
28		#include <boost/corosio/native/detail/validate_fd.hpp>
29		#include <boost/capy/ex/executor_ref.hpp>
30		#include <boost/capy/error.hpp>
31		#include <boost/capy/buffers.hpp>
32
33		#include <atomic>
34		#include <coroutine>
35		#include <cstddef>
36		#include <cstdint>
37		#include <filesystem>
38		#include <limits>
39		#include <memory>
40		#include <mutex>
41		#include <optional>
42		#include <stop_token>
43		#include <system_error>
44
45		#include <errno.h>
46		#include <fcntl.h>
47		#include <sys/stat.h>
48		#include <sys/uio.h>
49		#include <unistd.h>
50
51		/*
52		POSIX Random-Access File Implementation
53		========================================
54
55		Each async read/write heap-allocates an raf_op that serves
56		as both the thread-pool work item and the scheduler completion
57		op. This allows unlimited concurrent operations on the same
58		file object, matching Asio's per-op allocation model.
59
60		The raf_op self-deletes on completion or shutdown.
61		*/
62
63		namespace boost::corosio::detail {
64
65		struct scheduler;
66		class posix_random_access_file_service;
67
68		/** Random-access file implementation for POSIX backends. */
69		class posix_random_access_file final
70		: public random_access_file::implementation
71		, public std::enable_shared_from_this<posix_random_access_file>
72		, public intrusive_list<posix_random_access_file>::node
73		{
74		friend class posix_random_access_file_service;
75
76		public:
77		static constexpr std::size_t max_buffers = 16;
78
79		/** Per-operation state, heap-allocated for each async call.
80
81		Inherits from `coro_op` (for scheduler completion plus the shared
82		coroutine, cancellation and keepalive machinery) and
83		`pool_work_item` (for thread-pool dispatch). Linked into the
84		file's outstanding_ops_ list for cancellation tracking. `coro_op`
85		leads the base list so a `scheduler_op*` round-trips.
86		*/
87		struct raf_op final
88		: coro_op
89		, pool_work_item
90		, intrusive_list<raf_op>::node
91		{
92		iovec iovecs[max_buffers];
93		int iovec_count = 0;
94		std::uint64_t offset = 0;
95
96		int errn = 0;
97		std::size_t bytes_transferred = 0;
98
99		// Raw back-pointer for the typed work; `impl_ptr` is the keepalive.
100		posix_random_access_file* file_ = nullptr;
101
102		void operator()() override;
103		void destroy() override;
104
105		/// Thread-pool work function: executes preadv/pwritev.
106		static void do_work(pool_work_item*) noexcept;
107		};
108
109		explicit posix_random_access_file(
110		posix_random_access_file_service& svc) noexcept;
111
112		// -- random_access_file::implementation --
113
114		std::coroutine_handle<> read_some_at(
115		std::uint64_t offset,
116		std::coroutine_handle<>,
117		capy::executor_ref,
118		buffer_param,
119		std::stop_token,
120		std::error_code*,
121		std::size_t*) override;
122
123		std::coroutine_handle<> write_some_at(
124		std::uint64_t offset,
125		std::coroutine_handle<>,
126		capy::executor_ref,
127		buffer_param,
128		std::stop_token,
129		std::error_code*,
130		std::size_t*) override;
131
132	1105x	native_handle_type native_handle() const noexcept override
133		{
134	1105x	return fd_;
135		}
136
137	645x	void cancel() noexcept override
138		{
139	645x	std::lock_guard<std::mutex> lock(ops_mutex_);
140	645x	outstanding_ops_.for_each([](raf_op* op) {
141	8x	op->cancelled.store(true, std::memory_order_release);
142	8x	});
143	645x	}
144
145		std::uint64_t size() const override;
146		std::error_code resize(std::uint64_t new_size) noexcept override;
147		std::error_code sync_data() noexcept override;
148		std::error_code sync_all() noexcept override;
149		native_handle_type release() override;
150		std::error_code assign(native_handle_type handle) noexcept override;
151
152		std::error_code
153		open_file(std::filesystem::path const& path, file_base::flags mode);
154		void close_file() noexcept;
155
156		private:
157		posix_random_access_file_service& svc_;
158		int fd_ = -1;
159		std::mutex ops_mutex_;
160		intrusive_list<raf_op> outstanding_ops_;
161		};
162
163		// ---------------------------------------------------------------------------
164		// Inline implementation
165		// ---------------------------------------------------------------------------
166
167	221x	inline posix_random_access_file::posix_random_access_file(
168	221x	posix_random_access_file_service& svc) noexcept
169	221x	: svc_(svc)
170		{
171	221x	}
172
173		inline std::error_code
174	205x	posix_random_access_file::open_file(
175		std::filesystem::path const& path, file_base::flags mode)
176		{
177	205x	close_file();
178
179	205x	int oflags = 0;
180
181	205x	unsigned access = static_cast<unsigned>(mode) & 3u;
182	205x	if (access == static_cast<unsigned>(file_base::read_write))
183	35x	oflags \|= O_RDWR;
184	170x	else if (access == static_cast<unsigned>(file_base::write_only))
185	64x	oflags \|= O_WRONLY;
186		else
187	106x	oflags \|= O_RDONLY;
188
189	205x	if ((mode & file_base::create) != file_base::flags(0))
190	32x	oflags \|= O_CREAT;
191	205x	if ((mode & file_base::exclusive) != file_base::flags(0))
192	4x	oflags \|= O_EXCL;
193	205x	if ((mode & file_base::truncate) != file_base::flags(0))
194	14x	oflags \|= O_TRUNC;
195	205x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
196	2x	oflags \|= O_SYNC;
197		// Note: no O_APPEND for random access files
198
199	205x	int fd = ::open(path.c_str(), oflags, 0666);
200	205x	if (fd < 0)
201	9x	return make_err(errno);
202
203	196x	fd_ = fd;
204
205		#ifdef POSIX_FADV_RANDOM
206	196x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
207		#endif
208
209	196x	return {};
210		}
211
212		inline void
213	846x	posix_random_access_file::close_file() noexcept
214		{
215	846x	if (fd_ >= 0)
216		{
217	200x	::close(fd_);
218	200x	fd_ = -1;
219		}
220	846x	}
221
222		inline std::uint64_t
223	13x	posix_random_access_file::size() const
224		{
225		struct stat st;
226	13x	if (::fstat(fd_, &st) < 0)
227	5x	throw_system_error(make_err(errno), "random_access_file::size");
228	8x	return static_cast<std::uint64_t>(st.st_size);
229		}
230
231		inline std::error_code
232	13x	posix_random_access_file::resize(std::uint64_t new_size) noexcept
233		{
234	13x	if (new_size >
235	13x	static_cast<std::uint64_t>((std::numeric_limits<off_t>::max)()))
236	2x	return make_err(EOVERFLOW);
237	11x	if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
238	7x	return make_err(errno);
239	4x	return {};
240		}
241
242		inline std::error_code
243	7x	posix_random_access_file::sync_data() noexcept
244		{
245		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
246	7x	if (::fdatasync(fd_) < 0)
247		#else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
248		if (::fsync(fd_) < 0)
249		#endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
250	5x	return make_err(errno);
251	2x	return {};
252		}
253
254		inline std::error_code
255	7x	posix_random_access_file::sync_all() noexcept
256		{
257	7x	if (::fsync(fd_) < 0)
258	5x	return make_err(errno);
259	2x	return {};
260		}
261
262		inline native_handle_type
263	3x	posix_random_access_file::release()
264		{
265	3x	int fd = fd_;
266	3x	fd_ = -1;
267	3x	return fd;
268		}
269
270		inline std::error_code
271	13x	posix_random_access_file::assign(native_handle_type handle) noexcept
272		{
273		// handle >= 0 guard: an unset impl reports native_handle() == -1, and a
274		// caller-supplied -1 must fail as a bad fd, not a self-assign.
275	13x	if (handle >= 0 && handle == fd_)
276	2x	return std::make_error_code(std::errc::invalid_argument);
277
278		// Validate before touching the held fd: a failed assign must leave
279		// this object unchanged and the caller still owning handle.
280	11x	if (auto ec = validate_file_fd(handle))
281	4x	return ec;
282
283		// cancel() first: an in-flight read_at/write_at's pool-thread
284		// completion reads fd_ at execution time, not at post time, so
285		// without this a pending op silently completes against the newly
286		// adopted file instead of being cancelled. The service's
287		// close(handle) / destroy() normally pair cancel()+close_file();
288		// assign() bypasses that path and must do the same pairing itself.
289	7x	cancel();
290	7x	close_file();
291	7x	fd_ = handle;
292	7x	return {};
293		}
294
295		// read_some_at, write_some_at are defined in
296		// posix_random_access_file_service.hpp after the service.
297
298		// -- raf_op completion handler (scheduler thread) --
299
300		inline void
301	422x	posix_random_access_file::raf_op::operator()()
302		{
303	422x	stop_cb.reset();
304
305		// Empty buffers never reach the pool (diverted at initiation), so
306		// empty_buffer stays false and a 0-byte read is a genuine EOF.
307	773x	decode_io_result(
308	422x	ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
309	422x	errn != 0 ? make_err(errn) : std::error_code{}, is_read,
310		bytes_transferred, /empty_buffer=/false);
311
312		{
313	422x	std::lock_guard<std::mutex> lock(file_->ops_mutex_);
314	422x	file_->outstanding_ops_.remove(this);
315	422x	}
316
317	422x	impl_ptr.reset();
318
319	422x	auto coro = h;
320	422x	ex.on_work_finished();
321	422x	delete this;
322	422x	coro.resume();
323	422x	}
324
325		// -- raf_op shutdown cleanup --
326
327		inline void
328	6x	posix_random_access_file::raf_op::destroy()
329		{
330	6x	stop_cb.reset();
331		{
332	6x	std::lock_guard<std::mutex> lock(file_->ops_mutex_);
333	6x	file_->outstanding_ops_.remove(this);
334	6x	}
335	6x	impl_ptr.reset();
336	6x	ex.on_work_finished();
337	6x	delete this;
338	6x	}
339
340		} // namespace boost::corosio::detail
341
342		#endif // BOOST_COROSIO_POSIX
343
344		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
345