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

96.6% Lines (168 / 174) 100.0% Functions (19 / 19)

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

posix_random_access_file_service.hpp

f(x) Functions (19)

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/object_pool.hpp>
21		#include <boost/corosio/detail/object_ref.hpp>
22		#include <boost/corosio/detail/thread_pool.hpp>
23
24		#include <limits>
25		#include <vector>
26
27		namespace boost::corosio::detail {
28
29		/** Random-access file service for POSIX backends. */
30		class BOOST_COROSIO_DECL posix_random_access_file_service final
31		: public random_access_file_service
32		{
33		friend class posix_random_access_file;
34
35		public:
36	189x	explicit posix_random_access_file_service(capy::execution_context& ctx)
37	567x	: sched_(&get_scheduler(ctx))
38	189x	, pool_(ctx)
39		{
40	189x	}
41
42	378x	~posix_random_access_file_service() override = default;
43
44		posix_random_access_file_service(posix_random_access_file_service const&) =
45		delete;
46		posix_random_access_file_service&
47		operator=(posix_random_access_file_service const&) = delete;
48
49	244x	io_object::implementation* construct() override
50		{
51	244x	return object_pool_.acquire(*this);
52		}
53
54	244x	void destroy(io_object::implementation* p) override
55		{
56	244x	auto& impl = static_cast<posix_random_access_file&>(*p);
57	244x	impl.cancel();
58	244x	impl.close_file();
59	244x	release(&impl);
60	244x	}
61
62	455x	void close(io_object::handle& h) override
63		{
64	455x	if (h.get())
65		{
66	455x	auto& impl = static_cast<posix_random_access_file&>(*h.get());
67	455x	impl.cancel();
68	455x	impl.close_file();
69		}
70	455x	}
71
72	224x	std::error_code open_file(
73		random_access_file::implementation& impl,
74		std::filesystem::path const& path,
75		file_base::flags mode) override
76		{
77	224x	if (pool_unavailable())
78	✗	return std::make_error_code(std::errc::operation_not_supported);
79	224x	return static_cast<posix_random_access_file&>(impl).open_file(
80	224x	path, mode);
81		}
82
83		/** Check whether file I/O is unavailable on this context.
84
85		The file thread pool completes cross-thread, which the
86		lockless scheduler of the unsafe tier cannot accept.
87		*/
88	233x	bool pool_unavailable() const noexcept
89		{
90	233x	return sched_->scheduler_locking_disabled();
91		}
92
93	189x	void shutdown() override
94		{
95		// See uring_socket_service_base::shutdown(): snapshot under an
96		// acquired reference, cancel+close without the pool lock held;
97		// shutdown() sets shutting-down and takes the snapshot in one
98		// critical section, so a close that drops the last ref deletes
99		// rather than recycles.
100	189x	std::vector<posix_random_access_file*> live;
101	189x	object_pool_.shutdown(
102	189x	[&](posix_random_access_file* f)
103		{
104	6x	acquire(f);
105	6x	live.push_back(f);
106	6x	});
107	195x	for (auto* f : live)
108		{
109	6x	f->cancel();
110	6x	f->close_file();
111	6x	release(f);
112		}
113	189x	}
114
115	505x	void post(scheduler_op* op)
116		{
117	505x	sched_->post(op);
118	505x	}
119
120	515x	void work_started() noexcept
121		{
122	515x	sched_->work_started();
123	515x	}
124
125	515x	void work_finished() noexcept
126		{
127	515x	sched_->work_finished();
128	515x	}
129
130		/** Return the thread pool that runs this service's file work.
131
132		The pool's service is created on first use, so this can fail
133		where a plain accessor could not. Its workers start later, on
134		the first post, and a thread the system refuses there is
135		reported by that post rather than thrown here.
136
137		@throws std::bad_alloc If the service cannot be allocated.
138
139		@return The context's shared blocking-I/O pool.
140
141		@see thread_pool_ref::get
142		*/
143	515x	thread_pool& pool()
144		{
145	515x	return pool_.get();
146		}
147
148		private:
149		scheduler* sched_;
150		thread_pool_ref pool_;
151		object_pool<posix_random_access_file> object_pool_;
152		};
153
154		// ---------------------------------------------------------------------------
155		// posix_random_access_file inline implementations (require complete service)
156		// ---------------------------------------------------------------------------
157
158		inline std::coroutine_handle<>
159	391x	posix_random_access_file::read_some_at(
160		std::uint64_t offset,
161		capy::continuation& cont,
162		capy::executor_ref ex,
163		buffer_param param,
164		std::stop_token token,
165		std::error_code* ec,
166		std::size_t* bytes_out)
167		{
168		// Closed-object contract outranks the zero-length no-op.
169	391x	if (fd_ < 0)
170		{
171	4x	*ec = make_error_code(std::errc::bad_file_descriptor);
172	4x	*bytes_out = 0;
173	4x	return cont.h;
174		}
175
176	387x	capy::mutable_buffer bufs[max_buffers];
177	387x	auto count = param.copy_to(bufs, max_buffers);
178
179	387x	if (count == 0)
180		{
181	2x	*ec = {};
182	2x	*bytes_out = 0;
183	2x	return cont.h;
184		}
185
186	385x	auto* op = acquire_op();
187	385x	op->is_read = true;
188	385x	op->offset = offset;
189	385x	op->fd = fd_;
190
191	385x	op->iovec_count = static_cast<int>(count);
192	770x	for (int i = 0; i < op->iovec_count; ++i)
193		{
194	385x	op->iovecs[i].iov_base = bufs[i].data();
195	385x	op->iovecs[i].iov_len = bufs[i].size();
196		}
197
198	385x	op->cont = &cont;
199	385x	op->ex = ex;
200	385x	op->ec_out = ec;
201	385x	op->bytes_out = bytes_out;
202	385x	op->file_ = this;
203	385x	op->object_ref_ = detail::object_ref(this);
204	385x	op->start(token);
205
206	385x	svc_.work_started();
207
208		{
209	385x	std::lock_guard<std::mutex> lock(ops_mutex_);
210	385x	outstanding_ops_.push_back(op);
211	385x	}
212
213	385x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
214	385x	static_cast<pool_work_item*>(op)->destroy_ = &raf_op::destroy_work;
215	385x	if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
216		{
217		// The pool is shutting down, or the system refused it a thread.
218		// Nothing of this read went cross-thread, so it answers here
219		// like the closed-descriptor and zero-length exits above rather
220		// than through a completion the scheduler has to carry back.
221		// The op never reached the queue: it unlinks, unwinds the
222		// work count and recycles, and the coroutine resumes below.
223	2x	op->discard();
224	2x	*ec = pec;
225	2x	*bytes_out = 0;
226	2x	return cont.h;
227		}
228	383x	return std::noop_coroutine();
229		}
230
231		inline std::coroutine_handle<>
232	134x	posix_random_access_file::write_some_at(
233		std::uint64_t offset,
234		capy::continuation& cont,
235		capy::executor_ref ex,
236		buffer_param param,
237		std::stop_token token,
238		std::error_code* ec,
239		std::size_t* bytes_out)
240		{
241		// Closed-object contract outranks the zero-length no-op.
242	134x	if (fd_ < 0)
243		{
244	2x	*ec = make_error_code(std::errc::bad_file_descriptor);
245	2x	*bytes_out = 0;
246	2x	return cont.h;
247		}
248
249	132x	capy::mutable_buffer bufs[max_buffers];
250	132x	auto count = param.copy_to(bufs, max_buffers);
251
252	132x	if (count == 0)
253		{
254	2x	*ec = {};
255	2x	*bytes_out = 0;
256	2x	return cont.h;
257		}
258
259	130x	auto* op = acquire_op();
260	130x	op->is_read = false;
261	130x	op->offset = offset;
262	130x	op->fd = fd_;
263
264	130x	op->iovec_count = static_cast<int>(count);
265	260x	for (int i = 0; i < op->iovec_count; ++i)
266		{
267	130x	op->iovecs[i].iov_base = bufs[i].data();
268	130x	op->iovecs[i].iov_len = bufs[i].size();
269		}
270
271	130x	op->cont = &cont;
272	130x	op->ex = ex;
273	130x	op->ec_out = ec;
274	130x	op->bytes_out = bytes_out;
275	130x	op->file_ = this;
276	130x	op->object_ref_ = detail::object_ref(this);
277	130x	op->start(token);
278
279	130x	svc_.work_started();
280
281		{
282	130x	std::lock_guard<std::mutex> lock(ops_mutex_);
283	130x	outstanding_ops_.push_back(op);
284	130x	}
285
286	130x	static_cast<pool_work_item*>(op)->func_ = &raf_op::do_work;
287	130x	static_cast<pool_work_item*>(op)->destroy_ = &raf_op::destroy_work;
288	130x	if (auto pec = svc_.pool().post(static_cast<pool_work_item*>(op)))
289		{
290		// The pool is shutting down, or the system refused it a thread.
291		// Nothing of this write went cross-thread, so it answers here
292		// like the closed-descriptor and zero-length exits above rather
293		// than through a completion the scheduler has to carry back.
294		// The op never reached the queue: it unlinks, unwinds the
295		// work count and recycles, and the coroutine resumes below.
296	2x	op->discard();
297	2x	*ec = pec;
298	2x	*bytes_out = 0;
299	2x	return cont.h;
300		}
301	128x	return std::noop_coroutine();
302		}
303
304		// -- raf_op thread-pool work function --
305
306		inline void
307	505x	posix_random_access_file::raf_op::do_work(pool_work_item* w) noexcept
308		{
309	505x	auto* op = static_cast<raf_op*>(w);
310	505x	auto* self = op->file_;
311
312	505x	if (op->cancelled.load(std::memory_order_acquire))
313		{
314	68x	op->errn = ECANCELED;
315	68x	op->bytes_transferred = 0;
316		}
317	437x	else if (
318	874x	op->offset >
319	437x	static_cast<std::uint64_t>(std::numeric_limits<file_off_t>::max()))
320		{
321	2x	op->errn = EOVERFLOW;
322	2x	op->bytes_transferred = 0;
323		}
324	435x	else if (!self->gate_.enter(op->fd))
325		{
326		// Closed or replaced since this was queued.
327	✗	op->errn = ECANCELED;
328	✗	op->bytes_transferred = 0;
329		}
330	435x	else if (op->cancelled.load(std::memory_order_acquire))
331		{
332		// Closed and reopened onto the same number since the check
333		// above: the gate admits the number, but not this op.
334	✗	self->gate_.leave();
335	✗	op->errn = ECANCELED;
336	✗	op->bytes_transferred = 0;
337		}
338		else
339		{
340		ssize_t n;
341	435x	if (op->is_read)
342		{
343		do
344		{
345	323x	n = file_preadv(
346	323x	op->fd, op->iovecs, op->iovec_count,
347	323x	static_cast<file_off_t>(op->offset));
348		}
349	323x	while (n < 0 && errno == EINTR);
350		}
351		else
352		{
353		do
354		{
355	112x	n = file_pwritev(
356	112x	op->fd, op->iovecs, op->iovec_count,
357	112x	static_cast<file_off_t>(op->offset));
358		}
359	112x	while (n < 0 && errno == EINTR);
360		}
361
362	435x	if (n >= 0)
363		{
364	421x	op->errn = 0;
365	421x	op->bytes_transferred = static_cast<std::size_t>(n);
366		}
367		else
368		{
369	14x	op->errn = errno;
370	14x	op->bytes_transferred = 0;
371		}
372	435x	self->gate_.leave();
373		}
374
375	505x	self->svc_.post(static_cast<scheduler_op*>(op));
376	505x	}
377
378		inline std::error_code
379	9x	posix_random_access_file::assign(native_handle_type handle) noexcept
380		{
381		// The public assign() guarantees the object is closed.
382	9x	if (svc_.pool_unavailable())
383	2x	return std::make_error_code(std::errc::operation_not_supported);
384	7x	if (auto ec = validate_file_fd(handle))
385	4x	return ec;
386
387	3x	fd_ = handle;
388	3x	gate_.open(handle);
389	3x	return {};
390		}
391
392		inline void
393	244x	posix_random_access_file::retire() noexcept
394		{
395	244x	svc_.object_pool_.recycle(this);
396	244x	}
397
398		inline void
399	515x	posix_random_access_file::release_work(
400		posix_random_access_file_service& svc) noexcept
401		{
402	515x	svc.work_finished();
403	515x	}
404
405		} // namespace boost::corosio::detail
406
407		#endif // BOOST_COROSIO_POSIX
408
409		#endif // BOOST_COROSIO_NATIVE_DETAIL_POSIX_POSIX_RANDOM_ACCESS_FILE_SERVICE_HPP
410