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

93.0% Lines (199 / 214) 100.0% Functions (29 / 29)

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

uring_descriptor.hpp

f(x) Functions (29)

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_URING_URING_DESCRIPTOR_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_DESCRIPTOR_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_URING
16
17		#include <boost/corosio/posix_stream_descriptor.hpp>
18		#include <boost/corosio/wait_type.hpp>
19		#include <boost/corosio/detail/intrusive.hpp>
20		#include <boost/corosio/native/detail/uring/uring_file_ops.hpp>
21		#include <boost/corosio/native/detail/uring/uring_scheduler.hpp>
22		#include <boost/corosio/native/detail/uring/uring_socket_ops.hpp>
23		#include <boost/corosio/native/detail/validate_fd.hpp>
24
25		#include <atomic>
26		#include <coroutine>
27		#include <cstddef>
28		#include <cstdint>
29		#include <memory>
30		#include <system_error>
31
32		#include <errno.h>
33		#include <poll.h>
34		#include <sys/epoll.h>
35		#include <unistd.h>
36
37		/* io_uring-backed implementation of posix_stream_descriptor.
38
39		Three things differ from the reactor backends and from the other
40		io_uring services:
41
42		Transfers submit READV/WRITEV at offset -1 so the kernel uses (and
43		advances) the descriptor's own file position. The file services
44		pass a real offset; a pipe, tty or character device has none.
45
46		The descriptor's flags are never modified, as in asio. A transfer
47		on a blocking fd the kernel can poll parks in its internal poll and
48		ASYNC_CANCEL removes it. One the kernel punts to an io-wq worker
49		(no poll support, or no FMODE_NOWAIT) holds that worker; the cancel
50		interrupts it only if the driver's wait is interruptible, and the
51		resulting -EINTR is reported as canceled. A caller who made the fd
52		non-blocking gets EAGAIN completions, which the two-phase shape
53		below turns into a poll and a retry. A file with no poll support
54		is ready to every poll, so assign probes for it (fd_is_pollable)
55		and its EAGAIN fails with EOPNOTSUPP instead, as on the reactors.
56
57		An O_NONBLOCK descriptor the kernel cannot retry internally
58		completes with -EAGAIN; the op then re-arms itself as a poll_add on
59		the same descriptor and re-submits the transfer when the poll says
60		ready. The handler makes that decision before coro_drain_if_shutdown,
61		which disarms stop_cb: an op going round again keeps its
62		cancellation wiring, so a stop_token firing between phases still
63		reaches the kernel.
64
65		The gap between a CQE and its dispatch is the whole difficulty of
66		that shape, and one epoch counter closes it. Nothing of a transfer
67		waiting in that gap is in the ring, so a kernel cancel cannot reach
68		it. cancel() and every descriptor change bump epoch_ before they
69		take ring_mutex_, and do_prep, which runs under ring_mutex_,
70		compares the op's snapshot against it: an op that no longer
71		matches preps a NOP instead of a transfer and completes canceled.
72		Either the prep sees the bump, or the cancel's SQE queues behind
73		the transfer's and finds it. The epoch is also what makes the
74		staleness check exact rather than heuristic: a closed fd number
75		the next assign() gets back would satisfy a bare fd comparison.
76
77		There is no adopt-time registration. assign() validates and takes
78		the descriptor; a kernel that refuses it says so at the first
79		operation, as the public docstring promises.
80		*/
81
82		namespace boost::corosio::detail {
83
84		class uring_descriptor;
85
86		/** Advance a two-phase transfer op, or report that it is finished.
87
88		A kernel `EAGAIN` becomes a `poll_add` on the same descriptor, and
89		the poll's completion re-submits the transfer. Every path that
90		stops instead leaves @a op carrying a result the completion decode
91		can read as terminal.
92
93		@param op The op whose CQE just arrived.
94		@return True when a fresh SQE was submitted, in which case the
95		caller must neither complete nor disarm the op.
96		*/
97		template<class Op>
98		bool uring_descriptor_continue(Op& op) noexcept;
99
100		/// True when @a op no longer belongs to its owner's current intent.
101		/// Called from do_prep, under ring_mutex_.
102		template<class Op>
103		bool uring_descriptor_abandoned(Op const& op) noexcept;
104
105		/** Scatter read via `IORING_OP_READV` at the descriptor's own offset.
106
107		@see uring_descriptor_continue for the `polling` phase.
108		*/
109		struct uring_descriptor_read_op final : uring_file_read_op_base
110		{
111		uring_descriptor* desc = nullptr;
112		/// True while the submitted SQE is the readiness poll, not the read.
113		bool polling = false;
114		/// Owner epoch snapshotted by arm_slot; see uring_descriptor_continue.
115		std::uint32_t epoch = 0;
116		/// Set by do_prep when the op was abandoned before its SQE was built.
117		bool abandoned = false;
118
119	92x	uring_descriptor_read_op() noexcept : uring_file_read_op_base(&do_handler)
120		{
121	92x	prep_func = &do_prep;
122	92x	}
123
124	64x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
125		{
126	64x	auto* self = static_cast<uring_descriptor_read_op*>(base);
127		// Decided here because every cancel path records its intent
128		// before taking ring_mutex_, which this runs under: either the
129		// intent is visible now, or the cancel SQE queues behind ours.
130	64x	if (uring_descriptor_abandoned(*self))
131		{
132	1x	self->abandoned = true;
133	1x	::io_uring_prep_nop(sqe);
134	1x	return;
135		}
136	63x	if (self->polling)
137	1x	::io_uring_prep_poll_add(sqe, self->fd, POLLIN);
138		else
139	62x	uring_file_read_op_base::do_prep(base, sqe);
140		}
141
142		static void do_handler(
143		void* owner,
144		scheduler_op* base,
145		std::uint32_t bytes,
146		std::uint32_t error) noexcept;
147		};
148
149		/// Gather write via `IORING_OP_WRITEV` at the descriptor's own offset.
150		struct uring_descriptor_write_op final : uring_file_write_op_base
151		{
152		uring_descriptor* desc = nullptr;
153		/// True while the submitted SQE is the readiness poll, not the write.
154		bool polling = false;
155		/// Owner epoch snapshotted by arm_slot; see uring_descriptor_continue.
156		std::uint32_t epoch = 0;
157		/// Set by do_prep when the op was abandoned before its SQE was built.
158		bool abandoned = false;
159
160	83x	uring_descriptor_write_op() noexcept : uring_file_write_op_base(&do_handler)
161		{
162	83x	prep_func = &do_prep;
163	83x	}
164
165	12x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
166		{
167	12x	auto* self = static_cast<uring_descriptor_write_op*>(base);
168		// Decided here because every cancel path records its intent
169		// before taking ring_mutex_, which this runs under: either the
170		// intent is visible now, or the cancel SQE queues behind ours.
171	12x	if (uring_descriptor_abandoned(*self))
172		{
173	1x	self->abandoned = true;
174	1x	::io_uring_prep_nop(sqe);
175	1x	return;
176		}
177	11x	if (self->polling)
178	✗	::io_uring_prep_poll_add(sqe, self->fd, POLLOUT);
179		else
180	11x	uring_file_write_op_base::do_prep(base, sqe);
181		}
182
183		static void do_handler(
184		void* owner,
185		scheduler_op* base,
186		std::uint32_t bytes,
187		std::uint32_t error) noexcept;
188		};
189
190		/** Native io_uring implementation of @ref posix_stream_descriptor.
191
192		Holds the adopted descriptor and the five embedded op slots: one
193		transfer per direction, and one wait per direction.
194
195		@par Thread Safety
196		Distinct objects: Safe.@n
197		Shared objects: Unsafe. Each slot carries a single pending
198		operation, so a descriptor must not have two operations of the
199		same kind in flight.
200		*/
201		/** Return whether the kernel can poll @p fd.
202
203		epoll refuses a file with no poll support with EPERM; io_uring
204		instead reports such a file ready to every poll. Nothing is left
205		registered and @p fd is not modified. When the probe itself fails,
206		the descriptor is assumed pollable.
207		*/
208		inline bool
209	95x	fd_is_pollable(int fd) noexcept
210		{
211	95x	int ep = ::epoll_create1(EPOLL_CLOEXEC);
212	95x	if (ep < 0)
213	✗	return true;
214	95x	::epoll_event ev{};
215		bool const pollable =
216	95x	::epoll_ctl(ep, EPOLL_CTL_ADD, fd, &ev) == 0 \|\| errno != EPERM;
217	95x	::close(ep);
218	95x	return pollable;
219		}
220
221		class uring_descriptor_service;
222
223		class BOOST_COROSIO_DECL uring_descriptor final
224		: public posix_stream_descriptor::implementation
225		, public intrusive_list<uring_descriptor>::node
226		{
227		friend class uring_descriptor_service;
228
229		uring_descriptor_service* svc_ = nullptr;
230		uring_scheduler* sched_ = nullptr;
231		int fd_ = -1;
232		bool pollable_ = true;
233
234		// Bumped by cancel() and by every descriptor change. A transfer op
235		// between its EAGAIN CQE and its dispatch is invisible to the ring;
236		// its snapshot of this is what stops it re-arming.
237		std::atomic<std::uint32_t> epoch_{0};
238
239		uring_descriptor_read_op rd_;
240		uring_descriptor_write_op wr_;
241		uring_wait_op wait_rd_;
242		uring_wait_op wait_wr_;
243		uring_wait_op wait_er_;
244
245		public:
246	80x	uring_descriptor(
247		uring_descriptor_service& svc, uring_scheduler& sched) noexcept
248	160x	: svc_(&svc)
249	80x	, sched_(&sched)
250		{
251	80x	}
252
253	160x	~uring_descriptor() override
254	80x	{
255	80x	close_descriptor();
256	160x	}
257
258		/// Recycle into the owning service's pool. Defined out-of-line
259		/// after uring_descriptor_service for its complete type.
260		void retire() noexcept override;
261
262		/** Assert the closed state a recycled descriptor starts from.
263
264		`close_descriptor()` already drove fd_ to its closed value
265		before the refcount reached zero; each op's own `prepare()`
266		overwrites its state before the next use.
267
268		@pre refs_ == 0, fd closed, no op in flight.
269		*/
270	33x	void reuse() noexcept
271		{
272	33x	BOOST_COROSIO_ASSERT(fd_ == -1);
273	33x	BOOST_COROSIO_ASSERT(!rd_.stop_cb);
274	33x	BOOST_COROSIO_ASSERT(!wr_.stop_cb);
275	33x	BOOST_COROSIO_ASSERT(!wait_rd_.stop_cb);
276	33x	BOOST_COROSIO_ASSERT(!wait_wr_.stop_cb);
277	33x	BOOST_COROSIO_ASSERT(!wait_er_.stop_cb);
278	33x	}
279
280		// -- io_stream::implementation --
281
282	62x	std::coroutine_handle<> read_some(
283		capy::continuation& cont,
284		capy::executor_ref ex,
285		buffer_param buffers,
286		std::stop_token token,
287		std::error_code* ec,
288		std::size_t* bytes) override
289		{
290	62x	rd_.prepare(
291		cont, ex, ec, bytes, fd_, /file_offset=/-1, sched_,
292	124x	detail::object_ref(this), buffers, token);
293	62x	arm_slot(rd_);
294	62x	sched_->work_started();
295
296		// Closed-object contract outranks the zero-length no-op.
297	62x	if (fd_ < 0)
298		{
299	✗	rd_.empty_buffer = false;
300	✗	rd_.res = -EBADF;
301	✗	push_completed(&rd_);
302	✗	return std::noop_coroutine();
303		}
304
305	62x	if (rd_.empty_buffer \|\| rd_.cancelled.load(std::memory_order_acquire))
306		{
307	1x	push_completed(&rd_);
308	1x	return std::noop_coroutine();
309		}
310
311	61x	uring_submit_op(*sched_, &rd_);
312	61x	return std::noop_coroutine();
313		}
314
315	11x	std::coroutine_handle<> write_some(
316		capy::continuation& cont,
317		capy::executor_ref ex,
318		buffer_param buffers,
319		std::stop_token token,
320		std::error_code* ec,
321		std::size_t* bytes) override
322		{
323	11x	wr_.prepare(
324		cont, ex, ec, bytes, fd_, /file_offset=/-1, sched_,
325	22x	detail::object_ref(this), buffers, token);
326	11x	arm_slot(wr_);
327	11x	sched_->work_started();
328
329	11x	if (fd_ < 0)
330		{
331	✗	wr_.empty_buffer = false;
332	✗	wr_.res = -EBADF;
333	✗	push_completed(&wr_);
334	✗	return std::noop_coroutine();
335		}
336
337	11x	if (wr_.empty_buffer \|\| wr_.cancelled.load(std::memory_order_acquire))
338		{
339	✗	push_completed(&wr_);
340	✗	return std::noop_coroutine();
341		}
342
343	11x	uring_submit_op(*sched_, &wr_);
344	11x	return std::noop_coroutine();
345		}
346
347		// -- posix_stream_descriptor::implementation --
348
349	15x	std::coroutine_handle<> wait(
350		capy::continuation& cont,
351		capy::executor_ref ex,
352		wait_type w,
353		std::stop_token token,
354		std::error_code* ec) override
355		{
356	15x	uring_wait_op* op = nullptr;
357	15x	int poll_flags = 0;
358	15x	switch (w)
359		{
360	7x	case wait_type::read:
361	7x	op = &wait_rd_;
362	7x	poll_flags = POLLIN;
363	7x	break;
364	3x	case wait_type::write:
365	3x	op = &wait_wr_;
366	3x	poll_flags = POLLOUT;
367	3x	break;
368	5x	case wait_type::error:
369	5x	op = &wait_er_;
370		// POLLERR, POLLHUP and POLLNVAL are reported whether or not
371		// they are asked for, so the error wait names only POLLPRI.
372	5x	poll_flags = POLLPRI;
373	5x	break;
374		}
375
376	15x	op->prepare(
377	30x	cont, ex, ec, fd_, sched_, detail::object_ref(this), poll_flags,
378		token);
379	15x	sched_->work_started();
380
381	15x	if (fd_ < 0)
382		{
383	1x	op->res = -EBADF;
384	1x	push_completed(op);
385	1x	return std::noop_coroutine();
386		}
387
388		// Without poll support there is no error condition to watch;
389		// the kernel would refuse the poll with EINVAL.
390	14x	if (w == wait_type::error && !pollable_)
391		{
392	1x	op->res = -EOPNOTSUPP;
393	1x	push_completed(op);
394	1x	return std::noop_coroutine();
395		}
396
397	13x	if (op->cancelled.load(std::memory_order_acquire))
398		{
399	✗	push_completed(op);
400	✗	return std::noop_coroutine();
401		}
402
403	13x	uring_submit_op(*sched_, op);
404	13x	return std::noop_coroutine();
405		}
406
407	238x	native_handle_type native_handle() const noexcept override
408		{
409	238x	return fd_;
410		}
411
412	3x	native_handle_type release_descriptor() noexcept override
413		{
414		// Bump before the flush, as close_descriptor does. Do NOT
415		// close -- the caller takes ownership.
416	3x	epoch_.fetch_add(1, std::memory_order_release);
417	3x	cancel_waits();
418	3x	if (fd_ >= 0)
419	3x	sched_->flush_after_cancel(
420		fd_, {&rd_, &wr_, &wait_rd_, &wait_wr_, &wait_er_});
421	3x	native_handle_type released = fd_;
422	3x	fd_ = -1;
423	3x	return released;
424		}
425
426	25x	void cancel() noexcept override
427		{
428		// Bump before the SQE: the cancel reaches only what the ring
429		// currently holds, and an op waiting for its handler to run
430		// holds nothing there. The epoch is what that op consults.
431	25x	epoch_.fetch_add(1, std::memory_order_release);
432	25x	if (fd_ >= 0)
433	19x	sched_->cancel_ops({&rd_, &wr_, &wait_rd_, &wait_wr_, &wait_er_});
434	25x	}
435
436		/// Epoch bumped by every @ref cancel and every descriptor change.
437	100x	std::uint32_t epoch() const noexcept
438		{
439	200x	return epoch_.load(std::memory_order_acquire);
440		}
441
442		// -- Service-facing (non-virtual) --
443
444		/** Adopt an already-validated descriptor.
445
446		@param fd The descriptor to adopt.
447		@param pollable Whether the kernel can poll @p fd; see
448		@ref fd_is_pollable.
449		*/
450	98x	void set_descriptor(int fd, bool pollable = true) noexcept
451		{
452	98x	epoch_.fetch_add(1, std::memory_order_release);
453	98x	fd_ = fd;
454	98x	pollable_ = pollable;
455	98x	}
456
457		/// Whether the kernel can poll the held descriptor.
458	2x	bool pollable() const noexcept
459		{
460	2x	return pollable_;
461		}
462
463		/// Teardown hook named by uring_file_service_base.
464	304x	void close_file() noexcept
465		{
466	304x	close_descriptor();
467	304x	}
468
469		/// Cancel pending operations and close the descriptor. No-op when
470		/// already closed.
471	385x	void close_descriptor() noexcept
472		{
473	385x	if (fd_ < 0)
474	292x	return;
475		// Bump before the flush: an op prepping concurrently must see
476		// the change, or its SQE could land on a recycled fd number.
477	93x	epoch_.fetch_add(1, std::memory_order_release);
478	93x	cancel_waits();
479		// Both kernel entries below can run a queued pipe write as task
480		// work; with the reader already gone that raises SIGPIPE.
481	93x	scoped_sigpipe_block no_sigpipe;
482	93x	sched_->flush_after_cancel(
483		fd_, {&rd_, &wr_, &wait_rd_, &wait_wr_, &wait_er_});
484	93x	::close(fd_);
485	93x	fd_ = -1;
486	93x	}
487
488		private:
489		// A wait completing after a close or release must not probe the
490		// old number for SO_ERROR; see uring_wait_op.
491	96x	void cancel_waits() noexcept
492		{
493	96x	wait_rd_.request_cancel();
494	96x	wait_wr_.request_cancel();
495	96x	wait_er_.request_cancel();
496	96x	}
497
498		/** Bind a transfer slot to this descriptor for a fresh submission.
499
500		The epoch snapshot taken here is what every later prep compares
501		against; see uring_descriptor_continue.
502		*/
503		template<class Op>
504	73x	void arm_slot(Op& op) noexcept
505		{
506	73x	op.desc = this;
507	73x	op.polling = false;
508	73x	op.abandoned = false;
509	73x	op.epoch = epoch_.load(std::memory_order_acquire);
510	73x	}
511
512		/// Queue an already-counted op for the next dispatch cycle.
513	3x	void push_completed(scheduler_op* op) noexcept
514		{
515	3x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
516	3x	sched_->push_completed_locked(op);
517	3x	}
518		};
519
520		// --- Deferred implementations (need uring_descriptor complete) ---
521
522		template<class Op>
523		bool
524	86x	uring_descriptor_abandoned(Op const& op) noexcept
525		{
526	171x	return op.cancelled.load(std::memory_order_acquire) \|\|
527	171x	op.desc->epoch() != op.epoch;
528		}
529
530		template<class Op>
531		bool
532	83x	uring_descriptor_continue(Op& op) noexcept
533		{
534		// A NOP completed: the op was abandoned at prep.
535	83x	if (op.abandoned)
536		{
537	1x	op.res = -ECANCELED;
538	1x	return false;
539		}
540
541		// A failed poll, a transfer error or a byte count is the operation's
542		// answer, and bytes beat a cancel that landed after them.
543	164x	bool const rearm = op.polling
544	158x	? op.res >= 0
545	76x	: (op.res == -EAGAIN \|\| op.res == -EWOULDBLOCK);
546	82x	if (!rearm)
547		{
548		// A poll the full SQ never took comes back as -EAGAIN, and a
549		// transfer the cancel interrupted in an io-wq worker as -EINTR;
550		// either one abandoned meanwhile is owed canceled, as above.
551	75x	bool const interrupted =
552	75x	op.polling ? op.res == -EAGAIN : op.res == -EINTR;
553	75x	if (interrupted && uring_descriptor_abandoned(op))
554	2x	op.res = -ECANCELED;
555	75x	return false;
556		}
557
558		// Abandoned since the CQE arrived: the caller is owed canceled,
559		// never the poll's revents or the transfer's EAGAIN.
560	7x	if (uring_descriptor_abandoned(op))
561		{
562	4x	op.res = -ECANCELED;
563	4x	return false;
564		}
565
566		// A file with no poll support is ready to every poll, so arming
567		// one would retry the refused transfer on a CPU forever. The
568		// reactors report the same refusal.
569	3x	if (!op.polling && !op.desc->pollable())
570		{
571	1x	op.res = -EOPNOTSUPP;
572	1x	return false;
573		}
574	2x	op.polling = !op.polling;
575
576		// do_one spends a work_finished() on every op it dispatches, so an
577		// op going round again has to be counted again. Nothing may touch
578		// op after the submit: another thread can complete and free it.
579	2x	op.sched_->work_started();
580	2x	uring_submit_op(*op.sched_, &op);
581	2x	return true;
582		}
583
584		inline void
585	64x	uring_descriptor_read_op::do_handler(
586		void* owner,
587		scheduler_op* base,
588		std::uint32_t /bytes/,
589		std::uint32_t /error/) noexcept
590		{
591	64x	auto* self = static_cast<uring_descriptor_read_op*>(base);
592	64x	if (owner != nullptr && uring_descriptor_continue(*self))
593	1x	return;
594
595	63x	if (coro_drain_if_shutdown(owner, self))
596	3x	return;
597
598	60x	if (self->sched_)
599	60x	self->sched_->reset_inline_budget();
600
601	60x	uring_set_result(self, /is_read=/true, self->empty_buffer);
602	60x	if (self->bytes_out)
603	60x	*self->bytes_out =
604	60x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
605	60x	coro_resume(self);
606		}
607
608		inline void
609	11x	uring_descriptor_write_op::do_handler(
610		void* owner,
611		scheduler_op* base,
612		std::uint32_t /bytes/,
613		std::uint32_t /error/) noexcept
614		{
615	11x	auto* self = static_cast<uring_descriptor_write_op*>(base);
616	11x	if (owner != nullptr && uring_descriptor_continue(*self))
617	✗	return;
618
619	11x	if (coro_drain_if_shutdown(owner, self))
620	3x	return;
621
622	8x	if (self->sched_)
623	8x	self->sched_->reset_inline_budget();
624
625	8x	uring_set_result(self, /is_read=/false, self->empty_buffer);
626	8x	if (self->bytes_out)
627	8x	*self->bytes_out =
628	8x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
629	8x	coro_resume(self);
630		}
631
632		} // namespace boost::corosio::detail
633
634		#endif // BOOST_COROSIO_HAS_URING
635
636		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_DESCRIPTOR_HPP
637