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

90.3% Lines (131/145) 100.0% List of functions (16/16)

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

posix_stream_file_service.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_SERVICE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_POSIX
16
17		#include <boost/corosio/native/detail/posix/posix_stream_file.hpp>
18		#include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
19		#include <boost/corosio/detail/file_service.hpp>
20		#include <boost/corosio/detail/thread_pool.hpp>
21
22		#include <mutex>
23		#include <unordered_map>
24
25		namespace boost::corosio::detail {
26
27		/** Stream file service for POSIX backends.
28
29		Owns all posix_stream_file instances. Thread lifecycle is
30		managed by the thread_pool service (shared with resolver).
31		*/
32		class BOOST_COROSIO_DECL posix_stream_file_service final
33		: public file_service
34		{
35		public:
36	1370x	posix_stream_file_service(
37		capy::execution_context& ctx, scheduler& sched)
38	2740x	: sched_(&sched)
39	1370x	, pool_(get_or_create_pool(ctx))
40		{
41	1370x	}
42
43	2740x	~posix_stream_file_service() override = default;
44
45		posix_stream_file_service(posix_stream_file_service const&) = delete;
46		posix_stream_file_service& operator=(posix_stream_file_service const&) = delete;
47
48	83x	io_object::implementation* construct() override
49		{
50	83x	auto ptr = std::make_shared<posix_stream_file>(*this);
51	83x	auto* impl = ptr.get();
52
53		{
54	83x	std::lock_guard<std::mutex> lock(mutex_);
55	83x	file_list_.push_back(impl);
56	83x	file_ptrs_[impl] = std::move(ptr);
57	83x	}
58
59	83x	return impl;
60	83x	}
61
62	83x	void destroy(io_object::implementation* p) override
63		{
64	83x	auto& impl = static_cast<posix_stream_file&>(*p);
65	83x	impl.cancel();
66	83x	impl.close_file();
67	83x	destroy_impl(impl);
68	83x	}
69
70	144x	void close(io_object::handle& h) override
71		{
72	144x	if (h.get())
73		{
74	144x	auto& impl = static_cast<posix_stream_file&>(*h.get());
75	144x	impl.cancel();
76	144x	impl.close_file();
77		}
78	144x	}
79
80	67x	std::error_code open_file(
81		stream_file::implementation& impl,
82		std::filesystem::path const& path,
83		file_base::flags mode) override
84		{
85		// Unavailable in the unsafe tier: the file thread pool completes
86		// cross-thread, which the lockless scheduler cannot accept.
87	67x	if (sched_->scheduler_locking_disabled())
88	2x	return std::make_error_code(std::errc::operation_not_supported);
89	65x	return static_cast<posix_stream_file&>(impl).open_file(path, mode);
90		}
91
92	1370x	void shutdown() override
93		{
94	1370x	std::lock_guard<std::mutex> lock(mutex_);
95	1370x	for (auto* impl = file_list_.pop_front(); impl != nullptr;
96	✗	impl = file_list_.pop_front())
97		{
98	✗	impl->cancel();
99	✗	impl->close_file();
100		}
101	1370x	file_ptrs_.clear();
102	1370x	}
103
104	83x	void destroy_impl(posix_stream_file& impl)
105		{
106	83x	std::lock_guard<std::mutex> lock(mutex_);
107	83x	file_list_.remove(&impl);
108	83x	file_ptrs_.erase(&impl);
109	83x	}
110
111	35x	void post(scheduler_op* op)
112		{
113	35x	sched_->post(op);
114	35x	}
115
116		void work_started() noexcept
117		{
118		sched_->work_started();
119		}
120
121		void work_finished() noexcept
122		{
123		sched_->work_finished();
124		}
125
126	35x	thread_pool& pool() noexcept
127		{
128	35x	return pool_;
129		}
130
131		private:
132	1370x	static thread_pool& get_or_create_pool(capy::execution_context& ctx)
133		{
134	1370x	auto* p = ctx.find_service<thread_pool>();
135	1370x	if (p)
136	1370x	return *p;
137	✗	return ctx.make_service<thread_pool>();
138		}
139
140		scheduler* sched_;
141		thread_pool& pool_;
142		std::mutex mutex_;
143		intrusive_list<posix_stream_file> file_list_;
144		std::unordered_map<posix_stream_file*, std::shared_ptr<posix_stream_file>>
145		file_ptrs_;
146		};
147
148		/** Get or create the stream file service for the given context. */
149		inline posix_stream_file_service&
150	1370x	get_stream_file_service(capy::execution_context& ctx, scheduler& sched)
151		{
152	1370x	return ctx.make_service<posix_stream_file_service>(sched);
153		}
154
155		// ---------------------------------------------------------------------------
156		// posix_stream_file inline implementations (require complete service type)
157		// ---------------------------------------------------------------------------
158
159		inline std::coroutine_handle<>
160	20x	posix_stream_file::read_some(
161		std::coroutine_handle<> h,
162		capy::executor_ref ex,
163		buffer_param param,
164		std::stop_token token,
165		std::error_code* ec,
166		std::size_t* bytes_out)
167		{
168	20x	auto& op = read_op_;
169	20x	op.reset();
170	20x	op.is_read = true;
171
172	20x	capy::mutable_buffer bufs[max_buffers];
173	20x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
174
175	20x	if (op.iovec_count == 0)
176		{
177	2x	*ec = {};
178	2x	*bytes_out = 0;
179	2x	op.cont.h = h;
180	2x	return dispatch_coro(ex, op.cont);
181		}
182
183	36x	for (int i = 0; i < op.iovec_count; ++i)
184		{
185	18x	op.iovecs[i].iov_base = bufs[i].data();
186	18x	op.iovecs[i].iov_len = bufs[i].size();
187		}
188
189	18x	op.h = h;
190	18x	op.ex = ex;
191	18x	op.ec_out = ec;
192	18x	op.bytes_out = bytes_out;
193	18x	op.start(token);
194
195	18x	op.ex.on_work_started();
196
197	18x	read_pool_op_.file_ = this;
198	18x	read_pool_op_.ref_ = this->shared_from_this();
199	18x	read_pool_op_.func_ = &posix_stream_file::do_read_work;
200	18x	if (!svc_.pool().post(&read_pool_op_))
201		{
202	✗	op.impl_ref = std::move(read_pool_op_.ref_);
203	✗	op.cancelled.store(true, std::memory_order_release);
204	✗	svc_.post(&read_op_);
205		}
206	18x	return std::noop_coroutine();
207		}
208
209		inline void
210	18x	posix_stream_file::do_read_work(pool_work_item* w) noexcept
211		{
212	18x	auto* pw = static_cast<pool_op*>(w);
213	18x	auto* self = pw->file_;
214	18x	auto& op = self->read_op_;
215
216	18x	if (!op.cancelled.load(std::memory_order_acquire))
217		{
218		ssize_t n;
219		do
220		{
221	32x	n = ::preadv(self->fd_, op.iovecs, op.iovec_count,
222	16x	static_cast<off_t>(self->offset_));
223		}
224	16x	while (n < 0 && errno == EINTR);
225
226	16x	if (n >= 0)
227		{
228	16x	op.errn = 0;
229	16x	op.bytes_transferred = static_cast<std::size_t>(n);
230	16x	self->offset_ += static_cast<std::uint64_t>(n);
231		}
232		else
233		{
234	✗	op.errn = errno;
235	✗	op.bytes_transferred = 0;
236		}
237		}
238
239	18x	op.impl_ref = std::move(pw->ref_);
240	18x	self->svc_.post(&op);
241	18x	}
242
243		inline std::coroutine_handle<>
244	19x	posix_stream_file::write_some(
245		std::coroutine_handle<> h,
246		capy::executor_ref ex,
247		buffer_param param,
248		std::stop_token token,
249		std::error_code* ec,
250		std::size_t* bytes_out)
251		{
252	19x	auto& op = write_op_;
253	19x	op.reset();
254	19x	op.is_read = false;
255
256	19x	capy::mutable_buffer bufs[max_buffers];
257	19x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
258
259	19x	if (op.iovec_count == 0)
260		{
261	2x	*ec = {};
262	2x	*bytes_out = 0;
263	2x	op.cont.h = h;
264	2x	return dispatch_coro(ex, op.cont);
265		}
266
267	34x	for (int i = 0; i < op.iovec_count; ++i)
268		{
269	17x	op.iovecs[i].iov_base = bufs[i].data();
270	17x	op.iovecs[i].iov_len = bufs[i].size();
271		}
272
273	17x	op.h = h;
274	17x	op.ex = ex;
275	17x	op.ec_out = ec;
276	17x	op.bytes_out = bytes_out;
277	17x	op.start(token);
278
279	17x	op.ex.on_work_started();
280
281	17x	write_pool_op_.file_ = this;
282	17x	write_pool_op_.ref_ = this->shared_from_this();
283	17x	write_pool_op_.func_ = &posix_stream_file::do_write_work;
284	17x	if (!svc_.pool().post(&write_pool_op_))
285		{
286	✗	op.impl_ref = std::move(write_pool_op_.ref_);
287	✗	op.cancelled.store(true, std::memory_order_release);
288	✗	svc_.post(&write_op_);
289		}
290	17x	return std::noop_coroutine();
291		}
292
293		inline void
294	17x	posix_stream_file::do_write_work(pool_work_item* w) noexcept
295		{
296	17x	auto* pw = static_cast<pool_op*>(w);
297	17x	auto* self = pw->file_;
298	17x	auto& op = self->write_op_;
299
300	17x	if (!op.cancelled.load(std::memory_order_acquire))
301		{
302		ssize_t n;
303		do
304		{
305	34x	n = ::pwritev(self->fd_, op.iovecs, op.iovec_count,
306	17x	static_cast<off_t>(self->offset_));
307		}
308	17x	while (n < 0 && errno == EINTR);
309
310	17x	if (n >= 0)
311		{
312	17x	op.errn = 0;
313	17x	op.bytes_transferred = static_cast<std::size_t>(n);
314	17x	self->offset_ += static_cast<std::uint64_t>(n);
315		}
316		else
317		{
318	✗	op.errn = errno;
319	✗	op.bytes_transferred = 0;
320		}
321		}
322
323	17x	op.impl_ref = std::move(pw->ref_);
324	17x	self->svc_.post(&op);
325	17x	}
326
327		} // namespace boost::corosio::detail
328
329		#endif // BOOST_COROSIO_POSIX
330
331		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
332