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

99.3% Lines (148/149) 100.0% List of functions (21/21) 82.9% Branches (58/70)

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

posix_stream_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_STREAM_FILE_HPP
11			#define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_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/stream_file.hpp>
19			#include <boost/corosio/file_base.hpp>
20			#include <boost/corosio/detail/intrusive.hpp>
21			#include <boost/corosio/detail/dispatch_coro.hpp>
22			#include <boost/corosio/detail/scheduler_op.hpp>
23			#include <boost/corosio/detail/thread_pool.hpp>
24			#include <boost/corosio/detail/scheduler.hpp>
25			#include <boost/corosio/detail/buffer_param.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/validate_fd.hpp>
31			#include <boost/capy/ex/executor_ref.hpp>
32			#include <boost/capy/error.hpp>
33			#include <boost/capy/buffers.hpp>
34
35			#include <atomic>
36			#include <coroutine>
37			#include <cstddef>
38			#include <cstdint>
39			#include <filesystem>
40			#include <limits>
41			#include <memory>
42			#include <mutex>
43			#include <optional>
44			#include <stop_token>
45			#include <system_error>
46
47			#include <errno.h>
48			#include <fcntl.h>
49			#include <sys/stat.h>
50			#include <sys/uio.h>
51			#include <unistd.h>
52
53			/*
54			POSIX Stream File Implementation
55			=================================
56
57			Regular files cannot be monitored by epoll/kqueue/select — the kernel
58			always reports them as ready. Blocking I/O (pread/pwrite) is dispatched
59			to a shared thread pool, with completion posted back to the scheduler.
60
61			This follows the same pattern as posix_resolver: pool_work_item for
62			dispatch, scheduler_op for completion, shared_from_this for lifetime.
63
64			Completion Flow
65			---------------
66			1. read_some() sets up file_read_op, posts to thread pool
67			2. Pool thread runs preadv() (blocking)
68			3. Pool thread stores results, posts scheduler_op to scheduler
69			4. Scheduler invokes op() which resumes the coroutine
70
71			Single-Inflight Constraint
72			--------------------------
73			Only one asynchronous operation may be in flight at a time on a
74			given file object. Concurrent read and write is not supported
75			because both share offset_ without synchronization.
76			*/
77
78			namespace boost::corosio::detail {
79
80			struct scheduler;
81			class posix_stream_file_service;
82
83			/** Stream file implementation for POSIX backends.
84
85			Each instance contains embedded operation objects (read_op_, write_op_)
86			that are reused across calls. This avoids per-operation heap allocation.
87			*/
88			class posix_stream_file final
89			: public stream_file::implementation
90			, public std::enable_shared_from_this<posix_stream_file>
91			, public intrusive_list<posix_stream_file>::node
92			{
93			friend class posix_stream_file_service;
94
95			public:
96			static constexpr std::size_t max_buffers = 16;
97
98			/** Operation state for a single file read or write.
99
100			The coroutine, cancellation and keepalive machinery is inherited
101			from `coro_op`; only the pool-path result state lives here.
102			*/
103			struct file_op : coro_op
104			{
105			// Buffer data (copied from buffer_param at submission time)
106			iovec iovecs[max_buffers];
107		1270x	int iovec_count = 0;
108
109			// Snapshotted at submission: assign() may replace fd_ and
110			// offset_ while the worker runs.
111		1270x	int fd = -1;
112		1270x	std::uint64_t offset = 0;
113		1270x	std::uint64_t generation = 0;
114
115			// Result storage (populated by worker thread)
116		1270x	int errn = 0;
117		1270x	std::size_t bytes_transferred = 0;
118
119		3810x	file_op() = default;
120
121		636x	void reset() noexcept
122			{
123		636x	iovec_count = 0;
124		636x	errn = 0;
125		636x	bytes_transferred = 0;
126		636x	is_read = false;
127		636x	cancelled.store(false, std::memory_order_relaxed);
128		636x	stop_cb.reset();
129		636x	impl_ptr.reset();
130		636x	ec_out = nullptr;
131		636x	bytes_out = nullptr;
132		636x	}
133
134			void operator()() override;
135			void destroy() override;
136			};
137
138			/** Pool work item for thread pool dispatch. */
139		1270x	struct pool_op : pool_work_item
140			{
141		1270x	posix_stream_file* file_ = nullptr;
142			std::shared_ptr<posix_stream_file> ref_;
143			};
144
145			explicit posix_stream_file(posix_stream_file_service& svc) noexcept;
146
147			// -- io_stream::implementation --
148
149			std::coroutine_handle<> read_some(
150			std::coroutine_handle<>,
151			capy::executor_ref,
152			buffer_param,
153			std::stop_token,
154			std::error_code*,
155			std::size_t*) override;
156
157			std::coroutine_handle<> write_some(
158			std::coroutine_handle<>,
159			capy::executor_ref,
160			buffer_param,
161			std::stop_token,
162			std::error_code*,
163			std::size_t*) override;
164
165			// -- stream_file::implementation --
166
167		2714x	native_handle_type native_handle() const noexcept override
168			{
169		2714x	return fd_;
170			}
171
172		2275x	void cancel() noexcept override
173			{
174		2275x	read_op_.request_cancel();
175		2275x	write_op_.request_cancel();
176		2275x	}
177
178			std::uint64_t size() const override;
179			std::error_code resize(std::uint64_t new_size) noexcept override;
180			std::error_code sync_data() noexcept override;
181			std::error_code sync_all() noexcept override;
182			native_handle_type release() override;
183			std::error_code assign(native_handle_type handle) noexcept override;
184			capy::io_result<std::uint64_t>
185			seek(std::int64_t offset, file_base::seek_basis origin) noexcept override;
186
187			// -- Internal --
188
189			/** Open the file and store the fd. */
190			std::error_code
191			open_file(std::filesystem::path const& path, file_base::flags mode);
192
193			/** Close the file descriptor. */
194			void close_file() noexcept;
195
196			private:
197			posix_stream_file_service& svc_;
198		635x	int fd_ = -1;
199		635x	std::uint64_t offset_ = 0;
200
201			// Bumped whenever the held file changes. A worker advances offset_
202			// only if this still matches its op's snapshot; the mutex makes the
203			// check and the advance one step against a concurrent assign().
204			std::mutex offset_mutex_;
205		635x	std::uint64_t generation_ = 0;
206
207			file_op read_op_;
208			file_op write_op_;
209			pool_op read_pool_op_;
210			pool_op write_pool_op_;
211
212			void bump_generation() noexcept;
213
214			static void do_read_work(pool_work_item*) noexcept;
215			static void do_write_work(pool_work_item*) noexcept;
216			};
217
218			// ---------------------------------------------------------------------------
219			// Inline implementation
220			// ---------------------------------------------------------------------------
221
222		3810x	inline posix_stream_file::posix_stream_file(
223			posix_stream_file_service& svc) noexcept
224		635x	: svc_(svc)
225		2540x	{
226		1270x	}
227
228			inline std::error_code
229		611x	posix_stream_file::open_file(
230			std::filesystem::path const& path, file_base::flags mode)
231			{
232		611x	close_file();
233
234		611x	int oflags = 0;
235
236			// Access mode
237		611x	unsigned access = static_cast<unsigned>(mode) & 3u;
238	2/2 ✓ Branch 0 taken 14 times. ✓ Branch 1 taken 597 times.	611x	if (access == static_cast<unsigned>(file_base::read_write))
239		14x	oflags \|= O_RDWR;
240	2/2 ✓ Branch 0 taken 78 times. ✓ Branch 1 taken 519 times.	597x	else if (access == static_cast<unsigned>(file_base::write_only))
241		78x	oflags \|= O_WRONLY;
242			else
243		519x	oflags \|= O_RDONLY;
244
245			// Creation flags
246	2/2 ✓ Branch 0 taken 583 times. ✓ Branch 1 taken 28 times.	611x	if ((mode & file_base::create) != file_base::flags(0))
247		28x	oflags \|= O_CREAT;
248	2/2 ✓ Branch 0 taken 609 times. ✓ Branch 1 taken 2 times.	611x	if ((mode & file_base::exclusive) != file_base::flags(0))
249		2x	oflags \|= O_EXCL;
250	2/2 ✓ Branch 0 taken 594 times. ✓ Branch 1 taken 17 times.	611x	if ((mode & file_base::truncate) != file_base::flags(0))
251		17x	oflags \|= O_TRUNC;
252	2/2 ✓ Branch 0 taken 606 times. ✓ Branch 1 taken 5 times.	611x	if ((mode & file_base::append) != file_base::flags(0))
253		5x	oflags \|= O_APPEND;
254	2/2 ✓ Branch 0 taken 609 times. ✓ Branch 1 taken 2 times.	611x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
255		2x	oflags \|= O_SYNC;
256
257		611x	int fd = ::open(path.c_str(), oflags \| large_file_open_flag, 0666);
258	2/2 ✓ Branch 0 taken 605 times. ✓ Branch 1 taken 6 times.	611x	if (fd < 0)
259		6x	return make_err(errno);
260
261		605x	fd_ = fd;
262		605x	offset_ = 0;
263
264			// Append mode: position at end-of-file (preadv/pwritev use
265			// explicit offsets, so O_APPEND alone is not sufficient).
266	2/2 ✓ Branch 0 taken 600 times. ✓ Branch 1 taken 5 times.	605x	if ((mode & file_base::append) != file_base::flags(0))
267			{
268			file_stat_t st;
269	2/2 ✓ Branch 0 taken 3 times. ✓ Branch 1 taken 2 times.	5x	if (file_fstat(fd, &st) < 0)
270			{
271		2x	int err = errno;
272		2x	::close(fd);
273		2x	fd_ = -1;
274		2x	return make_err(err);
275			}
276		3x	offset_ = static_cast<std::uint64_t>(st.st_size);
277		3x	}
278
279			#ifdef POSIX_FADV_SEQUENTIAL
280			::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
281			#endif
282
283		603x	return {};
284		611x	}
285
286			inline void
287		2884x	posix_stream_file::bump_generation() noexcept
288			{
289	1/2 ✓ Branch 0 taken 2884 times. ✗ Branch 1 not taken.	2884x	std::lock_guard<std::mutex> lock(offset_mutex_);
290		2884x	++generation_;
291		2884x	}
292
293			inline void
294		2880x	posix_stream_file::close_file() noexcept
295			{
296		2880x	bump_generation();
297	2/2 ✓ Branch 0 taken 1877 times. ✓ Branch 1 taken 1003 times.	2880x	if (fd_ >= 0)
298			{
299	1/2 ✓ Branch 0 taken 1003 times. ✗ Branch 1 not taken.	1003x	::close(fd_);
300		1003x	fd_ = -1;
301		1003x	}
302		2880x	}
303
304			inline std::uint64_t
305		12x	posix_stream_file::size() const
306			{
307			file_stat_t st;
308	2/2 ✓ Branch 0 taken 10 times. ✓ Branch 1 taken 2 times.	12x	if (file_fstat(fd_, &st) < 0)
309		2x	throw_system_error(make_err(errno), "stream_file::size");
310		10x	return static_cast<std::uint64_t>(st.st_size);
311			}
312
313			inline std::error_code
314		9x	posix_stream_file::resize(std::uint64_t new_size) noexcept
315			{
316	4/4 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 7 times. ✓ Branch 2 taken 2 times. ✓ Branch 3 taken 7 times.	18x	if (new_size >
317		9x	static_cast<std::uint64_t>((std::numeric_limits<file_off_t>::max)()))
318		2x	return make_err(EOVERFLOW);
319	2/2 ✓ Branch 0 taken 3 times. ✓ Branch 1 taken 4 times.	7x	if (file_ftruncate(fd_, static_cast<file_off_t>(new_size)) < 0)
320	1/2 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken.	4x	return make_err(errno);
321		3x	return {};
322		9x	}
323
324			inline std::error_code
325		5x	posix_stream_file::sync_data() noexcept
326			{
327			#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
328			if (::fdatasync(fd_) < 0)
329			#else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
330	3/4 ✓ Branch 0 taken 5 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 2 times. ✓ Branch 3 taken 3 times.	5x	if (::fsync(fd_) < 0)
331			#endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
332	1/2 ✓ Branch 0 taken 2 times. ✗ Branch 1 not taken.	2x	return make_err(errno);
333		3x	return {};
334		5x	}
335
336			inline std::error_code
337		5x	posix_stream_file::sync_all() noexcept
338			{
339	3/4 ✓ Branch 0 taken 5 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 2 times. ✓ Branch 3 taken 3 times.	5x	if (::fsync(fd_) < 0)
340	1/2 ✓ Branch 0 taken 2 times. ✗ Branch 1 not taken.	2x	return make_err(errno);
341		3x	return {};
342		5x	}
343
344			inline native_handle_type
345		4x	posix_stream_file::release()
346			{
347			// A queued op has already copied the fd number; it must not run
348			// after the caller closes it and the number is recycled.
349		4x	cancel();
350		4x	bump_generation();
351		4x	int fd = fd_;
352		4x	fd_ = -1;
353		4x	offset_ = 0;
354		4x	return fd;
355			}
356
357			inline std::error_code
358		408x	posix_stream_file::assign(native_handle_type handle) noexcept
359			{
360			// The public assign() guarantees the object is closed.
361	2/2 ✓ Branch 0 taken 4 times. ✓ Branch 1 taken 404 times.	408x	if (auto ec = validate_file_fd(handle))
362		4x	return ec;
363
364		404x	fd_ = handle;
365		404x	offset_ = 0;
366		404x	return {};
367		408x	}
368
369			inline capy::io_result<std::uint64_t>
370		426x	posix_stream_file::seek(
371			std::int64_t offset, file_base::seek_basis origin) noexcept
372			{
373			// We track offset_ ourselves (not the kernel fd offset)
374			// because preadv/pwritev use explicit offsets.
375			std::int64_t new_pos;
376
377	2/2 ✓ Branch 0 taken 415 times. ✓ Branch 1 taken 11 times.	426x	if (origin == file_base::seek_set)
378			{
379		11x	new_pos = offset;
380		11x	}
381	2/2 ✓ Branch 0 taken 407 times. ✓ Branch 1 taken 8 times.	415x	else if (origin == file_base::seek_cur)
382			{
383		407x	new_pos = static_cast<std::int64_t>(offset_) + offset;
384		407x	}
385			else
386			{
387			file_stat_t st;
388	2/2 ✓ Branch 0 taken 6 times. ✓ Branch 1 taken 2 times.	8x	if (file_fstat(fd_, &st) < 0)
389	1/2 ✓ Branch 0 taken 2 times. ✗ Branch 1 not taken.	2x	return {make_err(errno), 0};
390		6x	new_pos = st.st_size + offset;
391			}
392
393	2/2 ✓ Branch 0 taken 418 times. ✓ Branch 1 taken 6 times.	424x	if (new_pos < 0)
394		6x	return {make_err(EINVAL), 0};
395	2/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 418 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 418 times.	836x	if (new_pos >
396		418x	static_cast<std::int64_t>((std::numeric_limits<file_off_t>::max)()))
397		✗	return {make_err(EOVERFLOW), 0};
398
399		418x	offset_ = static_cast<std::uint64_t>(new_pos);
400
401		418x	return {std::error_code{}, offset_};
402		426x	}
403
404			// -- file_op completion handler --
405			// (read_some, write_some, do_read_work, do_write_work are
406			// defined in posix_stream_file_service.hpp after the service)
407
408			inline void
409		609x	posix_stream_file::file_op::operator()()
410			{
411		609x	stop_cb.reset();
412
413			// Empty buffers never reach the pool (diverted at initiation), so
414			// empty_buffer stays false and a 0-byte read is a genuine EOF.
415		609x	decode_io_result(
416		609x	ec_out, bytes_out, cancelled.load(std::memory_order_acquire),
417	2/2 ✓ Branch 0 taken 16 times. ✓ Branch 1 taken 593 times.	609x	errn != 0 ? make_err(errn) : std::error_code{}, is_read,
418		609x	bytes_transferred, /empty_buffer=/false);
419
420			// Move impl_ptr to a local so members remain valid through
421			// dispatch — impl_ptr may be the last shared_ptr keeping
422			// the parent posix_stream_file (which embeds this file_op) alive.
423		609x	auto prevent_destroy = std::move(impl_ptr);
424		609x	ex.on_work_finished();
425		609x	cont.h = h;
426	2/4 ✓ Branch 0 taken 609 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 609 times. ✗ Branch 3 not taken.	609x	dispatch_coro(ex, cont).resume();
427		609x	}
428
429			inline void
430		2x	posix_stream_file::file_op::destroy()
431			{
432		2x	stop_cb.reset();
433		2x	auto local_ex = ex;
434		2x	impl_ptr.reset();
435		2x	local_ex.on_work_finished();
436		2x	}
437
438			} // namespace boost::corosio::detail
439
440			#endif // BOOST_COROSIO_POSIX
441
442			#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_HPP
443