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

93.9% Lines (138/0/147) 100.0% List of functions (14/0/14)

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

posix_random_access_file_service.hpp

f(x) Functions (14)

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	1790x	posix_random_access_file_service(
34		capy::execution_context& ctx, scheduler& sched)
35	3580x	: sched_(&sched)
36	1790x	, pool_(ctx)
37		{
38	1790x	}
39
40	3580x	~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	105x	io_object::implementation* construct() override
48		{
49	105x	auto ptr = std::make_shared<posix_random_access_file>(*this);
50	105x	auto* impl = ptr.get();
51
52		{
53	105x	std::lock_guard<std::mutex> lock(mutex_);
54	105x	file_list_.push_back(impl);
55	105x	file_ptrs_[impl] = std::move(ptr);
56	105x	}
57
58	105x	return impl;
59	105x	}
60
61	103x	void destroy(io_object::implementation* p) override
62		{
63	103x	auto& impl = static_cast<posix_random_access_file&>(*p);
64	103x	impl.cancel();
65	103x	impl.close_file();
66	103x	destroy_impl(impl);
67	103x	}
68
69	187x	void close(io_object::handle& h) override
70		{
71	187x	if (h.get())
72		{
73	187x	auto& impl = static_cast<posix_random_access_file&>(*h.get());
74	187x	impl.cancel();
75	187x	impl.close_file();
76		}
77	187x	}
78
79	91x	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	91x	if (sched_->scheduler_locking_disabled())
87	✗	return std::make_error_code(std::errc::operation_not_supported);
88	91x	return static_cast<posix_random_access_file&>(impl).open_file(
89	91x	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	103x	void destroy_impl(posix_random_access_file& impl)
105		{
106	103x	std::lock_guard<std::mutex> lock(mutex_);
107	103x	file_list_.remove(&impl);
108	103x	file_ptrs_.erase(&impl);
109	103x	}
110
111	324x	void post(scheduler_op* op)
112		{
113	324x	sched_->post(op);
114	324x	}
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	324x	thread_pool& pool()
140		{
141	324x	return pool_.get();
142		}
143
144		private:
145		scheduler* sched_;
146		thread_pool_ref pool_;
147		std::mutex mutex_;
148		intrusive_list<posix_random_access_file> file_list_;
149		std::unordered_map<
150		posix_random_access_file*,
151		std::shared_ptr<posix_random_access_file>>
152		file_ptrs_;
153		};
154
155		/** Get or create the random-access file service for the given context. */
156		inline posix_random_access_file_service&
157	1790x	get_random_access_file_service(capy::execution_context& ctx, scheduler& sched)
158		{
159	1790x	return ctx.make_service<posix_random_access_file_service>(sched);
160		}
161
162		// ---------------------------------------------------------------------------
163		// posix_random_access_file inline implementations (require complete service)
164		// ---------------------------------------------------------------------------
165
166		inline std::coroutine_handle<>
167	293x	posix_random_access_file::read_some_at(
168		std::uint64_t offset,
169		std::coroutine_handle<> h,
170		capy::executor_ref ex,
171		buffer_param param,
172		std::stop_token token,
173		std::error_code* ec,
174		std::size_t* bytes_out)
175		{
176		// Closed-object contract outranks the zero-length no-op.
177	293x	if (fd_ < 0)
178		{
179	4x	*ec = make_error_code(std::errc::bad_file_descriptor);
180	4x	*bytes_out = 0;
181	4x	return h;
182		}
183
184	289x	capy::mutable_buffer bufs[max_buffers];
185	289x	auto count = param.copy_to(bufs, max_buffers);
186
187	289x	if (count == 0)
188		{
189	2x	*ec = {};
190	2x	*bytes_out = 0;
191	2x	return h;
192		}
193
194	287x	auto* op = new raf_op();
195	287x	op->is_read = true;
196	287x	op->offset = offset;
197
198	287x	op->iovec_count = static_cast<int>(count);
199	574x	for (int i = 0; i < op->iovec_count; ++i)
200		{
201	287x	op->iovecs[i].iov_base = bufs[i].data();
202	287x	op->iovecs[i].iov_len = bufs[i].size();
203		}
204
205	287x	op->h = h;
206	287x	op->ex = ex;
207	287x	op->ec_out = ec;
208	287x	op->bytes_out = bytes_out;
209	287x	op->file_ = this;
210	287x	op->impl_ptr = this->shared_from_this();
211	287x	op->start(token);
212
213	287x	op->ex.on_work_started();
214
215		{
216	287x	std::lock_guard<std::mutex> lock(ops_mutex_);
217	287x	outstanding_ops_.push_back(op);
218	287x	}
219
220	287x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
221	287x	if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
222		{
223		// The pool is shutting down, or the system refused it a thread.
224		// Nothing of this read went cross-thread, so it answers here
225		// like the closed-descriptor and zero-length exits above rather
226		// than through a completion the scheduler has to carry back.
227		// destroy() is the discard the op never reaching the queue
228		// needs: it unlinks, unwinds the work count and frees.
229	✗	op->destroy();
230	✗	*ec = pec;
231	✗	*bytes_out = 0;
232	✗	return h;
233		}
234	287x	return std::noop_coroutine();
235		}
236
237		inline std::coroutine_handle<>
238	41x	posix_random_access_file::write_some_at(
239		std::uint64_t offset,
240		std::coroutine_handle<> h,
241		capy::executor_ref ex,
242		buffer_param param,
243		std::stop_token token,
244		std::error_code* ec,
245		std::size_t* bytes_out)
246		{
247		// Closed-object contract outranks the zero-length no-op.
248	41x	if (fd_ < 0)
249		{
250	2x	*ec = make_error_code(std::errc::bad_file_descriptor);
251	2x	*bytes_out = 0;
252	2x	return h;
253		}
254
255	39x	capy::mutable_buffer bufs[max_buffers];
256	39x	auto count = param.copy_to(bufs, max_buffers);
257
258	39x	if (count == 0)
259		{
260	2x	*ec = {};
261	2x	*bytes_out = 0;
262	2x	return h;
263		}
264
265	37x	auto* op = new raf_op();
266	37x	op->is_read = false;
267	37x	op->offset = offset;
268
269	37x	op->iovec_count = static_cast<int>(count);
270	74x	for (int i = 0; i < op->iovec_count; ++i)
271		{
272	37x	op->iovecs[i].iov_base = bufs[i].data();
273	37x	op->iovecs[i].iov_len = bufs[i].size();
274		}
275
276	37x	op->h = h;
277	37x	op->ex = ex;
278	37x	op->ec_out = ec;
279	37x	op->bytes_out = bytes_out;
280	37x	op->file_ = this;
281	37x	op->impl_ptr = this->shared_from_this();
282	37x	op->start(token);
283
284	37x	op->ex.on_work_started();
285
286		{
287	37x	std::lock_guard<std::mutex> lock(ops_mutex_);
288	37x	outstanding_ops_.push_back(op);
289	37x	}
290
291	37x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
292	37x	if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
293		{
294		// The pool is shutting down, or the system refused it a thread.
295		// Nothing of this write went cross-thread, so it answers here
296		// like the closed-descriptor and zero-length exits above rather
297		// than through a completion the scheduler has to carry back.
298		// destroy() is the discard the op never reaching the queue
299		// needs: it unlinks, unwinds the work count and frees.
300	✗	op->destroy();
301	✗	*ec = pec;
302	✗	*bytes_out = 0;
303	✗	return h;
304		}
305	37x	return std::noop_coroutine();
306		}
307
308		// -- raf_op thread-pool work function --
309
310		inline void
311	324x	posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
312		{
313	324x	auto* op = static_cast<raf_op*>(w);
314	324x	auto* self = op->file_;
315
316	324x	if (op->cancelled.load(std::memory_order_acquire))
317		{
318	4x	op->errn = ECANCELED;
319	4x	op->bytes_transferred = 0;
320		}
321	640x	else if (op->offset >
322	320x	static_cast<std::uint64_t>(std::numeric_limits<off_t>::max()))
323		{
324	2x	op->errn = EOVERFLOW;
325	2x	op->bytes_transferred = 0;
326		}
327		else
328		{
329		ssize_t n;
330	318x	if (op->is_read)
331		{
332		do
333		{
334	562x	n = ::preadv(self->fd_, op->iovecs, op->iovec_count,
335	281x	static_cast<off_t>(op->offset));
336		}
337	281x	while (n < 0 && errno == EINTR);
338		}
339		else
340		{
341		do
342		{
343	74x	n = ::pwritev(self->fd_, op->iovecs, op->iovec_count,
344	37x	static_cast<off_t>(op->offset));
345		}
346	37x	while (n < 0 && errno == EINTR);
347		}
348
349	318x	if (n >= 0)
350		{
351	304x	op->errn = 0;
352	304x	op->bytes_transferred = static_cast<std::size_t>(n);
353		}
354		else
355		{
356	14x	op->errn = errno;
357	14x	op->bytes_transferred = 0;
358		}
359		}
360
361	324x	self->svc_.post(static_cast<scheduler_op*>(op));
362	324x	}
363
364		} // namespace boost::corosio::detail
365
366		#endif // BOOST_COROSIO_POSIX
367
368		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
369