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

96.5% Lines (139 / 144) 64.5% Branches (49 / 76) 95.2% Functions (20 / 21)

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

posix_random_access_file.hpp

f(x) Functions (21)

Function Calls Lines Branches Blocks

Line	Branch	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/detail/dispatch_coro.hpp>
26			#include <boost/corosio/native/detail/coro_op.hpp>
27			#include <boost/corosio/native/detail/coro_op_complete.hpp>
28			#include <boost/corosio/native/detail/make_err.hpp>
29			#include <boost/corosio/native/detail/posix/large_file.hpp>
30			#include <boost/corosio/native/detail/posix/posix_fd_gate.hpp>
31			#include <boost/corosio/native/detail/validate_fd.hpp>
32			#include <boost/capy/ex/abandon.hpp>
33			#include <boost/capy/ex/executor_ref.hpp>
34			#include <boost/capy/error.hpp>
35			#include <boost/capy/buffers.hpp>
36
37			#include <atomic>
38			#include <coroutine>
39			#include <cstddef>
40			#include <cstdint>
41			#include <filesystem>
42			#include <limits>
43			#include <mutex>
44			#include <optional>
45			#include <stop_token>
46			#include <system_error>
47			#include <utility>
48
49			#include <errno.h>
50			#include <fcntl.h>
51			#include <sys/stat.h>
52			#include <sys/uio.h>
53			#include <unistd.h>
54
55			/*
56			POSIX Random-Access File Implementation
57			========================================
58
59			Each async read/write acquires an raf_op that serves as both the
60			thread-pool work item and the scheduler completion op. This
61			allows unlimited concurrent operations on the same file object.
62
63			Ops recycle through the file's free_ops_ list on completion or
64			shutdown, so steady-state reads and writes allocate nothing; the
65			file destructor frees the recycled storage.
66			*/
67
68			namespace boost::corosio::detail {
69
70			class scheduler;
71			class posix_random_access_file_service;
72
73			/** Random-access file implementation for POSIX backends. */
74			class posix_random_access_file final
75			: public random_access_file::implementation
76			, public intrusive_list<posix_random_access_file>::node
77			{
78			friend class posix_random_access_file_service;
79
80			public:
81			static constexpr std::size_t max_buffers = 16;
82
83			/** Per-operation state, acquired from the file's free list.
84
85			Inherits from `coro_op` (for scheduler completion plus the shared
86			coroutine, cancellation and keepalive machinery) and
87			`pool_work_item` (for thread-pool dispatch). The intrusive hook
88			links it into `outstanding_ops_` while in flight and `free_ops_`
89			once recycled — membership is strictly sequential, so one hook
90			serves both under `ops_mutex_`. `coro_op` leads the base list so
91			a `scheduler_op*` round-trips.
92			*/
93		418x	struct raf_op final
94			: coro_op
95			, pool_work_item
96			, intrusive_list<raf_op>::node
97			{
98			iovec iovecs[max_buffers];
99		418x	int iovec_count = 0;
100		418x	std::uint64_t offset = 0;
101			// Snapshotted at submission: assign() may replace fd_ while the
102			// worker runs.
103		418x	int fd = -1;
104
105		418x	int errn = 0;
106		418x	std::size_t bytes_transferred = 0;
107
108			// Raw back-pointer for the typed work; `object_ref_` is the keepalive.
109		418x	posix_random_access_file* file_ = nullptr;
110
111			void operator()() override;
112			void destroy() override;
113
114			/// Recycle an op the pool refused, leaving its coroutine to
115			/// the initiator that answers it.
116			void discard() noexcept;
117
118			/// Thread-pool work function: executes preadv/pwritev.
119			static void do_work(pool_work_item*) noexcept;
120
121			/// Destroy an op the pool shut down before running.
122			static void destroy_work(pool_work_item*) noexcept;
123
124			/// Return this op to its file's free list.
125			void recycle() noexcept;
126			};
127
128			/// Release the work an op holds on its service's context.
129			static void
130			release_work(posix_random_access_file_service& svc) noexcept;
131
132			explicit posix_random_access_file(
133			posix_random_access_file_service& svc) noexcept;
134
135			/// Recycle into the owning service's pool. Defined out-of-line
136			/// after posix_random_access_file_service for its complete type.
137			void retire() noexcept override;
138
139			/** Reset for recycling.
140
141			`close_file()` already drove fd_ to its closed value before
142			the refcount reached zero. Every `raf_op` holds its own
143			`object_ref` while in flight, so the refcount cannot reach
144			zero while one is outstanding — asserting `outstanding_ops_`
145			is empty is therefore a precondition check, not a defensive
146			one. `free_ops_` deliberately survives recycling: the storage
147			belongs to this impl and only the destructor frees it.
148
149			@pre refs_ == 0, fd closed, no op in flight.
150			*/
151		✗	void reuse() noexcept
152			{
153	0/4 ✗ Branch 0 not taken. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✗ Branch 3 not taken.	✗	BOOST_COROSIO_ASSERT(fd_ == -1);
154	0/4 ✗ Branch 0 not taken. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✗ Branch 3 not taken.	✗	BOOST_COROSIO_ASSERT(gate_.idle());
155	0/4 ✗ Branch 0 not taken. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✗ Branch 3 not taken.	✗	BOOST_COROSIO_ASSERT(outstanding_ops_.empty());
156		✗	}
157
158		484x	~posix_random_access_file() override
159		242x	{
160	2/2 ✓ Branch 0 taken 418 times. ✓ Branch 1 taken 242 times.	660x	while (auto* op = free_ops_.pop_front())
161	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 418 times.	418x	delete op;
162		484x	}
163
164			// -- random_access_file::implementation --
165
166			std::coroutine_handle<> read_some_at(
167			std::uint64_t offset,
168			capy::continuation&,
169			capy::executor_ref,
170			buffer_param,
171			std::stop_token,
172			std::error_code*,
173			std::size_t*) override;
174
175			std::coroutine_handle<> write_some_at(
176			std::uint64_t offset,
177			capy::continuation&,
178			capy::executor_ref,
179			buffer_param,
180			std::stop_token,
181			std::error_code*,
182			std::size_t*) override;
183
184		1227x	native_handle_type native_handle() const noexcept override
185			{
186		1227x	return fd_;
187			}
188
189		712x	void cancel() noexcept override
190			{
191	1/2 ✓ Branch 0 taken 712 times. ✗ Branch 1 not taken.	712x	std::lock_guard<std::mutex> lock(ops_mutex_);
192	1/2 ✓ Branch 0 taken 712 times. ✗ Branch 1 not taken.	742x	outstanding_ops_.for_each([](raf_op* op) {
193		30x	op->cancelled.store(true, std::memory_order_release);
194		30x	});
195		712x	}
196
197			capy::results<std::uint64_t> size() const noexcept override;
198			std::error_code resize(std::uint64_t new_size) noexcept override;
199			std::error_code sync_data() noexcept override;
200			std::error_code sync_all() noexcept override;
201			capy::results<native_handle_type> release() noexcept override;
202			std::error_code assign(native_handle_type handle) noexcept override;
203
204			std::error_code
205			open_file(std::filesystem::path const& path, file_base::flags mode);
206			void close_file() noexcept;
207
208			private:
209			/** Pop a recycled op, or allocate on the cold path.
210
211			Per-use fields are filled by the caller; the fields `prepare`-
212			style reuse must not inherit from the previous run (`errn`,
213			`bytes_transferred`) are reset here. `start()` resets the
214			cancellation machinery.
215			*/
216		504x	raf_op* acquire_op()
217			{
218			raf_op* op;
219			{
220		504x	std::lock_guard<std::mutex> lock(ops_mutex_);
221		504x	op = free_ops_.pop_front();
222		504x	}
223	2/2 ✓ Branch 0 taken 86 times. ✓ Branch 1 taken 418 times.	504x	if (!op)
224		418x	op = new raf_op();
225		504x	op->errn = 0;
226		504x	op->bytes_transferred = 0;
227		504x	return op;
228			}
229
230			posix_random_access_file_service& svc_;
231		242x	int fd_ = -1;
232			/// Keeps fd_'s number allocated while a pool worker uses it.
233			posix_fd_gate gate_;
234			std::mutex ops_mutex_;
235			intrusive_list<raf_op> outstanding_ops_;
236			intrusive_list<raf_op> free_ops_;
237			};
238
239			// ---------------------------------------------------------------------------
240			// Inline implementation
241			// ---------------------------------------------------------------------------
242
243		968x	inline posix_random_access_file::posix_random_access_file(
244			posix_random_access_file_service& svc) noexcept
245		242x	: svc_(svc)
246		726x	{
247		484x	}
248
249			inline std::error_code
250		219x	posix_random_access_file::open_file(
251			std::filesystem::path const& path, file_base::flags mode)
252			{
253		219x	close_file();
254
255		219x	int oflags = 0;
256
257		219x	unsigned access = static_cast<unsigned>(mode) & 3u;
258	2/2 ✓ Branch 0 taken 31 times. ✓ Branch 1 taken 188 times.	219x	if (access == static_cast<unsigned>(file_base::read_write))
259		31x	oflags \|= O_RDWR;
260	2/2 ✓ Branch 0 taken 66 times. ✓ Branch 1 taken 122 times.	188x	else if (access == static_cast<unsigned>(file_base::write_only))
261		66x	oflags \|= O_WRONLY;
262			else
263		122x	oflags \|= O_RDONLY;
264
265	2/2 ✓ Branch 0 taken 197 times. ✓ Branch 1 taken 22 times.	219x	if ((mode & file_base::create) != file_base::flags(0))
266		22x	oflags \|= O_CREAT;
267	2/2 ✓ Branch 0 taken 215 times. ✓ Branch 1 taken 4 times.	219x	if ((mode & file_base::exclusive) != file_base::flags(0))
268		4x	oflags \|= O_EXCL;
269	2/2 ✓ Branch 0 taken 205 times. ✓ Branch 1 taken 14 times.	219x	if ((mode & file_base::truncate) != file_base::flags(0))
270		14x	oflags \|= O_TRUNC;
271	2/2 ✓ Branch 0 taken 217 times. ✓ Branch 1 taken 2 times.	219x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
272		2x	oflags \|= O_SYNC;
273			// Note: no O_APPEND for random access files
274
275		219x	int fd = ::open(path.c_str(), oflags \| large_file_open_flag, 0666);
276	2/2 ✓ Branch 0 taken 213 times. ✓ Branch 1 taken 6 times.	219x	if (fd < 0)
277		6x	return make_err(errno);
278
279		213x	fd_ = fd;
280		213x	gate_.open(fd);
281
282			#ifdef POSIX_FADV_RANDOM
283			::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
284			#endif
285
286		213x	return {};
287		219x	}
288
289			inline void
290		918x	posix_random_access_file::close_file() noexcept
291			{
292		918x	gate_.close();
293		918x	fd_ = -1;
294		918x	}
295
296			inline capy::results<std::uint64_t>
297		8x	posix_random_access_file::size() const noexcept
298			{
299			file_stat_t st;
300	2/2 ✓ Branch 0 taken 6 times. ✓ Branch 1 taken 2 times.	8x	if (file_fstat(fd_, &st) < 0)
301	1/2 ✓ Branch 0 taken 2 times. ✗ Branch 1 not taken.	2x	return {make_err(errno), 0};
302		6x	return {std::error_code{}, static_cast<std::uint64_t>(st.st_size)};
303		8x	}
304
305			inline std::error_code
306		10x	posix_random_access_file::resize(std::uint64_t new_size) noexcept
307			{
308	4/4 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 8 times. ✓ Branch 2 taken 2 times. ✓ Branch 3 taken 8 times.	20x	if (new_size >
309		10x	static_cast<std::uint64_t>((std::numeric_limits<file_off_t>::max)()))
310		2x	return make_err(EOVERFLOW);
311	2/2 ✓ Branch 0 taken 4 times. ✓ Branch 1 taken 4 times.	8x	if (file_ftruncate(fd_, static_cast<file_off_t>(new_size)) < 0)
312	1/2 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken.	4x	return make_err(errno);
313		4x	return {};
314		10x	}
315
316			inline std::error_code
317		4x	posix_random_access_file::sync_data() noexcept
318			{
319			#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
320			if (::fdatasync(fd_) < 0)
321			#else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
322	3/4 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 2 times. ✓ Branch 3 taken 2 times.	4x	if (::fsync(fd_) < 0)
323			#endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
324	1/2 ✓ Branch 0 taken 2 times. ✗ Branch 1 not taken.	2x	return make_err(errno);
325		2x	return {};
326		4x	}
327
328			inline std::error_code
329		4x	posix_random_access_file::sync_all() noexcept
330			{
331	3/4 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 2 times. ✓ Branch 3 taken 2 times.	4x	if (::fsync(fd_) < 0)
332	1/2 ✓ Branch 0 taken 2 times. ✗ Branch 1 not taken.	2x	return make_err(errno);
333		2x	return {};
334		4x	}
335
336			inline capy::results<native_handle_type>
337		9x	posix_random_access_file::release() noexcept
338			{
339			// A queued op has already copied the fd number; it must not run
340			// after the caller closes it and the number is recycled.
341		9x	cancel();
342	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 7 times.	9x	if (!gate_.release())
343		2x	return {
344		2x	std::make_error_code(std::errc::device_or_resource_busy),
345			invalid_native_handle};
346		7x	int fd = fd_;
347		7x	fd_ = -1;
348		7x	return {std::error_code{}, fd};
349		9x	}
350
351			// read_some_at, write_some_at are defined in
352			// posix_random_access_file_service.hpp after the service.
353
354			// -- raf_op completion handler (scheduler thread) --
355
356			inline void
357		492x	posix_random_access_file::raf_op::operator()()
358			{
359		492x	stop_cb.reset();
360
361			// Empty buffers never reach the pool (diverted at initiation), so
362			// empty_buffer stays false and a 0-byte read is a genuine EOF.
363		492x	decode_results(
364		492x	ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
365	2/2 ✓ Branch 0 taken 111 times. ✓ Branch 1 taken 381 times.	492x	errn != 0 ? make_err(errn) : std::error_code{}, is_read,
366		492x	bytes_transferred, /empty_buffer=/false);
367
368			// Copy out everything needed after recycling: once this op is on
369			// free_ops_ a concurrent initiation may pop and refill it. The
370			// keepalive drops after the push so a final release never runs
371			// under ops_mutex_.
372		492x	auto keep = std::move(object_ref_);
373		492x	auto* c = cont;
374		492x	auto exec = ex;
375		492x	auto& svc = file_->svc_;
376			{
377	1/2 ✓ Branch 0 taken 492 times. ✗ Branch 1 not taken.	492x	std::lock_guard<std::mutex> lock(file_->ops_mutex_);
378		492x	file_->outstanding_ops_.remove(this);
379		492x	file_->free_ops_.push_front(this);
380		492x	}
381		492x	keep.reset();
382		492x	release_work(svc);
383	1/2 ✓ Branch 0 taken 492 times. ✗ Branch 1 not taken.	492x	dispatch_coro(exec, *c).resume();
384		492x	}
385
386			// -- raf_op shutdown cleanup --
387
388			inline void
389		12x	posix_random_access_file::raf_op::recycle() noexcept
390			{
391	1/2 ✓ Branch 0 taken 12 times. ✗ Branch 1 not taken.	12x	std::lock_guard<std::mutex> lock(file_->ops_mutex_);
392		12x	file_->outstanding_ops_.remove(this);
393		12x	file_->free_ops_.push_front(this);
394		12x	}
395
396			inline void
397		8x	posix_random_access_file::raf_op::destroy()
398			{
399		8x	stop_cb.reset();
400		8x	auto keep = std::move(object_ref_);
401		8x	auto& svc = file_->svc_;
402		8x	auto* c = std::exchange(cont, nullptr);
403		8x	recycle();
404			// The frame may own the file, so it goes while the keepalive
405			// still holds the implementation the op was recycled into.
406	2/4 ✓ Branch 0 taken 8 times. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 8 times.	8x	if (c && c->h)
407		8x	capy::abandon(*c);
408		8x	keep.reset();
409		8x	release_work(svc);
410		8x	}
411
412			inline void
413		4x	posix_random_access_file::raf_op::discard() noexcept
414			{
415		4x	stop_cb.reset();
416		4x	auto keep = std::move(object_ref_);
417		4x	auto& svc = file_->svc_;
418		4x	recycle();
419		4x	keep.reset();
420		4x	release_work(svc);
421		4x	}
422
423			inline void
424		8x	posix_random_access_file::raf_op::destroy_work(pool_work_item* w) noexcept
425			{
426	1/2 ✓ Branch 0 taken 8 times. ✗ Branch 1 not taken.	8x	static_cast<raf_op*>(w)->destroy();
427		8x	}
428
429			} // namespace boost::corosio::detail
430
431			#endif // BOOST_COROSIO_POSIX
432
433			#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_HPP
434