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

96.4% Lines (108 / 112) 94.1% Functions (16 / 17)

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

posix_random_access_file.hpp

f(x) Functions (17)

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 <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 acquires an raf_op that serves as both the
54		thread-pool work item and the scheduler completion op. This
55		allows unlimited concurrent operations on the same file object.
56
57		Ops recycle through the file's free_ops_ list on completion or
58		shutdown, so steady-state reads and writes allocate nothing; the
59		file destructor frees the recycled storage.
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 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, acquired from the file's free list.
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). The intrusive hook
82		links it into `outstanding_ops_` while in flight and `free_ops_`
83		once recycled — membership is strictly sequential, so one hook
84		serves both under `ops_mutex_`. `coro_op` leads the base list so
85		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; `object_ref_` 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		/// Recycle into the owning service's pool. Defined out-of-line
113		/// after posix_random_access_file_service for its complete type.
114		void retire() noexcept override;
115
116		/** Reset for recycling.
117
118		`close_file()` already drove fd_ to its closed value before
119		the refcount reached zero. Every `raf_op` holds its own
120		`object_ref` while in flight, so the refcount cannot reach
121		zero while one is outstanding — asserting `outstanding_ops_`
122		is empty is therefore a precondition check, not a defensive
123		one. `free_ops_` deliberately survives recycling: the storage
124		belongs to this impl and only the destructor frees it.
125
126		@pre refs_ == 0, fd closed, no op in flight.
127		*/
128	✗	void reuse() noexcept
129		{
130	✗	BOOST_COROSIO_ASSERT(fd_ == -1);
131	✗	BOOST_COROSIO_ASSERT(outstanding_ops_.empty());
132	✗	}
133
134	438x	~posix_random_access_file() override
135	219x	{
136	621x	while (auto* op = free_ops_.pop_front())
137	402x	delete op;
138	438x	}
139
140		// -- random_access_file::implementation --
141
142		std::coroutine_handle<> read_some_at(
143		std::uint64_t offset,
144		std::coroutine_handle<>,
145		capy::executor_ref,
146		buffer_param,
147		std::stop_token,
148		std::error_code*,
149		std::size_t*) override;
150
151		std::coroutine_handle<> write_some_at(
152		std::uint64_t offset,
153		std::coroutine_handle<>,
154		capy::executor_ref,
155		buffer_param,
156		std::stop_token,
157		std::error_code*,
158		std::size_t*) override;
159
160	1160x	native_handle_type native_handle() const noexcept override
161		{
162	1160x	return fd_;
163		}
164
165	638x	void cancel() noexcept override
166		{
167	638x	std::lock_guard<std::mutex> lock(ops_mutex_);
168	638x	outstanding_ops_.for_each([](raf_op* op) {
169	8x	op->cancelled.store(true, std::memory_order_release);
170	8x	});
171	638x	}
172
173		std::uint64_t size() const override;
174		std::error_code resize(std::uint64_t new_size) noexcept override;
175		std::error_code sync_data() noexcept override;
176		std::error_code sync_all() noexcept override;
177		native_handle_type release() override;
178		std::error_code assign(native_handle_type handle) noexcept override;
179
180		std::error_code
181		open_file(std::filesystem::path const& path, file_base::flags mode);
182		void close_file() noexcept;
183
184		private:
185		/** Pop a recycled op, or allocate on the cold path.
186
187		Per-use fields are filled by the caller; the fields `prepare`-
188		style reuse must not inherit from the previous run (`errn`,
189		`bytes_transferred`) are reset here. `start()` resets the
190		cancellation machinery.
191		*/
192	494x	raf_op* acquire_op()
193		{
194		raf_op* op;
195		{
196	494x	std::lock_guard<std::mutex> lock(ops_mutex_);
197	494x	op = free_ops_.pop_front();
198	494x	}
199	494x	if (!op)
200	402x	op = new raf_op();
201	494x	op->errn = 0;
202	494x	op->bytes_transferred = 0;
203	494x	return op;
204		}
205
206		posix_random_access_file_service& svc_;
207		int fd_ = -1;
208		std::mutex ops_mutex_;
209		intrusive_list<raf_op> outstanding_ops_;
210		intrusive_list<raf_op> free_ops_;
211		};
212
213		// ---------------------------------------------------------------------------
214		// Inline implementation
215		// ---------------------------------------------------------------------------
216
217	219x	inline posix_random_access_file::posix_random_access_file(
218	219x	posix_random_access_file_service& svc) noexcept
219	219x	: svc_(svc)
220		{
221	219x	}
222
223		inline std::error_code
224	203x	posix_random_access_file::open_file(
225		std::filesystem::path const& path, file_base::flags mode)
226		{
227	203x	close_file();
228
229	203x	int oflags = 0;
230
231	203x	unsigned access = static_cast<unsigned>(mode) & 3u;
232	203x	if (access == static_cast<unsigned>(file_base::read_write))
233	33x	oflags \|= O_RDWR;
234	170x	else if (access == static_cast<unsigned>(file_base::write_only))
235	64x	oflags \|= O_WRONLY;
236		else
237	106x	oflags \|= O_RDONLY;
238
239	203x	if ((mode & file_base::create) != file_base::flags(0))
240	28x	oflags \|= O_CREAT;
241	203x	if ((mode & file_base::exclusive) != file_base::flags(0))
242	4x	oflags \|= O_EXCL;
243	203x	if ((mode & file_base::truncate) != file_base::flags(0))
244	14x	oflags \|= O_TRUNC;
245	203x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
246	2x	oflags \|= O_SYNC;
247		// Note: no O_APPEND for random access files
248
249	203x	int fd = ::open(path.c_str(), oflags, 0666);
250	203x	if (fd < 0)
251	9x	return make_err(errno);
252
253	194x	fd_ = fd;
254
255		#ifdef POSIX_FADV_RANDOM
256	194x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
257		#endif
258
259	194x	return {};
260		}
261
262		inline void
263	844x	posix_random_access_file::close_file() noexcept
264		{
265	844x	if (fd_ >= 0)
266		{
267	198x	::close(fd_);
268	198x	fd_ = -1;
269		}
270	844x	}
271
272		inline std::uint64_t
273	13x	posix_random_access_file::size() const
274		{
275		struct stat st;
276	13x	if (::fstat(fd_, &st) < 0)
277	5x	throw_system_error(make_err(errno), "random_access_file::size");
278	8x	return static_cast<std::uint64_t>(st.st_size);
279		}
280
281		inline std::error_code
282	13x	posix_random_access_file::resize(std::uint64_t new_size) noexcept
283		{
284	13x	if (new_size >
285	13x	static_cast<std::uint64_t>((std::numeric_limits<off_t>::max)()))
286	2x	return make_err(EOVERFLOW);
287	11x	if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
288	7x	return make_err(errno);
289	4x	return {};
290		}
291
292		inline std::error_code
293	9x	posix_random_access_file::sync_data() noexcept
294		{
295		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
296	9x	if (::fdatasync(fd_) < 0)
297		#else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
298		if (::fsync(fd_) < 0)
299		#endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
300	7x	return make_err(errno);
301	2x	return {};
302		}
303
304		inline std::error_code
305	9x	posix_random_access_file::sync_all() noexcept
306		{
307	9x	if (::fsync(fd_) < 0)
308	7x	return make_err(errno);
309	2x	return {};
310		}
311
312		inline native_handle_type
313	3x	posix_random_access_file::release()
314		{
315	3x	int fd = fd_;
316	3x	fd_ = -1;
317	3x	return fd;
318		}
319
320		inline std::error_code
321	7x	posix_random_access_file::assign(native_handle_type handle) noexcept
322		{
323	7x	close_file();
324	7x	fd_ = handle;
325	7x	return {};
326		}
327
328		// read_some_at, write_some_at are defined in
329		// posix_random_access_file_service.hpp after the service.
330
331		// -- raf_op completion handler (scheduler thread) --
332
333		inline void
334	488x	posix_random_access_file::raf_op::operator()()
335		{
336	488x	stop_cb.reset();
337
338		// Empty buffers never reach the pool (diverted at initiation), so
339		// empty_buffer stays false and a 0-byte read is a genuine EOF.
340	901x	decode_io_result(
341	488x	ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
342	488x	errn != 0 ? make_err(errn) : std::error_code{}, is_read,
343		bytes_transferred, /empty_buffer=/false);
344
345		// Copy out everything needed after recycling: once this op is on
346		// free_ops_ a concurrent initiation may pop and refill it. The
347		// keepalive drops after the push so a final release never runs
348		// under ops_mutex_.
349	488x	auto keep = std::move(object_ref_);
350	488x	auto coro = h;
351	488x	auto exec = ex;
352		{
353	488x	std::lock_guard<std::mutex> lock(file_->ops_mutex_);
354	488x	file_->outstanding_ops_.remove(this);
355	488x	file_->free_ops_.push_front(this);
356	488x	}
357	488x	keep.reset();
358	488x	exec.on_work_finished();
359	488x	coro.resume();
360	488x	}
361
362		// -- raf_op shutdown cleanup --
363
364		inline void
365	6x	posix_random_access_file::raf_op::destroy()
366		{
367	6x	stop_cb.reset();
368	6x	auto keep = std::move(object_ref_);
369	6x	auto exec = ex;
370		{
371	6x	std::lock_guard<std::mutex> lock(file_->ops_mutex_);
372	6x	file_->outstanding_ops_.remove(this);
373	6x	file_->free_ops_.push_front(this);
374	6x	}
375	6x	keep.reset();
376	6x	exec.on_work_finished();
377	6x	}
378
379		} // namespace boost::corosio::detail
380
381		#endif // BOOST_COROSIO_POSIX
382
383		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
384