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

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