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

94.6% Lines (123 / 130) 94.7% Functions (18 / 19)

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

posix_random_access_file.hpp

f(x) Functions (19)

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/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		struct raf_op final
94		: coro_op
95		, pool_work_item
96		, intrusive_list<raf_op>::node
97		{
98		iovec iovecs[max_buffers];
99		int iovec_count = 0;
100		std::uint64_t offset = 0;
101		// Snapshotted at submission: assign() may replace fd_ while the
102		// worker runs.
103		int fd = -1;
104
105		int errn = 0;
106		std::size_t bytes_transferred = 0;
107
108		// Raw back-pointer for the typed work; `object_ref_` is the keepalive.
109		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	✗	BOOST_COROSIO_ASSERT(fd_ == -1);
154	✗	BOOST_COROSIO_ASSERT(gate_.idle());
155	✗	BOOST_COROSIO_ASSERT(outstanding_ops_.empty());
156	✗	}
157
158	488x	~posix_random_access_file() override
159	244x	{
160	664x	while (auto* op = free_ops_.pop_front())
161	420x	delete op;
162	488x	}
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	1246x	native_handle_type native_handle() const noexcept override
185		{
186	1246x	return fd_;
187		}
188
189	714x	void cancel() noexcept override
190		{
191	714x	std::lock_guard<std::mutex> lock(ops_mutex_);
192	714x	outstanding_ops_.for_each([](raf_op* op) {
193	23x	op->cancelled.store(true, std::memory_order_release);
194	23x	});
195	714x	}
196
197		std::uint64_t size() const 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		native_handle_type release() 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	515x	raf_op* acquire_op()
217		{
218		raf_op* op;
219		{
220	515x	std::lock_guard<std::mutex> lock(ops_mutex_);
221	515x	op = free_ops_.pop_front();
222	515x	}
223	515x	if (!op)
224	420x	op = new raf_op();
225	515x	op->errn = 0;
226	515x	op->bytes_transferred = 0;
227	515x	return op;
228		}
229
230		posix_random_access_file_service& svc_;
231		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	244x	inline posix_random_access_file::posix_random_access_file(
244	244x	posix_random_access_file_service& svc) noexcept
245	244x	: svc_(svc)
246		{
247	244x	}
248
249		inline std::error_code
250	224x	posix_random_access_file::open_file(
251		std::filesystem::path const& path, file_base::flags mode)
252		{
253	224x	close_file();
254
255	224x	int oflags = 0;
256
257	224x	unsigned access = static_cast<unsigned>(mode) & 3u;
258	224x	if (access == static_cast<unsigned>(file_base::read_write))
259	37x	oflags \|= O_RDWR;
260	187x	else if (access == static_cast<unsigned>(file_base::write_only))
261	69x	oflags \|= O_WRONLY;
262		else
263	118x	oflags \|= O_RDONLY;
264
265	224x	if ((mode & file_base::create) != file_base::flags(0))
266	28x	oflags \|= O_CREAT;
267	224x	if ((mode & file_base::exclusive) != file_base::flags(0))
268	4x	oflags \|= O_EXCL;
269	224x	if ((mode & file_base::truncate) != file_base::flags(0))
270	14x	oflags \|= O_TRUNC;
271	224x	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	224x	int fd = ::open(path.c_str(), oflags \| large_file_open_flag, 0666);
276	224x	if (fd < 0)
277	9x	return make_err(errno);
278
279	215x	fd_ = fd;
280	215x	gate_.open(fd);
281
282		#ifdef POSIX_FADV_RANDOM
283	215x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
284		#endif
285
286	215x	return {};
287		}
288
289		inline void
290	929x	posix_random_access_file::close_file() noexcept
291		{
292	929x	gate_.close();
293	929x	fd_ = -1;
294	929x	}
295
296		inline std::uint64_t
297	11x	posix_random_access_file::size() const
298		{
299		file_stat_t st;
300	11x	if (file_fstat(fd_, &st) < 0)
301	5x	throw_system_error(make_err(errno), "random_access_file::size");
302	6x	return static_cast<std::uint64_t>(st.st_size);
303		}
304
305		inline std::error_code
306	13x	posix_random_access_file::resize(std::uint64_t new_size) noexcept
307		{
308	13x	if (new_size >
309	13x	static_cast<std::uint64_t>((std::numeric_limits<file_off_t>::max)()))
310	2x	return make_err(EOVERFLOW);
311	11x	if (file_ftruncate(fd_, static_cast<file_off_t>(new_size)) < 0)
312	7x	return make_err(errno);
313	4x	return {};
314		}
315
316		inline std::error_code
317	7x	posix_random_access_file::sync_data() noexcept
318		{
319		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
320	7x	if (::fdatasync(fd_) < 0)
321		#else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
322		if (::fsync(fd_) < 0)
323		#endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
324	5x	return make_err(errno);
325	2x	return {};
326		}
327
328		inline std::error_code
329	7x	posix_random_access_file::sync_all() noexcept
330		{
331	7x	if (::fsync(fd_) < 0)
332	5x	return make_err(errno);
333	2x	return {};
334		}
335
336		inline native_handle_type
337	5x	posix_random_access_file::release()
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	5x	cancel();
342	5x	if (!gate_.release())
343	✗	throw_system_error(
344	✗	std::make_error_code(std::errc::device_or_resource_busy),
345		"random_access_file::release");
346	5x	int fd = fd_;
347	5x	fd_ = -1;
348	5x	return fd;
349		}
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	505x	posix_random_access_file::raf_op::operator()()
358		{
359	505x	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	926x	decode_results(
364	505x	ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
365	505x	errn != 0 ? make_err(errn) : std::error_code{}, is_read,
366		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	505x	auto keep = std::move(object_ref_);
373	505x	auto* c = cont;
374	505x	auto exec = ex;
375	505x	auto& svc = file_->svc_;
376		{
377	505x	std::lock_guard<std::mutex> lock(file_->ops_mutex_);
378	505x	file_->outstanding_ops_.remove(this);
379	505x	file_->free_ops_.push_front(this);
380	505x	}
381	505x	keep.reset();
382	505x	release_work(svc);
383	505x	dispatch_coro(exec, *c).resume();
384	505x	}
385
386		// -- raf_op shutdown cleanup --
387
388		inline void
389	10x	posix_random_access_file::raf_op::recycle() noexcept
390		{
391	10x	std::lock_guard<std::mutex> lock(file_->ops_mutex_);
392	10x	file_->outstanding_ops_.remove(this);
393	10x	file_->free_ops_.push_front(this);
394	10x	}
395
396		inline void
397	6x	posix_random_access_file::raf_op::destroy()
398		{
399	6x	stop_cb.reset();
400	6x	auto keep = std::move(object_ref_);
401	6x	auto& svc = file_->svc_;
402	6x	auto* c = std::exchange(cont, nullptr);
403	6x	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	6x	if (c && c->h)
407	6x	capy::abandon(*c);
408	6x	keep.reset();
409	6x	release_work(svc);
410	6x	}
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	6x	posix_random_access_file::raf_op::destroy_work(pool_work_item* w) noexcept
425		{
426	6x	static_cast<raf_op*>(w)->destroy();
427	6x	}
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