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

90.6% Lines (126/0/139) 94.7% List of functions (18/0/19)

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

posix_stream_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_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		std::error_code* ec_out = nullptr;
112		std::size_t* bytes_out = nullptr;
113
114		// Buffer data (copied from buffer_param at submission time)
115		iovec iovecs[max_buffers];
116		int iovec_count = 0;
117
118		// Result storage (populated by worker thread)
119		int errn = 0;
120		std::size_t bytes_transferred = 0;
121		bool is_read = false;
122
123		// Thread coordination
124		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	190x	file_op() = default;
131
132	55x	void reset() noexcept
133		{
134	55x	iovec_count = 0;
135	55x	errn = 0;
136	55x	bytes_transferred = 0;
137	55x	is_read = false;
138	55x	cancelled.store(false, std::memory_order_relaxed);
139	55x	stop_cb.reset();
140	55x	impl_ref.reset();
141	55x	ec_out = nullptr;
142	55x	bytes_out = nullptr;
143	55x	}
144
145		void operator()() override;
146		void destroy() override;
147
148	532x	void request_cancel() noexcept
149		{
150	532x	cancelled.store(true, std::memory_order_release);
151	532x	}
152
153	39x	void start(std::stop_token const& token)
154		{
155	39x	cancelled.store(false, std::memory_order_release);
156	39x	stop_cb.reset();
157	39x	if (token.stop_possible())
158	2x	stop_cb.emplace(token, canceller{this});
159	39x	}
160		};
161
162		/** Pool work item for thread pool dispatch. */
163		struct pool_op : pool_work_item
164		{
165		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	302x	native_handle_type native_handle() const noexcept override
192		{
193	302x	return fd_;
194		}
195
196	265x	void cancel() noexcept override
197		{
198	265x	read_op_.request_cancel();
199	265x	write_op_.request_cancel();
200	265x	}
201
202		std::uint64_t size() const override;
203		std::error_code resize(std::uint64_t new_size) noexcept override;
204		std::error_code sync_data() noexcept override;
205		std::error_code sync_all() noexcept override;
206		native_handle_type release() override;
207		std::error_code assign(native_handle_type handle) noexcept override;
208		capy::io_result<std::uint64_t>
209		seek(std::int64_t offset, file_base::seek_basis origin) noexcept override;
210
211		// -- Internal --
212
213		/** Open the file and store the fd. */
214		std::error_code open_file(
215		std::filesystem::path const& path, file_base::flags mode);
216
217		/** Close the file descriptor. */
218		void close_file() noexcept;
219
220		private:
221		posix_stream_file_service& svc_;
222		int fd_ = -1;
223		std::uint64_t offset_ = 0;
224
225		file_op read_op_;
226		file_op write_op_;
227		pool_op read_pool_op_;
228		pool_op write_pool_op_;
229
230		static void do_read_work(pool_work_item*) noexcept;
231		static void do_write_work(pool_work_item*) noexcept;
232		};
233
234		// ---------------------------------------------------------------------------
235		// Inline implementation
236		// ---------------------------------------------------------------------------
237
238		inline
239	95x	posix_stream_file::posix_stream_file(posix_stream_file_service& svc) noexcept
240	95x	: svc_(svc)
241		{
242	95x	}
243
244		inline std::error_code
245	73x	posix_stream_file::open_file(
246		std::filesystem::path const& path, file_base::flags mode)
247		{
248	73x	close_file();
249
250	73x	int oflags = 0;
251
252		// Access mode
253	73x	unsigned access = static_cast<unsigned>(mode) & 3u;
254	73x	if (access == static_cast<unsigned>(file_base::read_write))
255	4x	oflags \|= O_RDWR;
256	69x	else if (access == static_cast<unsigned>(file_base::write_only))
257	24x	oflags \|= O_WRONLY;
258		else
259	45x	oflags \|= O_RDONLY;
260
261		// Creation flags
262	73x	if ((mode & file_base::create) != file_base::flags(0))
263	20x	oflags \|= O_CREAT;
264	73x	if ((mode & file_base::exclusive) != file_base::flags(0))
265	2x	oflags \|= O_EXCL;
266	73x	if ((mode & file_base::truncate) != file_base::flags(0))
267	17x	oflags \|= O_TRUNC;
268	73x	if ((mode & file_base::append) != file_base::flags(0))
269	3x	oflags \|= O_APPEND;
270	73x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
271	2x	oflags \|= O_SYNC;
272
273	73x	int fd = ::open(path.c_str(), oflags, 0666);
274	73x	if (fd < 0)
275	4x	return make_err(errno);
276
277	69x	fd_ = fd;
278	69x	offset_ = 0;
279
280		// Append mode: position at end-of-file (preadv/pwritev use
281		// explicit offsets, so O_APPEND alone is not sufficient).
282	69x	if ((mode & file_base::append) != file_base::flags(0))
283		{
284		struct stat st;
285	3x	if (::fstat(fd, &st) < 0)
286		{
287	✗	int err = errno;
288	✗	::close(fd);
289	✗	fd_ = -1;
290	✗	return make_err(err);
291		}
292	3x	offset_ = static_cast<std::uint64_t>(st.st_size);
293		}
294
295		#ifdef POSIX_FADV_SEQUENTIAL
296	69x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_SEQUENTIAL);
297		#endif
298
299	69x	return {};
300		}
301
302		inline void
303	342x	posix_stream_file::close_file() noexcept
304		{
305	342x	if (fd_ >= 0)
306		{
307	73x	::close(fd_);
308	73x	fd_ = -1;
309		}
310	342x	}
311
312		inline std::uint64_t
313	12x	posix_stream_file::size() const
314		{
315		struct stat st;
316	12x	if (::fstat(fd_, &st) < 0)
317	✗	throw_system_error(make_err(errno), "stream_file::size");
318	12x	return static_cast<std::uint64_t>(st.st_size);
319		}
320
321		inline std::error_code
322	7x	posix_stream_file::resize(std::uint64_t new_size) noexcept
323		{
324	7x	if (new_size >
325	7x	static_cast<std::uint64_t>((std::numeric_limits<off_t>::max)()))
326	2x	return make_err(EOVERFLOW);
327	5x	if (::ftruncate(fd_, static_cast<off_t>(new_size)) < 0)
328	2x	return make_err(errno);
329	3x	return {};
330		}
331
332		inline std::error_code
333	5x	posix_stream_file::sync_data() noexcept
334		{
335		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
336	5x	if (::fdatasync(fd_) < 0)
337		#else // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
338		if (::fsync(fd_) < 0)
339		#endif // BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
340	2x	return make_err(errno);
341	3x	return {};
342		}
343
344		inline std::error_code
345	5x	posix_stream_file::sync_all() noexcept
346		{
347	5x	if (::fsync(fd_) < 0)
348	2x	return make_err(errno);
349	3x	return {};
350		}
351
352		inline native_handle_type
353	2x	posix_stream_file::release()
354		{
355	2x	int fd = fd_;
356	2x	fd_ = -1;
357	2x	offset_ = 0;
358	2x	return fd;
359		}
360
361		inline std::error_code
362	6x	posix_stream_file::assign(native_handle_type handle) noexcept
363		{
364	6x	close_file();
365	6x	fd_ = handle;
366	6x	offset_ = 0;
367	6x	return {};
368		}
369
370		inline capy::io_result<std::uint64_t>
371	20x	posix_stream_file::seek(
372		std::int64_t offset, file_base::seek_basis origin) noexcept
373		{
374		// We track offset_ ourselves (not the kernel fd offset)
375		// because preadv/pwritev use explicit offsets.
376		std::int64_t new_pos;
377
378	20x	if (origin == file_base::seek_set)
379		{
380	9x	new_pos = offset;
381		}
382	11x	else if (origin == file_base::seek_cur)
383		{
384	5x	new_pos = static_cast<std::int64_t>(offset_) + offset;
385		}
386		else
387		{
388		struct stat st;
389	6x	if (::fstat(fd_, &st) < 0)
390	✗	return {make_err(errno), 0};
391	6x	new_pos = st.st_size + offset;
392		}
393
394	20x	if (new_pos < 0)
395	6x	return {make_err(EINVAL), 0};
396	14x	if (new_pos >
397	14x	static_cast<std::int64_t>((std::numeric_limits<off_t>::max)()))
398	✗	return {make_err(EOVERFLOW), 0};
399
400	14x	offset_ = static_cast<std::uint64_t>(new_pos);
401
402	14x	return {std::error_code{}, offset_};
403		}
404
405		// -- file_op completion handler --
406		// (read_some, write_some, do_read_work, do_write_work are
407		// defined in posix_stream_file_service.hpp after the service)
408
409		inline void
410	39x	posix_stream_file::file_op::operator()()
411		{
412	39x	stop_cb.reset();
413
414	39x	bool const was_cancelled = cancelled.load(std::memory_order_acquire);
415
416	39x	if (ec_out)
417		{
418	39x	if (was_cancelled)
419	2x	*ec_out = capy::error::canceled;
420	37x	else if (errn != 0)
421	4x	*ec_out = make_err(errn);
422	33x	else if (is_read && bytes_transferred == 0)
423	3x	*ec_out = capy::error::eof;
424		else
425	30x	*ec_out = {};
426		}
427
428	39x	if (bytes_out)
429	39x	*bytes_out = was_cancelled ? 0 : bytes_transferred;
430
431		// Move impl_ref to a local so members remain valid through
432		// dispatch — impl_ref may be the last shared_ptr keeping
433		// the parent posix_stream_file (which embeds this file_op) alive.
434	39x	auto prevent_destroy = std::move(impl_ref);
435	39x	ex.on_work_finished();
436	39x	cont.h = h;
437	39x	dispatch_coro(ex, cont).resume();
438	39x	}
439
440		inline void
441	✗	posix_stream_file::file_op::destroy()
442		{
443	✗	stop_cb.reset();
444	✗	auto local_ex = ex;
445	✗	impl_ref.reset();
446	✗	local_ex.on_work_finished();
447	✗	}
448
449		} // namespace boost::corosio::detail
450
451		#endif // BOOST_COROSIO_POSIX
452
453		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_HPP
454