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

87.2% Lines (1101 / 1262, 25 excl) 100.0% Functions (114 / 114, 1 excl)

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

uring_types.hpp

f(x) Functions (115)

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_TYPES_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_URING_URING_TYPES_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_URING
17
18		#include <boost/corosio/detail/config.hpp>
19		#include <boost/corosio/detail/intrusive.hpp>
20		#include <boost/corosio/native/detail/uring/uring_acceptor_ops.hpp>
21		#include <boost/corosio/native/detail/uring/uring_buffer.hpp>
22		#include <boost/corosio/native/detail/uring/uring_dgram_ops.hpp>
23		#include <boost/corosio/native/detail/uring/uring_op.hpp>
24		#include <boost/corosio/native/detail/uring/uring_scheduler.hpp>
25		#include <boost/corosio/native/detail/uring/uring_multishot_acceptor.hpp>
26		#include <boost/corosio/native/detail/uring/uring_socket_ops.hpp>
27		#include <boost/corosio/native/detail/uring/uring_socket_service_base.hpp>
28		#include <boost/corosio/native/detail/native_socket_base.hpp>
29		#include <boost/corosio/native/detail/make_err.hpp>
30		#include <boost/corosio/native/detail/msg_flags.hpp>
31		#include <boost/corosio/native/detail/validate_fd.hpp>
32		#include <boost/corosio/detail/local_datagram_service.hpp>
33		#include <boost/corosio/detail/local_stream_acceptor_service.hpp>
34		#include <boost/corosio/detail/local_stream_service.hpp>
35		#include <boost/corosio/detail/tcp_acceptor_service.hpp>
36		#include <boost/corosio/detail/tcp_service.hpp>
37		#include <boost/corosio/detail/udp_service.hpp>
38		#include <boost/corosio/local_endpoint.hpp>
39		#include <boost/corosio/local_datagram_socket.hpp>
40		#include <boost/corosio/local_stream_acceptor.hpp>
41		#include <boost/corosio/local_stream_socket.hpp>
42		#include <boost/corosio/tcp_acceptor.hpp>
43		#include <boost/corosio/tcp_socket.hpp>
44		#include <boost/corosio/udp_socket.hpp>
45
46		#include <memory>
47		#include <mutex>
48		#include <optional>
49		#include <vector>
50
51		#include <fcntl.h>
52		#include <netinet/in.h>
53		#include <sys/socket.h>
54		#include <sys/un.h>
55		#include <unistd.h>
56
57		namespace boost::corosio::detail {
58
59		class uring_tcp_service;
60		class uring_tcp_acceptor_service; // Task 18
61		class uring_local_stream_service;
62		class uring_local_stream_acceptor_service;
63		class uring_udp_service;
64		class uring_local_datagram_service;
65
66		/** TCP socket implementation for io_uring.
67
68		Implements `tcp_socket::implementation` using a proactor model:
69		read, write, and connect operations are submitted to the kernel
70		via `uring_submit_op` and complete through the ring's CQE path.
71
72		The service holds one intrusive reference for as long as the impl
73		is live; in-flight ops hold an additional `object_ref_` keepalive so
74		the kernel's user-data pointer remains valid until the CQE arrives.
75
76		@par Thread Safety
77		Distinct objects: Safe.
78		Shared objects: Unsafe. A socket must not have two operations of
79		the same type in flight simultaneously.
80		*/
81		class BOOST_COROSIO_DECL uring_tcp_socket final
82		: public native_socket_base<
83		uring_tcp_socket,
84		tcp_socket::implementation,
85		endpoint>
86		, public intrusive_list<uring_tcp_socket>::node
87		{
88		friend uring_tcp_service;
89
90		uring_scheduler* sched_ = nullptr;
91		uring_tcp_service* svc_ = nullptr;
92
93		// fd_ and local_endpoint_ are provided by native_socket_base (the
94		// readiness/completion-agnostic socket base shared with the reactor
95		// sockets). native_handle()/is_open()/set_option()/get_option() come
96		// from there too; local_endpoint() is overridden below for lazy
97		// getsockname resolution.
98		using endpoint_state = uring_endpoint_state;
99		mutable std::atomic<endpoint_state> local_endpoint_state_{
100		endpoint_state::unresolved};
101		endpoint remote_endpoint_;
102
103		// Per-fd op slots — embedded to eliminate per-call heap allocation.
104		// Single-pending invariant per slot: at most one read, write, or
105		// connect in flight on this socket at any time (the awaitable
106		// contract).
107		uring_read_op rd_;
108		uring_write_op wr_;
109		uring_connect_op conn_;
110		uring_wait_op wait_op_;
111
112		mutable detail::speculative_state spec_;
113
114		public:
115		/** Construct with service and scheduler references.
116
117		Both refs must outlive this socket. `sched_` and `svc_` are
118		intentionally separate so service subclasses can pass a
119		different scheduler if needed.
120
121		@param svc The owning service (Task 13).
122		@param sched The io_uring scheduler owned by the context.
123		*/
124	1908x	explicit uring_tcp_socket(
125		uring_tcp_service& svc, uring_scheduler& sched) noexcept
126	3816x	: sched_(&sched)
127	1908x	, svc_(&svc)
128		{
129	1908x	}
130
131	3816x	~uring_tcp_socket() override
132	1908x	{
133	1908x	if (fd_ >= 0)
134	✗	::close(
135		fd_); // LCOV_EXCL_LINE backstop: close_socket() clears fd_ before destroy
136	3816x	}
137
138		// ----------------------------------------------------------------
139		// io_stream::implementation
140		// ----------------------------------------------------------------
141
142	158899x	std::coroutine_handle<> read_some(
143		capy::continuation& cont,
144		capy::executor_ref ex,
145		buffer_param buffers,
146		std::stop_token token,
147		std::error_code* ec,
148		std::size_t* bytes) override
149		{
150		iovec iovecs[uring_max_iov];
151	158899x	int iovec_count = copy_to_iovec(buffers, iovecs);
152	158899x	bool stop_now = token.stop_possible() && token.stop_requested();
153	158899x	bool empty_buf = (iovec_count == 0);
154
155	158899x	ssize_t n = 0;
156	158899x	int err = 0;
157	158899x	bool have_sync_res = stop_now \|\| empty_buf;
158	158899x	if (!have_sync_res && spec_.may_speculate_read())
159		{
160	158708x	n = uring_direct_read(fd_, iovecs, iovec_count);
161	158708x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
162		{
163	158208x	have_sync_res = true;
164	158208x	if (n < 0)
165	274x	err = errno;
166		// Speculative read produced a definitive answer (data
167		// or non-EAGAIN error); reset the failure streak so a
168		// burst of past EAGAINs doesn't latch perma-off when
169		// the workload is in fact speculation-friendly.
170	158208x	if (n >= 0)
171	157934x	spec_.on_read_success();
172		}
173		else
174		{
175	500x	spec_.on_read_exhausted();
176		}
177		}
178
179	158899x	if (have_sync_res)
180		{
181	158209x	if (sched_->try_consume_inline_budget())
182		{
183	148925x	decode_results(
184		ec, bytes, stop_now,
185	274x	err ? make_err(err) : std::error_code{},
186	148925x	/is_read=/true, n < 0 ? 0u : static_cast<std::size_t>(n),
187		empty_buf);
188	148925x	return dispatch_coro(ex, cont);
189		}
190	9284x	rd_.prepare(
191	18568x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this),
192		&spec_, buffers, token);
193	9284x	if (stop_now)
194	✗	rd_.cancelled.store(true, std::memory_order_release);
195		else
196	9284x	rd_.res = (n < 0) ? -err : static_cast<int>(n);
197	9284x	sched_->work_started();
198		{
199	9284x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
200	9284x	sched_->push_completed_locked(&rd_);
201	9284x	}
202	9284x	return std::noop_coroutine();
203		}
204
205	690x	rd_.prepare(
206	1380x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this), &spec_,
207		buffers, token);
208	690x	sched_->work_started();
209	690x	if (rd_.cancelled.load(std::memory_order_acquire))
210		{
211	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
212	✗	sched_->push_completed_locked(&rd_);
213	✗	return std::noop_coroutine();
214	✗	}
215	690x	uring_submit_op(*sched_, &rd_);
216	690x	return std::noop_coroutine();
217		}
218
219	158563x	std::coroutine_handle<> write_some(
220		capy::continuation& cont,
221		capy::executor_ref ex,
222		buffer_param buffers,
223		std::stop_token token,
224		std::error_code* ec,
225		std::size_t* bytes) override
226		{
227		iovec iovecs[uring_max_iov];
228	158563x	int iovec_count = copy_to_iovec(buffers, iovecs);
229	158563x	bool stop_now = token.stop_possible() && token.stop_requested();
230	158563x	bool empty_buf = (iovec_count == 0);
231
232	158563x	ssize_t n = 0;
233	158563x	int err = 0;
234	158563x	bool have_sync_res = stop_now \|\| empty_buf;
235		// A queued batch enters the kernel anyway; joining it spares
236		// this write a syscall of its own.
237	316848x	if (!have_sync_res && spec_.may_speculate_write() &&
238	158285x	!sched_->write_may_join_batch())
239		{
240	158048x	n = uring_direct_write(fd_, iovecs, iovec_count);
241	158048x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
242		{
243	158045x	have_sync_res = true;
244	158045x	if (n < 0)
245	5x	err = errno;
246		}
247		else
248		{
249	3x	spec_.on_write_exhausted();
250		}
251		}
252
253	158563x	if (have_sync_res)
254		{
255	158322x	if (sched_->try_consume_inline_budget())
256		{
257	149041x	decode_results(
258		ec, bytes, stop_now,
259	5x	err ? make_err(err) : std::error_code{},
260	149041x	/is_read=/false, n < 0 ? 0u : static_cast<std::size_t>(n),
261		/empty_buffer=/false);
262	149041x	return dispatch_coro(ex, cont);
263		}
264	9281x	wr_.prepare(
265	18562x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this),
266		&spec_, buffers, token);
267	9281x	if (stop_now)
268	✗	wr_.cancelled.store(true, std::memory_order_release);
269		else
270	9281x	wr_.res = (n < 0) ? -err : static_cast<int>(n);
271	9281x	sched_->work_started();
272		{
273	9281x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
274	9281x	sched_->push_completed_locked(&wr_);
275	9281x	}
276	9281x	return std::noop_coroutine();
277		}
278
279	241x	wr_.prepare(
280	482x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this), &spec_,
281		buffers, token);
282	241x	sched_->work_started();
283	241x	if (wr_.cancelled.load(std::memory_order_acquire))
284		{
285	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
286	✗	sched_->push_completed_locked(&wr_);
287	✗	return std::noop_coroutine();
288	✗	}
289	241x	uring_submit_op(*sched_, &wr_);
290	241x	return std::noop_coroutine();
291		}
292
293		// ----------------------------------------------------------------
294		// tcp_socket::implementation
295		// ----------------------------------------------------------------
296
297	3292x	std::coroutine_handle<> connect(
298		capy::continuation& cont,
299		capy::executor_ref ex,
300		endpoint ep,
301		std::stop_token token,
302		std::error_code* ec) override
303		{
304	3292x	bool stop_now = token.stop_possible() && token.stop_requested();
305	3292x	if (stop_now)
306		{
307	✗	if (sched_->try_consume_inline_budget())
308		{
309	✗	if (ec)
310	✗	*ec = capy::error::canceled;
311	✗	return dispatch_coro(ex, cont);
312		}
313	✗	conn_.addrlen = to_sockaddr(ep, this->native_family(), conn_.addr);
314	✗	conn_.prepare(
315	✗	cont, ex, ec, fd_, sched_, detail::object_ref(this), ep,
316		&remote_endpoint_, nullptr, token);
317	✗	conn_.cancelled.store(true, std::memory_order_release);
318	✗	sched_->work_started();
319		{
320	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
321	✗	sched_->push_completed_locked(&conn_);
322	✗	}
323	✗	return std::noop_coroutine();
324		}
325
326		// A speculative ::connect would leave the fd in EINPROGRESS and
327		// a subsequent IORING_OP_CONNECT would see EALREADY — avoid.
328	3292x	conn_.addrlen = to_sockaddr(ep, this->native_family(), conn_.addr);
329	3292x	conn_.local_state_out = &local_endpoint_state_;
330	3292x	conn_.prepare(
331	6584x	cont, ex, ec, fd_, sched_, detail::object_ref(this), ep,
332		&remote_endpoint_, nullptr, token);
333	3292x	sched_->work_started();
334	3292x	if (conn_.cancelled.load(std::memory_order_acquire))
335		{
336	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
337	✗	sched_->push_completed_locked(&conn_);
338	✗	return std::noop_coroutine();
339	✗	}
340	3292x	uring_submit_op(*sched_, &conn_);
341	3292x	return std::noop_coroutine();
342		}
343
344	25x	std::coroutine_handle<> wait(
345		capy::continuation& cont,
346		capy::executor_ref ex,
347		wait_type w,
348		std::stop_token token,
349		std::error_code* ec) override
350		{
351	25x	int poll_flags = 0;
352	25x	switch (w)
353		{
354	15x	case wait_type::read:
355	15x	poll_flags = POLLIN;
356	15x	break;
357	7x	case wait_type::write:
358	7x	poll_flags = POLLOUT;
359	7x	break;
360	3x	case wait_type::error:
361	3x	poll_flags = POLLPRI \| POLLERR \| POLLHUP;
362	3x	break;
363		}
364	25x	wait_op_.prepare(
365	50x	cont, ex, ec, fd_, sched_, detail::object_ref(this), poll_flags,
366		token);
367	25x	sched_->work_started();
368	25x	if (wait_op_.cancelled.load(std::memory_order_acquire))
369		{
370	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
371	✗	sched_->push_completed_locked(&wait_op_);
372	✗	return std::noop_coroutine();
373	✗	}
374	25x	uring_submit_op(*sched_, &wait_op_);
375	25x	return std::noop_coroutine();
376		}
377
378	4x	std::error_code shutdown(tcp_socket::shutdown_type what) noexcept override
379		{
380	4x	if (::shutdown(fd_, static_cast<int>(what)) != 0)
381	1x	return make_err(errno);
382	3x	return {};
383		}
384
385		// native_handle() / set_option() / get_option() are inherited from
386		// native_socket_base.
387
388	3x	native_handle_type release_socket() noexcept override
389		{
390		// Flush while the fd is still open so the kernel resolves
391		// pending SQEs before the caller can close and recycle the
392		// number (same reasoning as close_socket).
393		// A connect completing after this belongs to the descriptor
394		// the caller takes; see close_socket().
395	3x	conn_.request_cancel();
396	3x	wait_op_.request_cancel();
397	3x	if (fd_ >= 0)
398	3x	sched_->flush_after_cancel(fd_, {&rd_, &wr_, &conn_, &wait_op_});
399	3x	int fd = fd_;
400	3x	fd_ = -1;
401	3x	family_ = AF_UNSPEC;
402	3x	local_endpoint_ = endpoint{};
403	3x	remote_endpoint_ = endpoint{};
404	3x	local_endpoint_state_.store(
405		endpoint_state::unresolved, std::memory_order_release);
406	3x	return fd;
407		}
408
409	897x	void cancel() noexcept override
410		{
411	897x	if (fd_ >= 0)
412	485x	sched_->cancel_ops({&rd_, &wr_, &conn_, &wait_op_});
413	897x	}
414
415		/// Cancel in-flight ops, close the fd, and reset cached endpoints.
416		/// Called by the service on close()/teardown. The flush runs while
417		/// the fd is still open, so no queued SQE can name a recycled number.
418	17662x	void close_socket() noexcept
419		{
420		// A connect completing after this must not write endpoints into
421		// a closed impl that may be recycled.
422	17662x	conn_.request_cancel();
423	17662x	wait_op_.request_cancel();
424	17662x	if (fd_ >= 0)
425		{
426	6925x	sched_->flush_after_cancel(fd_, {&rd_, &wr_, &conn_, &wait_op_});
427	6925x	::close(fd_);
428	6925x	fd_ = -1;
429		}
430	17662x	family_ = AF_UNSPEC;
431	17662x	local_endpoint_ = endpoint{};
432	17662x	remote_endpoint_ = endpoint{};
433	17662x	local_endpoint_state_.store(
434		endpoint_state::unresolved, std::memory_order_release);
435	17662x	}
436
437		/** Recycle into the owning service's pool at zero references.
438
439		Defined out-of-line after `uring_tcp_service` — its body needs
440		the service's complete type to reach the private `pool_`
441		member (reachable because this impl is a friend), and this
442		non-template member would otherwise require it right here,
443		before the service class exists in the file.
444		*/
445		void retire() noexcept override;
446
447		/** Reset op slots, cached endpoints, and the speculation hint for
448		recycling.
449
450		`close_socket()` already closed fd_ before the refcount reached
451		zero; each op's own `prepare()` overwrites its iovec/addr scratch
452		before the next use. What is left: the endpoints and their
453		resolution state (a connect completing on another thread can
454		write them after close), `family_` (a recycled socket must not
455		report the prior session's family before it is set again), the
456		speculation hint (a per-connection performance heuristic
457		that must not leak into the next logical connection), and the
458		`stop_cb` invariant on every op slot.
459
460		@pre refs_ == 0, fd closed, no op in flight.
461		*/
462	8833x	void reuse() noexcept
463		{
464	8833x	BOOST_COROSIO_ASSERT(fd_ == -1);
465		// A connect completing on another thread while the socket closed
466		// can still have written these.
467	8833x	local_endpoint_ = endpoint{};
468	8833x	remote_endpoint_ = endpoint{};
469	8833x	local_endpoint_state_.store(
470		endpoint_state::unresolved, std::memory_order_relaxed);
471	8833x	BOOST_COROSIO_ASSERT(!rd_.stop_cb);
472	8833x	BOOST_COROSIO_ASSERT(!wr_.stop_cb);
473	8833x	BOOST_COROSIO_ASSERT(!conn_.stop_cb);
474	8833x	BOOST_COROSIO_ASSERT(!wait_op_.stop_cb);
475	8833x	family_ = AF_UNSPEC;
476	8833x	spec_.reset();
477	8833x	}
478
479	36x	endpoint local_endpoint() const noexcept override
480		{
481		// Lazy resolution: only fire the getsockname syscall when
482		// an accept or connect marked the endpoint "lazy_pending". For
483		// unbound/disconnected sockets the state remains unresolved
484		// and the accessor returns the default endpoint without a
485		// syscall. The mutable update races benignly with concurrent
486		// readers — both threads would compute the same value from
487		// the same fd.
488	36x	if (local_endpoint_state_.load(std::memory_order_acquire) ==
489	52x	endpoint_state::lazy_pending &&
490	16x	fd_ >= 0)
491		{
492	16x	sockaddr_storage local{};
493	16x	socklen_t len = sizeof(local);
494	16x	if (::getsockname(fd_, reinterpret_cast<sockaddr*>(&local), &len) ==
495		0)
496	16x	local_endpoint_ = sockaddr_to_endpoint(local);
497	16x	local_endpoint_state_.store(
498		endpoint_state::resolved, std::memory_order_release);
499		}
500	36x	return local_endpoint_;
501		}
502
503	43x	endpoint remote_endpoint() const noexcept override
504		{
505	43x	return remote_endpoint_;
506		}
507		};
508
509		/** TCP socket service for io_uring.
510
511		Owns all `uring_tcp_socket` implementations for an `io_context`.
512		Satisfies the `tcp_service` interface so the generic `tcp_socket`
513		front-end can call `open_socket` and `bind_socket` transparently.
514
515		Socket impls live in the service's `object_pool`. `construct()`
516		hands out a pooled impl with one reference already held; `destroy()`
517		releases it, and `retire()` recycles it onto the free list
518		instead of freeing it.
519
520		@par Thread Safety
521		All public member functions are thread-safe.
522		*/
523		class BOOST_COROSIO_DECL uring_tcp_service final
524		: public uring_socket_service_base<
525		uring_tcp_service,
526		tcp_service,
527		uring_tcp_socket>
528		{
529		using base_service = uring_socket_service_base<
530		uring_tcp_service,
531		tcp_service,
532		uring_tcp_socket>;
533
534		public:
535		/// Identifies this service for `execution_context` lookup.
536		using key_type = tcp_service;
537
538		/** Construct the TCP service.
539
540		@param ctx The owning execution context. The io_uring scheduler
541		must already be registered.
542		*/
543	457x	explicit uring_tcp_service(capy::execution_context& ctx) : base_service(ctx)
544		{
545	457x	}
546
547		// construct / destroy / shutdown / close / scheduler() are inherited
548		// from uring_socket_service_base. The methods below are TCP-specific.
549
550		/** Open a socket fd and associate it with an impl.
551
552		Creates a non-blocking, close-on-exec socket via `socket(2)`.
553
554		@param impl The socket implementation to initialise.
555		@param family Address family (e.g. `AF_INET`, `AF_INET6`).
556		@param type Socket type (e.g. `SOCK_STREAM`).
557		@param protocol Protocol number (e.g. `IPPROTO_TCP`).
558		@return Error code on failure, empty on success.
559		*/
560	3351x	std::error_code open_socket(
561		tcp_socket::implementation& impl,
562		int family,
563		int type,
564		int protocol) override
565		{
566	3351x	auto& sock = static_cast<uring_tcp_socket&>(impl);
567		int fd =
568	3351x	::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
569	3351x	if (fd < 0)
570	2x	return make_err(errno);
571		// LCOV_EXCL_START: dead — open() guards is_open(), so open_socket
572		// never runs against an already-open fd (assign_socket handles that).
573	−	if (sock.fd_ >= 0)
574	−	sock.close_socket();
575		// LCOV_EXCL_STOP
576	3349x	sock.fd_ = fd;
577	3349x	sock.family_ = family;
578		// Mirror epoll/select: IPv6 sockets default to v6-only so they
579		// behave consistently across platforms regardless of the kernel
580		// default for /proc/sys/net/ipv6/bindv6only.
581	3349x	if (family == AF_INET6)
582		{
583	14x	int one = 1;
584	14x	::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &one, sizeof(one));
585		}
586	3349x	return {};
587		}
588
589		/** Adopt a pre-created fd into an impl.
590
591		Takes ownership of `fd` on success; the caller retains
592		ownership on failure.
593
594		@param impl The socket implementation to assign to.
595		@param fd A valid, open, non-blocking IP stream fd.
596		@return Error code on failure, empty on success.
597		*/
598	11x	std::error_code assign_socket(
599		tcp_socket::implementation& impl, native_handle_type fd) override
600		{
601	11x	auto& sock = static_cast<uring_tcp_socket&>(impl);
602	11x	int nfd = static_cast<int>(fd);
603		// The public assign() guarantees the object is closed.
604	11x	if (auto ec = validate_socket_fd(nfd, SOCK_STREAM, true))
605	8x	return ec;
606
607	3x	sock.fd_ = nfd;
608
609	3x	sock.local_endpoint_ = endpoint{};
610	3x	sock.remote_endpoint_ = endpoint{};
611
612	3x	sockaddr_storage local{};
613	3x	socklen_t local_len = sizeof(local);
614	3x	if (::getsockname(
615	3x	sock.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
616		{
617	3x	sock.local_endpoint_ = sockaddr_to_endpoint(local);
618	3x	sock.family_ = local.ss_family;
619		}
620	3x	sock.local_endpoint_state_.store(
621		uring_tcp_socket::endpoint_state::resolved,
622		std::memory_order_release);
623
624	3x	sockaddr_storage remote{};
625	3x	socklen_t remote_len = sizeof(remote);
626	3x	if (::getpeername(
627	3x	sock.fd_, reinterpret_cast<sockaddr*>(&remote), &remote_len) ==
628		0)
629	3x	sock.remote_endpoint_ = sockaddr_to_endpoint(remote);
630
631	3x	return {};
632		}
633
634		/** Bind the socket and capture the local endpoint via `getsockname`.
635
636		@param impl The socket implementation to bind.
637		@param ep The local endpoint to bind to.
638		@return Error code on failure, empty on success.
639		*/
640		std::error_code
641	9x	bind_socket(tcp_socket::implementation& impl, endpoint ep) override
642		{
643	9x	auto& sock = static_cast<uring_tcp_socket&>(impl);
644	9x	sockaddr_storage addr{};
645	9x	socklen_t len = endpoint_to_sockaddr(ep, addr);
646	9x	if (::bind(sock.fd_, reinterpret_cast<sockaddr*>(&addr), len) < 0)
647	4x	return make_err(errno);
648
649	5x	sockaddr_storage local{};
650	5x	socklen_t local_len = sizeof(local);
651	5x	if (::getsockname(
652	5x	sock.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
653	5x	sock.local_endpoint_ = sockaddr_to_endpoint(local);
654	5x	sock.local_endpoint_state_.store(
655		uring_tcp_socket::endpoint_state::resolved,
656		std::memory_order_release);
657	5x	return {};
658		}
659
660		/** Wrap an already-accepted fd as a new socket impl.
661
662		Called by the acceptor service (Task 17) after `accept(2)`
663		returns a connected fd. Captures both endpoints via the provided
664		peer address and a `getsockname` call.
665
666		@param fd Accepted file descriptor (must be non-blocking).
667		@param peer Peer endpoint from `accept(2)`.
668		@param family Address family of the accepted socket.
669		@return Raw pointer to the registered impl.
670		*/
671	3576x	uring_tcp_socket* adopt_fd(int fd, endpoint const& peer, int family)
672		{
673	3576x	auto* p = this->acquire_impl();
674	3576x	p->fd_ = fd;
675	3576x	p->family_ = family;
676	3576x	p->remote_endpoint_ = peer;
677		// Mark the local endpoint as authoritative-but-unresolved.
678		// The accessor will fetch it via getsockname on first call.
679		// Accept-heavy workloads that never query the local endpoint
680		// skip the syscall entirely.
681	3576x	p->local_endpoint_state_.store(
682		uring_tcp_socket::endpoint_state::lazy_pending,
683		std::memory_order_release);
684
685	3576x	return p;
686		}
687		};
688
689		inline void
690	10741x	uring_tcp_socket::retire() noexcept
691		{
692	10741x	svc_->pool_.recycle(this);
693	10741x	}
694
695		/** TCP acceptor implementation for io_uring.
696
697		Inherits the multishot machinery (parked-fd queue, waiter queue,
698		CQE drain on destruction) from `uring_multishot_acceptor_base`.
699		This class adds only the `accept()` override (matching
700		`tcp_acceptor::implementation`'s exact signature) and the
701		`adopt_thunk` static that wraps an accepted fd via
702		`uring_tcp_service::adopt_fd`.
703		*/
704		class BOOST_COROSIO_DECL uring_tcp_acceptor final
705		: public uring_multishot_acceptor_base<
706		uring_tcp_acceptor,
707		tcp_acceptor::implementation,
708		endpoint,
709		uring_tcp_service,
710		uring_tcp_acceptor_service>
711		{
712		friend uring_tcp_acceptor_service;
713
714		using base_type = uring_multishot_acceptor_base<
715		uring_tcp_acceptor,
716		tcp_acceptor::implementation,
717		endpoint,
718		uring_tcp_service,
719		uring_tcp_acceptor_service>;
720
721		// Readiness-wait slot. The multishot accept op delivers accepted
722		// fds, but `wait()` reports raw poll readiness on the listening fd
723		// without consuming a connection — see the wait() override.
724		uring_wait_op wait_op_;
725
726		public:
727	813x	explicit uring_tcp_acceptor(
728		uring_tcp_acceptor_service& svc,
729		uring_scheduler& sched,
730		uring_tcp_service& peer_svc) noexcept
731	813x	: base_type(svc, sched, peer_svc)
732		{
733	813x	}
734
735		/** Extend the base reset with this acceptor's own `wait_op_` slot.
736
737		@pre refs_ == 0, no wait in flight (see base_type::reuse()).
738		*/
739	466x	void reuse() noexcept
740		{
741	466x	base_type::reuse();
742	466x	BOOST_COROSIO_ASSERT(!wait_op_.stop_cb);
743	466x	}
744
745		/// Cancel this acceptor's readiness wait, if one is in flight.
746	1822x	void cancel_wait_op() noexcept
747		{
748	1822x	wait_op_.on_cancel();
749	1822x	}
750
751	4029x	std::coroutine_handle<> accept(
752		capy::continuation& cont,
753		capy::executor_ref ex,
754		std::stop_token token,
755		std::error_code* ec,
756		io_object::implementation** impl_out) override
757		{
758	4029x	base_type::dispatch_or_queue(cont, ex, token, ec, impl_out);
759	4029x	return std::noop_coroutine();
760		}
761
762	12x	std::coroutine_handle<> wait(
763		capy::continuation& cont,
764		capy::executor_ref ex,
765		wait_type w,
766		std::stop_token token,
767		std::error_code* ec) override
768		{
769		// Closed-object contract: complete with bad_file_descriptor
770		// instead of parking a waiter no accept machinery will signal.
771	12x	if (this->fd_ < 0)
772		{
773	1x	auto* op = this->acquire_node();
774	1x	op->cont = &cont;
775	1x	op->ex = ex;
776	1x	op->ec_out = ec;
777	1x	op->err = EBADF;
778	1x	this->sched_->post(op);
779	1x	return std::noop_coroutine();
780		}
781		// Multishot accepting drains the kernel queue as connections
782		// arrive, so a poll on the listener never reports it
783		// readable; read waits complete from the delivery queue.
784	11x	if (w == wait_type::read)
785		{
786	9x	this->park_read_wait(cont, ex, token, ec);
787	9x	return std::noop_coroutine();
788		}
789		// Writability carries no meaning for a listening socket;
790		// fail uniformly instead of never completing.
791	2x	if (w == wait_type::write)
792		{
793	1x	auto* op = this->acquire_node();
794	1x	op->cont = &cont;
795	1x	op->ex = ex;
796	1x	op->ec_out = ec;
797	1x	op->err = ENOTSUP;
798	1x	this->sched_->post(op);
799	1x	return std::noop_coroutine();
800		}
801		// Errors are not consumed by the accept machinery, so the
802		// error wait still polls the descriptor.
803	1x	wait_op_.prepare(
804	2x	cont, ex, ec, this->fd_, this->sched_, detail::object_ref(this),
805		POLLPRI \| POLLERR \| POLLHUP, token);
806	1x	this->sched_->work_started();
807	1x	if (wait_op_.cancelled.load(std::memory_order_acquire))
808		{
809	✗	uring_scheduler::lock_type lock(this->sched_->dispatch_mutex());
810	✗	this->sched_->push_completed_locked(&wait_op_);
811	✗	return std::noop_coroutine();
812	✗	}
813	1x	uring_submit_op(*this->sched_, &wait_op_);
814	1x	return std::noop_coroutine();
815		}
816
817	3576x	static io_object::implementation* adopt_thunk(
818		void* peer_service,
819		int fd,
820		sockaddr_storage const& peer,
821		socklen_t /peer_len/) noexcept
822		{
823	3576x	auto* svc = static_cast<uring_tcp_service*>(peer_service);
824	3576x	return svc->adopt_fd(fd, sockaddr_to_endpoint(peer), peer.ss_family);
825		}
826		};
827
828		/** TCP acceptor service for io_uring.
829
830		Owns all `uring_tcp_acceptor` implementations for an `io_context`.
831		Satisfies the `tcp_acceptor_service` interface so the generic
832		`tcp_acceptor` front-end can call `open_acceptor_socket`,
833		`bind_acceptor`, and `listen_acceptor` transparently.
834
835		Acceptor impls live in the service's `object_pool`. `construct()`
836		hands out a pooled impl with one reference already held; `destroy()`
837		releases it, and `retire()` recycles it onto the free list
838		instead of freeing it.
839
840		@par Thread Safety
841		All public member functions are thread-safe.
842		*/
843		class BOOST_COROSIO_DECL uring_tcp_acceptor_service final
844		: public tcp_acceptor_service
845		{
846		template<class, class, class, class, class>
847		friend class uring_multishot_acceptor_base;
848
849		public:
850		/// Identifies this service for `execution_context` lookup.
851		using key_type = tcp_acceptor_service;
852
853		/** Construct the TCP acceptor service.
854
855		@param ctx The owning execution context. Both the io_uring scheduler
856		and the TCP socket service must already be registered.
857		*/
858	390x	explicit uring_tcp_acceptor_service(capy::execution_context& ctx)
859	1170x	: sched_(&ctx.use_service<uring_scheduler>())
860	390x	, peer_svc_(&ctx.use_service<uring_tcp_service>())
861		{
862	390x	}
863
864	390x	void shutdown() override
865		{
866		// See uring_socket_service_base::shutdown(): snapshot under an
867		// acquired reference, cancel without the pool lock held;
868		// shutdown() sets shutting-down and takes the snapshot in one
869		// critical section, so a cancel that drops the last ref deletes
870		// rather than recycles.
871	390x	std::vector<uring_tcp_acceptor*> live;
872	390x	pool_.shutdown(
873	390x	[&](uring_tcp_acceptor* a)
874		{
875	569x	acquire(a);
876	569x	live.push_back(a);
877	569x	});
878	959x	for (auto* a : live)
879		{
880	569x	a->abort_all();
881	569x	release(a);
882		}
883	390x	}
884
885	1279x	io_object::implementation* construct() override
886		{
887	1279x	return acquire_impl();
888		}
889
890	1279x	void destroy(io_object::implementation* p) override
891		{
892	1279x	if (!p)
893	✗	return;
894	1279x	release(static_cast<uring_tcp_acceptor*>(p));
895		}
896
897		// Close the fd eagerly when tcp_acceptor::close() is called, before
898		// destroy() releases the service's reference and retire() may
899		// recycle this impl.
900	2522x	void close(io_object::handle& h) override
901		{
902	2522x	auto* acc = static_cast<uring_tcp_acceptor*>(h.get());
903	2522x	if (acc && acc->fd_ >= 0)
904		{
905		// Close only once no queued SQE can name the listener.
906	1243x	acc->flush_listener();
907	1243x	::close(acc->fd_);
908	1243x	acc->fd_ = -1;
909	1243x	acc->local_endpoint_ = endpoint{};
910		}
911	2522x	}
912
913		/** Create a non-blocking, close-on-exec socket for accepting.
914
915		@param impl The acceptor implementation to initialise.
916		@param family Address family (e.g. `AF_INET`, `AF_INET6`).
917		@param type Socket type (e.g. `SOCK_STREAM`).
918		@param protocol Protocol number (e.g. `IPPROTO_TCP`).
919		@return Error code on failure, empty on success.
920		*/
921	1247x	std::error_code open_acceptor_socket(
922		tcp_acceptor::implementation& impl,
923		int family,
924		int type,
925		int protocol) override
926		{
927	1247x	auto& acc = static_cast<uring_tcp_acceptor&>(impl);
928		int fd =
929	1247x	::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
930	1247x	if (fd < 0)
931	5x	return make_err(errno);
932		// LCOV_EXCL_START: dead — open() guards is_open(), so open_socket
933		// never runs against an already-open fd (assign_socket handles that).
934	−	if (acc.fd_ >= 0)
935		{
936	−	acc.flush_listener();
937	−	::close(acc.fd_);
938		}
939		// LCOV_EXCL_STOP
940	1242x	acc.fd_ = fd;
941		// Match epoll/select: IPv6 acceptors default to dual-stack
942		// (v6-only=false) so they accept both IPv4 and IPv6 connections.
943	1242x	if (family == AF_INET6)
944		{
945	12x	int zero = 0;
946	12x	::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &zero, sizeof(zero));
947		}
948	1242x	return {};
949		}
950
951		/** Adopt an already-listening descriptor.
952
953		@param impl The acceptor implementation to assign to.
954		@param fd The native socket to adopt.
955		@return Error code on failure, empty on success.
956		*/
957	9x	std::error_code assign_socket(
958		tcp_acceptor::implementation& impl, native_handle_type fd) override
959		{
960	9x	auto& acc = static_cast<uring_tcp_acceptor&>(impl);
961	9x	int nfd = static_cast<int>(fd);
962		// The public assign() guarantees the object is closed.
963	9x	if (auto ec = validate_socket_fd(nfd, SOCK_STREAM, true))
964	3x	return ec;
965
966	6x	acc.adopt_listening_fd(nfd);
967
968	6x	acc.local_endpoint_ = endpoint{};
969	6x	sockaddr_storage local{};
970	6x	socklen_t local_len = sizeof(local);
971	6x	if (::getsockname(
972	6x	nfd, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
973	6x	acc.local_endpoint_ = sockaddr_to_endpoint(local);
974
975	6x	if (fd_is_listening(nfd))
976	6x	acc.start_multishot();
977	6x	return {};
978		}
979
980		/** Bind an open acceptor and capture the local endpoint.
981
982		@param impl The acceptor implementation to bind.
983		@param ep The local endpoint to bind to.
984		@return Error code on failure, empty on success.
985		*/
986		std::error_code
987	1231x	bind_acceptor(tcp_acceptor::implementation& impl, endpoint ep) override
988		{
989	1231x	auto& acc = static_cast<uring_tcp_acceptor&>(impl);
990	1231x	sockaddr_storage addr{};
991	1231x	socklen_t len = endpoint_to_sockaddr(ep, addr);
992	1231x	if (::bind(acc.fd_, reinterpret_cast<sockaddr*>(&addr), len) < 0)
993	7x	return make_err(errno);
994
995	1224x	sockaddr_storage local{};
996	1224x	socklen_t local_len = sizeof(local);
997	1224x	if (::getsockname(
998	1224x	acc.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
999	1224x	acc.local_endpoint_ = sockaddr_to_endpoint(local);
1000	1224x	return {};
1001		}
1002
1003		/** Start listening and submit the multishot accept SQE.
1004
1005		Calls `::listen(2)` then arms the io_uring multishot accept
1006		operation that delivers one CQE per accepted connection.
1007
1008		@param impl The acceptor implementation to listen on.
1009		@param backlog Maximum pending-connection queue length.
1010		@return Error code on failure, empty on success.
1011		*/
1012		std::error_code
1013	1219x	listen_acceptor(tcp_acceptor::implementation& impl, int backlog) override
1014		{
1015	1219x	auto& acc = static_cast<uring_tcp_acceptor&>(impl);
1016	1219x	if (::listen(acc.fd_, backlog) < 0)
1017	3x	return make_err(errno);
1018	1216x	if (acc.prepare_listen_arm())
1019	1214x	acc.start_multishot();
1020	1216x	return {};
1021		}
1022
1023		/// Return the scheduler used by acceptors created by this service.
1024		uring_scheduler& scheduler() noexcept
1025		{
1026		return *sched_;
1027		}
1028
1029		private:
1030		/// Pop a recycled impl or news one, with the service reference held.
1031	1279x	uring_tcp_acceptor* acquire_impl()
1032		{
1033	1279x	return pool_.acquire(this, sched_, *peer_svc_);
1034		}
1035
1036		uring_scheduler* sched_;
1037		uring_tcp_service* peer_svc_;
1038		object_pool<uring_tcp_acceptor> pool_;
1039		};
1040
1041		/** Unix domain stream socket implementation for io_uring.
1042
1043		Implements `local_stream_socket::implementation` using a proactor
1044		model: read, write, and connect operations are submitted to the
1045		kernel via `uring_submit_op` and complete through the ring's
1046		CQE path.
1047
1048		The service holds one intrusive reference for as long as the impl
1049		is live; in-flight ops hold an additional `object_ref_` keepalive so
1050		the kernel's user-data pointer remains valid until the CQE arrives.
1051
1052		@par Thread Safety
1053		Distinct objects: Safe.
1054		Shared objects: Unsafe. A socket must not have two operations of
1055		the same type in flight simultaneously.
1056		*/
1057		class BOOST_COROSIO_DECL uring_local_stream_socket final
1058		: public native_socket_base<
1059		uring_local_stream_socket,
1060		local_stream_socket::implementation,
1061		corosio::local_endpoint>
1062		, public intrusive_list<uring_local_stream_socket>::node
1063		{
1064		friend uring_local_stream_service;
1065
1066		uring_scheduler* sched_ = nullptr;
1067		uring_local_stream_service* svc_ = nullptr;
1068
1069		// fd_ and local_endpoint_ live in native_socket_base, which also
1070		// provides native_handle/is_open/set_option/get_option/local_endpoint.
1071		corosio::local_endpoint remote_endpoint_;
1072
1073		// Per-fd op slots — embedded to eliminate per-call heap allocation.
1074		// Single-pending invariant per slot.
1075		uring_read_op rd_;
1076		uring_write_op wr_;
1077		uring_local_connect_op conn_;
1078		uring_wait_op wait_op_;
1079
1080		mutable detail::speculative_state spec_;
1081
1082		public:
1083		/** Construct with service and scheduler references.
1084
1085		Both refs must outlive this socket.
1086
1087		@param svc The owning service.
1088		@param sched The io_uring scheduler owned by the context.
1089		*/
1090	166x	explicit uring_local_stream_socket(
1091		uring_local_stream_service& svc, uring_scheduler& sched) noexcept
1092	332x	: sched_(&sched)
1093	166x	, svc_(&svc)
1094		{
1095	166x	}
1096
1097	332x	~uring_local_stream_socket() override
1098	166x	{
1099	166x	if (fd_ >= 0)
1100	✗	::close(
1101		fd_); // LCOV_EXCL_LINE backstop: close_socket() clears fd_ before destroy
1102	332x	}
1103
1104		// ----------------------------------------------------------------
1105		// io_stream::implementation
1106		// ----------------------------------------------------------------
1107
1108	103x	std::coroutine_handle<> read_some(
1109		capy::continuation& cont,
1110		capy::executor_ref ex,
1111		buffer_param buffers,
1112		std::stop_token token,
1113		std::error_code* ec,
1114		std::size_t* bytes) override
1115		{
1116		iovec iovecs[uring_max_iov];
1117	103x	int iovec_count = copy_to_iovec(buffers, iovecs);
1118	103x	bool stop_now = token.stop_possible() && token.stop_requested();
1119	103x	bool empty_buf = (iovec_count == 0);
1120
1121	103x	ssize_t n = 0;
1122	103x	int err = 0;
1123	103x	bool have_sync_res = stop_now \|\| empty_buf;
1124	103x	if (!have_sync_res && spec_.may_speculate_read())
1125		{
1126	101x	n = uring_direct_read(fd_, iovecs, iovec_count);
1127	101x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
1128		{
1129	30x	have_sync_res = true;
1130	30x	if (n < 0)
1131	2x	err = errno;
1132		// Speculative read produced a definitive answer (data
1133		// or non-EAGAIN error); reset the failure streak so a
1134		// burst of past EAGAINs doesn't latch perma-off when
1135		// the workload is in fact speculation-friendly.
1136	30x	if (n >= 0)
1137	28x	spec_.on_read_success();
1138		}
1139		else
1140		{
1141	71x	spec_.on_read_exhausted();
1142		}
1143		}
1144
1145	103x	if (have_sync_res)
1146		{
1147	31x	if (sched_->try_consume_inline_budget())
1148		{
1149	30x	decode_results(
1150		ec, bytes, stop_now,
1151	2x	err ? make_err(err) : std::error_code{},
1152	30x	/is_read=/true, n < 0 ? 0u : static_cast<std::size_t>(n),
1153		empty_buf);
1154	30x	return dispatch_coro(ex, cont);
1155		}
1156	1x	rd_.prepare(
1157	2x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this),
1158		&spec_, buffers, token);
1159	1x	if (stop_now)
1160	✗	rd_.cancelled.store(true, std::memory_order_release);
1161		else
1162	1x	rd_.res = (n < 0) ? -err : static_cast<int>(n);
1163	1x	sched_->work_started();
1164		{
1165	1x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
1166	1x	sched_->push_completed_locked(&rd_);
1167	1x	}
1168	1x	return std::noop_coroutine();
1169		}
1170
1171	72x	rd_.prepare(
1172	144x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this), &spec_,
1173		buffers, token);
1174	72x	sched_->work_started();
1175	72x	if (rd_.cancelled.load(std::memory_order_acquire))
1176		{
1177	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
1178	✗	sched_->push_completed_locked(&rd_);
1179	✗	return std::noop_coroutine();
1180	✗	}
1181	72x	uring_submit_op(*sched_, &rd_);
1182	72x	return std::noop_coroutine();
1183		}
1184
1185	41x	std::coroutine_handle<> write_some(
1186		capy::continuation& cont,
1187		capy::executor_ref ex,
1188		buffer_param buffers,
1189		std::stop_token token,
1190		std::error_code* ec,
1191		std::size_t* bytes) override
1192		{
1193		iovec iovecs[uring_max_iov];
1194	41x	int iovec_count = copy_to_iovec(buffers, iovecs);
1195	41x	bool stop_now = token.stop_possible() && token.stop_requested();
1196	41x	bool empty_buf = (iovec_count == 0);
1197
1198	41x	ssize_t n = 0;
1199	41x	int err = 0;
1200	41x	bool have_sync_res = stop_now \|\| empty_buf;
1201		// A queued batch enters the kernel anyway; joining it spares
1202		// this write a syscall of its own.
1203	81x	if (!have_sync_res && spec_.may_speculate_write() &&
1204	40x	!sched_->write_may_join_batch())
1205		{
1206	31x	n = uring_direct_write(fd_, iovecs, iovec_count);
1207	31x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
1208		{
1209	29x	have_sync_res = true;
1210	29x	if (n < 0)
1211	2x	err = errno;
1212		}
1213		else
1214		{
1215	2x	spec_.on_write_exhausted();
1216		}
1217		}
1218
1219	41x	if (have_sync_res)
1220		{
1221	30x	if (sched_->try_consume_inline_budget())
1222		{
1223	29x	decode_results(
1224		ec, bytes, stop_now,
1225	2x	err ? make_err(err) : std::error_code{},
1226	29x	/is_read=/false, n < 0 ? 0u : static_cast<std::size_t>(n),
1227		/empty_buffer=/false);
1228	29x	return dispatch_coro(ex, cont);
1229		}
1230	1x	wr_.prepare(
1231	2x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this),
1232		&spec_, buffers, token);
1233	1x	if (stop_now)
1234	✗	wr_.cancelled.store(true, std::memory_order_release);
1235		else
1236	1x	wr_.res = (n < 0) ? -err : static_cast<int>(n);
1237	1x	sched_->work_started();
1238		{
1239	1x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
1240	1x	sched_->push_completed_locked(&wr_);
1241	1x	}
1242	1x	return std::noop_coroutine();
1243		}
1244
1245	11x	wr_.prepare(
1246	22x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this), &spec_,
1247		buffers, token);
1248	11x	sched_->work_started();
1249	11x	if (wr_.cancelled.load(std::memory_order_acquire))
1250		{
1251	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
1252	✗	sched_->push_completed_locked(&wr_);
1253	✗	return std::noop_coroutine();
1254	✗	}
1255	11x	uring_submit_op(*sched_, &wr_);
1256	11x	return std::noop_coroutine();
1257		}
1258
1259		// ----------------------------------------------------------------
1260		// local_stream_socket::implementation
1261		// ----------------------------------------------------------------
1262
1263	17x	std::coroutine_handle<> connect(
1264		capy::continuation& cont,
1265		capy::executor_ref ex,
1266		corosio::local_endpoint ep,
1267		std::stop_token token,
1268		std::error_code* ec) override
1269		{
1270	17x	bool stop_now = token.stop_possible() && token.stop_requested();
1271	17x	if (stop_now)
1272		{
1273	✗	if (sched_->try_consume_inline_budget())
1274		{
1275	✗	if (ec)
1276	✗	*ec = capy::error::canceled;
1277	✗	return dispatch_coro(ex, cont);
1278		}
1279	✗	conn_.addrlen = to_sockaddr(ep, conn_.addr);
1280	✗	conn_.prepare(
1281	✗	cont, ex, ec, fd_, sched_, detail::object_ref(this), ep,
1282		&remote_endpoint_, &local_endpoint_, token);
1283	✗	conn_.cancelled.store(true, std::memory_order_release);
1284	✗	sched_->work_started();
1285		{
1286	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
1287	✗	sched_->push_completed_locked(&conn_);
1288	✗	}
1289	✗	return std::noop_coroutine();
1290		}
1291
1292		// A speculative ::connect would leave the fd in EINPROGRESS and
1293		// a subsequent IORING_OP_CONNECT would see EALREADY — avoid.
1294	17x	conn_.addrlen = to_sockaddr(ep, conn_.addr);
1295	17x	conn_.prepare(
1296	34x	cont, ex, ec, fd_, sched_, detail::object_ref(this), ep,
1297		&remote_endpoint_, &local_endpoint_, token);
1298	17x	sched_->work_started();
1299	17x	if (conn_.cancelled.load(std::memory_order_acquire))
1300		{
1301	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
1302	✗	sched_->push_completed_locked(&conn_);
1303	✗	return std::noop_coroutine();
1304	✗	}
1305	17x	uring_submit_op(*sched_, &conn_);
1306	17x	return std::noop_coroutine();
1307		}
1308
1309	8x	std::coroutine_handle<> wait(
1310		capy::continuation& cont,
1311		capy::executor_ref ex,
1312		wait_type w,
1313		std::stop_token token,
1314		std::error_code* ec) override
1315		{
1316	8x	int poll_flags = 0;
1317	8x	switch (w)
1318		{
1319	6x	case wait_type::read:
1320	6x	poll_flags = POLLIN;
1321	6x	break;
1322	1x	case wait_type::write:
1323	1x	poll_flags = POLLOUT;
1324	1x	break;
1325	1x	case wait_type::error:
1326	1x	poll_flags = POLLPRI \| POLLERR \| POLLHUP;
1327	1x	break;
1328		}
1329	8x	wait_op_.prepare(
1330	16x	cont, ex, ec, fd_, sched_, detail::object_ref(this), poll_flags,
1331		token);
1332	8x	sched_->work_started();
1333	8x	if (wait_op_.cancelled.load(std::memory_order_acquire))
1334		{
1335	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
1336	✗	sched_->push_completed_locked(&wait_op_);
1337	✗	return std::noop_coroutine();
1338	✗	}
1339	8x	uring_submit_op(*sched_, &wait_op_);
1340	8x	return std::noop_coroutine();
1341		}
1342
1343		std::error_code
1344	3x	shutdown(local_stream_socket::shutdown_type what) noexcept override
1345		{
1346	3x	if (::shutdown(fd_, static_cast<int>(what)) != 0)
1347	1x	return make_err(errno);
1348	2x	return {};
1349		}
1350
1351		// native_handle / is_open / set_option / get_option / local_endpoint
1352		// are inherited from native_socket_base.
1353
1354	3x	native_handle_type release_socket() noexcept override
1355		{
1356		// Flush while the fd is still open so the kernel resolves
1357		// pending SQEs before the caller can close and recycle the
1358		// number (same reasoning as close_socket).
1359		// A connect completing after this belongs to the descriptor
1360		// the caller takes; see close_socket().
1361	3x	conn_.request_cancel();
1362	3x	wait_op_.request_cancel();
1363	3x	if (fd_ >= 0)
1364	3x	sched_->flush_after_cancel(fd_, {&rd_, &wr_, &conn_, &wait_op_});
1365	3x	int fd = fd_;
1366	3x	fd_ = -1;
1367	3x	family_ = AF_UNSPEC;
1368	3x	local_endpoint_ = corosio::local_endpoint{};
1369	3x	remote_endpoint_ = corosio::local_endpoint{};
1370	3x	return fd;
1371		}
1372
1373	11x	void cancel() noexcept override
1374		{
1375	11x	if (fd_ >= 0)
1376	3x	sched_->cancel_ops({&rd_, &wr_, &conn_, &wait_op_});
1377	11x	}
1378
1379		/// Cancel in-flight ops, close the fd, and reset cached endpoints.
1380		/// Called by the service on close()/teardown.
1381	428x	void close_socket() noexcept
1382		{
1383		// A connect completing after this must not write endpoints into
1384		// a closed impl that may be recycled.
1385	428x	conn_.request_cancel();
1386	428x	wait_op_.request_cancel();
1387	428x	if (fd_ >= 0)
1388		{
1389	168x	sched_->flush_after_cancel(fd_, {&rd_, &wr_, &conn_, &wait_op_});
1390	168x	::close(fd_);
1391	168x	fd_ = -1;
1392		}
1393	428x	family_ = AF_UNSPEC;
1394	428x	local_endpoint_ = corosio::local_endpoint{};
1395	428x	remote_endpoint_ = corosio::local_endpoint{};
1396	428x	}
1397
1398		/// Recycle into the owning service's pool. See
1399		/// uring_tcp_socket::retire() for why this is out-of-line.
1400		void retire() noexcept override;
1401
1402		/** Reset op slots, cached endpoints, and the speculation hint for
1403		recycling. See uring_tcp_socket::reuse() for the rationale.
1404
1405		@pre refs_ == 0, fd closed, no op in flight.
1406		*/
1407	98x	void reuse() noexcept
1408		{
1409	98x	BOOST_COROSIO_ASSERT(fd_ == -1);
1410		// A connect completing on another thread while the socket closed
1411		// can still have written these.
1412	98x	local_endpoint_ = corosio::local_endpoint{};
1413	98x	remote_endpoint_ = corosio::local_endpoint{};
1414	98x	BOOST_COROSIO_ASSERT(!rd_.stop_cb);
1415	98x	BOOST_COROSIO_ASSERT(!wr_.stop_cb);
1416	98x	BOOST_COROSIO_ASSERT(!conn_.stop_cb);
1417	98x	BOOST_COROSIO_ASSERT(!wait_op_.stop_cb);
1418	98x	spec_.reset();
1419	98x	}
1420
1421	2x	corosio::local_endpoint remote_endpoint() const noexcept override
1422		{
1423	2x	return remote_endpoint_;
1424		}
1425		};
1426
1427		/** Unix domain stream socket service for io_uring.
1428
1429		Owns all `uring_local_stream_socket` implementations for an
1430		`io_context`. Satisfies the `local_stream_service` interface so the
1431		generic `local_stream_socket` front-end can call `open_socket` and
1432		`assign_socket` transparently.
1433
1434		Socket impls live in the service's `object_pool`. `construct()`
1435		hands out a pooled impl with one reference already held; `destroy()`
1436		releases it, and `retire()` recycles it onto the free list
1437		instead of freeing it.
1438
1439		@par Thread Safety
1440		All public member functions are thread-safe.
1441		*/
1442		class BOOST_COROSIO_DECL uring_local_stream_service final
1443		: public uring_socket_service_base<
1444		uring_local_stream_service,
1445		local_stream_service,
1446		uring_local_stream_socket>
1447		{
1448		using base_service = uring_socket_service_base<
1449		uring_local_stream_service,
1450		local_stream_service,
1451		uring_local_stream_socket>;
1452
1453		public:
1454		/// Identifies this service for `execution_context` lookup.
1455		using key_type = local_stream_service;
1456
1457		/** Construct the local stream service.
1458
1459		@param ctx The owning execution context. The io_uring scheduler
1460		must already be registered.
1461		*/
1462	115x	explicit uring_local_stream_service(capy::execution_context& ctx)
1463	115x	: base_service(ctx)
1464		{
1465	115x	}
1466
1467		// construct / destroy / shutdown / close / scheduler() are inherited
1468		// from uring_socket_service_base.
1469
1470		/** Open an AF_UNIX stream socket and associate it with an impl.
1471
1472		Creates a non-blocking, close-on-exec socket via `socket(2)`.
1473		`family` is always `AF_UNIX` for local stream sockets.
1474
1475		@param impl The socket implementation to initialise.
1476		@param family Address family (`AF_UNIX`).
1477		@param type Socket type (`SOCK_STREAM`).
1478		@param protocol Protocol number (typically 0).
1479		@return Error code on failure, empty on success.
1480		*/
1481	30x	std::error_code open_socket(
1482		local_stream_socket::implementation& impl,
1483		int family,
1484		int type,
1485		int protocol) override
1486		{
1487	30x	auto& sock = static_cast<uring_local_stream_socket&>(impl);
1488		int fd =
1489	30x	::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
1490	30x	if (fd < 0)
1491	1x	return make_err(errno);
1492		// LCOV_EXCL_START: dead — open() guards is_open(), so open_socket
1493		// never runs against an already-open fd (assign_socket handles that).
1494	−	if (sock.fd_ >= 0)
1495	−	sock.close_socket();
1496		// LCOV_EXCL_STOP
1497	29x	sock.fd_ = fd;
1498	29x	sock.family_ = AF_UNIX;
1499	29x	return {};
1500		}
1501
1502		/** Adopt a pre-created fd into an impl (e.g. from `socketpair`).
1503
1504		Takes ownership of `fd` on success; the caller retains ownership
1505		on failure.
1506
1507		@param impl The socket implementation to assign to.
1508		@param fd A valid, open, non-blocking AF_UNIX stream fd.
1509		@return Error code on failure, empty on success.
1510		*/
1511	131x	std::error_code assign_socket(
1512		local_stream_socket::implementation& impl,
1513		native_handle_type fd) override
1514		{
1515	131x	auto& sock = static_cast<uring_local_stream_socket&>(impl);
1516	131x	int nfd = static_cast<int>(fd);
1517		// The public assign() guarantees the object is closed.
1518	131x	if (auto ec = validate_socket_fd(nfd, SOCK_STREAM, false))
1519	6x	return ec;
1520
1521	125x	sock.fd_ = nfd;
1522	125x	sock.family_ = AF_UNIX;
1523
1524	125x	sockaddr_storage local{};
1525	125x	socklen_t local_len = sizeof(local);
1526	125x	if (::getsockname(
1527	125x	sock.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
1528	125x	sock.local_endpoint_ = sockaddr_to_local_endpoint(local, local_len);
1529
1530	125x	sockaddr_storage remote{};
1531	125x	socklen_t remote_len = sizeof(remote);
1532	125x	if (::getpeername(
1533	125x	sock.fd_, reinterpret_cast<sockaddr*>(&remote), &remote_len) ==
1534		0)
1535		sock.remote_endpoint_ =
1536	125x	sockaddr_to_local_endpoint(remote, remote_len);
1537
1538	125x	return {};
1539		}
1540
1541		/** Wrap an already-accepted fd as a new socket impl.
1542
1543		Called by the acceptor service after `accept(2)` returns a
1544		connected fd. Captures both endpoints via the provided peer
1545		address and a `getsockname` call.
1546
1547		@param fd Accepted file descriptor (must be non-blocking).
1548		@param peer Peer endpoint from `accept(2)`.
1549		@return Raw pointer to the registered impl.
1550		*/
1551		uring_local_stream_socket*
1552	17x	adopt_fd(int fd, corosio::local_endpoint const& peer)
1553		{
1554	17x	auto* p = this->acquire_impl();
1555	17x	p->fd_ = fd;
1556	17x	p->family_ = AF_UNIX;
1557	17x	p->remote_endpoint_ = peer;
1558
1559	17x	sockaddr_storage local{};
1560	17x	socklen_t len = sizeof(local);
1561	17x	if (::getsockname(fd, reinterpret_cast<sockaddr*>(&local), &len) == 0)
1562	17x	p->local_endpoint_ = sockaddr_to_local_endpoint(local, len);
1563
1564	17x	return p;
1565		}
1566		};
1567
1568		inline void
1569	264x	uring_local_stream_socket::retire() noexcept
1570		{
1571	264x	svc_->pool_.recycle(this);
1572	264x	}
1573
1574		/** Local-stream (Unix domain) acceptor for io_uring.
1575
1576		Inherits all multishot machinery (parked-fd queue, waiter queue,
1577		descriptor release, CQE drain on destruction) from
1578		`uring_multishot_acceptor_base`. Adds only the `accept()`
1579		override and the `adopt_thunk` static that wraps an accepted fd
1580		via `uring_local_stream_service::adopt_fd`.
1581		*/
1582		class BOOST_COROSIO_DECL uring_local_stream_acceptor final
1583		: public uring_multishot_acceptor_base<
1584		uring_local_stream_acceptor,
1585		local_stream_acceptor::implementation,
1586		corosio::local_endpoint,
1587		uring_local_stream_service,
1588		uring_local_stream_acceptor_service>
1589		{
1590		friend uring_local_stream_acceptor_service;
1591
1592		using base_type = uring_multishot_acceptor_base<
1593		uring_local_stream_acceptor,
1594		local_stream_acceptor::implementation,
1595		corosio::local_endpoint,
1596		uring_local_stream_service,
1597		uring_local_stream_acceptor_service>;
1598
1599		// Readiness-wait slot. See uring_tcp_acceptor::wait_op_.
1600		uring_wait_op wait_op_;
1601
1602		public:
1603	60x	explicit uring_local_stream_acceptor(
1604		uring_local_stream_acceptor_service& svc,
1605		uring_scheduler& sched,
1606		uring_local_stream_service& peer_svc) noexcept
1607	60x	: base_type(svc, sched, peer_svc)
1608		{
1609	60x	}
1610
1611		/** Extend the base reset with this acceptor's own `wait_op_` slot.
1612
1613		@pre refs_ == 0, no wait in flight (see base_type::reuse()).
1614		*/
1615	36x	void reuse() noexcept
1616		{
1617	36x	base_type::reuse();
1618	36x	BOOST_COROSIO_ASSERT(!wait_op_.stop_cb);
1619	36x	}
1620
1621		/// Cancel this acceptor's readiness wait, if one is in flight.
1622	87x	void cancel_wait_op() noexcept
1623		{
1624	87x	wait_op_.on_cancel();
1625	87x	}
1626
1627	22x	std::coroutine_handle<> accept(
1628		capy::continuation& cont,
1629		capy::executor_ref ex,
1630		std::stop_token token,
1631		std::error_code* ec,
1632		io_object::implementation** impl_out) override
1633		{
1634	22x	base_type::dispatch_or_queue(cont, ex, token, ec, impl_out);
1635	22x	return std::noop_coroutine();
1636		}
1637
1638	5x	std::coroutine_handle<> wait(
1639		capy::continuation& cont,
1640		capy::executor_ref ex,
1641		wait_type w,
1642		std::stop_token token,
1643		std::error_code* ec) override
1644		{
1645		// Closed-object contract: complete with bad_file_descriptor
1646		// instead of parking a waiter no accept machinery will signal.
1647	5x	if (this->fd_ < 0)
1648		{
1649	1x	auto* op = this->acquire_node();
1650	1x	op->cont = &cont;
1651	1x	op->ex = ex;
1652	1x	op->ec_out = ec;
1653	1x	op->err = EBADF;
1654	1x	this->sched_->post(op);
1655	1x	return std::noop_coroutine();
1656		}
1657		// Multishot accepting drains the kernel queue as connections
1658		// arrive, so a poll on the listener never reports it
1659		// readable; read waits complete from the delivery queue.
1660	4x	if (w == wait_type::read)
1661		{
1662	2x	this->park_read_wait(cont, ex, token, ec);
1663	2x	return std::noop_coroutine();
1664		}
1665		// Writability carries no meaning for a listening socket;
1666		// fail uniformly instead of never completing.
1667	2x	if (w == wait_type::write)
1668		{
1669	1x	auto* op = this->acquire_node();
1670	1x	op->cont = &cont;
1671	1x	op->ex = ex;
1672	1x	op->ec_out = ec;
1673	1x	op->err = ENOTSUP;
1674	1x	this->sched_->post(op);
1675	1x	return std::noop_coroutine();
1676		}
1677		// Errors are not consumed by the accept machinery, so the
1678		// error wait still polls the descriptor.
1679	1x	wait_op_.prepare(
1680	2x	cont, ex, ec, this->fd_, this->sched_, detail::object_ref(this),
1681		POLLPRI \| POLLERR \| POLLHUP, token);
1682	1x	this->sched_->work_started();
1683	1x	if (wait_op_.cancelled.load(std::memory_order_acquire))
1684		{
1685	✗	uring_scheduler::lock_type lock(this->sched_->dispatch_mutex());
1686	✗	this->sched_->push_completed_locked(&wait_op_);
1687	✗	return std::noop_coroutine();
1688	✗	}
1689	1x	uring_submit_op(*this->sched_, &wait_op_);
1690	1x	return std::noop_coroutine();
1691		}
1692
1693	17x	static io_object::implementation* adopt_thunk(
1694		void* peer_service,
1695		int fd,
1696		sockaddr_storage const& peer,
1697		socklen_t peer_len) noexcept
1698		{
1699	17x	auto* svc = static_cast<uring_local_stream_service*>(peer_service);
1700	17x	return svc->adopt_fd(fd, sockaddr_to_local_endpoint(peer, peer_len));
1701		}
1702		};
1703
1704		/** Unix domain stream acceptor service for io_uring.
1705
1706		Owns all `uring_local_stream_acceptor` implementations for an
1707		`io_context`. Satisfies the `local_stream_acceptor_service` interface
1708		so the generic `local_stream_acceptor` front-end can call
1709		`open_acceptor_socket`, `bind_acceptor`, and `listen_acceptor`
1710		transparently.
1711
1712		Acceptor impls live in the service's `object_pool`. `construct()`
1713		hands out a pooled impl with one reference already held; `destroy()`
1714		releases it, and `retire()` recycles it onto the free list
1715		instead of freeing it.
1716
1717		@par Thread Safety
1718		All public member functions are thread-safe.
1719		*/
1720		class BOOST_COROSIO_DECL uring_local_stream_acceptor_service final
1721		: public local_stream_acceptor_service
1722		{
1723		template<class, class, class, class, class>
1724		friend class uring_multishot_acceptor_base;
1725
1726		public:
1727		/// Identifies this service for `execution_context` lookup.
1728		using key_type = local_stream_acceptor_service;
1729
1730		/** Construct the local stream acceptor service.
1731
1732		@param ctx The owning execution context. Both the io_uring scheduler
1733		and the local stream socket service must already be registered.
1734		*/
1735	54x	explicit uring_local_stream_acceptor_service(capy::execution_context& ctx)
1736	162x	: sched_(&ctx.use_service<uring_scheduler>())
1737	54x	, peer_svc_(&ctx.use_service<uring_local_stream_service>())
1738		{
1739	54x	}
1740
1741	54x	void shutdown() override
1742		{
1743		// See uring_tcp_acceptor_service::shutdown() for the ordering
1744		// rationale.
1745	54x	std::vector<uring_local_stream_acceptor*> live;
1746	54x	pool_.shutdown(
1747	54x	[&](uring_local_stream_acceptor* a)
1748		{
1749	32x	acquire(a);
1750	32x	live.push_back(a);
1751	32x	});
1752	86x	for (auto* a : live)
1753		{
1754	32x	a->abort_all();
1755	32x	release(a);
1756		}
1757	54x	}
1758
1759	96x	io_object::implementation* construct() override
1760		{
1761	96x	return acquire_impl();
1762		}
1763
1764	96x	void destroy(io_object::implementation* p) override
1765		{
1766	96x	if (!p)
1767	✗	return;
1768	96x	release(static_cast<uring_local_stream_acceptor*>(p));
1769		}
1770
1771		// Close the fd eagerly when local_stream_acceptor::close() is called,
1772		// before destroy() drops the service's reference and recycling runs.
1773	144x	void close(io_object::handle& h) override
1774		{
1775	144x	auto* acc = static_cast<uring_local_stream_acceptor*>(h.get());
1776	144x	if (acc && acc->fd_ >= 0)
1777		{
1778		// See uring_tcp_acceptor_service::close.
1779	48x	acc->flush_listener();
1780	48x	::close(acc->fd_);
1781	48x	acc->fd_ = -1;
1782	48x	acc->local_endpoint_ = corosio::local_endpoint{};
1783		}
1784	144x	}
1785
1786		/** Create a non-blocking, close-on-exec AF_UNIX socket for accepting.
1787
1788		@param impl The acceptor implementation to initialise.
1789		@param family Address family (`AF_UNIX`).
1790		@param type Socket type (`SOCK_STREAM`).
1791		@param protocol Protocol number (typically 0).
1792		@return Error code on failure, empty on success.
1793		*/
1794	53x	std::error_code open_acceptor_socket(
1795		local_stream_acceptor::implementation& impl,
1796		int family,
1797		int type,
1798		int protocol) override
1799		{
1800	53x	auto& acc = static_cast<uring_local_stream_acceptor&>(impl);
1801		int fd =
1802	53x	::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
1803	53x	if (fd < 0)
1804	3x	return make_err(errno);
1805		// LCOV_EXCL_START: dead — open() guards is_open(), so open_socket
1806		// never runs against an already-open fd (assign_socket handles that).
1807	−	if (acc.fd_ >= 0)
1808		{
1809	−	acc.flush_listener();
1810	−	::close(acc.fd_);
1811		}
1812		// LCOV_EXCL_STOP
1813	50x	acc.fd_ = fd;
1814	50x	return {};
1815		}
1816
1817		/** Adopt an already-listening descriptor.
1818
1819		@param impl The acceptor implementation to assign to.
1820		@param fd The native socket to adopt.
1821		@return Error code on failure, empty on success.
1822		*/
1823	3x	std::error_code assign_socket(
1824		local_stream_acceptor::implementation& impl,
1825		native_handle_type fd) override
1826		{
1827	3x	auto& acc = static_cast<uring_local_stream_acceptor&>(impl);
1828	3x	int nfd = static_cast<int>(fd);
1829		// The public assign() guarantees the object is closed.
1830	3x	if (auto ec = validate_socket_fd(nfd, SOCK_STREAM, false))
1831	1x	return ec;
1832
1833	2x	acc.adopt_listening_fd(nfd);
1834
1835	2x	acc.local_endpoint_ = corosio::local_endpoint{};
1836	2x	sockaddr_storage local{};
1837	2x	socklen_t local_len = sizeof(local);
1838	2x	if (::getsockname(
1839	2x	nfd, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
1840	2x	acc.local_endpoint_ = sockaddr_to_local_endpoint(local, local_len);
1841
1842	2x	if (fd_is_listening(nfd))
1843	2x	acc.start_multishot();
1844	2x	return {};
1845		}
1846
1847		/** Bind an open acceptor and capture the local endpoint.
1848
1849		@param impl The acceptor implementation to bind.
1850		@param ep The local endpoint (path) to bind to.
1851		@return Error code on failure, empty on success.
1852		*/
1853	45x	std::error_code bind_acceptor(
1854		local_stream_acceptor::implementation& impl,
1855		corosio::local_endpoint ep) override
1856		{
1857	45x	auto& acc = static_cast<uring_local_stream_acceptor&>(impl);
1858	45x	sockaddr_storage addr{};
1859	45x	socklen_t len = endpoint_to_sockaddr(ep, addr);
1860	45x	if (::bind(acc.fd_, reinterpret_cast<sockaddr*>(&addr), len) < 0)
1861	3x	return make_err(errno);
1862
1863	42x	sockaddr_storage local{};
1864	42x	socklen_t local_len = sizeof(local);
1865	42x	if (::getsockname(
1866	42x	acc.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
1867	42x	acc.local_endpoint_ = sockaddr_to_local_endpoint(local, local_len);
1868	42x	return {};
1869		}
1870
1871		/** Start listening and submit the multishot accept SQE.
1872
1873		Calls `::listen(2)` then arms the io_uring multishot accept
1874		operation that delivers one CQE per accepted connection.
1875
1876		@param impl The acceptor implementation to listen on.
1877		@param backlog Maximum pending-connection queue length.
1878		@return Error code on failure, empty on success.
1879		*/
1880	37x	std::error_code listen_acceptor(
1881		local_stream_acceptor::implementation& impl, int backlog) override
1882		{
1883	37x	auto& acc = static_cast<uring_local_stream_acceptor&>(impl);
1884	37x	if (::listen(acc.fd_, backlog) < 0)
1885	3x	return make_err(errno);
1886	34x	if (acc.prepare_listen_arm())
1887	34x	acc.start_multishot();
1888	34x	return {};
1889		}
1890
1891		/// Return the scheduler used by acceptors created by this service.
1892		uring_scheduler& scheduler() noexcept
1893		{
1894		return *sched_;
1895		}
1896
1897		private:
1898		/// Pop a recycled impl or news one, with the service reference held.
1899	96x	uring_local_stream_acceptor* acquire_impl()
1900		{
1901	96x	return pool_.acquire(this, sched_, *peer_svc_);
1902		}
1903
1904		uring_scheduler* sched_;
1905		uring_local_stream_service* peer_svc_;
1906		object_pool<uring_local_stream_acceptor> pool_;
1907		};
1908
1909		/** UDP socket implementation for io_uring.
1910
1911		Implements `udp_socket::implementation` using a proactor model:
1912		send_to, recv_from, send, recv, and connect operations are submitted
1913		to the kernel via `uring_submit_op` and complete through the ring's
1914		CQE path.
1915
1916		The service holds one intrusive reference for as long as the impl
1917		is live; in-flight ops hold an additional `object_ref_` keepalive so
1918		the kernel's user-data pointer remains valid until the CQE arrives.
1919
1920		@par Thread Safety
1921		Distinct objects: Safe.
1922		Shared objects: Unsafe. One send and one recv may be in flight
1923		simultaneously, but two sends or two recvs must not overlap.
1924		*/
1925		class BOOST_COROSIO_DECL uring_udp_socket final
1926		: public native_socket_base<
1927		uring_udp_socket,
1928		udp_socket::implementation,
1929		corosio::endpoint>
1930		, public intrusive_list<uring_udp_socket>::node
1931		{
1932		friend uring_udp_service;
1933
1934		uring_scheduler* sched_ = nullptr;
1935		uring_udp_service* svc_ = nullptr;
1936
1937		// fd_ and local_endpoint_ live in native_socket_base, which also
1938		// provides native_handle/is_open/set_option/get_option/local_endpoint.
1939		corosio::endpoint remote_endpoint_;
1940
1941		// Per-fd op slots — embedded to eliminate per-call heap allocation.
1942		// Single-pending invariant per slot.
1943		uring_connect_op conn_;
1944		uring_dgram_send_op send_;
1945		uring_dgram_recv_op recv_;
1946		uring_wait_op wait_op_;
1947
1948		mutable detail::speculative_state spec_;
1949
1950		public:
1951		/** Construct with service and scheduler references.
1952
1953		Both refs must outlive this socket.
1954
1955		@param svc The owning service.
1956		@param sched The io_uring scheduler owned by the context.
1957		*/
1958	149x	explicit uring_udp_socket(
1959		uring_udp_service& svc, uring_scheduler& sched) noexcept
1960	298x	: sched_(&sched)
1961	149x	, svc_(&svc)
1962		{
1963	149x	}
1964
1965	298x	~uring_udp_socket() override
1966	149x	{
1967	149x	if (fd_ >= 0)
1968	✗	::close(
1969		fd_); // LCOV_EXCL_LINE backstop: close_socket() clears fd_ before destroy
1970	298x	}
1971
1972		// ----------------------------------------------------------------
1973		// udp_socket::implementation
1974		// ----------------------------------------------------------------
1975
1976	33x	std::coroutine_handle<> send_to(
1977		capy::continuation& cont,
1978		capy::executor_ref ex,
1979		buffer_param buf,
1980		endpoint dest,
1981		int flags,
1982		std::stop_token token,
1983		std::error_code* ec,
1984		std::size_t* bytes_out) override
1985		{
1986	33x	sockaddr_storage addr{};
1987	33x	socklen_t len = endpoint_to_sockaddr(dest, addr);
1988	33x	return submit_send(
1989	66x	cont, ex, buf, len, addr, flags, token, ec, bytes_out);
1990		}
1991
1992	86x	std::coroutine_handle<> recv_from(
1993		capy::continuation& cont,
1994		capy::executor_ref ex,
1995		buffer_param buf,
1996		endpoint* source,
1997		int flags,
1998		std::stop_token token,
1999		std::error_code* ec,
2000		std::size_t* bytes_out) override
2001		{
2002	86x	return submit_recv(
2003		cont, ex, buf, source != nullptr, source, flags, token, ec,
2004	86x	bytes_out);
2005		}
2006
2007	128x	std::coroutine_handle<> send(
2008		capy::continuation& cont,
2009		capy::executor_ref ex,
2010		buffer_param buf,
2011		int flags,
2012		std::stop_token token,
2013		std::error_code* ec,
2014		std::size_t* bytes_out) override
2015		{
2016	128x	sockaddr_storage empty{};
2017	128x	return submit_send(
2018	256x	cont, ex, buf, 0, empty, flags, token, ec, bytes_out);
2019		}
2020
2021	47x	std::coroutine_handle<> recv(
2022		capy::continuation& cont,
2023		capy::executor_ref ex,
2024		buffer_param buf,
2025		int flags,
2026		std::stop_token token,
2027		std::error_code* ec,
2028		std::size_t* bytes_out) override
2029		{
2030	47x	return submit_recv(
2031	47x	cont, ex, buf, false, nullptr, flags, token, ec, bytes_out);
2032		}
2033
2034	41x	std::coroutine_handle<> connect(
2035		capy::continuation& cont,
2036		capy::executor_ref ex,
2037		endpoint ep,
2038		std::stop_token token,
2039		std::error_code* ec) override
2040		{
2041	41x	bool stop_now = token.stop_possible() && token.stop_requested();
2042	41x	if (stop_now)
2043		{
2044	✗	if (sched_->try_consume_inline_budget())
2045		{
2046	✗	if (ec)
2047	✗	*ec = capy::error::canceled;
2048	✗	return dispatch_coro(ex, cont);
2049		}
2050	✗	conn_.addrlen = to_sockaddr(ep, this->native_family(), conn_.addr);
2051	✗	conn_.prepare(
2052	✗	cont, ex, ec, fd_, sched_, detail::object_ref(this), ep,
2053		&remote_endpoint_, &local_endpoint_, token);
2054	✗	conn_.cancelled.store(true, std::memory_order_release);
2055	✗	sched_->work_started();
2056		{
2057	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2058	✗	sched_->push_completed_locked(&conn_);
2059	✗	}
2060	✗	return std::noop_coroutine();
2061		}
2062
2063		// io_uring's IORING_OP_CONNECT re-invokes connect(2) internally;
2064		// a prior speculative ::connect would leave EINPROGRESS → EALREADY.
2065	41x	conn_.addrlen = to_sockaddr(ep, this->native_family(), conn_.addr);
2066	41x	conn_.prepare(
2067	82x	cont, ex, ec, fd_, sched_, detail::object_ref(this), ep,
2068		&remote_endpoint_, &local_endpoint_, token);
2069	41x	sched_->work_started();
2070	41x	if (conn_.cancelled.load(std::memory_order_acquire))
2071		{
2072	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2073	✗	sched_->push_completed_locked(&conn_);
2074	✗	return std::noop_coroutine();
2075	✗	}
2076	41x	uring_submit_op(*sched_, &conn_);
2077	41x	return std::noop_coroutine();
2078		}
2079
2080	9x	std::coroutine_handle<> wait(
2081		capy::continuation& cont,
2082		capy::executor_ref ex,
2083		wait_type w,
2084		std::stop_token token,
2085		std::error_code* ec) override
2086		{
2087	9x	int poll_flags = 0;
2088	9x	switch (w)
2089		{
2090	7x	case wait_type::read:
2091	7x	poll_flags = POLLIN;
2092	7x	break;
2093	1x	case wait_type::write:
2094	1x	poll_flags = POLLOUT;
2095	1x	break;
2096	1x	case wait_type::error:
2097	1x	poll_flags = POLLPRI \| POLLERR \| POLLHUP;
2098	1x	break;
2099		}
2100	9x	wait_op_.prepare(
2101	18x	cont, ex, ec, fd_, sched_, detail::object_ref(this), poll_flags,
2102		token);
2103	9x	sched_->work_started();
2104	9x	if (wait_op_.cancelled.load(std::memory_order_acquire))
2105		{
2106	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2107	✗	sched_->push_completed_locked(&wait_op_);
2108	✗	return std::noop_coroutine();
2109	✗	}
2110	9x	uring_submit_op(*sched_, &wait_op_);
2111	9x	return std::noop_coroutine();
2112		}
2113
2114		// native_handle / is_open / set_option / get_option / local_endpoint
2115		// are inherited from native_socket_base.
2116
2117	2x	std::error_code shutdown(udp_socket::shutdown_type what) noexcept override
2118		{
2119	2x	if (::shutdown(fd_, static_cast<int>(what)) != 0)
2120	1x	return make_err(errno);
2121	1x	return {};
2122		}
2123
2124	4x	native_handle_type release_socket() noexcept override
2125		{
2126		// Flush while the fd is still open so the kernel resolves
2127		// pending SQEs before the caller can close and recycle the
2128		// number (same reasoning as close_socket).
2129		// A connect completing after this belongs to the descriptor
2130		// the caller takes; see close_socket().
2131	4x	conn_.request_cancel();
2132	4x	wait_op_.request_cancel();
2133	4x	if (fd_ >= 0)
2134	4x	sched_->flush_after_cancel(
2135		fd_, {&conn_, &send_, &recv_, &wait_op_});
2136	4x	int fd = fd_;
2137	4x	fd_ = -1;
2138	4x	family_ = AF_UNSPEC;
2139	4x	local_endpoint_ = endpoint{};
2140	4x	remote_endpoint_ = endpoint{};
2141	4x	return fd;
2142		}
2143
2144	9x	void cancel() noexcept override
2145		{
2146	9x	if (fd_ >= 0)
2147	5x	sched_->cancel_ops({&conn_, &send_, &recv_, &wait_op_});
2148	9x	}
2149
2150		/// Cancel in-flight ops, close the fd, and reset cached endpoints.
2151		/// Called by the service on close()/teardown.
2152	314x	void close_socket() noexcept
2153		{
2154		// A connect completing after this must not write endpoints into
2155		// a closed impl that may be recycled.
2156	314x	conn_.request_cancel();
2157	314x	wait_op_.request_cancel();
2158	314x	if (fd_ >= 0)
2159		{
2160	132x	sched_->flush_after_cancel(
2161		fd_, {&conn_, &send_, &recv_, &wait_op_});
2162	132x	::close(fd_);
2163	132x	fd_ = -1;
2164		}
2165	314x	family_ = AF_UNSPEC;
2166	314x	local_endpoint_ = endpoint{};
2167	314x	remote_endpoint_ = endpoint{};
2168	314x	}
2169
2170		/// Recycle into the owning service's pool. See
2171		/// uring_tcp_socket::retire() for why this is out-of-line.
2172		void retire() noexcept override;
2173
2174		/** Reset op slots, cached endpoints, the speculation hint, and
2175		`family_` for recycling. `family_` is set by `open_socket()`/
2176		`assign_socket()`, not by this reset, so resetting it to
2177		`AF_UNSPEC` here is defensive — it guards any future entry
2178		path that constructs a usable socket without going through
2179		either of those.
2180
2181		@pre refs_ == 0, fd closed, no op in flight.
2182		*/
2183	33x	void reuse() noexcept
2184		{
2185	33x	BOOST_COROSIO_ASSERT(fd_ == -1);
2186		// A connect completing on another thread while the socket closed
2187		// can still have written these.
2188	33x	local_endpoint_ = endpoint{};
2189	33x	remote_endpoint_ = endpoint{};
2190	33x	BOOST_COROSIO_ASSERT(!conn_.stop_cb);
2191	33x	BOOST_COROSIO_ASSERT(!send_.stop_cb);
2192	33x	BOOST_COROSIO_ASSERT(!recv_.stop_cb);
2193	33x	BOOST_COROSIO_ASSERT(!wait_op_.stop_cb);
2194	33x	family_ = AF_UNSPEC;
2195	33x	spec_.reset();
2196	33x	}
2197
2198	3x	endpoint remote_endpoint() const noexcept override
2199		{
2200	3x	return remote_endpoint_;
2201		}
2202
2203		private:
2204	161x	std::coroutine_handle<> submit_send(
2205		capy::continuation& cont,
2206		capy::executor_ref ex,
2207		buffer_param buffers,
2208		socklen_t dest_len,
2209		sockaddr_storage const& dest_storage,
2210		int flags,
2211		std::stop_token const& token,
2212		std::error_code* ec,
2213		std::size_t* bytes)
2214		{
2215		iovec iovecs[uring_max_iov];
2216	161x	int iovec_count = copy_to_iovec(buffers, iovecs);
2217	161x	bool stop_now = token.stop_possible() && token.stop_requested();
2218	161x	bool empty_buf = (iovec_count == 0);
2219
2220	161x	ssize_t n = 0;
2221	161x	int err = 0;
2222	161x	bool have_sync_res = stop_now \|\| empty_buf;
2223	161x	if (!have_sync_res && spec_.may_speculate_write())
2224		{
2225	160x	msghdr msg{};
2226	160x	msg.msg_iov = iovecs;
2227	160x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(iovec_count);
2228	160x	sockaddr_storage dest_copy = dest_storage;
2229	160x	if (dest_len > 0)
2230		{
2231	32x	msg.msg_name = &dest_copy;
2232	32x	msg.msg_namelen = dest_len;
2233		}
2234	160x	int native_flags = to_native_msg_flags(flags) \| MSG_NOSIGNAL;
2235		do
2236		{
2237	160x	n = ::sendmsg(fd_, &msg, native_flags);
2238		}
2239	160x	while (n < 0 && errno == EINTR);
2240	160x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
2241		{
2242	159x	have_sync_res = true;
2243	159x	if (n < 0)
2244	1x	err = errno;
2245		}
2246		else
2247		{
2248	1x	spec_.on_write_exhausted();
2249		}
2250		}
2251
2252	161x	if (have_sync_res)
2253		{
2254	160x	if (sched_->try_consume_inline_budget())
2255		{
2256	154x	decode_results(
2257		ec, bytes, stop_now,
2258	1x	err ? make_err(err) : std::error_code{},
2259	154x	/is_read=/false, n < 0 ? 0u : static_cast<std::size_t>(n),
2260		/empty_buffer=/false);
2261	154x	return dispatch_coro(ex, cont);
2262		}
2263	12x	send_.prepare(
2264	12x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this),
2265		&spec_, buffers, dest_len, dest_storage,
2266		to_native_msg_flags(flags), token);
2267	6x	if (stop_now)
2268	✗	send_.cancelled.store(true, std::memory_order_release);
2269		else
2270	6x	send_.res = (n < 0) ? -err : static_cast<int>(n);
2271	6x	sched_->work_started();
2272		{
2273	6x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2274	6x	sched_->push_completed_locked(&send_);
2275	6x	}
2276	6x	return std::noop_coroutine();
2277		}
2278
2279	2x	send_.prepare(
2280	2x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this), &spec_,
2281		buffers, dest_len, dest_storage, to_native_msg_flags(flags), token);
2282	1x	sched_->work_started();
2283	1x	if (send_.cancelled.load(std::memory_order_acquire))
2284		{
2285	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2286	✗	sched_->push_completed_locked(&send_);
2287	✗	return std::noop_coroutine();
2288	✗	}
2289	1x	uring_submit_op(*sched_, &send_);
2290	1x	return std::noop_coroutine();
2291		}
2292
2293	133x	std::coroutine_handle<> submit_recv(
2294		capy::continuation& cont,
2295		capy::executor_ref ex,
2296		buffer_param buffers,
2297		bool want_source,
2298		corosio::endpoint* source_out,
2299		int flags,
2300		std::stop_token const& token,
2301		std::error_code* ec,
2302		std::size_t* bytes)
2303		{
2304		iovec iovecs[uring_max_iov];
2305	133x	int iovec_count = copy_to_iovec(buffers, iovecs);
2306	133x	bool stop_now = token.stop_possible() && token.stop_requested();
2307	133x	bool empty_buf = (iovec_count == 0);
2308
2309	133x	ssize_t n = 0;
2310	133x	int err = 0;
2311	133x	bool have_sync_res = stop_now \|\| empty_buf;
2312	133x	sockaddr_storage src_storage{};
2313	133x	socklen_t src_namelen = 0;
2314	133x	if (!have_sync_res && spec_.may_speculate_read())
2315		{
2316	132x	msghdr msg{};
2317	132x	msg.msg_iov = iovecs;
2318	132x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(iovec_count);
2319	132x	if (want_source)
2320		{
2321	85x	msg.msg_name = &src_storage;
2322	85x	msg.msg_namelen = sizeof(src_storage);
2323		}
2324	132x	int native_flags = to_native_msg_flags(flags);
2325		do
2326		{
2327	132x	n = ::recvmsg(fd_, &msg, native_flags);
2328		}
2329	132x	while (n < 0 && errno == EINTR);
2330	132x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
2331		{
2332	109x	have_sync_res = true;
2333	109x	if (n < 0)
2334	1x	err = errno;
2335	109x	src_namelen = (n >= 0) ? msg.msg_namelen : 0;
2336		}
2337		else
2338		{
2339	23x	spec_.on_read_exhausted();
2340		}
2341		}
2342
2343	133x	if (have_sync_res)
2344		{
2345	110x	if (sched_->try_consume_inline_budget())
2346		{
2347	106x	decode_results(
2348		ec, bytes, stop_now,
2349	1x	err ? make_err(err) : std::error_code{},
2350	106x	/is_read=/false, n < 0 ? 0u : static_cast<std::size_t>(n),
2351		/empty_buffer=/false);
2352	106x	if (n >= 0 && want_source && source_out && !empty_buf)
2353	62x	*source_out = sockaddr_to_endpoint(src_storage);
2354	106x	return dispatch_coro(ex, cont);
2355		}
2356	8x	recv_.prepare(
2357	8x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this),
2358		&spec_, buffers, source_out,
2359		want_source ? &write_ip_source : nullptr,
2360		to_native_msg_flags(flags), token);
2361	4x	if (stop_now)
2362	✗	recv_.cancelled.store(true, std::memory_order_release);
2363		else
2364		{
2365	4x	recv_.res = (n < 0) ? -err : static_cast<int>(n);
2366		// Hand the speculative source over to do_handler's
2367		// source_writer so it translates into source_out the same
2368		// way the kernel-completed path would.
2369	4x	if (n >= 0 && want_source)
2370		{
2371	2x	recv_.source_storage = src_storage;
2372	2x	recv_.source_len = src_namelen;
2373		}
2374		}
2375	4x	sched_->work_started();
2376		{
2377	4x	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2378	4x	sched_->push_completed_locked(&recv_);
2379	4x	}
2380	4x	return std::noop_coroutine();
2381		}
2382
2383	46x	recv_.prepare(
2384	46x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this), &spec_,
2385		buffers, source_out, want_source ? &write_ip_source : nullptr,
2386		to_native_msg_flags(flags), token);
2387	23x	sched_->work_started();
2388	46x	if (recv_.iovec_count == 0 \|\|
2389	23x	recv_.cancelled.load(std::memory_order_acquire))
2390		{
2391	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2392	✗	sched_->push_completed_locked(&recv_);
2393	✗	return std::noop_coroutine();
2394	✗	}
2395	23x	uring_submit_op(*sched_, &recv_);
2396	23x	return std::noop_coroutine();
2397		}
2398
2399	8x	static void write_ip_source(
2400		void* ctx, sockaddr_storage const& s, socklen_t /len/) noexcept
2401		{
2402	8x	if (auto* out = static_cast<corosio::endpoint*>(ctx))
2403	8x	*out = sockaddr_to_endpoint(s);
2404	8x	}
2405		};
2406
2407		/** UDP socket service for io_uring.
2408
2409		Owns all `uring_udp_socket` implementations for an `io_context`.
2410		Satisfies the `udp_service` interface so the generic `udp_socket`
2411		front-end can call `open_datagram_socket` and `bind_datagram`
2412		transparently.
2413
2414		Socket impls live in the service's `object_pool`. `construct()`
2415		hands out a pooled impl with one reference already held; `destroy()`
2416		releases it, and `retire()` recycles it onto the free list
2417		instead of freeing it.
2418
2419		@par Thread Safety
2420		All public member functions are thread-safe.
2421		*/
2422		class BOOST_COROSIO_DECL uring_udp_service final
2423		: public uring_socket_service_base<
2424		uring_udp_service,
2425		udp_service,
2426		uring_udp_socket>
2427		{
2428		using base_service = uring_socket_service_base<
2429		uring_udp_service,
2430		udp_service,
2431		uring_udp_socket>;
2432
2433		public:
2434		/// Identifies this service for `execution_context` lookup.
2435		using key_type = udp_service;
2436
2437		/** Construct the UDP service.
2438
2439		@param ctx The owning execution context. The io_uring scheduler
2440		must already be registered.
2441		*/
2442	100x	explicit uring_udp_service(capy::execution_context& ctx) : base_service(ctx)
2443		{
2444	100x	}
2445
2446		// construct / destroy / shutdown / close / scheduler() are inherited
2447		// from uring_socket_service_base.
2448
2449		/** Open a datagram socket and associate it with an impl.
2450
2451		Creates a non-blocking, close-on-exec socket via `socket(2)`.
2452
2453		@param impl The socket implementation to initialise.
2454		@param family Address family (e.g. `AF_INET`, `AF_INET6`).
2455		@param type Socket type (`SOCK_DGRAM`).
2456		@param protocol Protocol number (`IPPROTO_UDP`).
2457		@return Error code on failure, empty on success.
2458		*/
2459	133x	std::error_code open_datagram_socket(
2460		udp_socket::implementation& impl,
2461		int family,
2462		int type,
2463		int protocol) override
2464		{
2465	133x	auto& sock = static_cast<uring_udp_socket&>(impl);
2466		int fd =
2467	133x	::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
2468	133x	if (fd < 0)
2469	1x	return make_err(errno);
2470		// LCOV_EXCL_START: dead — open() guards is_open(), so open_socket
2471		// never runs against an already-open fd (assign_socket handles that).
2472	−	if (sock.fd_ >= 0)
2473	−	sock.close_socket();
2474		// LCOV_EXCL_STOP
2475	132x	sock.fd_ = fd;
2476	132x	sock.family_ = family;
2477	132x	if (family == AF_INET6)
2478		{
2479	13x	int one = 1;
2480	13x	::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &one, sizeof(one));
2481		}
2482	132x	return {};
2483		}
2484
2485		/** Adopt a pre-created fd into an impl.
2486
2487		Takes ownership of `fd` on success; the caller retains
2488		ownership on failure.
2489
2490		@param impl The socket implementation to assign to.
2491		@param fd A valid, open, non-blocking IP datagram fd.
2492		@return Error code on failure, empty on success.
2493		*/
2494	8x	std::error_code assign_socket(
2495		udp_socket::implementation& impl, native_handle_type fd) override
2496		{
2497	8x	auto& sock = static_cast<uring_udp_socket&>(impl);
2498	8x	int nfd = static_cast<int>(fd);
2499		// The public assign() guarantees the object is closed.
2500	8x	if (auto ec = validate_socket_fd(nfd, SOCK_DGRAM, true))
2501	4x	return ec;
2502
2503	4x	sock.fd_ = nfd;
2504
2505	4x	sock.local_endpoint_ = endpoint{};
2506	4x	sock.remote_endpoint_ = endpoint{};
2507
2508	4x	sockaddr_storage local{};
2509	4x	socklen_t local_len = sizeof(local);
2510	4x	if (::getsockname(
2511	4x	sock.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
2512		{
2513	4x	sock.local_endpoint_ = sockaddr_to_endpoint(local);
2514	4x	sock.family_ = local.ss_family;
2515		}
2516
2517	4x	sockaddr_storage remote{};
2518	4x	socklen_t remote_len = sizeof(remote);
2519	4x	if (::getpeername(
2520	4x	sock.fd_, reinterpret_cast<sockaddr*>(&remote), &remote_len) ==
2521		0)
2522	1x	sock.remote_endpoint_ = sockaddr_to_endpoint(remote);
2523
2524	4x	return {};
2525		}
2526
2527		/** Bind the socket and capture the local endpoint via `getsockname`.
2528
2529		@param impl The socket implementation to bind.
2530		@param ep The local endpoint to bind to.
2531		@return Error code on failure, empty on success.
2532		*/
2533		std::error_code
2534	82x	bind_datagram(udp_socket::implementation& impl, endpoint ep) override
2535		{
2536	82x	auto& sock = static_cast<uring_udp_socket&>(impl);
2537	82x	sockaddr_storage addr{};
2538	82x	socklen_t len = endpoint_to_sockaddr(ep, addr);
2539	82x	if (::bind(sock.fd_, reinterpret_cast<sockaddr*>(&addr), len) < 0)
2540	2x	return make_err(errno);
2541
2542	80x	sockaddr_storage local{};
2543	80x	socklen_t local_len = sizeof(local);
2544	80x	if (::getsockname(
2545	80x	sock.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
2546	80x	sock.local_endpoint_ = sockaddr_to_endpoint(local);
2547	80x	return {};
2548		}
2549		};
2550
2551		inline void
2552	182x	uring_udp_socket::retire() noexcept
2553		{
2554	182x	svc_->pool_.recycle(this);
2555	182x	}
2556
2557		/** Unix domain datagram socket implementation for io_uring.
2558
2559		Implements `local_datagram_socket::implementation` using a proactor
2560		model: send_to, recv_from, send, recv, and connect operations are
2561		submitted to the kernel via `uring_submit_op` and complete through
2562		the ring's CQE path.
2563
2564		The service holds one intrusive reference for as long as the impl
2565		is live; in-flight ops hold an additional `object_ref_` keepalive so
2566		the kernel's user-data pointer remains valid until the CQE arrives.
2567
2568		@par Thread Safety
2569		Distinct objects: Safe.
2570		Shared objects: Unsafe. One send and one recv may be in flight
2571		simultaneously, but two sends or two recvs must not overlap.
2572		*/
2573		class BOOST_COROSIO_DECL uring_local_datagram_socket final
2574		: public native_socket_base<
2575		uring_local_datagram_socket,
2576		local_datagram_socket::implementation,
2577		corosio::local_endpoint>
2578		, public intrusive_list<uring_local_datagram_socket>::node
2579		{
2580		friend uring_local_datagram_service;
2581
2582		uring_scheduler* sched_ = nullptr;
2583		uring_local_datagram_service* svc_ = nullptr;
2584
2585		// fd_ and local_endpoint_ live in native_socket_base, which also
2586		// provides native_handle/is_open/set_option/get_option/local_endpoint.
2587		corosio::local_endpoint remote_endpoint_;
2588
2589		// Per-fd op slots — embedded to eliminate per-call heap allocation.
2590		// Single-pending invariant per slot.
2591		uring_local_connect_op conn_;
2592		uring_dgram_send_op send_;
2593		uring_dgram_recv_op recv_;
2594		uring_wait_op wait_op_;
2595
2596		mutable detail::speculative_state spec_;
2597
2598		public:
2599		/** Construct with service and scheduler references.
2600
2601		Both refs must outlive this socket.
2602
2603		@param svc The owning service.
2604		@param sched The io_uring scheduler owned by the context.
2605		*/
2606	123x	explicit uring_local_datagram_socket(
2607		uring_local_datagram_service& svc, uring_scheduler& sched) noexcept
2608	246x	: sched_(&sched)
2609	123x	, svc_(&svc)
2610		{
2611	123x	}
2612
2613	246x	~uring_local_datagram_socket() override
2614	123x	{
2615	123x	if (fd_ >= 0)
2616	✗	::close(
2617		fd_); // LCOV_EXCL_LINE backstop: close_socket() clears fd_ before destroy
2618	246x	}
2619
2620		// ----------------------------------------------------------------
2621		// local_datagram_socket::implementation
2622		// ----------------------------------------------------------------
2623
2624	55x	std::coroutine_handle<> send_to(
2625		capy::continuation& cont,
2626		capy::executor_ref ex,
2627		buffer_param buf,
2628		corosio::local_endpoint dest,
2629		int flags,
2630		std::stop_token token,
2631		std::error_code* ec,
2632		std::size_t* bytes_out) override
2633		{
2634	55x	sockaddr_storage addr{};
2635	55x	socklen_t len = endpoint_to_sockaddr(dest, addr);
2636	55x	return submit_send(
2637	110x	cont, ex, buf, len, addr, flags, token, ec, bytes_out);
2638		}
2639
2640	45x	std::coroutine_handle<> recv_from(
2641		capy::continuation& cont,
2642		capy::executor_ref ex,
2643		buffer_param buf,
2644		corosio::local_endpoint* source,
2645		int flags,
2646		std::stop_token token,
2647		std::error_code* ec,
2648		std::size_t* bytes_out) override
2649		{
2650	45x	return submit_recv(
2651		cont, ex, buf, source != nullptr, source, flags, token, ec,
2652	45x	bytes_out);
2653		}
2654
2655	48x	std::coroutine_handle<> send(
2656		capy::continuation& cont,
2657		capy::executor_ref ex,
2658		buffer_param buf,
2659		int flags,
2660		std::stop_token token,
2661		std::error_code* ec,
2662		std::size_t* bytes_out) override
2663		{
2664	48x	sockaddr_storage empty{};
2665	48x	return submit_send(
2666	96x	cont, ex, buf, 0, empty, flags, token, ec, bytes_out);
2667		}
2668
2669	47x	std::coroutine_handle<> recv(
2670		capy::continuation& cont,
2671		capy::executor_ref ex,
2672		buffer_param buf,
2673		int flags,
2674		std::stop_token token,
2675		std::error_code* ec,
2676		std::size_t* bytes_out) override
2677		{
2678	47x	return submit_recv(
2679	47x	cont, ex, buf, false, nullptr, flags, token, ec, bytes_out);
2680		}
2681
2682	26x	std::coroutine_handle<> connect(
2683		capy::continuation& cont,
2684		capy::executor_ref ex,
2685		corosio::local_endpoint ep,
2686		std::stop_token token,
2687		std::error_code* ec) override
2688		{
2689	26x	bool stop_now = token.stop_possible() && token.stop_requested();
2690	26x	if (stop_now)
2691		{
2692	✗	if (sched_->try_consume_inline_budget())
2693		{
2694	✗	if (ec)
2695	✗	*ec = capy::error::canceled;
2696	✗	return dispatch_coro(ex, cont);
2697		}
2698	✗	conn_.addrlen = to_sockaddr(ep, conn_.addr);
2699	✗	conn_.prepare(
2700	✗	cont, ex, ec, fd_, sched_, detail::object_ref(this), ep,
2701		&remote_endpoint_, &local_endpoint_, token);
2702	✗	conn_.cancelled.store(true, std::memory_order_release);
2703	✗	sched_->work_started();
2704		{
2705	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2706	✗	sched_->push_completed_locked(&conn_);
2707	✗	}
2708	✗	return std::noop_coroutine();
2709		}
2710
2711		// io_uring's IORING_OP_CONNECT re-invokes connect(2) internally;
2712		// a prior speculative ::connect would leave EINPROGRESS → EALREADY.
2713	26x	conn_.addrlen = to_sockaddr(ep, conn_.addr);
2714	26x	conn_.prepare(
2715	52x	cont, ex, ec, fd_, sched_, detail::object_ref(this), ep,
2716		&remote_endpoint_, &local_endpoint_, token);
2717	26x	sched_->work_started();
2718	26x	if (conn_.cancelled.load(std::memory_order_acquire))
2719		{
2720	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2721	✗	sched_->push_completed_locked(&conn_);
2722	✗	return std::noop_coroutine();
2723	✗	}
2724	26x	uring_submit_op(*sched_, &conn_);
2725	26x	return std::noop_coroutine();
2726		}
2727
2728	5x	std::coroutine_handle<> wait(
2729		capy::continuation& cont,
2730		capy::executor_ref ex,
2731		wait_type w,
2732		std::stop_token token,
2733		std::error_code* ec) override
2734		{
2735	5x	int poll_flags = 0;
2736	5x	switch (w)
2737		{
2738	3x	case wait_type::read:
2739	3x	poll_flags = POLLIN;
2740	3x	break;
2741	1x	case wait_type::write:
2742	1x	poll_flags = POLLOUT;
2743	1x	break;
2744	1x	case wait_type::error:
2745	1x	poll_flags = POLLPRI \| POLLERR \| POLLHUP;
2746	1x	break;
2747		}
2748	5x	wait_op_.prepare(
2749	10x	cont, ex, ec, fd_, sched_, detail::object_ref(this), poll_flags,
2750		token);
2751	5x	sched_->work_started();
2752	5x	if (wait_op_.cancelled.load(std::memory_order_acquire))
2753		{
2754	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2755	✗	sched_->push_completed_locked(&wait_op_);
2756	✗	return std::noop_coroutine();
2757	✗	}
2758	5x	uring_submit_op(*sched_, &wait_op_);
2759	5x	return std::noop_coroutine();
2760		}
2761
2762		std::error_code
2763	3x	shutdown(local_datagram_socket::shutdown_type what) noexcept override
2764		{
2765	3x	if (::shutdown(fd_, static_cast<int>(what)) != 0)
2766	1x	return make_err(errno);
2767	2x	return {};
2768		}
2769
2770		// native_handle / is_open / set_option / get_option / local_endpoint
2771		// are inherited from native_socket_base.
2772
2773	2x	native_handle_type release_socket() noexcept override
2774		{
2775		// Flush while the fd is still open so the kernel resolves
2776		// pending SQEs before the caller can close and recycle the
2777		// number (same reasoning as close_socket).
2778		// A connect completing after this belongs to the descriptor
2779		// the caller takes; see close_socket().
2780	2x	conn_.request_cancel();
2781	2x	wait_op_.request_cancel();
2782	2x	if (fd_ >= 0)
2783	2x	sched_->flush_after_cancel(
2784		fd_, {&conn_, &send_, &recv_, &wait_op_});
2785	2x	int fd = fd_;
2786	2x	fd_ = -1;
2787	2x	family_ = AF_UNSPEC;
2788	2x	local_endpoint_ = corosio::local_endpoint{};
2789	2x	remote_endpoint_ = corosio::local_endpoint{};
2790	2x	return fd;
2791		}
2792
2793	6x	void cancel() noexcept override
2794		{
2795	6x	if (fd_ >= 0)
2796	2x	sched_->cancel_ops({&conn_, &send_, &recv_, &wait_op_});
2797	6x	}
2798
2799		/// Cancel in-flight ops, close the fd, and reset cached endpoints.
2800		/// Called by the service on close()/teardown.
2801	266x	void close_socket() noexcept
2802		{
2803		// A connect completing after this must not write endpoints into
2804		// a closed impl that may be recycled.
2805	266x	conn_.request_cancel();
2806	266x	wait_op_.request_cancel();
2807	266x	if (fd_ >= 0)
2808		{
2809	105x	sched_->flush_after_cancel(
2810		fd_, {&conn_, &send_, &recv_, &wait_op_});
2811	105x	::close(fd_);
2812	105x	fd_ = -1;
2813		}
2814	266x	family_ = AF_UNSPEC;
2815	266x	local_endpoint_ = corosio::local_endpoint{};
2816	266x	remote_endpoint_ = corosio::local_endpoint{};
2817	266x	}
2818
2819		/// Recycle into the owning service's pool. See
2820		/// uring_tcp_socket::retire() for why this is out-of-line.
2821		void retire() noexcept override;
2822
2823		/** Reset op slots, cached endpoints, and the speculation hint for
2824		recycling. See uring_tcp_socket::reuse() for the rationale.
2825
2826		@pre refs_ == 0, fd closed, no op in flight.
2827		*/
2828	38x	void reuse() noexcept
2829		{
2830	38x	BOOST_COROSIO_ASSERT(fd_ == -1);
2831		// A connect completing on another thread while the socket closed
2832		// can still have written these.
2833	38x	local_endpoint_ = corosio::local_endpoint{};
2834	38x	remote_endpoint_ = corosio::local_endpoint{};
2835	38x	BOOST_COROSIO_ASSERT(!conn_.stop_cb);
2836	38x	BOOST_COROSIO_ASSERT(!send_.stop_cb);
2837	38x	BOOST_COROSIO_ASSERT(!recv_.stop_cb);
2838	38x	BOOST_COROSIO_ASSERT(!wait_op_.stop_cb);
2839	38x	spec_.reset();
2840	38x	}
2841
2842	1x	corosio::local_endpoint remote_endpoint() const noexcept override
2843		{
2844	1x	return remote_endpoint_;
2845		}
2846
2847		// LCOV_EXCL_START: the public bind routes through the
2848		// service's bind_socket; nothing calls the implementation
2849		// interface's bind on this backend.
2850	−	std::error_code bind(corosio::local_endpoint ep) noexcept override
2851		{
2852	−	sockaddr_storage addr{};
2853	−	socklen_t len = endpoint_to_sockaddr(ep, addr);
2854	−	if (::bind(fd_, reinterpret_cast<sockaddr*>(&addr), len) != 0)
2855	−	return make_err(errno);
2856
2857	−	sockaddr_storage local{};
2858	−	socklen_t local_len = sizeof(local);
2859	−	if (::getsockname(
2860	−	fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
2861	−	local_endpoint_ = sockaddr_to_local_endpoint(local, local_len);
2862	−	return {};
2863		}
2864		// LCOV_EXCL_STOP
2865
2866		private:
2867	103x	std::coroutine_handle<> submit_send(
2868		capy::continuation& cont,
2869		capy::executor_ref ex,
2870		buffer_param buffers,
2871		socklen_t dest_len,
2872		sockaddr_storage const& dest_storage,
2873		int flags,
2874		std::stop_token const& token,
2875		std::error_code* ec,
2876		std::size_t* bytes)
2877		{
2878		iovec iovecs[uring_max_iov];
2879	103x	int iovec_count = copy_to_iovec(buffers, iovecs);
2880	103x	bool stop_now = token.stop_possible() && token.stop_requested();
2881	103x	bool empty_buf = (iovec_count == 0);
2882
2883	103x	ssize_t n = 0;
2884	103x	int err = 0;
2885	103x	bool have_sync_res = stop_now \|\| empty_buf;
2886	103x	if (!have_sync_res && spec_.may_speculate_write())
2887		{
2888	103x	msghdr msg{};
2889	103x	msg.msg_iov = iovecs;
2890	103x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(iovec_count);
2891	103x	sockaddr_storage dest_copy = dest_storage;
2892	103x	if (dest_len > 0)
2893		{
2894	55x	msg.msg_name = &dest_copy;
2895	55x	msg.msg_namelen = dest_len;
2896		}
2897	103x	int native_flags = to_native_msg_flags(flags) \| MSG_NOSIGNAL;
2898		do
2899		{
2900	103x	n = ::sendmsg(fd_, &msg, native_flags);
2901		}
2902	103x	while (n < 0 && errno == EINTR);
2903	103x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
2904		{
2905	64x	have_sync_res = true;
2906	64x	if (n < 0)
2907	1x	err = errno;
2908		}
2909		else
2910		{
2911	39x	spec_.on_write_exhausted();
2912		}
2913		}
2914
2915	103x	if (have_sync_res)
2916		{
2917	64x	if (sched_->try_consume_inline_budget())
2918		{
2919	64x	decode_results(
2920		ec, bytes, stop_now,
2921	1x	err ? make_err(err) : std::error_code{},
2922	64x	/is_read=/false, n < 0 ? 0u : static_cast<std::size_t>(n),
2923		/empty_buffer=/false);
2924	64x	return dispatch_coro(ex, cont);
2925		}
2926	✗	send_.prepare(
2927	✗	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this),
2928		&spec_, buffers, dest_len, dest_storage,
2929		to_native_msg_flags(flags), token);
2930	✗	if (stop_now)
2931	✗	send_.cancelled.store(true, std::memory_order_release);
2932		else
2933	✗	send_.res = (n < 0) ? -err : static_cast<int>(n);
2934	✗	sched_->work_started();
2935		{
2936	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2937	✗	sched_->push_completed_locked(&send_);
2938	✗	}
2939	✗	return std::noop_coroutine();
2940		}
2941
2942	78x	send_.prepare(
2943	78x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this), &spec_,
2944		buffers, dest_len, dest_storage, to_native_msg_flags(flags), token);
2945	39x	sched_->work_started();
2946	39x	if (send_.cancelled.load(std::memory_order_acquire))
2947		{
2948	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
2949	✗	sched_->push_completed_locked(&send_);
2950	✗	return std::noop_coroutine();
2951	✗	}
2952	39x	uring_submit_op(*sched_, &send_);
2953	39x	return std::noop_coroutine();
2954		}
2955
2956	92x	std::coroutine_handle<> submit_recv(
2957		capy::continuation& cont,
2958		capy::executor_ref ex,
2959		buffer_param buffers,
2960		bool want_source,
2961		corosio::local_endpoint* source_out,
2962		int flags,
2963		std::stop_token const& token,
2964		std::error_code* ec,
2965		std::size_t* bytes)
2966		{
2967		iovec iovecs[uring_max_iov];
2968	92x	int iovec_count = copy_to_iovec(buffers, iovecs);
2969	92x	bool stop_now = token.stop_possible() && token.stop_requested();
2970	92x	bool empty_buf = (iovec_count == 0);
2971
2972	92x	ssize_t n = 0;
2973	92x	int err = 0;
2974	92x	bool have_sync_res = stop_now \|\| empty_buf;
2975	92x	sockaddr_storage src_storage{};
2976	92x	socklen_t src_namelen = 0;
2977	92x	if (!have_sync_res && spec_.may_speculate_read())
2978		{
2979	60x	msghdr msg{};
2980	60x	msg.msg_iov = iovecs;
2981	60x	msg.msg_iovlen = static_cast<decltype(msg.msg_iovlen)>(iovec_count);
2982	60x	if (want_source)
2983		{
2984	29x	msg.msg_name = &src_storage;
2985	29x	msg.msg_namelen = sizeof(src_storage);
2986		}
2987	60x	int native_flags = to_native_msg_flags(flags);
2988		do
2989		{
2990	60x	n = ::recvmsg(fd_, &msg, native_flags);
2991		}
2992	60x	while (n < 0 && errno == EINTR);
2993	60x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
2994		{
2995	44x	have_sync_res = true;
2996	44x	if (n < 0)
2997	✗	err = errno;
2998	44x	src_namelen = (n >= 0) ? msg.msg_namelen : 0;
2999		}
3000		else
3001		{
3002	16x	spec_.on_read_exhausted();
3003		}
3004		}
3005
3006	92x	if (have_sync_res)
3007		{
3008	44x	if (sched_->try_consume_inline_budget())
3009		{
3010	44x	decode_results(
3011		ec, bytes, stop_now,
3012	✗	err ? make_err(err) : std::error_code{},
3013	44x	/is_read=/false, n < 0 ? 0u : static_cast<std::size_t>(n),
3014		/empty_buffer=/false);
3015	44x	if (n >= 0 && want_source && source_out && !empty_buf)
3016		*source_out =
3017	21x	sockaddr_to_local_endpoint(src_storage, src_namelen);
3018	44x	return dispatch_coro(ex, cont);
3019		}
3020	✗	recv_.prepare(
3021	✗	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this),
3022		&spec_, buffers, source_out,
3023		want_source ? &write_local_source : nullptr,
3024		to_native_msg_flags(flags), token);
3025	✗	if (stop_now)
3026	✗	recv_.cancelled.store(true, std::memory_order_release);
3027		else
3028		{
3029	✗	recv_.res = (n < 0) ? -err : static_cast<int>(n);
3030		// Hand the speculative source over to do_handler's
3031		// source_writer so it translates into source_out the same
3032		// way the kernel-completed path would.
3033	✗	if (n >= 0 && want_source)
3034		{
3035	✗	recv_.source_storage = src_storage;
3036	✗	recv_.source_len = src_namelen;
3037		}
3038		}
3039	✗	sched_->work_started();
3040		{
3041	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
3042	✗	sched_->push_completed_locked(&recv_);
3043	✗	}
3044	✗	return std::noop_coroutine();
3045		}
3046
3047	96x	recv_.prepare(
3048	96x	cont, ex, ec, bytes, fd_, sched_, detail::object_ref(this), &spec_,
3049		buffers, source_out, want_source ? &write_local_source : nullptr,
3050		to_native_msg_flags(flags), token);
3051	48x	sched_->work_started();
3052	96x	if (recv_.iovec_count == 0 \|\|
3053	48x	recv_.cancelled.load(std::memory_order_acquire))
3054		{
3055	✗	uring_scheduler::lock_type lock(sched_->dispatch_mutex());
3056	✗	sched_->push_completed_locked(&recv_);
3057	✗	return std::noop_coroutine();
3058	✗	}
3059	48x	uring_submit_op(*sched_, &recv_);
3060	48x	return std::noop_coroutine();
3061		}
3062
3063	21x	static void write_local_source(
3064		void* ctx, sockaddr_storage const& s, socklen_t len) noexcept
3065		{
3066	21x	if (auto* out = static_cast<corosio::local_endpoint*>(ctx))
3067	21x	*out = sockaddr_to_local_endpoint(s, len);
3068	21x	}
3069		};
3070
3071		/** Unix domain datagram socket service for io_uring.
3072
3073		Owns all `uring_local_datagram_socket` implementations for an
3074		`io_context`. Satisfies the `local_datagram_service` interface so the
3075		generic `local_datagram_socket` front-end can call `open_socket` and
3076		`bind_socket` transparently.
3077
3078		Socket impls live in the service's `object_pool`. `construct()`
3079		hands out a pooled impl with one reference already held; `destroy()`
3080		releases it, and `retire()` recycles it onto the free list
3081		instead of freeing it.
3082
3083		@par Thread Safety
3084		All public member functions are thread-safe.
3085		*/
3086		class BOOST_COROSIO_DECL uring_local_datagram_service final
3087		: public uring_socket_service_base<
3088		uring_local_datagram_service,
3089		local_datagram_service,
3090		uring_local_datagram_socket>
3091		{
3092		using base_service = uring_socket_service_base<
3093		uring_local_datagram_service,
3094		local_datagram_service,
3095		uring_local_datagram_socket>;
3096
3097		public:
3098		/// Identifies this service for `execution_context` lookup.
3099		using key_type = local_datagram_service;
3100
3101		/** Construct the local datagram service.
3102
3103		@param ctx The owning execution context. The io_uring scheduler
3104		must already be registered.
3105		*/
3106	73x	explicit uring_local_datagram_service(capy::execution_context& ctx)
3107	73x	: base_service(ctx)
3108		{
3109	73x	}
3110
3111		// construct / destroy / shutdown / close / scheduler() are inherited
3112		// from uring_socket_service_base.
3113
3114		/** Open an AF_UNIX datagram socket and associate it with an impl.
3115
3116		Creates a non-blocking, close-on-exec socket via `socket(2)`.
3117		`family` is always `AF_UNIX` for local datagram sockets.
3118
3119		@param impl The socket implementation to initialise.
3120		@param family Address family (`AF_UNIX`).
3121		@param type Socket type (`SOCK_DGRAM`).
3122		@param protocol Protocol number (typically 0).
3123		@return Error code on failure, empty on success.
3124		*/
3125	50x	std::error_code open_socket(
3126		local_datagram_socket::implementation& impl,
3127		int family,
3128		int type,
3129		int protocol) override
3130		{
3131	50x	auto& sock = static_cast<uring_local_datagram_socket&>(impl);
3132		int fd =
3133	50x	::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
3134	50x	if (fd < 0)
3135	1x	return make_err(errno);
3136		// LCOV_EXCL_START: dead — open() guards is_open(), so open_socket
3137		// never runs against an already-open fd (assign_socket handles that).
3138	−	if (sock.fd_ >= 0)
3139	−	sock.close_socket();
3140		// LCOV_EXCL_STOP
3141	49x	sock.fd_ = fd;
3142	49x	sock.family_ = AF_UNIX;
3143	49x	return {};
3144		}
3145
3146		/** Adopt a pre-created fd into an impl (e.g. from `socketpair`).
3147
3148		Takes ownership of `fd` on success; the caller retains ownership
3149		on failure.
3150
3151		@param impl The socket implementation to assign to.
3152		@param fd A valid, open, non-blocking AF_UNIX datagram fd.
3153		@return Error code on failure, empty on success.
3154		*/
3155	60x	std::error_code assign_socket(
3156		local_datagram_socket::implementation& impl,
3157		native_handle_type fd) override
3158		{
3159	60x	auto& sock = static_cast<uring_local_datagram_socket&>(impl);
3160	60x	int nfd = static_cast<int>(fd);
3161		// The public assign() guarantees the object is closed.
3162	60x	if (auto ec = validate_socket_fd(nfd, SOCK_DGRAM, false))
3163	2x	return ec;
3164
3165	58x	sock.fd_ = nfd;
3166	58x	sock.family_ = AF_UNIX;
3167
3168	58x	sockaddr_storage local{};
3169	58x	socklen_t local_len = sizeof(local);
3170	58x	if (::getsockname(
3171	58x	sock.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
3172	58x	sock.local_endpoint_ = sockaddr_to_local_endpoint(local, local_len);
3173
3174	58x	sockaddr_storage remote{};
3175	58x	socklen_t remote_len = sizeof(remote);
3176	58x	if (::getpeername(
3177	58x	sock.fd_, reinterpret_cast<sockaddr*>(&remote), &remote_len) ==
3178		0)
3179		sock.remote_endpoint_ =
3180	58x	sockaddr_to_local_endpoint(remote, remote_len);
3181
3182	58x	return {};
3183		}
3184
3185		/** Bind the socket and capture the local endpoint via `getsockname`.
3186
3187		@param impl The socket implementation to bind.
3188		@param ep The local endpoint (path) to bind to.
3189		@return Error code on failure, empty on success.
3190		*/
3191	37x	std::error_code bind_socket(
3192		local_datagram_socket::implementation& impl,
3193		corosio::local_endpoint ep) override
3194		{
3195	37x	auto& sock = static_cast<uring_local_datagram_socket&>(impl);
3196	37x	sockaddr_storage addr{};
3197	37x	socklen_t len = endpoint_to_sockaddr(ep, addr);
3198	37x	if (::bind(sock.fd_, reinterpret_cast<sockaddr*>(&addr), len) < 0)
3199	4x	return make_err(errno);
3200
3201	33x	sockaddr_storage local{};
3202	33x	socklen_t local_len = sizeof(local);
3203	33x	if (::getsockname(
3204	33x	sock.fd_, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
3205	33x	sock.local_endpoint_ = sockaddr_to_local_endpoint(local, local_len);
3206	33x	return {};
3207		}
3208		};
3209
3210		inline void
3211	161x	uring_local_datagram_socket::retire() noexcept
3212		{
3213	161x	svc_->pool_.recycle(this);
3214	161x	}
3215
3216		} // namespace boost::corosio::detail
3217
3218		#endif // BOOST_COROSIO_HAS_URING
3219
3220		#endif // BOOST_COROSIO_NATIVE_DETAIL_URING_URING_TYPES_HPP
3221