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

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