include/boost/corosio/native/detail/uring/uring_random_access_file.hpp

97.8% Lines (180 / 184) 96.9% Functions (31 / 32)

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

uring_random_access_file.hpp

f(x) Functions (32)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
3		// Copyright (c) 2026 Michael Vandeberg
4		//
5		// Distributed under the Boost Software License, Version 1.0. (See accompanying
6		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7		//
8		// Official repository: https://github.com/cppalliance/corosio
9		//
10
11		#ifndef BOOST_COROSIO_NATIVE_DETAIL_URING_URING_RANDOM_ACCESS_FILE_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_RANDOM_ACCESS_FILE_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_URING
17
18		#include <boost/corosio/detail/random_access_file_service.hpp>
19		#include <boost/corosio/detail/config.hpp>
20		#include <boost/corosio/detail/intrusive.hpp>
21		#include <boost/corosio/native/detail/uring/uring_file_ops.hpp>
22		#include <boost/corosio/native/detail/uring/uring_file_service_base.hpp>
23		#include <boost/corosio/native/detail/uring/uring_scheduler.hpp>
24		#include <boost/corosio/native/detail/make_err.hpp>
25		#include <boost/corosio/native/detail/posix/large_file.hpp>
26		#include <boost/corosio/native/detail/validate_fd.hpp>
27		#include <boost/corosio/random_access_file.hpp>
28		#include <boost/capy/ex/abandon.hpp>
29
30		#include <cstdint>
31		#include <filesystem>
32		#include <limits>
33		#include <mutex>
34		#include <system_error>
35		#include <utility>
36
37		#include <fcntl.h>
38		#include <sys/stat.h>
39		#include <sys/types.h>
40		#include <unistd.h>
41
42		namespace boost::corosio::detail {
43
44		class uring_random_access_file_service;
45
46		/** Native io_uring random-access-file implementation.
47
48		Async `read_some_at` / `write_some_at` submit `IORING_OP_READV`
49		/ `IORING_OP_WRITEV` with the caller-supplied offset. Metadata
50		operations (open, size, resize, sync, close) are synchronous
51		syscalls.
52
53		@par Thread Safety
54		Concurrent `read_some_at` / `write_some_at` calls on the same
55		file at distinct offsets are safe; ordering between two
56		submissions at the same offset is unspecified at the kernel
57		level (matches POSIX `pread(2)` / `pwrite(2)` semantics).
58		*/
59		class BOOST_COROSIO_DECL uring_random_access_file final
60		: public random_access_file::implementation
61		, public intrusive_list<uring_random_access_file>::node
62		{
63		friend class uring_random_access_file_service;
64
65		int fd_ = -1;
66		uring_scheduler* sched_ = nullptr;
67		uring_random_access_file_service* svc_ = nullptr;
68
69		// Random-access files legitimately support concurrent ops at
70		// different offsets on the same fd (e.g. parallel reads in
71		// testConcurrentReads). Embedding a single slot would smash
72		// state across calls; ops recycle through per-file free lists
73		// so concurrency is unbounded but steady state allocates
74		// nothing. The storage belongs to this impl (it survives
75		// recycling) and only the destructor frees it.
76
77		/** Pooled random-access op: completion recycles instead of
78		deleting.
79
80		Reuses the heap variant's submission/finish machinery but
81		swaps the completion handler (the func-pointer convention
82		makes that one protected assignment) for one that returns the
83		op to its file's free list. The inherited `object_ref_`
84		keepalive is dropped only after the push, so a final release
85		never runs under `ops_mutex_`.
86		*/
87		template<class Base>
88		struct pooled_op final
89		: Base
90		, intrusive_list<pooled_op<Base>>::node
91		{
92		uring_random_access_file* file_ = nullptr;
93
94	211x	pooled_op() noexcept
95	211x	{
96	211x	this->func_ = &do_pooled_handler;
97	211x	}
98
99	249x	static void do_pooled_handler(
100		void* owner,
101		scheduler_op* base,
102		std::uint32_t /bytes/,
103		std::uint32_t /error/) noexcept
104		{
105	249x	auto* self = static_cast<pooled_op*>(base);
106	249x	self->stop_cb.reset();
107	249x	auto keep = std::move(self->object_ref_);
108	249x	if (owner == nullptr)
109		{
110	4x	auto* c = std::exchange(self->cont, nullptr);
111	4x	self->file_->release_op(self);
112		// The frame may own the file; the keepalive outlives it.
113	4x	if (c && c->h)
114	4x	capy::abandon(*c);
115	4x	return;
116		}
117	245x	Base::finish(self);
118	245x	auto* c = self->cont;
119	245x	auto ex = self->ex;
120	245x	self->file_->release_op(self);
121	245x	dispatch_coro(ex, *c).resume();
122	249x	}
123		};
124
125		using pooled_read_op = pooled_op<uring_random_access_read_op>;
126		using pooled_write_op = pooled_op<uring_random_access_write_op>;
127
128		std::mutex ops_mutex_;
129		intrusive_list<pooled_read_op> free_reads_;
130		intrusive_list<pooled_write_op> free_writes_;
131		// In flight, so a cancel can name each one.
132		intrusive_list<pooled_read_op> live_reads_;
133		intrusive_list<pooled_write_op> live_writes_;
134
135		/// Pop a recycled op of the given kind, or allocate on the cold
136		/// path, and list it live; `prepare()` resets every per-use field.
137		template<class Op>
138	249x	Op* acquire_op(intrusive_list<Op>& free_list, intrusive_list<Op>& live)
139		{
140	249x	std::lock_guard<std::mutex> lock(ops_mutex_);
141	249x	Op* op = free_list.pop_front();
142	249x	if (!op)
143	211x	op = new Op();
144	249x	op->file_ = this;
145	249x	live.push_back(op);
146	249x	return op;
147	249x	}
148
149	188x	void release_op(pooled_read_op* op) noexcept
150		{
151	188x	std::lock_guard<std::mutex> lock(ops_mutex_);
152	188x	live_reads_.remove(op);
153	188x	free_reads_.push_front(op);
154	188x	}
155
156	61x	void release_op(pooled_write_op* op) noexcept
157		{
158	61x	std::lock_guard<std::mutex> lock(ops_mutex_);
159	61x	live_writes_.remove(op);
160	61x	free_writes_.push_front(op);
161	61x	}
162
163		/// Cancel every live op. @pre `ops_mutex_` and the ring mutex are held.
164	105x	void cancel_live_locked() noexcept
165		{
166	105x	live_reads_.for_each(
167	2x	[this](pooled_read_op* op) { sched_->cancel_op_locked(op); });
168	105x	live_writes_.for_each(
169	2x	[this](pooled_write_op* op) { sched_->cancel_op_locked(op); });
170	105x	}
171
172		public:
173	115x	explicit uring_random_access_file(
174		uring_random_access_file_service& svc,
175		uring_scheduler& sched) noexcept
176	230x	: sched_(&sched)
177	115x	, svc_(&svc)
178		{
179	115x	}
180
181	230x	~uring_random_access_file() override
182	115x	{
183	115x	close_file();
184	282x	while (auto* op = free_reads_.pop_front())
185	167x	delete op;
186	159x	while (auto* op = free_writes_.pop_front())
187	44x	delete op;
188	230x	}
189
190		/// Recycle into the owning service's pool. Defined out-of-line
191		/// after uring_random_access_file_service for its complete type.
192		void retire() noexcept override;
193
194		/** Reset for recycling.
195
196		`close_file()` already drove fd_ to its closed value before the
197		refcount reached zero; every op holds its own `object_ref`
198		while in flight, so none can be outstanding here. The free
199		lists deliberately survive recycling — the storage belongs to
200		this impl and only the destructor frees it.
201
202		@pre refs_ == 0, fd closed, no op in flight.
203		*/
204	✗	void reuse() noexcept
205		{
206	✗	BOOST_COROSIO_ASSERT(fd_ == -1);
207	✗	}
208
209		// -- random_access_file::implementation --
210
211		std::coroutine_handle<> read_some_at(
212		std::uint64_t,
213		capy::continuation&,
214		capy::executor_ref,
215		buffer_param,
216		std::stop_token,
217		std::error_code*,
218		std::size_t*) override;
219
220		std::coroutine_handle<> write_some_at(
221		std::uint64_t,
222		capy::continuation&,
223		capy::executor_ref,
224		buffer_param,
225		std::stop_token,
226		std::error_code*,
227		std::size_t*) override;
228
229	590x	native_handle_type native_handle() const noexcept override
230		{
231	590x	return fd_;
232		}
233
234	2x	void cancel() noexcept override
235		{
236	2x	if (fd_ < 0)
237	✗	return;
238	2x	std::lock_guard<std::mutex> lock(ops_mutex_);
239	4x	sched_->cancel_ops([this]() noexcept { cancel_live_locked(); });
240	2x	}
241
242	5x	capy::results<std::uint64_t> size() const noexcept override
243		{
244		file_stat_t st;
245	5x	if (file_fstat(fd_, &st) < 0)
246	2x	return {make_err(errno), 0};
247	3x	return {std::error_code{}, static_cast<std::uint64_t>(st.st_size)};
248		}
249
250	6x	std::error_code resize(std::uint64_t new_size) noexcept override
251		{
252	6x	if (new_size > static_cast<std::uint64_t>(
253	6x	(std::numeric_limits<file_off_t>::max)()))
254	1x	return make_err(EOVERFLOW);
255	5x	if (file_ftruncate(fd_, static_cast<file_off_t>(new_size)) < 0)
256	3x	return make_err(errno);
257	2x	return {};
258		}
259
260	3x	std::error_code sync_data() noexcept override
261		{
262		#if BOOST_COROSIO_HAS_POSIX_SYNCHRONIZED_IO
263	3x	if (::fdatasync(fd_) < 0)
264		#else
265		if (::fsync(fd_) < 0)
266		#endif
267	2x	return make_err(errno);
268	1x	return {};
269		}
270
271	3x	std::error_code sync_all() noexcept override
272		{
273	3x	if (::fsync(fd_) < 0)
274	2x	return make_err(errno);
275	1x	return {};
276		}
277
278	1x	capy::results<native_handle_type> release() noexcept override
279		{
280		// Flush while the fd is still open, so no queued SQE can name
281		// a number the caller closes and recycles.
282	1x	if (fd_ >= 0)
283	1x	flush_live();
284	1x	int fd = fd_;
285	1x	fd_ = -1;
286	1x	return {std::error_code{}, fd};
287		}
288
289	3x	std::error_code assign(native_handle_type handle) noexcept override
290		{
291		// The public assign() guarantees the object is closed.
292	3x	if (auto ec = validate_file_fd(handle))
293	2x	return ec;
294
295	1x	fd_ = handle;
296	1x	return {};
297		}
298
299		// -- Internal --
300
301		/// Open the file. Synchronous; sets `fd_`. Caller is the service.
302		std::error_code
303	106x	open_file(std::filesystem::path const& path, file_base::flags mode)
304		{
305	106x	close_file();
306
307	106x	int oflags = 0;
308	106x	unsigned access = static_cast<unsigned>(mode) & 3u;
309	106x	if (access == static_cast<unsigned>(file_base::read_write))
310	20x	oflags \|= O_RDWR;
311	86x	else if (access == static_cast<unsigned>(file_base::write_only))
312	32x	oflags \|= O_WRONLY;
313		else
314	54x	oflags \|= O_RDONLY;
315
316	106x	if ((mode & file_base::create) != file_base::flags(0))
317	14x	oflags \|= O_CREAT;
318	106x	if ((mode & file_base::exclusive) != file_base::flags(0))
319	2x	oflags \|= O_EXCL;
320	106x	if ((mode & file_base::truncate) != file_base::flags(0))
321	7x	oflags \|= O_TRUNC;
322	106x	if ((mode & file_base::sync_all_on_write) != file_base::flags(0))
323	1x	oflags \|= O_SYNC;
324
325	106x	oflags \|= O_CLOEXEC;
326
327	106x	int fd = ::open(path.c_str(), oflags \| large_file_open_flag, 0666);
328	106x	if (fd < 0)
329	4x	return make_err(errno);
330
331	102x	fd_ = fd;
332
333		#ifdef POSIX_FADV_RANDOM
334		// Hint the page cache that access will be random; matches
335		// the POSIX backend.
336	102x	::posix_fadvise(fd_, 0, 0, POSIX_FADV_RANDOM);
337		#endif
338
339	102x	return {};
340		}
341
342		/// Cancel the live ops and flush every SQE naming `fd_`.
343	103x	void flush_live() noexcept
344		{
345	103x	std::lock_guard<std::mutex> lock(ops_mutex_);
346	103x	sched_->flush_after_cancel(
347	206x	fd_, [this]() noexcept { cancel_live_locked(); });
348	103x	}
349
350		/// Cancel any in-flight ops and close the fd. Idempotent.
351	553x	void close_file() noexcept
352		{
353	553x	if (fd_ >= 0)
354		{
355		// The kernel may run a queued pipe write as task work at
356		// either kernel entry below; with the reader already gone
357		// that raises SIGPIPE.
358	102x	scoped_sigpipe_block no_sigpipe;
359	102x	flush_live();
360	102x	::close(fd_);
361	102x	fd_ = -1;
362	102x	}
363	553x	}
364		};
365
366		inline std::coroutine_handle<>
367	188x	uring_random_access_file::read_some_at(
368		std::uint64_t user_offset,
369		capy::continuation& cont,
370		capy::executor_ref ex,
371		buffer_param buffers,
372		std::stop_token token,
373		std::error_code* ec,
374		std::size_t* bytes)
375		{
376	188x	auto* op = acquire_op(free_reads_, live_reads_);
377	188x	op->prepare(
378		cont, ex, ec, bytes, fd_, static_cast<std::int64_t>(user_offset),
379	376x	sched_, detail::object_ref(this), buffers, token);
380	188x	sched_->work_started();
381
382		// Closed-object contract outranks the zero-length no-op.
383	188x	if (fd_ < 0)
384		{
385	2x	op->empty_buffer = false;
386	2x	op->res = -EBADF;
387	2x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
388	2x	sched_->push_completed_locked(op);
389	2x	return std::noop_coroutine();
390	2x	}
391
392	371x	if (op->empty_buffer \|\|
393	185x	op->cancelled.load(std::memory_order_acquire))
394		{
395	1x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
396	1x	sched_->push_completed_locked(op);
397	1x	return std::noop_coroutine();
398	1x	}
399
400	185x	uring_submit_op(*sched_, op);
401	185x	return std::noop_coroutine();
402		}
403
404		inline std::coroutine_handle<>
405	61x	uring_random_access_file::write_some_at(
406		std::uint64_t user_offset,
407		capy::continuation& cont,
408		capy::executor_ref ex,
409		buffer_param buffers,
410		std::stop_token token,
411		std::error_code* ec,
412		std::size_t* bytes)
413		{
414	61x	auto* op = acquire_op(free_writes_, live_writes_);
415	61x	op->prepare(
416		cont, ex, ec, bytes, fd_, static_cast<std::int64_t>(user_offset),
417	122x	sched_, detail::object_ref(this), buffers, token);
418	61x	sched_->work_started();
419
420		// Closed-object contract outranks the zero-length no-op.
421	61x	if (fd_ < 0)
422		{
423	1x	op->empty_buffer = false;
424	1x	op->res = -EBADF;
425	1x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
426	1x	sched_->push_completed_locked(op);
427	1x	return std::noop_coroutine();
428	1x	}
429
430	119x	if (op->empty_buffer \|\|
431	59x	op->cancelled.load(std::memory_order_acquire))
432		{
433	1x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
434	1x	sched_->push_completed_locked(op);
435	1x	return std::noop_coroutine();
436	1x	}
437
438	59x	uring_submit_op(*sched_, op);
439	59x	return std::noop_coroutine();
440		}
441
442		/** Native io_uring random-access-file service.
443
444		Owns all `uring_random_access_file` impls. Replaces
445		`posix_random_access_file_service` for the io_uring backend;
446		registered under the abstract `random_access_file_service` key
447		by `uring_t::construct`.
448		*/
449		class BOOST_COROSIO_DECL uring_random_access_file_service final
450		: public uring_file_service_base<
451		uring_random_access_file_service,
452		random_access_file_service,
453		uring_random_access_file>
454		{
455		using base_service = uring_file_service_base<
456		uring_random_access_file_service,
457		random_access_file_service,
458		uring_random_access_file>;
459
460		public:
461	86x	explicit uring_random_access_file_service(
462		capy::execution_context& ctx)
463	86x	: base_service(ctx.use_service<uring_scheduler>())
464		{
465	86x	}
466
467		// construct / destroy / close / shutdown / scheduler() are inherited
468		// from uring_file_service_base.
469
470	106x	std::error_code open_file(
471		random_access_file::implementation& impl,
472		std::filesystem::path const& path,
473		file_base::flags mode) override
474		{
475	106x	return static_cast<uring_random_access_file&>(impl).open_file(
476	106x	path, mode);
477		}
478		};
479
480		inline void
481	115x	uring_random_access_file::retire() noexcept
482		{
483	115x	svc_->pool_.recycle(this);
484	115x	}
485
486		} // namespace boost::corosio::detail
487
488		#endif // BOOST_COROSIO_HAS_URING
489
490		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_RANDOM_ACCESS_FILE_HPP
491