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

93.9% Lines (139/0/148) 100.0% List of functions (15/0/15)

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

posix_random_access_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_RANDOM_ACCESS_FILE_SERVICE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_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_random_access_file.hpp>
18		#include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
19		#include <boost/corosio/detail/random_access_file_service.hpp>
20		#include <boost/corosio/detail/thread_pool.hpp>
21
22		#include <limits>
23		#include <mutex>
24		#include <unordered_map>
25
26		namespace boost::corosio::detail {
27
28		/** Random-access file service for POSIX backends. */
29		class BOOST_COROSIO_DECL posix_random_access_file_service final
30		: public random_access_file_service
31		{
32		public:
33	1603x	posix_random_access_file_service(
34		capy::execution_context& ctx, scheduler& sched)
35	3206x	: sched_(&sched)
36	1603x	, pool_(get_or_create_pool(ctx))
37		{
38	1603x	}
39
40	3206x	~posix_random_access_file_service() override = default;
41
42		posix_random_access_file_service(
43		posix_random_access_file_service const&) = delete;
44		posix_random_access_file_service& operator=(
45		posix_random_access_file_service const&) = delete;
46
47	96x	io_object::implementation* construct() override
48		{
49	96x	auto ptr = std::make_shared<posix_random_access_file>(*this);
50	96x	auto* impl = ptr.get();
51
52		{
53	96x	std::lock_guard<std::mutex> lock(mutex_);
54	96x	file_list_.push_back(impl);
55	96x	file_ptrs_[impl] = std::move(ptr);
56	96x	}
57
58	96x	return impl;
59	96x	}
60
61	96x	void destroy(io_object::implementation* p) override
62		{
63	96x	auto& impl = static_cast<posix_random_access_file&>(*p);
64	96x	impl.cancel();
65	96x	impl.close_file();
66	96x	destroy_impl(impl);
67	96x	}
68
69	173x	void close(io_object::handle& h) override
70		{
71	173x	if (h.get())
72		{
73	173x	auto& impl = static_cast<posix_random_access_file&>(*h.get());
74	173x	impl.cancel();
75	173x	impl.close_file();
76		}
77	173x	}
78
79	77x	std::error_code open_file(
80		random_access_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	77x	if (sched_->scheduler_locking_disabled())
87	✗	return std::make_error_code(std::errc::operation_not_supported);
88	77x	return static_cast<posix_random_access_file&>(impl).open_file(
89	77x	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	96x	void destroy_impl(posix_random_access_file& impl)
105		{
106	96x	std::lock_guard<std::mutex> lock(mutex_);
107	96x	file_list_.remove(&impl);
108	96x	file_ptrs_.erase(&impl);
109	96x	}
110
111	302x	void post(scheduler_op* op)
112		{
113	302x	sched_->post(op);
114	302x	}
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	302x	thread_pool& pool() noexcept
127		{
128	302x	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_random_access_file> file_list_;
144		std::unordered_map<
145		posix_random_access_file*,
146		std::shared_ptr<posix_random_access_file>>
147		file_ptrs_;
148		};
149
150		/** Get or create the random-access file service for the given context. */
151		inline posix_random_access_file_service&
152	1603x	get_random_access_file_service(capy::execution_context& ctx, scheduler& sched)
153		{
154	1603x	return ctx.make_service<posix_random_access_file_service>(sched);
155		}
156
157		// ---------------------------------------------------------------------------
158		// posix_random_access_file inline implementations (require complete service)
159		// ---------------------------------------------------------------------------
160
161		inline std::coroutine_handle<>
162	281x	posix_random_access_file::read_some_at(
163		std::uint64_t offset,
164		std::coroutine_handle<> h,
165		capy::executor_ref ex,
166		buffer_param param,
167		std::stop_token token,
168		std::error_code* ec,
169		std::size_t* bytes_out)
170		{
171		// Closed-object contract outranks the zero-length no-op.
172	281x	if (fd_ < 0)
173		{
174	4x	*ec = make_error_code(std::errc::bad_file_descriptor);
175	4x	*bytes_out = 0;
176	4x	return h;
177		}
178
179	277x	capy::mutable_buffer bufs[max_buffers];
180	277x	auto count = param.copy_to(bufs, max_buffers);
181
182	277x	if (count == 0)
183		{
184	2x	*ec = {};
185	2x	*bytes_out = 0;
186	2x	return h;
187		}
188
189	275x	auto* op = new raf_op();
190	275x	op->is_read = true;
191	275x	op->offset = offset;
192
193	275x	op->iovec_count = static_cast<int>(count);
194	550x	for (int i = 0; i < op->iovec_count; ++i)
195		{
196	275x	op->iovecs[i].iov_base = bufs[i].data();
197	275x	op->iovecs[i].iov_len = bufs[i].size();
198		}
199
200	275x	op->h = h;
201	275x	op->ex = ex;
202	275x	op->ec_out = ec;
203	275x	op->bytes_out = bytes_out;
204	275x	op->file_ = this;
205	275x	op->file_ref = this->shared_from_this();
206	275x	op->start(token);
207
208	275x	op->ex.on_work_started();
209
210		{
211	275x	std::lock_guard<std::mutex> lock(ops_mutex_);
212	275x	outstanding_ops_.push_back(op);
213	275x	}
214
215	275x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
216	275x	if (!svc_.pool().post(static_cast<pool_work_item*>(op)))
217		{
218	✗	op->cancelled.store(true, std::memory_order_release);
219	✗	svc_.post(static_cast<scheduler_op*>(op));
220		}
221	275x	return std::noop_coroutine();
222		}
223
224		inline std::coroutine_handle<>
225	31x	posix_random_access_file::write_some_at(
226		std::uint64_t offset,
227		std::coroutine_handle<> h,
228		capy::executor_ref ex,
229		buffer_param param,
230		std::stop_token token,
231		std::error_code* ec,
232		std::size_t* bytes_out)
233		{
234		// Closed-object contract outranks the zero-length no-op.
235	31x	if (fd_ < 0)
236		{
237	2x	*ec = make_error_code(std::errc::bad_file_descriptor);
238	2x	*bytes_out = 0;
239	2x	return h;
240		}
241
242	29x	capy::mutable_buffer bufs[max_buffers];
243	29x	auto count = param.copy_to(bufs, max_buffers);
244
245	29x	if (count == 0)
246		{
247	2x	*ec = {};
248	2x	*bytes_out = 0;
249	2x	return h;
250		}
251
252	27x	auto* op = new raf_op();
253	27x	op->is_read = false;
254	27x	op->offset = offset;
255
256	27x	op->iovec_count = static_cast<int>(count);
257	54x	for (int i = 0; i < op->iovec_count; ++i)
258		{
259	27x	op->iovecs[i].iov_base = bufs[i].data();
260	27x	op->iovecs[i].iov_len = bufs[i].size();
261		}
262
263	27x	op->h = h;
264	27x	op->ex = ex;
265	27x	op->ec_out = ec;
266	27x	op->bytes_out = bytes_out;
267	27x	op->file_ = this;
268	27x	op->file_ref = this->shared_from_this();
269	27x	op->start(token);
270
271	27x	op->ex.on_work_started();
272
273		{
274	27x	std::lock_guard<std::mutex> lock(ops_mutex_);
275	27x	outstanding_ops_.push_back(op);
276	27x	}
277
278	27x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
279	27x	if (!svc_.pool().post(static_cast<pool_work_item*>(op)))
280		{
281	✗	op->cancelled.store(true, std::memory_order_release);
282	✗	svc_.post(static_cast<scheduler_op*>(op));
283		}
284	27x	return std::noop_coroutine();
285		}
286
287		// -- raf_op thread-pool work function --
288
289		inline void
290	302x	posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
291		{
292	302x	auto* op = static_cast<raf_op*>(w);
293	302x	auto* self = op->file_;
294
295	302x	if (op->cancelled.load(std::memory_order_acquire))
296		{
297	2x	op->errn = ECANCELED;
298	2x	op->bytes_transferred = 0;
299		}
300	600x	else if (op->offset >
301	300x	static_cast<std::uint64_t>(std::numeric_limits<off_t>::max()))
302		{
303	2x	op->errn = EOVERFLOW;
304	2x	op->bytes_transferred = 0;
305		}
306		else
307		{
308		ssize_t n;
309	298x	if (op->is_read)
310		{
311		do
312		{
313	542x	n = ::preadv(self->fd_, op->iovecs, op->iovec_count,
314	271x	static_cast<off_t>(op->offset));
315		}
316	271x	while (n < 0 && errno == EINTR);
317		}
318		else
319		{
320		do
321		{
322	54x	n = ::pwritev(self->fd_, op->iovecs, op->iovec_count,
323	27x	static_cast<off_t>(op->offset));
324		}
325	27x	while (n < 0 && errno == EINTR);
326		}
327
328	298x	if (n >= 0)
329		{
330	294x	op->errn = 0;
331	294x	op->bytes_transferred = static_cast<std::size_t>(n);
332		}
333		else
334		{
335	4x	op->errn = errno;
336	4x	op->bytes_transferred = 0;
337		}
338		}
339
340	302x	self->svc_.post(static_cast<scheduler_op*>(op));
341	302x	}
342
343		} // namespace boost::corosio::detail
344
345		#endif // BOOST_COROSIO_POSIX
346
347		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
348