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

99.2% Lines (126/0/127) 100.0% List of functions (16/0/16)

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

posix_stream_file.hpp

f(x) Functions (16)

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