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

91.1% Lines (144/0/158) 100.0% List of functions (15/0/15)

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

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