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

93.5% Lines (145/0/155) 100.0% List of functions (16/0/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	1603x	posix_stream_file_service(
37		capy::execution_context& ctx, scheduler& sched)
38	3206x	: sched_(&sched)
39	1603x	, pool_(get_or_create_pool(ctx))
40		{
41	1603x	}
42
43	3206x	~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	95x	io_object::implementation* construct() override
49		{
50	95x	auto ptr = std::make_shared<posix_stream_file>(*this);
51	95x	auto* impl = ptr.get();
52
53		{
54	95x	std::lock_guard<std::mutex> lock(mutex_);
55	95x	file_list_.push_back(impl);
56	95x	file_ptrs_[impl] = std::move(ptr);
57	95x	}
58
59	95x	return impl;
60	95x	}
61
62	95x	void destroy(io_object::implementation* p) override
63		{
64	95x	auto& impl = static_cast<posix_stream_file&>(*p);
65	95x	impl.cancel();
66	95x	impl.close_file();
67	95x	destroy_impl(impl);
68	95x	}
69
70	168x	void close(io_object::handle& h) override
71		{
72	168x	if (h.get())
73		{
74	168x	auto& impl = static_cast<posix_stream_file&>(*h.get());
75	168x	impl.cancel();
76	168x	impl.close_file();
77		}
78	168x	}
79
80	75x	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	75x	if (sched_->scheduler_locking_disabled())
88	2x	return std::make_error_code(std::errc::operation_not_supported);
89	73x	return static_cast<posix_stream_file&>(impl).open_file(path, mode);
90		}
91
92	1603x	void shutdown() override
93		{
94	1603x	std::lock_guard<std::mutex> lock(mutex_);
95	1603x	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	1603x	file_ptrs_.clear();
102	1603x	}
103
104	95x	void destroy_impl(posix_stream_file& impl)
105		{
106	95x	std::lock_guard<std::mutex> lock(mutex_);
107	95x	file_list_.remove(&impl);
108	95x	file_ptrs_.erase(&impl);
109	95x	}
110
111	39x	void post(scheduler_op* op)
112		{
113	39x	sched_->post(op);
114	39x	}
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	39x	thread_pool& pool() noexcept
127		{
128	39x	return pool_;
129		}
130
131		private:
132	1603x	static thread_pool& get_or_create_pool(capy::execution_context& ctx)
133		{
134	1603x	auto* p = ctx.find_service<thread_pool>();
135	1603x	if (p)
136	1603x	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	1603x	get_stream_file_service(capy::execution_context& ctx, scheduler& sched)
151		{
152	1603x	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	28x	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	28x	auto& op = read_op_;
169	28x	op.reset();
170	28x	op.is_read = true;
171
172		// Closed-object contract outranks the zero-length no-op.
173	28x	if (fd_ < 0)
174		{
175	6x	*ec = make_error_code(std::errc::bad_file_descriptor);
176	6x	*bytes_out = 0;
177	6x	op.cont.h = h;
178	6x	return dispatch_coro(ex, op.cont);
179		}
180
181	22x	capy::mutable_buffer bufs[max_buffers];
182	22x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
183
184	22x	if (op.iovec_count == 0)
185		{
186	2x	*ec = {};
187	2x	*bytes_out = 0;
188	2x	op.cont.h = h;
189	2x	return dispatch_coro(ex, op.cont);
190		}
191
192	40x	for (int i = 0; i < op.iovec_count; ++i)
193		{
194	20x	op.iovecs[i].iov_base = bufs[i].data();
195	20x	op.iovecs[i].iov_len = bufs[i].size();
196		}
197
198	20x	op.h = h;
199	20x	op.ex = ex;
200	20x	op.ec_out = ec;
201	20x	op.bytes_out = bytes_out;
202	20x	op.start(token);
203
204	20x	op.ex.on_work_started();
205
206	20x	read_pool_op_.file_ = this;
207	20x	read_pool_op_.ref_ = this->shared_from_this();
208	20x	read_pool_op_.func_ = &posix_stream_file::do_read_work;
209	20x	if (!svc_.pool().post(&read_pool_op_))
210		{
211	✗	op.impl_ref = std::move(read_pool_op_.ref_);
212	✗	op.cancelled.store(true, std::memory_order_release);
213	✗	svc_.post(&read_op_);
214		}
215	20x	return std::noop_coroutine();
216		}
217
218		inline void
219	20x	posix_stream_file::do_read_work(pool_work_item* w) noexcept
220		{
221	20x	auto* pw = static_cast<pool_op*>(w);
222	20x	auto* self = pw->file_;
223	20x	auto& op = self->read_op_;
224
225	20x	if (!op.cancelled.load(std::memory_order_acquire))
226		{
227		ssize_t n;
228		do
229		{
230	36x	n = ::preadv(self->fd_, op.iovecs, op.iovec_count,
231	18x	static_cast<off_t>(self->offset_));
232		}
233	18x	while (n < 0 && errno == EINTR);
234
235	18x	if (n >= 0)
236		{
237	16x	op.errn = 0;
238	16x	op.bytes_transferred = static_cast<std::size_t>(n);
239	16x	self->offset_ += static_cast<std::uint64_t>(n);
240		}
241		else
242		{
243	2x	op.errn = errno;
244	2x	op.bytes_transferred = 0;
245		}
246		}
247
248	20x	op.impl_ref = std::move(pw->ref_);
249	20x	self->svc_.post(&op);
250	20x	}
251
252		inline std::coroutine_handle<>
253	27x	posix_stream_file::write_some(
254		std::coroutine_handle<> h,
255		capy::executor_ref ex,
256		buffer_param param,
257		std::stop_token token,
258		std::error_code* ec,
259		std::size_t* bytes_out)
260		{
261	27x	auto& op = write_op_;
262	27x	op.reset();
263	27x	op.is_read = false;
264
265		// Closed-object contract outranks the zero-length no-op.
266	27x	if (fd_ < 0)
267		{
268	6x	*ec = make_error_code(std::errc::bad_file_descriptor);
269	6x	*bytes_out = 0;
270	6x	op.cont.h = h;
271	6x	return dispatch_coro(ex, op.cont);
272		}
273
274	21x	capy::mutable_buffer bufs[max_buffers];
275	21x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
276
277	21x	if (op.iovec_count == 0)
278		{
279	2x	*ec = {};
280	2x	*bytes_out = 0;
281	2x	op.cont.h = h;
282	2x	return dispatch_coro(ex, op.cont);
283		}
284
285	38x	for (int i = 0; i < op.iovec_count; ++i)
286		{
287	19x	op.iovecs[i].iov_base = bufs[i].data();
288	19x	op.iovecs[i].iov_len = bufs[i].size();
289		}
290
291	19x	op.h = h;
292	19x	op.ex = ex;
293	19x	op.ec_out = ec;
294	19x	op.bytes_out = bytes_out;
295	19x	op.start(token);
296
297	19x	op.ex.on_work_started();
298
299	19x	write_pool_op_.file_ = this;
300	19x	write_pool_op_.ref_ = this->shared_from_this();
301	19x	write_pool_op_.func_ = &posix_stream_file::do_write_work;
302	19x	if (!svc_.pool().post(&write_pool_op_))
303		{
304	✗	op.impl_ref = std::move(write_pool_op_.ref_);
305	✗	op.cancelled.store(true, std::memory_order_release);
306	✗	svc_.post(&write_op_);
307		}
308	19x	return std::noop_coroutine();
309		}
310
311		inline void
312	19x	posix_stream_file::do_write_work(pool_work_item* w) noexcept
313		{
314	19x	auto* pw = static_cast<pool_op*>(w);
315	19x	auto* self = pw->file_;
316	19x	auto& op = self->write_op_;
317
318	19x	if (!op.cancelled.load(std::memory_order_acquire))
319		{
320		ssize_t n;
321		do
322		{
323	38x	n = ::pwritev(self->fd_, op.iovecs, op.iovec_count,
324	19x	static_cast<off_t>(self->offset_));
325		}
326	19x	while (n < 0 && errno == EINTR);
327
328	19x	if (n >= 0)
329		{
330	17x	op.errn = 0;
331	17x	op.bytes_transferred = static_cast<std::size_t>(n);
332	17x	self->offset_ += static_cast<std::uint64_t>(n);
333		}
334		else
335		{
336	2x	op.errn = errno;
337	2x	op.bytes_transferred = 0;
338		}
339		}
340
341	19x	op.impl_ref = std::move(pw->ref_);
342	19x	self->svc_.post(&op);
343	19x	}
344
345		} // namespace boost::corosio::detail
346
347		#endif // BOOST_COROSIO_POSIX
348
349		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
350