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

92.8% Lines (155/167) 100.0% List of functions (15/15) 68.1% Branches (49/72)

[ ] 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		3888x	posix_random_access_file_service(
34			capy::execution_context& ctx, scheduler& sched)
35		1944x	: sched_(&sched)
36	1/2 ✓ Branch 0 taken 1944 times. ✗ Branch 1 not taken.	1944x	, pool_(get_or_create_pool(ctx))
37		3888x	{
38		3888x	}
39
40		5832x	~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	1/2 ✓ Branch 0 taken 96 times. ✗ Branch 1 not taken.	96x	std::lock_guard<std::mutex> lock(mutex_);
54		96x	file_list_.push_back(impl);
55	1/2 ✓ Branch 0 taken 96 times. ✗ Branch 1 not taken.	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	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 173 times.	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		173x	}
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	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 77 times.	77x	if (sched_->scheduler_locking_disabled())
87		✗	return std::make_error_code(std::errc::operation_not_supported);
88		154x	return static_cast<posix_random_access_file&>(impl).open_file(
89		77x	path, mode);
90		77x	}
91
92		1944x	void shutdown() override
93			{
94		1944x	std::lock_guard<std::mutex> lock(mutex_);
95	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 1944 times.	1944x	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		1944x	file_ptrs_.clear();
102		1944x	}
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	1/2 ✓ Branch 0 taken 96 times. ✗ Branch 1 not taken.	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		1944x	static thread_pool& get_or_create_pool(capy::execution_context& ctx)
133			{
134		1944x	auto* p = ctx.find_service<thread_pool>();
135	1/2 ✓ Branch 0 taken 1944 times. ✗ Branch 1 not taken.	1944x	if (p)
136		1944x	return *p;
137		✗	return ctx.make_service<thread_pool>();
138		1944x	}
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		1944x	get_random_access_file_service(capy::execution_context& ctx, scheduler& sched)
153			{
154		1944x	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	2/2 ✓ Branch 0 taken 277 times. ✓ Branch 1 taken 4 times.	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	2/2 ✓ Branch 0 taken 275 times. ✓ Branch 1 taken 2 times.	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	2/2 ✓ Branch 0 taken 275 times. ✓ Branch 1 taken 275 times.	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		275x	}
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	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 275 times.	275x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
216	2/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 275 times. ✓ Branch 2 taken 275 times. ✗ Branch 3 not taken.	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		281x	}
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	2/2 ✓ Branch 0 taken 29 times. ✓ Branch 1 taken 2 times.	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	2/2 ✓ Branch 0 taken 27 times. ✓ Branch 1 taken 2 times.	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	2/2 ✓ Branch 0 taken 27 times. ✓ Branch 1 taken 27 times.	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		27x	}
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	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 27 times.	27x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
279	2/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 27 times. ✓ Branch 2 taken 27 times. ✗ Branch 3 not taken.	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		31x	}
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	1/2 ✓ Branch 0 taken 302 times. ✗ Branch 1 not taken.	302x	auto* op = static_cast<raf_op*>(w);
293		302x	auto* self = op->file_;
294
295	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 300 times.	302x	if (op->cancelled.load(std::memory_order_acquire))
296			{
297		2x	op->errn = ECANCELED;
298		2x	op->bytes_transferred = 0;
299		2x	}
300	4/4 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 298 times. ✓ Branch 2 taken 2 times. ✓ Branch 3 taken 298 times.	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		2x	}
306			else
307			{
308			ssize_t n;
309	2/2 ✓ Branch 0 taken 271 times. ✓ Branch 1 taken 27 times.	298x	if (op->is_read)
310			{
311		271x	do
312			{
313	2/4 ✓ Branch 0 taken 271 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 271 times. ✗ Branch 3 not taken.	542x	n = ::preadv(self->fd_, op->iovecs, op->iovec_count,
314		271x	static_cast<off_t>(op->offset));
315		542x	}
316	3/4 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 269 times. ✓ Branch 2 taken 2 times. ✗ Branch 3 not taken.	271x	while (n < 0 && errno == EINTR);
317		271x	}
318			else
319			{
320		27x	do
321			{
322	2/4 ✓ Branch 0 taken 27 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 27 times. ✗ Branch 3 not taken.	54x	n = ::pwritev(self->fd_, op->iovecs, op->iovec_count,
323		27x	static_cast<off_t>(op->offset));
324		54x	}
325	3/4 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 25 times. ✓ Branch 2 taken 2 times. ✗ Branch 3 not taken.	27x	while (n < 0 && errno == EINTR);
326			}
327
328	2/2 ✓ Branch 0 taken 294 times. ✓ Branch 1 taken 4 times.	298x	if (n >= 0)
329			{
330		294x	op->errn = 0;
331		294x	op->bytes_transferred = static_cast<std::size_t>(n);
332		294x	}
333			else
334			{
335	1/2 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken.	4x	op->errn = errno;
336		4x	op->bytes_transferred = 0;
337			}
338			}
339
340	1/2 ✓ Branch 0 taken 302 times. ✗ Branch 1 not taken.	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