include/boost/corosio/native/detail/io_uring/io_uring_socket_ops.hpp

88.8% Lines (231/260) 92.9% List of functions (26/28)

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

io_uring_socket_ops.hpp

f(x) Functions (28)

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_IO_URING_IO_URING_SOCKET_OPS_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_SOCKET_OPS_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_IO_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/io_uring/io_uring_buffer.hpp>
26		#include <boost/corosio/native/detail/io_uring/io_uring_op.hpp>
27		#include <boost/corosio/native/detail/io_uring/io_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 <system_error>
33
34		#include <netinet/in.h>
35		#include <poll.h>
36		#include <sys/socket.h>
37		#include <sys/uio.h>
38
39		namespace boost::corosio::detail {
40
41		/// Maximum scatter/gather segments per read/write/dgram op.
42		///
43		/// Bounded well below `IOV_MAX` (1024 on Linux) so each op's
44		/// `iovec[io_uring_max_iov]` lives inside the io_uring_op object on
45		/// the same allocation as the rest of its state. Plan 4's registered-
46		/// buffer work will revisit; until then 16 covers typical scatter use
47		/// cases (fragmented buffers from buffer_sequence) without bloating
48		/// per-op memory.
49		inline constexpr std::size_t io_uring_max_iov = 16;
50
51		/** Fill an `iovec` array from a buffer sequence without type-punning.
52
53		`buffer_param::copy_to` writes `capy::mutable_buffer` objects. Those
54		cannot legally alias `iovec` storage: no `mutable_buffer` lives at
55		that address, and `mutable_buffer` is not an implicit-lifetime type,
56		so its lifetime cannot be started in place (which also rules out
57		`std::start_lifetime_as_array`). We copy into a real `mutable_buffer`
58		scratch array, then translate field by field into the caller's
59		`iovec` array. Zero-size buffers are already skipped by `copy_to`.
60
61		@param buffers The source buffer sequence.
62		@param iovecs Destination array, filled with the non-zero buffers.
63		@return The number of `iovec` entries written.
64		*/
65		inline int
66	381653x	copy_to_iovec(
67		buffer_param const& buffers,
68		iovec (&iovecs)[io_uring_max_iov]) noexcept
69		{
70	381653x	capy::mutable_buffer bufs[io_uring_max_iov];
71	381653x	std::size_t const n = buffers.copy_to(bufs, io_uring_max_iov);
72	763298x	for (std::size_t i = 0; i < n; ++i)
73		{
74	381645x	iovecs[i].iov_base = bufs[i].data();
75	381645x	iovecs[i].iov_len = bufs[i].size();
76		}
77	381653x	return static_cast<int>(n);
78		}
79
80		/** Resolve ec_out/bytes_out from a CQE result for a completed I/O op.
81
82		Shared by read, write, and connect handlers. For reads, `res == 0`
83		with a non-empty buffer means the peer closed the connection (EOF).
84
85		@param self The completed op.
86		@param is_read True if this is a receive/read operation.
87		@param empty_buf True if the submitted buffer was zero-length.
88		*/
89		inline void
90	24516x	uring_set_result(io_uring_op* self, bool is_read, bool empty_buf) noexcept
91		{
92	48916x	decode_io_result(
93		self->ec_out,
94	24516x	self->cancelled.load(std::memory_order_acquire),
95	116x	self->res < 0 ? make_err(-self->res) : std::error_code{},
96		is_read,
97	24516x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u,
98		empty_buf);
99	24516x	}
100
101		/** Scatter-gather read via `IORING_OP_READV`.
102
103		@par Handler dispatch
104		do_cqe captures `res`/`cqe_flags` and queues self into `local`;
105		do_handler runs from the scheduler queue and resumes the coroutine.
106		*/
107		struct uring_read_op : io_uring_op
108		{
109		iovec iovecs[io_uring_max_iov];
110		int iovec_count = 0;
111		int fd = -1;
112		detail::speculative_state* spec_state = nullptr;
113
114	9046x	uring_read_op() noexcept
115	9046x	: io_uring_op(&do_handler, &do_cqe, &do_prep)
116		{
117	9046x	is_read = true;
118	9046x	}
119
120		/** Reset and initialize for a new submission.
121
122		Embedded ops are reused across calls; every mutable field the
123		handler may read must be re-initialized here. `start(token)`
124		also resets `cancelled`, `sqe_set`, and `stop_cb`.
125
126		@pre This slot has no in-flight op (its prior op completed).
127		*/
128	10785x	void prepare(
129		std::coroutine_handle<> handle,
130		capy::executor_ref executor,
131		std::error_code* ec,
132		std::size_t* bytes,
133		int file_descriptor,
134		io_uring_scheduler* scheduler,
135		std::shared_ptr<void> impl,
136		detail::speculative_state* spec,
137		buffer_param buffers,
138		std::stop_token const& token) noexcept
139		{
140	10785x	h = handle;
141	10785x	ex = executor;
142	10785x	ec_out = ec;
143	10785x	bytes_out = bytes;
144	10785x	fd = file_descriptor;
145	10785x	sched_ = scheduler;
146	10785x	impl_ptr = std::move(impl);
147	10785x	spec_state = spec;
148	10785x	res = 0;
149	10785x	cqe_flags = 0;
150	10785x	iovec_count = copy_to_iovec(buffers, iovecs);
151	10785x	empty_buffer = (iovec_count == 0);
152	10785x	start(token);
153	10785x	}
154
155	214x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
156		{
157	214x	auto* self = static_cast<uring_read_op*>(base);
158		// Single-buffer fast path: IORING_OP_RECV with a flat
159		// (buffer, length) skips the iovec-array indirection that
160		// IORING_OP_READV pays. For multi-iovec scatter reads, fall
161		// back to readv.
162	214x	if (self->iovec_count == 1)
163		{
164	214x	::io_uring_prep_recv(
165		sqe, self->fd,
166		self->iovecs[0].iov_base,
167		self->iovecs[0].iov_len,
168		0);
169		}
170		else
171		{
172	✗	::io_uring_prep_readv(
173	✗	sqe, self->fd, self->iovecs, self->iovec_count, 0);
174		}
175	214x	}
176
177	214x	static void do_cqe(
178		io_uring_op* base, int res, unsigned flags,
179		ready_queue& local) noexcept
180		{
181	214x	auto* self = static_cast<uring_read_op*>(base);
182	214x	self->res = res;
183	214x	self->cqe_flags = flags;
184	214x	local.push(self);
185	214x	}
186
187	10785x	static void do_handler(
188		void* owner, scheduler_op* base,
189		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
190		{
191	10785x	auto* self = static_cast<uring_read_op*>(base);
192	10785x	if (coro_drain_if_shutdown(owner, self))
193	✗	return;
194
195	10785x	if (self->sched_)
196	10785x	self->sched_->reset_inline_budget();
197
198	10785x	uring_set_result(self, true, self->empty_buffer);
199
200	10785x	if (self->res > 0 && self->spec_state)
201		{
202		// Kernel signalled readiness — restore speculation.
203	10677x	self->spec_state->on_async_read_ready();
204		}
205
206	10785x	if (self->bytes_out)
207	10785x	*self->bytes_out =
208	10785x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
209
210	10785x	coro_resume(self);
211		// suicide drops here; may destroy impl + self.
212		}
213		};
214
215		/** Scatter-gather write via `IORING_OP_SENDMSG` with `MSG_NOSIGNAL`.
216
217		`MSG_NOSIGNAL` prevents `SIGPIPE` when the peer has closed the
218		connection; the error is surfaced as `EPIPE` instead.
219		*/
220		struct uring_write_op : io_uring_op
221		{
222		iovec iovecs[io_uring_max_iov];
223		int iovec_count = 0;
224		int fd = -1;
225		msghdr msg{};
226		detail::speculative_state* spec_state = nullptr;
227
228	9046x	uring_write_op() noexcept
229	9046x	: io_uring_op(&do_handler, &do_cqe, &do_prep)
230	9046x	{}
231
232		/** Reset and initialize for a new submission. See uring_read_op::prepare. */
233	10571x	void prepare(
234		std::coroutine_handle<> handle,
235		capy::executor_ref executor,
236		std::error_code* ec,
237		std::size_t* bytes,
238		int file_descriptor,
239		io_uring_scheduler* scheduler,
240		std::shared_ptr<void> impl,
241		detail::speculative_state* spec,
242		buffer_param buffers,
243		std::stop_token const& token) noexcept
244		{
245	10571x	h = handle;
246	10571x	ex = executor;
247	10571x	ec_out = ec;
248	10571x	bytes_out = bytes;
249	10571x	fd = file_descriptor;
250	10571x	sched_ = scheduler;
251	10571x	impl_ptr = std::move(impl);
252	10571x	spec_state = spec;
253	10571x	res = 0;
254	10571x	cqe_flags = 0;
255	10571x	iovec_count = copy_to_iovec(buffers, iovecs);
256	10571x	empty_buffer = (iovec_count == 0);
257	10571x	if (!empty_buffer)
258		{
259	10571x	msg = {};
260	10571x	msg.msg_iov = iovecs;
261	10571x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(iovec_count);
262		}
263	10571x	start(token);
264	10571x	}
265
266	✗	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
267		{
268	✗	auto* self = static_cast<uring_write_op*>(base);
269		// Single-buffer fast path: IORING_OP_SEND with MSG_NOSIGNAL
270		// skips the msghdr indirection that IORING_OP_SENDMSG pays.
271		// For multi-iovec scatter writes, fall back to sendmsg.
272	✗	if (self->iovec_count == 1)
273		{
274	✗	::io_uring_prep_send(
275		sqe, self->fd,
276	✗	self->iovecs[0].iov_base,
277		self->iovecs[0].iov_len,
278		MSG_NOSIGNAL);
279		}
280		else
281		{
282	✗	::io_uring_prep_sendmsg(
283	✗	sqe, self->fd, &self->msg, MSG_NOSIGNAL);
284		}
285	✗	}
286
287	✗	static void do_cqe(
288		io_uring_op* base, int res, unsigned flags,
289		ready_queue& local) noexcept
290		{
291	✗	auto* self = static_cast<uring_write_op*>(base);
292	✗	self->res = res;
293	✗	self->cqe_flags = flags;
294	✗	local.push(self);
295	✗	}
296
297	10571x	static void do_handler(
298		void* owner, scheduler_op* base,
299		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
300		{
301	10571x	auto* self = static_cast<uring_write_op*>(base);
302	10571x	if (coro_drain_if_shutdown(owner, self))
303	✗	return;
304
305	10571x	if (self->sched_)
306	10571x	self->sched_->reset_inline_budget();
307
308	10571x	uring_set_result(self, false, self->empty_buffer);
309
310	10571x	if (self->res > 0 && self->spec_state)
311		{
312		// Kernel signalled readiness — restore speculation.
313	10571x	self->spec_state->on_async_write_ready();
314		}
315
316	10571x	if (self->bytes_out)
317	10571x	*self->bytes_out =
318	10571x	self->res >= 0 ? static_cast<std::size_t>(self->res) : 0u;
319
320	10571x	coro_resume(self);
321		}
322		};
323
324		/** Non-blocking connect via `IORING_OP_CONNECT`.
325
326		Negative `res` is the connect error; zero means success.
327		`remote_endpoint_out` is written only on success so a failed
328		connect does not corrupt the socket's cached remote endpoint.
329		*/
330		struct uring_connect_op : io_uring_op
331		{
332		sockaddr_storage addr{};
333		socklen_t addrlen = 0;
334		int fd = -1;
335		endpoint target_endpoint{};
336		endpoint* remote_endpoint_out = nullptr;
337		endpoint* local_endpoint_out = nullptr;
338
339	9066x	uring_connect_op() noexcept
340	9066x	: io_uring_op(&do_handler, &do_cqe, &do_prep)
341	9066x	{}
342
343		/** Reset and initialize for a new submission.
344
345		The caller must fill `addr` and `addrlen` before calling this
346		(typically via `to_sockaddr(ep, family, conn_.addr)` which
347		returns the addrlen) — `to_sockaddr` is the family-aware
348		helper and requires the socket family which is known to the
349		caller, not the op.
350		*/
351	2979x	void prepare(
352		std::coroutine_handle<> handle,
353		capy::executor_ref executor,
354		std::error_code* ec,
355		int file_descriptor,
356		io_uring_scheduler* scheduler,
357		std::shared_ptr<void> impl,
358		endpoint target,
359		endpoint* remote_out,
360		endpoint* local_out,
361		std::stop_token const& token) noexcept
362		{
363	2979x	h = handle;
364	2979x	ex = executor;
365	2979x	ec_out = ec;
366	2979x	bytes_out = nullptr;
367	2979x	fd = file_descriptor;
368	2979x	sched_ = scheduler;
369	2979x	impl_ptr = std::move(impl);
370	2979x	res = 0;
371	2979x	cqe_flags = 0;
372	2979x	target_endpoint = target;
373	2979x	remote_endpoint_out = remote_out;
374	2979x	local_endpoint_out = local_out;
375		// addr / addrlen are pre-filled by the caller.
376	2979x	start(token);
377	2979x	}
378
379	2979x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
380		{
381	2979x	auto* self = static_cast<uring_connect_op*>(base);
382	2979x	::io_uring_prep_connect(
383		sqe, self->fd,
384	2979x	reinterpret_cast<sockaddr const*>(&self->addr),
385		self->addrlen);
386	2979x	}
387
388	2979x	static void do_cqe(
389		io_uring_op* base, int res, unsigned flags,
390		ready_queue& local) noexcept
391		{
392	2979x	auto* self = static_cast<uring_connect_op*>(base);
393	2979x	self->res = res;
394	2979x	self->cqe_flags = flags;
395	2979x	local.push(self);
396	2979x	}
397
398	2979x	static void do_handler(
399		void* owner, scheduler_op* base,
400		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
401		{
402	2979x	auto* self = static_cast<uring_connect_op*>(base);
403	2979x	if (coro_drain_if_shutdown(owner, self))
404	✗	return;
405
406	2979x	if (self->sched_)
407	2979x	self->sched_->reset_inline_budget();
408
409	2979x	uring_set_result(self, false, false);
410
411		// Write endpoints only on success.
412	2979x	if (self->res >= 0)
413		{
414	2970x	if (self->remote_endpoint_out)
415	2970x	*self->remote_endpoint_out = self->target_endpoint;
416	2970x	if (self->local_endpoint_out && self->fd >= 0)
417		{
418	2970x	sockaddr_storage local{};
419	2970x	socklen_t len = sizeof(local);
420	2970x	if (::getsockname(self->fd,
421	2970x	reinterpret_cast<sockaddr*>(&local), &len) == 0)
422	2970x	*self->local_endpoint_out = sockaddr_to_endpoint(local);
423		}
424		}
425
426	2979x	coro_resume(self);
427		}
428		};
429
430		/** Submit an `io_uring_op` whose `prep_func` is set.
431
432		Acquires the ring mutex, prepares the SQE, and (under the same
433		mutex) CAS-sets `submit_op_posted_`. The first submitter of a
434		batch wins the CAS and posts the scheduler's `submit_sqes_op`,
435		which later flushes all queued SQEs in a single
436		`io_uring_submit_and_get_events` call and drains any ready CQEs.
437		Subsequent submitters in the same batch piggyback — their SQEs
438		sit in the user-space SQ ring until that op dispatches.
439
440		On SQ-ring exhaustion (after one flush retry), surfaces `EAGAIN`
441		on `*op->ec_out` and queues the op as completed so its handler
442		dispatches on the next `do_one` cycle.
443
444		@pre `op->prep_func != nullptr`.
445
446		@par Exception Safety
447		Nothrow.
448		*/
449		inline void
450	3580x	io_uring_submit_op(io_uring_scheduler& sched, io_uring_op* op) noexcept
451		{
452	3580x	sched.lazy_init_ring();
453
454	3580x	bool need_post = false;
455		{
456	3580x	typename io_uring_scheduler::lock_type ring_lock(sched.ring_mutex());
457
458	3580x	::io_uring_sqe* sqe = ::io_uring_get_sqe(sched.ring());
459	3580x	if (!sqe)
460		{
461		// SQ ring full — flush to kernel and retry once.
462	✗	::io_uring_submit(sched.ring());
463	✗	sqe = ::io_uring_get_sqe(sched.ring());
464		}
465
466	3580x	if (!sqe)
467		{
468		// SQ stayed full after one flush — synchronous failure path.
469		// Surface EAGAIN and queue the op as completed so do_one
470		// dispatches the handler. The caller's work_started() already
471		// counted this op. (CAS path is not entered here.)
472	✗	if (op->ec_out)
473	✗	*op->ec_out = make_err(EAGAIN);
474	✗	typename io_uring_scheduler::lock_type lock(sched.dispatch_mutex());
475	✗	sched.push_completed_locked(op);
476	✗	return;
477	✗	}
478
479	3580x	op->prep_func(op, sqe);
480	3580x	::io_uring_sqe_set_data(sqe, op);
481		// Count this op against the in-flight gate in do_one: it
482		// expects exactly one F_MORE-less CQE per submitted SQE
483		// (multishot ops decrement only on the terminal CQE).
484	3580x	sched.inflight_inc();
485		// Release pairs with the acquire in io_uring_op::request_cancel:
486		// a stop_token firing after we release the mutex will see
487		// sqe_set==true and submit a cancel-by-user_data SQE.
488	3580x	op->sqe_set.store(true, std::memory_order_release);
489
490		// First submitter in a batch wins the CAS and will post
491		// submit_sqes_op; others piggyback on the same flush.
492	3580x	if (!sched.submit_op_posted_exchange(true))
493	188x	need_post = true;
494	3580x	}
495
496	3580x	if (need_post)
497		{
498		// Flush is deferred to submit_sqes_op; post() owns the wake.
499	188x	sched.post(&sched.submit_op_ref());
500		}
501		}
502
503		/** Readiness wait via `IORING_OP_POLL_ADD`.
504
505		Used to implement the `wait()` virtual for socket and acceptor
506		implementations. The op submits a one-shot poll on `fd` for the
507		requested set of poll flags (POLLIN / POLLOUT / POLLPRI\|POLLERR\|
508		POLLHUP) and reports completion without transferring any data.
509
510		The CQE's `res` carries the actual revents, but we surface only
511		success/cancel/error on `*ec_out` — callers of `wait()` just need
512		a readiness signal, not the specific event mask.
513		*/
514		struct uring_wait_op : io_uring_op
515		{
516		int fd = -1;
517		int poll_flags = 0;
518
519	9368x	uring_wait_op() noexcept
520	9368x	: io_uring_op(&do_handler, &do_cqe, &do_prep)
521	9368x	{}
522
523		/** Reset and initialize for a new submission. */
524	21x	void prepare(
525		std::coroutine_handle<> handle,
526		capy::executor_ref executor,
527		std::error_code* ec,
528		int file_descriptor,
529		io_uring_scheduler* scheduler,
530		std::shared_ptr<void> impl,
531		int flags,
532		std::stop_token const& token) noexcept
533		{
534	21x	h = handle;
535	21x	ex = executor;
536	21x	ec_out = ec;
537	21x	bytes_out = nullptr;
538	21x	fd = file_descriptor;
539	21x	sched_ = scheduler;
540	21x	impl_ptr = std::move(impl);
541	21x	poll_flags = flags;
542	21x	res = 0;
543	21x	cqe_flags = 0;
544	21x	start(token);
545	21x	}
546
547	17x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
548		{
549	17x	auto* self = static_cast<uring_wait_op*>(base);
550	17x	::io_uring_prep_poll_add(sqe, self->fd, self->poll_flags);
551	17x	}
552
553	17x	static void do_cqe(
554		io_uring_op* base, int res, unsigned flags,
555		ready_queue& local) noexcept
556		{
557	17x	auto* self = static_cast<uring_wait_op*>(base);
558	17x	self->res = res;
559	17x	self->cqe_flags = flags;
560	17x	local.push(self);
561	17x	}
562
563	21x	static void do_handler(
564		void* owner, scheduler_op* base,
565		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
566		{
567	21x	auto* self = static_cast<uring_wait_op*>(base);
568	21x	if (coro_drain_if_shutdown(owner, self))
569	✗	return;
570
571	21x	if (self->sched_)
572	21x	self->sched_->reset_inline_budget();
573
574		// Wait reports only success/cancel/error — no bytes, no EOF.
575	36x	decode_io_result(
576		self->ec_out,
577	21x	self->cancelled.load(std::memory_order_acquire),
578	21x	self->res < 0 ? make_err(-self->res) : std::error_code{},
579		/is_read=/false, /bytes=/0, /empty_buffer=/false);
580
581	21x	coro_resume(self);
582		}
583		};
584
585		/** Non-blocking connect for Unix domain sockets via `IORING_OP_CONNECT`.
586
587		Like `uring_connect_op` but stores `local_endpoint` for the target
588		and out-pointers, since `sockaddr_to_local_endpoint` returns
589		`local_endpoint`, not `endpoint`.
590		*/
591		struct uring_local_connect_op : io_uring_op
592		{
593		sockaddr_storage addr{};
594		socklen_t addrlen = 0;
595		int fd = -1;
596		corosio::local_endpoint target_endpoint{};
597		corosio::local_endpoint* remote_endpoint_out = nullptr;
598		corosio::local_endpoint* local_endpoint_out = nullptr;
599
600	149x	uring_local_connect_op() noexcept
601	149x	: io_uring_op(&do_handler, &do_cqe, &do_prep)
602	149x	{}
603
604		/** Reset and initialize for a new submission.
605
606		Caller pre-fills `addr` and `addrlen` (see uring_connect_op::prepare).
607		*/
608	14x	void prepare(
609		std::coroutine_handle<> handle,
610		capy::executor_ref executor,
611		std::error_code* ec,
612		int file_descriptor,
613		io_uring_scheduler* scheduler,
614		std::shared_ptr<void> impl,
615		corosio::local_endpoint target,
616		corosio::local_endpoint* remote_out,
617		corosio::local_endpoint* local_out,
618		std::stop_token const& token) noexcept
619		{
620	14x	h = handle;
621	14x	ex = executor;
622	14x	ec_out = ec;
623	14x	bytes_out = nullptr;
624	14x	fd = file_descriptor;
625	14x	sched_ = scheduler;
626	14x	impl_ptr = std::move(impl);
627	14x	res = 0;
628	14x	cqe_flags = 0;
629	14x	target_endpoint = target;
630	14x	remote_endpoint_out = remote_out;
631	14x	local_endpoint_out = local_out;
632	14x	start(token);
633	14x	}
634
635	14x	static void do_prep(io_uring_op* base, ::io_uring_sqe* sqe) noexcept
636		{
637	14x	auto* self = static_cast<uring_local_connect_op*>(base);
638	14x	::io_uring_prep_connect(
639		sqe, self->fd,
640	14x	reinterpret_cast<sockaddr const*>(&self->addr),
641		self->addrlen);
642	14x	}
643
644	14x	static void do_cqe(
645		io_uring_op* base, int res, unsigned flags,
646		ready_queue& local) noexcept
647		{
648	14x	auto* self = static_cast<uring_local_connect_op*>(base);
649	14x	self->res = res;
650	14x	self->cqe_flags = flags;
651	14x	local.push(self);
652	14x	}
653
654	14x	static void do_handler(
655		void* owner, scheduler_op* base,
656		std::uint32_t /bytes/, std::uint32_t /error/) noexcept
657		{
658	14x	auto* self = static_cast<uring_local_connect_op*>(base);
659	14x	if (coro_drain_if_shutdown(owner, self))
660	✗	return;
661
662	14x	if (self->sched_)
663	14x	self->sched_->reset_inline_budget();
664
665	14x	uring_set_result(self, false, false);
666
667		// Write endpoints only on success.
668	14x	if (self->res >= 0)
669		{
670	11x	if (self->remote_endpoint_out)
671	11x	*self->remote_endpoint_out = self->target_endpoint;
672	11x	if (self->local_endpoint_out && self->fd >= 0)
673		{
674	11x	sockaddr_storage local{};
675	11x	socklen_t len = sizeof(local);
676	11x	if (::getsockname(self->fd,
677	11x	reinterpret_cast<sockaddr*>(&local), &len) == 0)
678	11x	*self->local_endpoint_out =
679	11x	sockaddr_to_local_endpoint(local, len);
680		}
681		}
682
683	14x	coro_resume(self);
684		}
685		};
686
687		} // namespace boost::corosio::detail
688
689		#endif // BOOST_COROSIO_HAS_IO_URING
690
691		#endif // BOOST_COROSIO_NATIVE_DETAIL_IO_URING_IO_URING_SOCKET_OPS_HPP
692