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

97.7% Lines (126 / 129) 100.0% Functions (17 / 17)

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

posix_stream_file.hpp

f(x) Functions (17)

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