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

89.3% Lines (142/159) 100.0% List of functions (15/15) 54.4% Branches (37/68)

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

posix_random_access_file_service.hpp

f(x) Functions (15)

Function Calls Lines Branches Blocks

Line	Branch	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		2662x	posix_random_access_file_service(
34			capy::execution_context& ctx, scheduler& sched)
35		1331x	: sched_(&sched)
36	1/2 ✓ Branch 0 taken 1331 times. ✗ Branch 1 not taken.	1331x	, pool_(get_or_create_pool(ctx))
37		2662x	{
38		2662x	}
39
40		3993x	~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		76x	io_object::implementation* construct() override
48			{
49		76x	auto ptr = std::make_shared<posix_random_access_file>(*this);
50		76x	auto* impl = ptr.get();
51
52			{
53	1/2 ✓ Branch 0 taken 76 times. ✗ Branch 1 not taken.	76x	std::lock_guard<std::mutex> lock(mutex_);
54		76x	file_list_.push_back(impl);
55	1/2 ✓ Branch 0 taken 76 times. ✗ Branch 1 not taken.	76x	file_ptrs_[impl] = std::move(ptr);
56		76x	}
57
58		76x	return impl;
59		76x	}
60
61		76x	void destroy(io_object::implementation* p) override
62			{
63		76x	auto& impl = static_cast<posix_random_access_file&>(*p);
64		76x	impl.cancel();
65		76x	impl.close_file();
66		76x	destroy_impl(impl);
67		76x	}
68
69		136x	void close(io_object::handle& h) override
70			{
71	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 136 times.	136x	if (h.get())
72			{
73		136x	auto& impl = static_cast<posix_random_access_file&>(*h.get());
74		136x	impl.cancel();
75		136x	impl.close_file();
76		136x	}
77		136x	}
78
79		64x	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	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 64 times.	64x	if (sched_->scheduler_locking_disabled())
87		✗	return std::make_error_code(std::errc::operation_not_supported);
88		128x	return static_cast<posix_random_access_file&>(impl).open_file(
89		64x	path, mode);
90		64x	}
91
92		1331x	void shutdown() override
93			{
94		1331x	std::lock_guard<std::mutex> lock(mutex_);
95	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 1331 times.	1331x	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		1331x	file_ptrs_.clear();
102		1331x	}
103
104		76x	void destroy_impl(posix_random_access_file& impl)
105			{
106		76x	std::lock_guard<std::mutex> lock(mutex_);
107		76x	file_list_.remove(&impl);
108	1/2 ✓ Branch 0 taken 76 times. ✗ Branch 1 not taken.	76x	file_ptrs_.erase(&impl);
109		76x	}
110
111		294x	void post(scheduler_op* op)
112			{
113		294x	sched_->post(op);
114		294x	}
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		294x	thread_pool& pool() noexcept
127			{
128		294x	return pool_;
129			}
130
131			private:
132		1331x	static thread_pool& get_or_create_pool(capy::execution_context& ctx)
133			{
134		1331x	auto* p = ctx.find_service<thread_pool>();
135	1/2 ✓ Branch 0 taken 1331 times. ✗ Branch 1 not taken.	1331x	if (p)
136		1331x	return *p;
137		✗	return ctx.make_service<thread_pool>();
138		1331x	}
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		1331x	get_random_access_file_service(capy::execution_context& ctx, scheduler& sched)
153			{
154		1331x	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		272x	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		272x	capy::mutable_buffer bufs[max_buffers];
172		272x	auto count = param.copy_to(bufs, max_buffers);
173
174	2/2 ✓ Branch 0 taken 270 times. ✓ Branch 1 taken 2 times.	272x	if (count == 0)
175			{
176		2x	*ec = {};
177		2x	*bytes_out = 0;
178		2x	return h;
179			}
180
181		270x	auto* op = new raf_op();
182		270x	op->is_read = true;
183		270x	op->offset = offset;
184
185		270x	op->iovec_count = static_cast<int>(count);
186	2/2 ✓ Branch 0 taken 270 times. ✓ Branch 1 taken 270 times.	540x	for (int i = 0; i < op->iovec_count; ++i)
187			{
188		270x	op->iovecs[i].iov_base = bufs[i].data();
189		270x	op->iovecs[i].iov_len = bufs[i].size();
190		270x	}
191
192		270x	op->h = h;
193		270x	op->ex = ex;
194		270x	op->ec_out = ec;
195		270x	op->bytes_out = bytes_out;
196		270x	op->file_ = this;
197		270x	op->file_ref = this->shared_from_this();
198		270x	op->start(token);
199
200		270x	op->ex.on_work_started();
201
202			{
203		270x	std::lock_guard<std::mutex> lock(ops_mutex_);
204		270x	outstanding_ops_.push_back(op);
205		270x	}
206
207	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 270 times.	270x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
208	2/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 270 times. ✓ Branch 2 taken 270 times. ✗ Branch 3 not taken.	270x	if (!svc_.pool().post(static_cast<pool_work_item*>(op)))
209			{
210		✗	op->cancelled.store(true, std::memory_order_release);
211		✗	svc_.post(static_cast<scheduler_op*>(op));
212		✗	}
213		270x	return std::noop_coroutine();
214		272x	}
215
216			inline std::coroutine_handle<>
217		26x	posix_random_access_file::write_some_at(
218			std::uint64_t offset,
219			std::coroutine_handle<> h,
220			capy::executor_ref ex,
221			buffer_param param,
222			std::stop_token token,
223			std::error_code* ec,
224			std::size_t* bytes_out)
225			{
226		26x	capy::mutable_buffer bufs[max_buffers];
227		26x	auto count = param.copy_to(bufs, max_buffers);
228
229	2/2 ✓ Branch 0 taken 24 times. ✓ Branch 1 taken 2 times.	26x	if (count == 0)
230			{
231		2x	*ec = {};
232		2x	*bytes_out = 0;
233		2x	return h;
234			}
235
236		24x	auto* op = new raf_op();
237		24x	op->is_read = false;
238		24x	op->offset = offset;
239
240		24x	op->iovec_count = static_cast<int>(count);
241	2/2 ✓ Branch 0 taken 24 times. ✓ Branch 1 taken 24 times.	48x	for (int i = 0; i < op->iovec_count; ++i)
242			{
243		24x	op->iovecs[i].iov_base = bufs[i].data();
244		24x	op->iovecs[i].iov_len = bufs[i].size();
245		24x	}
246
247		24x	op->h = h;
248		24x	op->ex = ex;
249		24x	op->ec_out = ec;
250		24x	op->bytes_out = bytes_out;
251		24x	op->file_ = this;
252		24x	op->file_ref = this->shared_from_this();
253		24x	op->start(token);
254
255		24x	op->ex.on_work_started();
256
257			{
258		24x	std::lock_guard<std::mutex> lock(ops_mutex_);
259		24x	outstanding_ops_.push_back(op);
260		24x	}
261
262	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 24 times.	24x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
263	2/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 24 times. ✓ Branch 2 taken 24 times. ✗ Branch 3 not taken.	24x	if (!svc_.pool().post(static_cast<pool_work_item*>(op)))
264			{
265		✗	op->cancelled.store(true, std::memory_order_release);
266		✗	svc_.post(static_cast<scheduler_op*>(op));
267		✗	}
268		24x	return std::noop_coroutine();
269		26x	}
270
271			// -- raf_op thread-pool work function --
272
273			inline void
274		294x	posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
275			{
276	1/2 ✓ Branch 0 taken 294 times. ✗ Branch 1 not taken.	294x	auto* op = static_cast<raf_op*>(w);
277		294x	auto* self = op->file_;
278
279	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 292 times.	294x	if (op->cancelled.load(std::memory_order_acquire))
280			{
281		2x	op->errn = ECANCELED;
282		2x	op->bytes_transferred = 0;
283		2x	}
284	2/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 292 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 292 times.	584x	else if (op->offset >
285		292x	static_cast<std::uint64_t>(std::numeric_limits<off_t>::max()))
286			{
287		✗	op->errn = EOVERFLOW;
288		✗	op->bytes_transferred = 0;
289		✗	}
290			else
291			{
292			ssize_t n;
293	2/2 ✓ Branch 0 taken 268 times. ✓ Branch 1 taken 24 times.	292x	if (op->is_read)
294			{
295		268x	do
296			{
297	2/4 ✓ Branch 0 taken 268 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 268 times. ✗ Branch 3 not taken.	536x	n = ::preadv(self->fd_, op->iovecs, op->iovec_count,
298		268x	static_cast<off_t>(op->offset));
299		536x	}
300	1/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 268 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken.	268x	while (n < 0 && errno == EINTR);
301		268x	}
302			else
303			{
304		24x	do
305			{
306	2/4 ✓ Branch 0 taken 24 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 24 times. ✗ Branch 3 not taken.	48x	n = ::pwritev(self->fd_, op->iovecs, op->iovec_count,
307		24x	static_cast<off_t>(op->offset));
308		48x	}
309	1/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 24 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken.	24x	while (n < 0 && errno == EINTR);
310			}
311
312	1/2 ✓ Branch 0 taken 292 times. ✗ Branch 1 not taken.	292x	if (n >= 0)
313			{
314		292x	op->errn = 0;
315		292x	op->bytes_transferred = static_cast<std::size_t>(n);
316		292x	}
317			else
318			{
319		✗	op->errn = errno;
320		✗	op->bytes_transferred = 0;
321			}
322			}
323
324	1/2 ✓ Branch 0 taken 294 times. ✗ Branch 1 not taken.	294x	self->svc_.post(static_cast<scheduler_op*>(op));
325		294x	}
326
327			} // namespace boost::corosio::detail
328
329			#endif // BOOST_COROSIO_POSIX
330
331			#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
332