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

100.0% Lines (160/0/160) 100.0% List of functions (15/0/15)

[ ] prev/next file

posix_stream_file_service.hpp

f(x) Functions (15)

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 : public file_service
33		{
34		public:
35	2496x	posix_stream_file_service(capy::execution_context& ctx, scheduler& sched)
36	4992x	: sched_(&sched)
37	2496x	, pool_(ctx)
38		{
39	2496x	}
40
41	4992x	~posix_stream_file_service() override = default;
42
43		posix_stream_file_service(posix_stream_file_service const&) = delete;
44		posix_stream_file_service&
45		operator=(posix_stream_file_service const&) = delete;
46
47	171x	io_object::implementation* construct() override
48		{
49	171x	auto ptr = std::make_shared<posix_stream_file>(*this);
50	171x	auto* impl = ptr.get();
51
52		{
53	171x	std::lock_guard<std::mutex> lock(mutex_);
54	171x	file_list_.push_back(impl);
55	171x	file_ptrs_[impl] = std::move(ptr);
56	171x	}
57
58	171x	return impl;
59	171x	}
60
61	169x	void destroy(io_object::implementation* p) override
62		{
63	169x	auto& impl = static_cast<posix_stream_file&>(*p);
64	169x	impl.cancel();
65	169x	impl.close_file();
66	169x	destroy_impl(impl);
67	169x	}
68
69	309x	void close(io_object::handle& h) override
70		{
71	309x	if (h.get())
72		{
73	309x	auto& impl = static_cast<posix_stream_file&>(*h.get());
74	309x	impl.cancel();
75	309x	impl.close_file();
76		}
77	309x	}
78
79	158x	std::error_code open_file(
80		stream_file::implementation& impl,
81		std::filesystem::path const& path,
82		file_base::flags mode) override
83		{
84		// Unavailable in the unsafe tier: the file thread pool completes
85		// cross-thread, which the lockless scheduler cannot accept.
86	158x	if (sched_->scheduler_locking_disabled())
87	2x	return std::make_error_code(std::errc::operation_not_supported);
88	156x	return static_cast<posix_stream_file&>(impl).open_file(path, mode);
89		}
90
91	2496x	void shutdown() override
92		{
93	2496x	std::lock_guard<std::mutex> lock(mutex_);
94	2498x	for (auto* impl = file_list_.pop_front(); impl != nullptr;
95	2x	impl = file_list_.pop_front())
96		{
97	2x	impl->cancel();
98	2x	impl->close_file();
99		}
100	2496x	file_ptrs_.clear();
101	2496x	}
102
103	169x	void destroy_impl(posix_stream_file& impl)
104		{
105	169x	std::lock_guard<std::mutex> lock(mutex_);
106	169x	file_list_.remove(&impl);
107	169x	file_ptrs_.erase(&impl);
108	169x	}
109
110	116x	void post(scheduler_op* op)
111		{
112	116x	sched_->post(op);
113	116x	}
114
115		void work_started() noexcept
116		{
117		sched_->work_started();
118		}
119
120		void work_finished() noexcept
121		{
122		sched_->work_finished();
123		}
124
125		/** Return the thread pool that runs this service's file work.
126
127		The pool's service is created on first use, so this can fail
128		where a plain accessor could not. Its workers start later, on
129		the first post, and a thread the system refuses there is
130		reported by that post rather than thrown here.
131
132		@throws std::bad_alloc If the service cannot be allocated.
133
134		@return The context's shared blocking-I/O pool.
135
136		@see thread_pool_ref::get
137		*/
138	125x	thread_pool& pool()
139		{
140	125x	return pool_.get();
141		}
142
143		private:
144		scheduler* sched_;
145		thread_pool_ref pool_;
146		std::mutex mutex_;
147		intrusive_list<posix_stream_file> file_list_;
148		std::unordered_map<posix_stream_file*, std::shared_ptr<posix_stream_file>>
149		file_ptrs_;
150		};
151
152		/** Get or create the stream file service for the given context. */
153		inline posix_stream_file_service&
154	2496x	get_stream_file_service(capy::execution_context& ctx, scheduler& sched)
155		{
156	2496x	return ctx.make_service<posix_stream_file_service>(sched);
157		}
158
159		// ---------------------------------------------------------------------------
160		// posix_stream_file inline implementations (require complete service type)
161		// ---------------------------------------------------------------------------
162
163		inline std::coroutine_handle<>
164	96x	posix_stream_file::read_some(
165		std::coroutine_handle<> h,
166		capy::executor_ref ex,
167		buffer_param param,
168		std::stop_token token,
169		std::error_code* ec,
170		std::size_t* bytes_out)
171		{
172	96x	auto& op = read_op_;
173	96x	op.reset();
174	96x	op.is_read = true;
175
176		// Closed-object contract outranks the zero-length no-op.
177	96x	if (fd_ < 0)
178		{
179	6x	*ec = make_error_code(std::errc::bad_file_descriptor);
180	6x	*bytes_out = 0;
181	6x	op.cont.h = h;
182	6x	return dispatch_coro(ex, op.cont);
183		}
184
185	90x	capy::mutable_buffer bufs[max_buffers];
186	90x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
187
188	90x	if (op.iovec_count == 0)
189		{
190	2x	*ec = {};
191	2x	*bytes_out = 0;
192	2x	op.cont.h = h;
193	2x	return dispatch_coro(ex, op.cont);
194		}
195
196	176x	for (int i = 0; i < op.iovec_count; ++i)
197		{
198	88x	op.iovecs[i].iov_base = bufs[i].data();
199	88x	op.iovecs[i].iov_len = bufs[i].size();
200		}
201
202	88x	op.h = h;
203	88x	op.ex = ex;
204	88x	op.ec_out = ec;
205	88x	op.bytes_out = bytes_out;
206	88x	op.start(token);
207
208	88x	op.ex.on_work_started();
209
210	88x	read_pool_op_.file_ = this;
211	88x	read_pool_op_.ref_ = this->shared_from_this();
212	88x	read_pool_op_.func_ = &posix_stream_file::do_read_work;
213	88x	if (auto pec = svc_.pool().post(&read_pool_op_))
214		{
215		// The pool is shutting down, or the system refused it a thread.
216		// Nothing of this read went cross-thread, so it answers here
217		// like the closed-descriptor and zero-length exits above rather
218		// than through a completion the scheduler has to carry back.
219	7x	read_pool_op_.ref_.reset();
220	7x	op.stop_cb.reset();
221	7x	op.ex.on_work_finished();
222	7x	*ec = pec;
223	7x	*bytes_out = 0;
224	7x	op.cont.h = h;
225	7x	return dispatch_coro(ex, op.cont);
226		}
227	81x	return std::noop_coroutine();
228		}
229
230		inline void
231	81x	posix_stream_file::do_read_work(pool_work_item* w) noexcept
232		{
233	81x	auto* pw = static_cast<pool_op*>(w);
234	81x	auto* self = pw->file_;
235	81x	auto& op = self->read_op_;
236
237	81x	if (!op.cancelled.load(std::memory_order_acquire))
238		{
239		ssize_t n;
240		do
241		{
242	150x	n = ::preadv(
243	75x	self->fd_, op.iovecs, op.iovec_count,
244	75x	static_cast<off_t>(self->offset_));
245		}
246	75x	while (n < 0 && errno == EINTR);
247
248	75x	if (n >= 0)
249		{
250	66x	op.errn = 0;
251	66x	op.bytes_transferred = static_cast<std::size_t>(n);
252	66x	self->offset_ += static_cast<std::uint64_t>(n);
253		}
254		else
255		{
256	9x	op.errn = errno;
257	9x	op.bytes_transferred = 0;
258		}
259		}
260
261	81x	op.impl_ptr = std::move(pw->ref_);
262	81x	self->svc_.post(&op);
263	81x	}
264
265		inline std::coroutine_handle<>
266	45x	posix_stream_file::write_some(
267		std::coroutine_handle<> h,
268		capy::executor_ref ex,
269		buffer_param param,
270		std::stop_token token,
271		std::error_code* ec,
272		std::size_t* bytes_out)
273		{
274	45x	auto& op = write_op_;
275	45x	op.reset();
276	45x	op.is_read = false;
277
278		// Closed-object contract outranks the zero-length no-op.
279	45x	if (fd_ < 0)
280		{
281	6x	*ec = make_error_code(std::errc::bad_file_descriptor);
282	6x	*bytes_out = 0;
283	6x	op.cont.h = h;
284	6x	return dispatch_coro(ex, op.cont);
285		}
286
287	39x	capy::mutable_buffer bufs[max_buffers];
288	39x	op.iovec_count = static_cast<int>(param.copy_to(bufs, max_buffers));
289
290	39x	if (op.iovec_count == 0)
291		{
292	2x	*ec = {};
293	2x	*bytes_out = 0;
294	2x	op.cont.h = h;
295	2x	return dispatch_coro(ex, op.cont);
296		}
297
298	74x	for (int i = 0; i < op.iovec_count; ++i)
299		{
300	37x	op.iovecs[i].iov_base = bufs[i].data();
301	37x	op.iovecs[i].iov_len = bufs[i].size();
302		}
303
304	37x	op.h = h;
305	37x	op.ex = ex;
306	37x	op.ec_out = ec;
307	37x	op.bytes_out = bytes_out;
308	37x	op.start(token);
309
310	37x	op.ex.on_work_started();
311
312	37x	write_pool_op_.file_ = this;
313	37x	write_pool_op_.ref_ = this->shared_from_this();
314	37x	write_pool_op_.func_ = &posix_stream_file::do_write_work;
315	37x	if (auto pec = svc_.pool().post(&write_pool_op_))
316		{
317		// The pool is shutting down, or the system refused it a thread.
318		// Nothing of this write went cross-thread, so it answers here
319		// like the closed-descriptor and zero-length exits above rather
320		// than through a completion the scheduler has to carry back.
321	2x	write_pool_op_.ref_.reset();
322	2x	op.stop_cb.reset();
323	2x	op.ex.on_work_finished();
324	2x	*ec = pec;
325	2x	*bytes_out = 0;
326	2x	op.cont.h = h;
327	2x	return dispatch_coro(ex, op.cont);
328		}
329	35x	return std::noop_coroutine();
330		}
331
332		inline void
333	35x	posix_stream_file::do_write_work(pool_work_item* w) noexcept
334		{
335	35x	auto* pw = static_cast<pool_op*>(w);
336	35x	auto* self = pw->file_;
337	35x	auto& op = self->write_op_;
338
339	35x	if (!op.cancelled.load(std::memory_order_acquire))
340		{
341		ssize_t n;
342		do
343		{
344	66x	n = ::pwritev(
345	33x	self->fd_, op.iovecs, op.iovec_count,
346	33x	static_cast<off_t>(self->offset_));
347		}
348	33x	while (n < 0 && errno == EINTR);
349
350	33x	if (n >= 0)
351		{
352	24x	op.errn = 0;
353	24x	op.bytes_transferred = static_cast<std::size_t>(n);
354	24x	self->offset_ += static_cast<std::uint64_t>(n);
355		}
356		else
357		{
358	9x	op.errn = errno;
359	9x	op.bytes_transferred = 0;
360		}
361		}
362
363	35x	op.impl_ptr = std::move(pw->ref_);
364	35x	self->svc_.post(&op);
365	35x	}
366
367		} // namespace boost::corosio::detail
368
369		#endif // BOOST_COROSIO_POSIX
370
371		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_STREAM_FILE_SERVICE_HPP
372