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

89.0% Lines (137/154) 95.7% List of functions (22/23) 79.7% Branches (51/64)

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

posix_stream_file.hpp

f(x) Functions (23)

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