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

88.8% Lines (95/0/107) 87.5% List of functions (14/0/16)

[ ] prev/next file n p prev/next uncovered line

posix_random_access_file.hpp

f(x) Functions (16)

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