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

99.3% Lines (287 / 289) 97.1% Functions (34 / 35)

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

uring_socket_ops.hpp

f(x) Functions (35)

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_SOCKET_OPS_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_SOCKET_OPS_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_URING
17
18		#include <liburing.h>
19
20		#include <boost/capy/buffers.hpp>
21		#include <boost/capy/error.hpp>
22		#include <boost/corosio/detail/buffer_param.hpp>
23		#include <boost/corosio/detail/dispatch_coro.hpp>
24		#include <boost/corosio/local_endpoint.hpp>
25		#include <boost/corosio/native/detail/uring/uring_buffer.hpp>
26		#include <boost/corosio/native/detail/uring/uring_op.hpp>
27		#include <boost/corosio/native/detail/uring/uring_scheduler.hpp>
28		#include <boost/corosio/native/detail/coro_op_complete.hpp>
29		#include <boost/corosio/native/detail/make_err.hpp>
30		#include <boost/corosio/native/detail/speculative_state.hpp>
31
32		#include <atomic>
33		#include <system_error>
34
35		#include <errno.h>
36		#include <netinet/in.h>
37		#include <poll.h>
38		#include <sys/socket.h>
39		#include <sys/uio.h>
40
41		namespace boost::corosio::detail {
42
43		/** Read directly from a socket, retrying on `EINTR`.
44
45		A single buffer uses `recv`, which skips the VFS layer and the
46		iovec copy-in that `readv` pays on every call.
47
48		@return The byte count, or -1 with `errno` set.
49		*/
50		inline ssize_t
51	158809x	uring_direct_read(int fd, iovec* iovecs, int count) noexcept
52		{
53		ssize_t n;
54		do
55		{
56	158809x	n = count == 1 ? ::recv(fd, iovecs[0].iov_base, iovecs[0].iov_len, 0)
57	3x	: ::readv(fd, iovecs, count);
58		}
59	158809x	while (n < 0 && errno == EINTR);
60	158809x	return n;
61		}
62
63		/** Write directly to a socket, retrying on `EINTR`.
64
65		A single buffer uses `send`, which skips the msghdr and iovec
66		copy-in that `sendmsg` pays on every call.
67
68		@return The byte count, or -1 with `errno` set.
69		*/
70		inline ssize_t
71	158079x	uring_direct_write(int fd, iovec* iovecs, int count) noexcept
72		{
73		ssize_t n;
74		do
75		{
76	158079x	if (count == 1)
77		{
78	158076x	n = ::send(fd, iovecs[0].iov_base, iovecs[0].iov_len, MSG_NOSIGNAL);
79		}
80		else
81		{
82	3x	msghdr msg{};
83	3x	msg.msg_iov = iovecs;
84	3x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(count);
85	3x	n = ::sendmsg(fd, &msg, MSG_NOSIGNAL);
86		}
87		}
88	158079x	while (n < 0 && errno == EINTR);
89	158079x	return n;
90		}
91
92		/** Resolution state of a TCP socket's cached local endpoint.
93
94		@li `unresolved`: never set; the accessor returns the default
95		endpoint (open-but-unbound socket, failed connect).
96		@li `lazy_pending`: the socket has an authoritative local endpoint
97		that has not been fetched; the accessor calls `getsockname` on
98		first read.
99		@li `resolved`: the cached endpoint is authoritative.
100		*/
101		enum class uring_endpoint_state : int
102		{
103		unresolved,
104		lazy_pending,
105		resolved
106		};
107
108		/// Maximum scatter/gather segments per read/write/dgram op.
109		///
110		/// Bounded well below `IOV_MAX` (1024 on Linux) so each op's
111		/// `iovec[uring_max_iov]` lives inside the uring_op object on
112		/// the same allocation as the rest of its state. Plan 4's registered-
113		/// buffer work will revisit; until then 16 covers typical scatter use
114		/// cases (fragmented buffers from buffer_sequence) without bloating
115		/// per-op memory.
116		inline constexpr std::size_t uring_max_iov = 16;
117
118		/** Fill an `iovec` array from a buffer sequence without type-punning.
119
120		`buffer_param::copy_to` writes `capy::mutable_buffer` objects. Those
121		cannot legally alias `iovec` storage: no `mutable_buffer` lives at
122		that address, and `mutable_buffer` is not an implicit-lifetime type,
123		so its lifetime cannot be started in place (which also rules out
124		`std::start_lifetime_as_array`). We copy into a real `mutable_buffer`
125		scratch array, then translate field by field into the caller's
126		`iovec` array. Zero-size buffers are already skipped by `copy_to`.
127
128		@param buffers The source buffer sequence.
129		@param iovecs Destination array, filled with the non-zero buffers.
130		@return The number of `iovec` entries written.
131		*/
132		inline int
133	338199x	copy_to_iovec(
134		buffer_param const& buffers, iovec (&iovecs)[uring_max_iov]) noexcept
135		{
136	338199x	capy::mutable_buffer bufs[uring_max_iov];
137	338199x	std::size_t const n = buffers.copy_to(bufs, uring_max_iov);
138	676120x	for (std::size_t i = 0; i < n; ++i)
139		{
140	337921x	iovecs[i].iov_base = bufs[i].data();
141	337921x	iovecs[i].iov_len = bufs[i].size();
142		}
143	338199x	return static_cast<int>(n);
144		}
145
146		/** Resolve ec_out/bytes_out from a CQE result for a completed I/O op.
147
148		Shared by read, write, and connect handlers. For reads, `res == 0`
149		with a non-empty buffer means the peer closed the connection (EOF).
150
151		@param self The completed op.
152		@param is_read True if this is a receive/read operation.
153		@param empty_buf True if the submitted buffer was zero-length.
154		*/
155		inline void
156	23132x	uring_set_result(uring_op* self, bool is_read, bool empty_buf) noexcept
157		{
158	45971x	decode_results(
159		self->ec_out, self->bytes_out,
160	23132x	self->cancelled.load(std::memory_order_acquire),
161	293x	self->res < 0 ? make_err(-self->res) : std::error_code{}, is_read,
162	23132x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u, empty_buf);
163	23132x	}
164
165		/** Scatter-gather read via `IORING_OP_READV`.
166
167		@par Handler dispatch
168		do_cqe captures `res`/`cqe_flags` and queues self into `local`;
169		do_handler runs from the scheduler queue and resumes the coroutine.
170		*/
171		struct uring_read_op : uring_op
172		{
173		iovec iovecs[uring_max_iov];
174		int iovec_count = 0;
175		int fd = -1;
176		detail::speculative_state* spec_state = nullptr;
177
178	2074x	uring_read_op() noexcept : uring_op(&do_handler, &do_cqe, &do_prep)
179		{
180	2074x	is_read = true;
181	2074x	}
182
183		/** Reset and initialize for a new submission.
184
185		Embedded ops are reused across calls; every mutable field the
186		handler may read must be re-initialized here. `start(token)`
187		also resets `cancelled`, `sqe_set`, and `stop_cb`.
188
189		@pre This slot has no in-flight op (its prior op completed).
190		*/
191	10047x	void prepare(
192		capy::continuation& awaiting,
193		capy::executor_ref executor,
194		std::error_code* ec,
195		std::size_t* bytes,
196		int file_descriptor,
197		uring_scheduler* scheduler,
198		detail::object_ref impl,
199		detail::speculative_state* spec,
200		buffer_param buffers,
201		std::stop_token const& token) noexcept
202		{
203	10047x	cont = &awaiting;
204	10047x	ex = executor;
205	10047x	ec_out = ec;
206	10047x	bytes_out = bytes;
207	10047x	fd = file_descriptor;
208	10047x	sched_ = scheduler;
209	10047x	object_ref_ = std::move(impl);
210	10047x	spec_state = spec;
211	10047x	res = 0;
212	10047x	cqe_flags = 0;
213	10047x	iovec_count = copy_to_iovec(buffers, iovecs);
214	10047x	empty_buffer = (iovec_count == 0);
215	10047x	start(token);
216	10047x	}
217
218	761x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
219		{
220	761x	auto* self = static_cast<uring_read_op*>(base);
221		// Single-buffer fast path: IORING_OP_RECV with a flat
222		// (buffer, length) skips the iovec-array indirection that
223		// IORING_OP_READV pays. For multi-iovec scatter reads, fall
224		// back to readv.
225	761x	if (self->iovec_count == 1)
226		{
227	760x	::io_uring_prep_recv(
228		sqe, self->fd, self->iovecs[0].iov_base,
229		self->iovecs[0].iov_len, 0);
230		}
231		else
232		{
233	1x	::io_uring_prep_readv(
234	1x	sqe, self->fd, self->iovecs, self->iovec_count, 0);
235		}
236	761x	}
237
238		static void
239	761x	do_cqe(uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
240		{
241	761x	auto* self = static_cast<uring_read_op*>(base);
242	761x	self->res = res;
243	761x	self->cqe_flags = flags;
244	761x	local.push(self);
245	761x	}
246
247	10047x	static void do_handler(
248		void* owner,
249		scheduler_op* base,
250		std::uint32_t /bytes/,
251		std::uint32_t /error/) noexcept
252		{
253	10047x	auto* self = static_cast<uring_read_op*>(base);
254	10047x	if (coro_drain_if_shutdown(owner, self))
255	212x	return;
256
257	9835x	if (self->sched_)
258	9835x	self->sched_->reset_inline_budget();
259
260	9835x	uring_set_result(self, true, self->empty_buffer);
261
262	9835x	if (self->res > 0 && self->spec_state)
263		{
264		// Kernel signalled readiness — restore speculation.
265	9647x	self->spec_state->on_async_read_ready();
266		}
267
268	9835x	coro_resume(self);
269		// suicide drops here; may destroy impl + self.
270		}
271		};
272
273		/** Scatter-gather write via `IORING_OP_SENDMSG` with `MSG_NOSIGNAL`.
274
275		`MSG_NOSIGNAL` prevents `SIGPIPE` when the peer has closed the
276		connection; the error is surfaced as `EPIPE` instead.
277		*/
278		struct uring_write_op : uring_op
279		{
280		iovec iovecs[uring_max_iov];
281		int iovec_count = 0;
282		int fd = -1;
283		msghdr msg{};
284		detail::speculative_state* spec_state = nullptr;
285
286	2074x	uring_write_op() noexcept : uring_op(&do_handler, &do_cqe, &do_prep) {}
287
288		/** Reset and initialize for a new submission. See uring_read_op::prepare. */
289	9534x	void prepare(
290		capy::continuation& awaiting,
291		capy::executor_ref executor,
292		std::error_code* ec,
293		std::size_t* bytes,
294		int file_descriptor,
295		uring_scheduler* scheduler,
296		detail::object_ref impl,
297		detail::speculative_state* spec,
298		buffer_param buffers,
299		std::stop_token const& token) noexcept
300		{
301	9534x	cont = &awaiting;
302	9534x	ex = executor;
303	9534x	ec_out = ec;
304	9534x	bytes_out = bytes;
305	9534x	fd = file_descriptor;
306	9534x	sched_ = scheduler;
307	9534x	object_ref_ = std::move(impl);
308	9534x	spec_state = spec;
309	9534x	res = 0;
310	9534x	cqe_flags = 0;
311	9534x	iovec_count = copy_to_iovec(buffers, iovecs);
312	9534x	empty_buffer = (iovec_count == 0);
313	9534x	if (!empty_buffer)
314		{
315	9534x	msg = {};
316	9534x	msg.msg_iov = iovecs;
317	9534x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(iovec_count);
318		}
319	9534x	start(token);
320	9534x	}
321
322	251x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
323		{
324	251x	auto* self = static_cast<uring_write_op*>(base);
325		// Single-buffer fast path: IORING_OP_SEND with MSG_NOSIGNAL
326		// skips the msghdr indirection that IORING_OP_SENDMSG pays.
327		// For multi-iovec scatter writes, fall back to sendmsg.
328	251x	if (self->iovec_count == 1)
329		{
330	250x	::io_uring_prep_send(
331	250x	sqe, self->fd, self->iovecs[0].iov_base,
332		self->iovecs[0].iov_len, MSG_NOSIGNAL);
333		}
334		else
335		{
336	1x	::io_uring_prep_sendmsg(sqe, self->fd, &self->msg, MSG_NOSIGNAL);
337		}
338	251x	}
339
340		static void
341	251x	do_cqe(uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
342		{
343	251x	auto* self = static_cast<uring_write_op*>(base);
344	251x	self->res = res;
345	251x	self->cqe_flags = flags;
346	251x	local.push(self);
347	251x	}
348
349	9534x	static void do_handler(
350		void* owner,
351		scheduler_op* base,
352		std::uint32_t /bytes/,
353		std::uint32_t /error/) noexcept
354		{
355	9534x	auto* self = static_cast<uring_write_op*>(base);
356	9534x	if (coro_drain_if_shutdown(owner, self))
357	3x	return;
358
359	9531x	if (self->sched_)
360	9531x	self->sched_->reset_inline_budget();
361
362	9531x	uring_set_result(self, false, self->empty_buffer);
363
364	9531x	if (self->res > 0 && self->spec_state)
365		{
366		// Kernel signalled readiness — restore speculation.
367	9528x	self->spec_state->on_async_write_ready();
368		}
369
370	9531x	coro_resume(self);
371		}
372		};
373
374		/** Non-blocking connect via `IORING_OP_CONNECT`.
375
376		Negative `res` is the connect error; zero means success.
377		`remote_endpoint_out` is written only on success so a failed
378		connect does not corrupt the socket's cached remote endpoint.
379		*/
380		struct uring_connect_op : uring_op
381		{
382		sockaddr_storage addr{};
383		socklen_t addrlen = 0;
384		int fd = -1;
385		endpoint target_endpoint{};
386		endpoint* remote_endpoint_out = nullptr;
387		endpoint* local_endpoint_out = nullptr;
388		// Set instead of `local_endpoint_out` by a socket that fetches its
389		// local endpoint on demand, sparing each connect a `getsockname`.
390		std::atomic<uring_endpoint_state>* local_state_out = nullptr;
391
392	2057x	uring_connect_op() noexcept : uring_op(&do_handler, &do_cqe, &do_prep) {}
393
394		/** Reset and initialize for a new submission.
395
396		The caller must fill `addr` and `addrlen` before calling this
397		(typically via `to_sockaddr(ep, family, conn_.addr)` which
398		returns the addrlen) — `to_sockaddr` is the family-aware
399		helper and requires the socket family which is known to the
400		caller, not the op.
401		*/
402	3333x	void prepare(
403		capy::continuation& awaiting,
404		capy::executor_ref executor,
405		std::error_code* ec,
406		int file_descriptor,
407		uring_scheduler* scheduler,
408		detail::object_ref impl,
409		endpoint target,
410		endpoint* remote_out,
411		endpoint* local_out,
412		std::stop_token const& token) noexcept
413		{
414	3333x	cont = &awaiting;
415	3333x	ex = executor;
416	3333x	ec_out = ec;
417	3333x	bytes_out = nullptr;
418	3333x	fd = file_descriptor;
419	3333x	sched_ = scheduler;
420	3333x	object_ref_ = std::move(impl);
421	3333x	res = 0;
422	3333x	cqe_flags = 0;
423	3333x	target_endpoint = target;
424	3333x	remote_endpoint_out = remote_out;
425	3333x	local_endpoint_out = local_out;
426		// addr / addrlen are pre-filled by the caller.
427	3333x	start(token);
428	3333x	}
429
430	3333x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
431		{
432	3333x	auto* self = static_cast<uring_connect_op*>(base);
433	3333x	::io_uring_prep_connect(
434	3333x	sqe, self->fd, reinterpret_cast<sockaddr const*>(&self->addr),
435		self->addrlen);
436	3333x	}
437
438		static void
439	3333x	do_cqe(uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
440		{
441	3333x	auto* self = static_cast<uring_connect_op*>(base);
442	3333x	self->res = res;
443	3333x	self->cqe_flags = flags;
444	3333x	local.push(self);
445	3333x	}
446
447	3333x	static void do_handler(
448		void* owner,
449		scheduler_op* base,
450		std::uint32_t /bytes/,
451		std::uint32_t /error/) noexcept
452		{
453	3333x	auto* self = static_cast<uring_connect_op*>(base);
454	3333x	if (coro_drain_if_shutdown(owner, self))
455	2x	return;
456
457	3331x	if (self->sched_)
458	3331x	self->sched_->reset_inline_budget();
459
460	3331x	uring_set_result(self, false, false);
461
462		// Write endpoints only on success; a close cancels the op.
463	6650x	if (self->res >= 0 &&
464	3319x	!self->cancelled.load(std::memory_order_acquire))
465		{
466	3308x	if (self->remote_endpoint_out)
467	3308x	*self->remote_endpoint_out = self->target_endpoint;
468	3308x	if (self->local_state_out)
469	3267x	self->local_state_out->store(
470		uring_endpoint_state::lazy_pending,
471		std::memory_order_release);
472	3308x	if (self->local_endpoint_out && self->fd >= 0)
473		{
474	41x	sockaddr_storage local{};
475	41x	socklen_t len = sizeof(local);
476	41x	if (::getsockname(
477	41x	self->fd, reinterpret_cast<sockaddr*>(&local), &len) ==
478		0)
479	41x	*self->local_endpoint_out = sockaddr_to_endpoint(local);
480		}
481		}
482
483	3331x	coro_resume(self);
484		}
485		};
486
487		/** Prepare @p sqe for @p op and count it in flight.
488
489		@pre The caller holds the ring mutex.
490
491		@return True when this submitter must post the scheduler's
492		`submit_sqes_op` to flush the batch.
493		*/
494		inline bool
495	6207x	uring_link_sqe(
496		uring_scheduler& sched, uring_op* op, ::io_uring_sqe* sqe) noexcept
497		{
498	6207x	op->prep_func(op, sqe);
499	6207x	::io_uring_sqe_set_data(sqe, op);
500		// Count this op against the in-flight gate in do_one: it
501		// expects exactly one F_MORE-less CQE per submitted SQE
502		// (multishot ops decrement only on the terminal CQE).
503	6207x	sched.inflight_inc();
504		// Paired with uring_op::on_cancel: of a stop and this link, the
505		// later one sees the other and queues the cancel. A stop that came
506		// after the initiator's own check would otherwise find no SQE to
507		// cancel and leave the op armed.
508	6207x	op->sqe_set.store(true, std::memory_order_seq_cst);
509	6207x	if (op->cancelled.load(std::memory_order_seq_cst))
510	1x	sched.cancel_op_locked(op);
511		// First submitter in a batch wins the CAS and will post
512		// submit_sqes_op; others piggyback on the same flush.
513	6207x	return !sched.submit_op_posted_exchange(true);
514		}
515
516		/** Submit an `uring_op` a `work_started()` already paid for.
517
518		Acquires the ring mutex, prepares the SQE, and (under the same
519		mutex) CAS-sets `submit_op_posted_`. The first submitter of a
520		batch wins the CAS and posts the scheduler's `submit_sqes_op`,
521		which later flushes all queued SQEs in a single
522		`io_uring_submit_and_get_events` call and drains any ready CQEs.
523		Subsequent submitters in the same batch piggyback — their SQEs
524		sit in the user-space SQ ring until that op dispatches.
525
526		On SQ-ring exhaustion (after one flush retry), completes the op
527		with `EAGAIN` and queues it so its handler dispatches on the next
528		`do_one` cycle, exactly as if the kernel had returned that error.
529
530		@pre `op->prep_func != nullptr`.
531		@pre A `work_started()` backs this op, so the `work_finished()` the
532		scheduler spends on everything it dispatches is owed.
533
534		@par Exception Safety
535		Nothrow.
536
537		@param sched The scheduler owning the ring.
538		@param op The operation to submit.
539
540		@see uring_try_submit_op_if
541		*/
542		inline void
543	4953x	uring_submit_op(uring_scheduler& sched, uring_op* op) noexcept
544		{
545	4953x	sched.lazy_init_ring();
546
547		{
548	4953x	typename uring_scheduler::lock_type ring_lock(sched.ring_mutex());
549
550	4953x	::io_uring_sqe* sqe = sched.acquire_sqe();
551	4953x	if (!sqe)
552		{
553		// SQ stayed full after one flush — synchronous failure path.
554		// Report EAGAIN the way a CQE would, because every handler
555		// re-derives its result from `res`: a code written straight
556		// to ec_out is overwritten on the way out, and a res left at
557		// zero reads as end-of-file, a zero-byte write, or a
558		// successful connect that never happened. Queue the op as
559		// completed so do_one dispatches the handler; the caller's
560		// work_started() pays for the work_finished() do_one spends
561		// on it. (CAS path is not entered here.)
562	3x	op->res = -EAGAIN;
563	3x	typename uring_scheduler::lock_type lock(sched.dispatch_mutex());
564	3x	sched.push_completed_locked(op);
565	3x	return;
566	3x	}
567
568		// Posted under the ring lock: once it is released, the scheduler
569		// can complete this op and, if it belongs to another context,
570		// finish and be destroyed. Flush is deferred to submit_sqes_op;
571		// post() owns the wake.
572	4950x	if (uring_link_sqe(sched, op, sqe))
573	4363x	sched.post(&sched.submit_op_ref());
574	4953x	}
575		}
576
577		/// Outcome of @ref uring_try_submit_op_if.
578		enum class uring_guarded_submit
579		{
580		/// The SQE was prepared and will be flushed with the batch.
581		submitted,
582		/// `admit` declined; nothing was prepared.
583		refused,
584		/// `admit` accepted but the SQ stayed full after one flush.
585		sq_full
586		};
587
588		/** Submit an uncounted op if @p admit allows it.
589
590		`admit` runs under the ring mutex before an SQE is taken. A cancel
591		flushed under that mutex (`flush_after_cancel`) therefore either
592		runs before the check, which then sees the owner's closing state,
593		or follows this op's SQE in the submission queue and cancels it.
594
595		@pre `op->prep_func != nullptr`. No `work_started()` backs this
596		submission, so a full SQ is reported to the caller rather than
597		queued as the op's completion.
598
599		@param sched The scheduler owning the ring.
600		@param op The operation to submit.
601		@param admit Called as `admit()` under the ring mutex; returns
602		whether to submit. May take a lock ordered after the ring
603		mutex. On `sq_full` it has already returned true, so the caller
604		undoes whatever it recorded.
605
606		@return What happened to the op.
607		*/
608		template<class Admit>
609		[[nodiscard]] uring_guarded_submit
610	1259x	uring_try_submit_op_if(
611		uring_scheduler& sched, uring_op* op, Admit admit) noexcept
612		{
613	1259x	sched.lazy_init_ring();
614
615		{
616	1259x	typename uring_scheduler::lock_type ring_lock(sched.ring_mutex());
617	1259x	if (!admit())
618	✗	return uring_guarded_submit::refused;
619
620	1259x	::io_uring_sqe* sqe = sched.acquire_sqe();
621	1259x	if (!sqe)
622		{
623	2x	op->res = -EAGAIN;
624	2x	return uring_guarded_submit::sq_full;
625		}
626		// Posted under the ring lock, as in uring_submit_op.
627	1257x	if (uring_link_sqe(sched, op, sqe))
628	711x	sched.post(&sched.submit_op_ref());
629	1259x	}
630	1257x	return uring_guarded_submit::submitted;
631		}
632
633		/** Readiness wait via `IORING_OP_POLL_ADD`.
634
635		Used to implement the `wait()` virtual for socket and acceptor
636		implementations. The op submits a one-shot poll on `fd` for the
637		requested set of poll flags (POLLIN / POLLOUT / POLLPRI\|POLLERR\|
638		POLLHUP) and reports completion without transferring any data.
639
640		The CQE's `res` carries the actual revents, but we surface only
641		success/cancel/error on `*ec_out` — callers of `wait()` just need
642		a readiness signal, not the specific event mask.
643		*/
644		struct uring_wait_op : uring_op
645		{
646		int fd = -1;
647		int poll_flags = 0;
648
649	3459x	uring_wait_op() noexcept : uring_op(&do_handler, &do_cqe, &do_prep) {}
650
651		/** Reset and initialize for a new submission. */
652	64x	void prepare(
653		capy::continuation& awaiting,
654		capy::executor_ref executor,
655		std::error_code* ec,
656		int file_descriptor,
657		uring_scheduler* scheduler,
658		detail::object_ref impl,
659		int flags,
660		std::stop_token const& token) noexcept
661		{
662	64x	cont = &awaiting;
663	64x	ex = executor;
664	64x	ec_out = ec;
665	64x	bytes_out = nullptr;
666	64x	fd = file_descriptor;
667	64x	sched_ = scheduler;
668	64x	object_ref_ = std::move(impl);
669	64x	poll_flags = flags;
670	64x	res = 0;
671	64x	cqe_flags = 0;
672	64x	start(token);
673	64x	}
674
675	61x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
676		{
677	61x	auto* self = static_cast<uring_wait_op*>(base);
678	61x	::io_uring_prep_poll_add(sqe, self->fd, self->poll_flags);
679	61x	}
680
681		static void
682	61x	do_cqe(uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
683		{
684	61x	auto* self = static_cast<uring_wait_op*>(base);
685	61x	self->res = res;
686	61x	self->cqe_flags = flags;
687	61x	local.push(self);
688	61x	}
689
690	64x	static void do_handler(
691		void* owner,
692		scheduler_op* base,
693		std::uint32_t /bytes/,
694		std::uint32_t /error/) noexcept
695		{
696	64x	auto* self = static_cast<uring_wait_op*>(base);
697	64x	if (coro_drain_if_shutdown(owner, self))
698	4x	return;
699
700	60x	if (self->sched_)
701	60x	self->sched_->reset_inline_budget();
702
703		// A readiness wait (POLLIN/POLLOUT) reports readiness: POLLERR
704		// and POLLHUP mean the next read or write will not block, and
705		// that operation names the condition. Reading SO_ERROR here
706		// would clear it, turning a reset into a clean EOF for the read
707		// that follows. asio reports the same on every backend. Only a
708		// descriptor closed under the poll (POLLNVAL) fails the wait.
709		//
710		// wait(error) is different: its completion has to say why, so
711		// it names SO_ERROR, or EIO when the kernel has none. OOB
712		// (POLLPRI) is a readiness signal, not an error.
713	60x	bool const readiness_wait =
714	60x	(self->poll_flags & (POLLIN \| POLLOUT)) != 0;
715
716	60x	std::error_code ec{};
717	60x	if (self->res < 0)
718		{
719	22x	ec = make_err(-self->res);
720		}
721	38x	else if (self->res & POLLNVAL)
722		{
723	✗	ec = make_err(EBADF);
724		}
725	38x	else if (
726	45x	!readiness_wait && (self->res & (POLLERR \| POLLHUP)) &&
727	7x	!self->cancelled.load(std::memory_order_acquire))
728		{
729		// Not after a close: the number may name another socket now,
730		// and the probe would read and clear its error.
731	6x	int so_err = 0;
732	6x	socklen_t len = sizeof(so_err);
733	6x	if (::getsockopt(self->fd, SOL_SOCKET, SO_ERROR, &so_err, &len) < 0)
734		{
735		// A non-socket (pipe, chardev, ...) has no SO_ERROR and
736		// fails the probe with ENOTSOCK; reporting that would
737		// name the probe rather than the fault, so fall through
738		// to the EIO substitution below. Every other failure
739		// (EBADF from a concurrent close, say) still reports
740		// itself, unchanged on the socket hot path.
741	4x	so_err = (errno == ENOTSOCK) ? 0 : errno;
742		}
743	6x	if (so_err == 0)
744	4x	so_err = EIO;
745	6x	ec = make_err(so_err);
746		}
747
748		// Wait reports only success/cancel/error — no bytes, no EOF.
749	60x	decode_results(
750		self->ec_out, /bytes_out=/nullptr,
751	60x	self->cancelled.load(std::memory_order_acquire), ec,
752		/is_read=/false, /bytes=/0, /empty_buffer=/false);
753
754	60x	coro_resume(self);
755		}
756		};
757
758		/** Non-blocking connect for Unix domain sockets via `IORING_OP_CONNECT`.
759
760		Like `uring_connect_op` but stores `local_endpoint` for the target
761		and out-pointers, since `sockaddr_to_local_endpoint` returns
762		`local_endpoint`, not `endpoint`.
763		*/
764		struct uring_local_connect_op : uring_op
765		{
766		sockaddr_storage addr{};
767		socklen_t addrlen = 0;
768		int fd = -1;
769		corosio::local_endpoint target_endpoint{};
770		corosio::local_endpoint* remote_endpoint_out = nullptr;
771		corosio::local_endpoint* local_endpoint_out = nullptr;
772
773	289x	uring_local_connect_op() noexcept : uring_op(&do_handler, &do_cqe, &do_prep)
774		{
775	289x	}
776
777		/** Reset and initialize for a new submission.
778
779		Caller pre-fills `addr` and `addrlen` (see uring_connect_op::prepare).
780		*/
781	43x	void prepare(
782		capy::continuation& awaiting,
783		capy::executor_ref executor,
784		std::error_code* ec,
785		int file_descriptor,
786		uring_scheduler* scheduler,
787		detail::object_ref impl,
788		corosio::local_endpoint target,
789		corosio::local_endpoint* remote_out,
790		corosio::local_endpoint* local_out,
791		std::stop_token const& token) noexcept
792		{
793	43x	cont = &awaiting;
794	43x	ex = executor;
795	43x	ec_out = ec;
796	43x	bytes_out = nullptr;
797	43x	fd = file_descriptor;
798	43x	sched_ = scheduler;
799	43x	object_ref_ = std::move(impl);
800	43x	res = 0;
801	43x	cqe_flags = 0;
802	43x	target_endpoint = target;
803	43x	remote_endpoint_out = remote_out;
804	43x	local_endpoint_out = local_out;
805	43x	start(token);
806	43x	}
807
808	43x	static void do_prep(uring_op* base, ::io_uring_sqe* sqe) noexcept
809		{
810	43x	auto* self = static_cast<uring_local_connect_op*>(base);
811	43x	::io_uring_prep_connect(
812	43x	sqe, self->fd, reinterpret_cast<sockaddr const*>(&self->addr),
813		self->addrlen);
814	43x	}
815
816		static void
817	43x	do_cqe(uring_op* base, int res, unsigned flags, ready_queue& local) noexcept
818		{
819	43x	auto* self = static_cast<uring_local_connect_op*>(base);
820	43x	self->res = res;
821	43x	self->cqe_flags = flags;
822	43x	local.push(self);
823	43x	}
824
825	43x	static void do_handler(
826		void* owner,
827		scheduler_op* base,
828		std::uint32_t /bytes/,
829		std::uint32_t /error/) noexcept
830		{
831	43x	auto* self = static_cast<uring_local_connect_op*>(base);
832	43x	if (coro_drain_if_shutdown(owner, self))
833	2x	return;
834
835	41x	if (self->sched_)
836	41x	self->sched_->reset_inline_budget();
837
838	41x	uring_set_result(self, false, false);
839
840		// Write endpoints only on success; a close cancels the op.
841	79x	if (self->res >= 0 &&
842	38x	!self->cancelled.load(std::memory_order_acquire))
843		{
844	38x	if (self->remote_endpoint_out)
845	38x	*self->remote_endpoint_out = self->target_endpoint;
846	38x	if (self->local_endpoint_out && self->fd >= 0)
847		{
848	38x	sockaddr_storage local{};
849	38x	socklen_t len = sizeof(local);
850	38x	if (::getsockname(
851	38x	self->fd, reinterpret_cast<sockaddr*>(&local), &len) ==
852		0)
853	38x	*self->local_endpoint_out =
854	38x	sockaddr_to_local_endpoint(local, len);
855		}
856		}
857
858	41x	coro_resume(self);
859		}
860		};
861
862		} // namespace boost::corosio::detail
863
864		#endif // BOOST_COROSIO_HAS_URING
865
866		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_SOCKET_OPS_HPP
867