include/boost/corosio/native/detail/reactor/reactor_datagram_socket.hpp

98.6% Lines (363 / 368) 100.0% Functions (112 / 112)

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

reactor_datagram_socket.hpp

f(x) Functions (112)

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_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
13
14		#include <boost/corosio/udp_socket.hpp>
15		#include <boost/corosio/shutdown_type.hpp>
16		#include <boost/corosio/wait_type.hpp>
17		#include <boost/corosio/detail/config.hpp>
18		#include <boost/corosio/native/detail/reactor/reactor_basic_socket.hpp>
19		#include <boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp>
20		#include <boost/corosio/native/detail/msg_flags.hpp>
21		#include <boost/corosio/detail/dispatch_coro.hpp>
22		#include <boost/capy/continuation.hpp>
23		#include <boost/capy/buffers.hpp>
24
25		#include <coroutine>
26		#include <cstring>
27
28		#include <errno.h>
29		#include <sys/socket.h>
30		#include <sys/uio.h>
31
32		namespace boost::corosio::detail {
33
34		/** CRTP base for reactor-backed datagram socket implementations.
35
36		Inherits shared data members and cancel/close/register logic
37		from reactor_basic_socket. Adds datagram-specific I/O dispatch
38		for both connectionless (send_to, recv_from) and connected
39		(connect, send, recv) modes.
40
41		@tparam Derived The concrete socket type (CRTP).
42		@tparam Service The backend's datagram service type.
43		@tparam ConnOp The backend's connect op type.
44		@tparam SendToOp The backend's send_to op type.
45		@tparam RecvFromOp The backend's recv_from op type.
46		@tparam SendOp The backend's connected send op type.
47		@tparam RecvOp The backend's connected recv op type.
48		@tparam WaitOp The backend's wait op type.
49		@tparam DescState The backend's descriptor_state type.
50		@tparam ImplBase The public vtable base
51		(udp_socket::implementation or
52		local_datagram_socket::implementation).
53		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
54		*/
55		template<
56		class Derived,
57		class Service,
58		class ConnOp,
59		class SendToOp,
60		class RecvFromOp,
61		class SendOp,
62		class RecvOp,
63		class WaitOp,
64		class DescState,
65		class ImplBase = udp_socket::implementation,
66		class Endpoint = endpoint>
67		class reactor_datagram_socket
68		: public reactor_basic_socket<
69		Derived,
70		ImplBase,
71		Service,
72		DescState,
73		Endpoint>
74		{
75		using base_type =
76		reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>;
77		using self_type = reactor_datagram_socket<
78		Derived,
79		Service,
80		ConnOp,
81		SendToOp,
82		RecvFromOp,
83		SendOp,
84		RecvOp,
85		WaitOp,
86		DescState,
87		ImplBase,
88		Endpoint>;
89		friend base_type;
90		friend reactor_io_core<Derived, Service, DescState>;
91		friend Derived;
92
93		protected:
94		// NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
95	636x	explicit reactor_datagram_socket(Service& svc) noexcept : base_type(svc) {}
96
97		protected:
98		Endpoint remote_endpoint_;
99
100		public:
101		/// Pending connect operation slot.
102		ConnOp conn_;
103
104		/// Pending send_to operation slot.
105		SendToOp wr_;
106
107		/// Pending recv_from operation slot.
108		RecvFromOp rd_;
109
110		/// Pending connected send operation slot.
111		SendOp send_wr_;
112
113		/// Pending connected recv operation slot.
114		RecvOp recv_rd_;
115
116		/// Pending wait-for-read operation slot.
117		WaitOp wait_rd_;
118
119		/// Pending wait-for-write operation slot.
120		WaitOp wait_wr_;
121
122		/// Pending wait-for-error operation slot.
123		WaitOp wait_er_;
124
125	636x	~reactor_datagram_socket() override = default;
126
127		/// Return the cached remote endpoint.
128	8x	Endpoint remote_endpoint() const noexcept override
129		{
130	8x	return remote_endpoint_;
131		}
132
133		// --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
134
135	167x	std::coroutine_handle<> send_to(
136		capy::continuation& cont,
137		capy::executor_ref ex,
138		buffer_param buf,
139		Endpoint dest,
140		int flags,
141		std::stop_token token,
142		std::error_code* ec,
143		std::size_t* bytes_out) override
144		{
145	167x	return do_send_to(cont, ex, buf, dest, flags, token, ec, bytes_out);
146		}
147
148	187x	std::coroutine_handle<> recv_from(
149		capy::continuation& cont,
150		capy::executor_ref ex,
151		buffer_param buf,
152		Endpoint* source,
153		int flags,
154		std::stop_token token,
155		std::error_code* ec,
156		std::size_t* bytes_out) override
157		{
158	187x	return do_recv_from(cont, ex, buf, source, flags, token, ec, bytes_out);
159		}
160
161	60x	std::coroutine_handle<> connect(
162		capy::continuation& cont,
163		capy::executor_ref ex,
164		Endpoint ep,
165		std::stop_token token,
166		std::error_code* ec) override
167		{
168	60x	return do_connect(cont, ex, ep, token, ec);
169		}
170
171	123x	std::coroutine_handle<> send(
172		capy::continuation& cont,
173		capy::executor_ref ex,
174		buffer_param buf,
175		int flags,
176		std::stop_token token,
177		std::error_code* ec,
178		std::size_t* bytes_out) override
179		{
180	123x	return do_send(cont, ex, buf, flags, token, ec, bytes_out);
181		}
182
183	165x	std::coroutine_handle<> recv(
184		capy::continuation& cont,
185		capy::executor_ref ex,
186		buffer_param buf,
187		int flags,
188		std::stop_token token,
189		std::error_code* ec,
190		std::size_t* bytes_out) override
191		{
192	165x	return do_recv(cont, ex, buf, flags, token, ec, bytes_out);
193		}
194
195	63x	void cancel() noexcept override
196		{
197	63x	this->do_cancel();
198	63x	}
199
200	44x	std::coroutine_handle<> wait(
201		capy::continuation& cont,
202		capy::executor_ref ex,
203		wait_type w,
204		std::stop_token token,
205		std::error_code* ec) override
206		{
207	44x	return do_wait(cont, ex, w, token, ec);
208		}
209
210		// --- End virtual overrides ---
211
212		/// Close the socket (non-virtual, called by the service).
213	2648x	void close_socket() noexcept
214		{
215	2648x	do_close_socket();
216	2648x	}
217
218		/// Cache local and remote endpoints.
219	186x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
220		{
221	186x	this->local_endpoint_ = std::move(local);
222	186x	remote_endpoint_ = std::move(remote);
223	186x	}
224
225		/** Shared send_to dispatch.
226
227		Tries sendmsg() speculatively. On success or hard error,
228		returns via inline budget or posts through queue.
229		On EAGAIN, registers with the reactor.
230		*/
231		std::coroutine_handle<> do_send_to(
232		capy::continuation&,
233		capy::executor_ref,
234		buffer_param,
235		Endpoint const&,
236		int flags,
237		std::stop_token const&,
238		std::error_code*,
239		std::size_t*);
240
241		/** Shared recv_from dispatch.
242
243		Tries recvmsg() speculatively. On success or hard error,
244		returns via inline budget or posts through queue.
245		On EAGAIN, registers with the reactor.
246		*/
247		std::coroutine_handle<> do_recv_from(
248		capy::continuation&,
249		capy::executor_ref,
250		buffer_param,
251		Endpoint*,
252		int flags,
253		std::stop_token const&,
254		std::error_code*,
255		std::size_t*);
256
257		/** Shared connect dispatch.
258
259		Tries connect() speculatively. On synchronous completion,
260		returns via inline budget or posts through queue.
261		The result is always synchronous: a datagram connect only
262		records the peer address in the kernel.
263		*/
264		std::coroutine_handle<> do_connect(
265		capy::continuation&,
266		capy::executor_ref,
267		Endpoint const&,
268		std::stop_token const&,
269		std::error_code*);
270
271		/** Shared connected send dispatch.
272
273		Like do_send_to but uses send_wr_ slot and sendmsg()
274		with msg_name=nullptr.
275		*/
276		std::coroutine_handle<> do_send(
277		capy::continuation&,
278		capy::executor_ref,
279		buffer_param,
280		int flags,
281		std::stop_token const&,
282		std::error_code*,
283		std::size_t*);
284
285		/** Shared connected recv dispatch.
286
287		Like do_recv_from but uses recv_rd_ slot and recvmsg()
288		with msg_name=nullptr.
289		*/
290		std::coroutine_handle<> do_recv(
291		capy::continuation&,
292		capy::executor_ref,
293		buffer_param,
294		int flags,
295		std::stop_token const&,
296		std::error_code*,
297		std::size_t*);
298
299		/** Shared readiness-wait dispatch.
300
301		Every wait type probes the descriptor with a zero-timeout
302		`poll()` and completes at once if the condition already
303		holds; otherwise the op re-probes under the descriptor mutex
304		and parks, completing when a reactor event arrives and a
305		fresh probe confirms the condition. A write wait therefore
306		completes only while a non-blocking write can make progress.
307		*/
308		std::coroutine_handle<> do_wait(
309		capy::continuation&,
310		capy::executor_ref,
311		wait_type,
312		std::stop_token const&,
313		std::error_code*);
314
315		/** Close the socket and cancel pending operations.
316
317		Extends the base do_close_socket() to also reset
318		the remote endpoint.
319		*/
320	2648x	void do_close_socket() noexcept
321		{
322	2648x	base_type::do_close_socket();
323	2648x	remote_endpoint_ = Endpoint{};
324	2648x	}
325
326	8x	native_handle_type do_release_socket() noexcept
327		{
328	8x	auto fd = base_type::do_release_socket();
329	8x	remote_endpoint_ = Endpoint{};
330	8x	return fd;
331		}
332
333		/** Reset op slots and the cached peer for recycling.
334
335		See reactor_stream_socket::reuse() — same rationale: op state
336		is overwritten by the next do_* call, so only `remote_endpoint_`
337		needs an active reset here; every other slot only asserts
338		`stop_cb` disengaged.
339
340		@pre refs_ == 0, fd closed and deregistered, no op in flight.
341		*/
342	143x	void reuse() noexcept
343		{
344	143x	base_type::reuse();
345	143x	remote_endpoint_ = Endpoint{};
346	143x	BOOST_COROSIO_ASSERT(!conn_.stop_cb);
347	143x	BOOST_COROSIO_ASSERT(!wr_.stop_cb);
348	143x	BOOST_COROSIO_ASSERT(!rd_.stop_cb);
349	143x	BOOST_COROSIO_ASSERT(!send_wr_.stop_cb);
350	143x	BOOST_COROSIO_ASSERT(!recv_rd_.stop_cb);
351	143x	BOOST_COROSIO_ASSERT(!wait_rd_.stop_cb);
352	143x	BOOST_COROSIO_ASSERT(!wait_wr_.stop_cb);
353	143x	BOOST_COROSIO_ASSERT(!wait_er_.stop_cb);
354	143x	}
355
356		#if !defined(NDEBUG)
357		/// Extend base_type::poison() to the cached peer endpoint.
358	779x	void poison() noexcept
359		{
360	779x	base_type::poison();
361	779x	std::memset(
362	779x	static_cast<void*>(&remote_endpoint_), 0xDB,
363		sizeof(remote_endpoint_));
364	779x	}
365		#endif
366
367		/** Shut down part or all of the full-duplex connection.
368
369		Not an override — concrete backends forward here.
370
371		@param what 0 = receive, 1 = send, 2 = both.
372		*/
373	14x	std::error_code do_shutdown(int what) noexcept
374		{
375		int how;
376	14x	switch (what)
377		{
378	2x	case 0:
379	2x	how = SHUT_RD;
380	2x	break;
381	8x	case 1:
382	8x	how = SHUT_WR;
383	8x	break;
384	4x	case 2:
385	4x	how = SHUT_RDWR;
386	4x	break;
387	✗	default:
388	✗	return make_err(EINVAL);
389		}
390	14x	if (::shutdown(this->fd_, how) != 0)
391	4x	return make_err(errno);
392	10x	return {};
393		}
394
395		private:
396		// CRTP callbacks for reactor_io_core cancel/close
397
398		template<class Op>
399	22x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
400		{
401	22x	if (&op == static_cast<void*>(&conn_))
402	✗	return &this->desc_state_.connect_op;
403	22x	if (&op == static_cast<void*>(&rd_))
404	6x	return &this->desc_state_.read_op;
405	16x	if (&op == static_cast<void*>(&wr_))
406	2x	return &this->desc_state_.write_op;
407	14x	if (&op == static_cast<void*>(&recv_rd_))
408	4x	return &this->desc_state_.read_op;
409	10x	if (&op == static_cast<void*>(&send_wr_))
410	2x	return &this->desc_state_.write_op;
411	8x	if (&op == static_cast<void*>(&wait_rd_))
412	4x	return &this->desc_state_.wait_read_op;
413	4x	if (&op == static_cast<void*>(&wait_wr_))
414	✗	return &this->desc_state_.wait_write_op;
415	4x	if (&op == static_cast<void*>(&wait_er_))
416	4x	return &this->desc_state_.wait_error_op;
417	✗	return nullptr;
418		}
419
420		template<class Fn>
421	2719x	void for_each_op(Fn fn) noexcept
422		{
423	2719x	fn(conn_);
424	2719x	fn(rd_);
425	2719x	fn(wr_);
426	2719x	fn(recv_rd_);
427	2719x	fn(send_wr_);
428	2719x	fn(wait_rd_);
429	2719x	fn(wait_wr_);
430	2719x	fn(wait_er_);
431	2719x	}
432
433		template<class Fn>
434	3298x	void for_each_desc_entry(Fn fn) noexcept
435		{
436	3298x	fn(conn_, this->desc_state_.connect_op);
437	3298x	fn(rd_, this->desc_state_.read_op);
438	3298x	fn(wr_, this->desc_state_.write_op);
439	3298x	fn(recv_rd_, this->desc_state_.read_op);
440	3298x	fn(send_wr_, this->desc_state_.write_op);
441	3298x	fn(wait_rd_, this->desc_state_.wait_read_op);
442	3298x	fn(wait_wr_, this->desc_state_.wait_write_op);
443	3298x	fn(wait_er_, this->desc_state_.wait_error_op);
444	3298x	}
445		};
446
447		// do_send_to
448
449		template<
450		class Derived,
451		class Service,
452		class ConnOp,
453		class SendToOp,
454		class RecvFromOp,
455		class SendOp,
456		class RecvOp,
457		class WaitOp,
458		class DescState,
459		class ImplBase,
460		class Endpoint>
461		std::coroutine_handle<>
462	167x	reactor_datagram_socket<
463		Derived,
464		Service,
465		ConnOp,
466		SendToOp,
467		RecvFromOp,
468		SendOp,
469		RecvOp,
470		WaitOp,
471		DescState,
472		ImplBase,
473		Endpoint>::
474		do_send_to(
475		capy::continuation& cont,
476		capy::executor_ref ex,
477		buffer_param param,
478		Endpoint const& dest,
479		int flags,
480		std::stop_token const& token,
481		std::error_code* ec,
482		std::size_t* bytes_out)
483		{
484	167x	auto& op = wr_;
485	167x	op.reset();
486
487	167x	capy::mutable_buffer bufs[SendToOp::max_buffers];
488	167x	op.iovec_count =
489	167x	static_cast<int>(param.copy_to(bufs, SendToOp::max_buffers));
490
491	328x	for (int i = 0; i < op.iovec_count; ++i)
492		{
493	161x	op.iovecs[i].iov_base = bufs[i].data();
494	161x	op.iovecs[i].iov_len = bufs[i].size();
495		}
496
497		// Set up destination address
498	167x	op.dest_len = to_sockaddr(dest, socket_family(this->fd_), op.dest_storage);
499	167x	op.fd = this->fd_;
500	167x	op.msg_flags = to_native_msg_flags(flags);
501
502		// Speculative sendmsg
503	167x	msghdr msg{};
504	167x	msg.msg_name = &op.dest_storage;
505	167x	msg.msg_namelen = op.dest_len;
506	167x	msg.msg_iov = op.iovecs;
507	167x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
508
509		#ifdef MSG_NOSIGNAL
510	167x	int send_flags = op.msg_flags \| MSG_NOSIGNAL;
511		#else
512		int send_flags = op.msg_flags;
513		#endif
514
515		ssize_t n;
516		do
517		{
518	167x	n = ::sendmsg(this->fd_, &msg, send_flags);
519		}
520	167x	while (n < 0 && errno == EINTR);
521
522	167x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
523		{
524	133x	int err = (n < 0) ? errno : 0;
525	133x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
526
527	133x	if (this->svc_.scheduler().try_consume_inline_budget())
528		{
529	51x	*ec = err ? make_err(err) : std::error_code{};
530	51x	*bytes_out = bytes;
531	51x	return dispatch_coro(ex, cont);
532		}
533	82x	op.cont = &cont;
534	82x	op.ex = ex;
535	82x	op.ec_out = ec;
536	82x	op.bytes_out = bytes_out;
537	82x	op.start(token, static_cast<Derived*>(this));
538	82x	op.object_ref_ = detail::object_ref(this);
539	82x	op.complete(err, bytes);
540	82x	this->svc_.post(&op);
541	82x	return std::noop_coroutine();
542		}
543
544		// EAGAIN — register with reactor
545	34x	op.cont = &cont;
546	34x	op.ex = ex;
547	34x	op.ec_out = ec;
548	34x	op.bytes_out = bytes_out;
549	34x	op.start(token, static_cast<Derived*>(this));
550	34x	op.object_ref_ = detail::object_ref(this);
551
552	34x	this->register_op(
553	34x	op, this->desc_state_.write_op, this->desc_state_.write_ready, true);
554	34x	return std::noop_coroutine();
555		}
556
557		// do_recv_from
558
559		template<
560		class Derived,
561		class Service,
562		class ConnOp,
563		class SendToOp,
564		class RecvFromOp,
565		class SendOp,
566		class RecvOp,
567		class WaitOp,
568		class DescState,
569		class ImplBase,
570		class Endpoint>
571		std::coroutine_handle<>
572	187x	reactor_datagram_socket<
573		Derived,
574		Service,
575		ConnOp,
576		SendToOp,
577		RecvFromOp,
578		SendOp,
579		RecvOp,
580		WaitOp,
581		DescState,
582		ImplBase,
583		Endpoint>::
584		do_recv_from(
585		capy::continuation& cont,
586		capy::executor_ref ex,
587		buffer_param param,
588		Endpoint* source,
589		int flags,
590		std::stop_token const& token,
591		std::error_code* ec,
592		std::size_t* bytes_out)
593		{
594	187x	auto& op = rd_;
595	187x	op.reset();
596
597	187x	capy::mutable_buffer bufs[RecvFromOp::max_buffers];
598	187x	op.iovec_count =
599	187x	static_cast<int>(param.copy_to(bufs, RecvFromOp::max_buffers));
600
601	187x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
602		{
603	4x	op.cont = &cont;
604	4x	op.ex = ex;
605	4x	op.ec_out = ec;
606	4x	op.bytes_out = bytes_out;
607	4x	op.start(token, static_cast<Derived*>(this));
608	4x	op.object_ref_ = detail::object_ref(this);
609	4x	op.complete(0, 0);
610	4x	this->svc_.post(&op);
611	4x	return std::noop_coroutine();
612		}
613
614	366x	for (int i = 0; i < op.iovec_count; ++i)
615		{
616	183x	op.iovecs[i].iov_base = bufs[i].data();
617	183x	op.iovecs[i].iov_len = bufs[i].size();
618		}
619
620	183x	op.fd = this->fd_;
621	183x	op.source_out = source;
622	183x	op.msg_flags = to_native_msg_flags(flags);
623
624		// Speculative recvmsg
625	183x	msghdr msg{};
626	183x	msg.msg_name = &op.source_storage;
627	183x	msg.msg_namelen = sizeof(op.source_storage);
628	183x	msg.msg_iov = op.iovecs;
629	183x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
630
631		ssize_t n;
632		do
633		{
634	183x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
635		}
636	183x	while (n < 0 && errno == EINTR);
637
638	183x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
639		{
640	106x	int err = (n < 0) ? errno : 0;
641	106x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
642	106x	if (n >= 0)
643	104x	op.source_addrlen = msg.msg_namelen;
644
645	106x	if (this->svc_.scheduler().try_consume_inline_budget())
646		{
647	89x	*ec = err ? make_err(err) : std::error_code{};
648	89x	*bytes_out = bytes;
649	89x	if (source && !err && n >= 0)
650	174x	*source = from_sockaddr_as(
651	87x	op.source_storage, op.source_addrlen, Endpoint{});
652	89x	return dispatch_coro(ex, cont);
653		}
654	17x	op.cont = &cont;
655	17x	op.ex = ex;
656	17x	op.ec_out = ec;
657	17x	op.bytes_out = bytes_out;
658	17x	op.start(token, static_cast<Derived*>(this));
659	17x	op.object_ref_ = detail::object_ref(this);
660	17x	op.complete(err, bytes);
661	17x	this->svc_.post(&op);
662	17x	return std::noop_coroutine();
663		}
664
665		// EAGAIN — register with reactor
666	77x	op.cont = &cont;
667	77x	op.ex = ex;
668	77x	op.ec_out = ec;
669	77x	op.bytes_out = bytes_out;
670	77x	op.start(token, static_cast<Derived*>(this));
671	77x	op.object_ref_ = detail::object_ref(this);
672
673	77x	this->register_op(
674	77x	op, this->desc_state_.read_op, this->desc_state_.read_ready);
675	77x	return std::noop_coroutine();
676		}
677
678		// do_connect
679
680		template<
681		class Derived,
682		class Service,
683		class ConnOp,
684		class SendToOp,
685		class RecvFromOp,
686		class SendOp,
687		class RecvOp,
688		class WaitOp,
689		class DescState,
690		class ImplBase,
691		class Endpoint>
692		std::coroutine_handle<>
693	60x	reactor_datagram_socket<
694		Derived,
695		Service,
696		ConnOp,
697		SendToOp,
698		RecvFromOp,
699		SendOp,
700		RecvOp,
701		WaitOp,
702		DescState,
703		ImplBase,
704		Endpoint>::
705		do_connect(
706		capy::continuation& cont,
707		capy::executor_ref ex,
708		Endpoint const& ep,
709		std::stop_token const& token,
710		std::error_code* ec)
711		{
712	60x	auto& op = conn_;
713
714	60x	sockaddr_storage storage{};
715	60x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
716		int result =
717	60x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
718
719	60x	if (result == 0)
720		{
721	56x	sockaddr_storage local_storage{};
722	56x	socklen_t local_len = sizeof(local_storage);
723	56x	if (::getsockname(
724		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
725	56x	&local_len) == 0)
726	48x	this->local_endpoint_ =
727	56x	from_sockaddr_as(local_storage, local_len, Endpoint{});
728	56x	remote_endpoint_ = ep;
729		}
730
731		// A datagram connect cannot block — it only records the peer
732		// address — so the result is handled synchronously; the deferred
733		// arm exists for a spent inline budget, not for a parked op.
734	60x	int err = (result < 0) ? errno : 0;
735	60x	if (this->svc_.scheduler().try_consume_inline_budget())
736		{
737	14x	*ec = err ? make_err(err) : std::error_code{};
738	14x	return dispatch_coro(ex, cont);
739		}
740	46x	op.reset();
741	46x	op.cont = &cont;
742	46x	op.ex = ex;
743	46x	op.ec_out = ec;
744	46x	op.fd = this->fd_;
745	46x	op.target_endpoint = ep;
746	46x	op.start(token, static_cast<Derived*>(this));
747	46x	op.object_ref_ = detail::object_ref(this);
748	46x	op.complete(err, 0);
749	46x	this->svc_.post(&op);
750	46x	return std::noop_coroutine();
751		}
752
753		// do_send (connected mode)
754
755		template<
756		class Derived,
757		class Service,
758		class ConnOp,
759		class SendToOp,
760		class RecvFromOp,
761		class SendOp,
762		class RecvOp,
763		class WaitOp,
764		class DescState,
765		class ImplBase,
766		class Endpoint>
767		std::coroutine_handle<>
768	123x	reactor_datagram_socket<
769		Derived,
770		Service,
771		ConnOp,
772		SendToOp,
773		RecvFromOp,
774		SendOp,
775		RecvOp,
776		WaitOp,
777		DescState,
778		ImplBase,
779		Endpoint>::
780		do_send(
781		capy::continuation& cont,
782		capy::executor_ref ex,
783		buffer_param param,
784		int flags,
785		std::stop_token const& token,
786		std::error_code* ec,
787		std::size_t* bytes_out)
788		{
789	123x	auto& op = send_wr_;
790	123x	op.reset();
791
792	123x	capy::mutable_buffer bufs[SendOp::max_buffers];
793	123x	op.iovec_count = static_cast<int>(param.copy_to(bufs, SendOp::max_buffers));
794
795	242x	for (int i = 0; i < op.iovec_count; ++i)
796		{
797	119x	op.iovecs[i].iov_base = bufs[i].data();
798	119x	op.iovecs[i].iov_len = bufs[i].size();
799		}
800
801	123x	op.fd = this->fd_;
802	123x	op.msg_flags = to_native_msg_flags(flags);
803
804		// Speculative sendmsg with no destination (connected mode)
805	123x	msghdr msg{};
806	123x	msg.msg_iov = op.iovecs;
807	123x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
808
809		#ifdef MSG_NOSIGNAL
810	123x	int send_flags = op.msg_flags \| MSG_NOSIGNAL;
811		#else
812		int send_flags = op.msg_flags;
813		#endif
814
815		ssize_t n;
816		do
817		{
818	123x	n = ::sendmsg(this->fd_, &msg, send_flags);
819		}
820	123x	while (n < 0 && errno == EINTR);
821
822	123x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
823		{
824	89x	int err = (n < 0) ? errno : 0;
825	89x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
826
827	89x	if (this->svc_.scheduler().try_consume_inline_budget())
828		{
829	58x	*ec = err ? make_err(err) : std::error_code{};
830	58x	*bytes_out = bytes;
831	58x	return dispatch_coro(ex, cont);
832		}
833	31x	op.cont = &cont;
834	31x	op.ex = ex;
835	31x	op.ec_out = ec;
836	31x	op.bytes_out = bytes_out;
837	31x	op.start(token, static_cast<Derived*>(this));
838	31x	op.object_ref_ = detail::object_ref(this);
839	31x	op.complete(err, bytes);
840	31x	this->svc_.post(&op);
841	31x	return std::noop_coroutine();
842		}
843
844		// EAGAIN — register with reactor
845	34x	op.cont = &cont;
846	34x	op.ex = ex;
847	34x	op.ec_out = ec;
848	34x	op.bytes_out = bytes_out;
849	34x	op.start(token, static_cast<Derived*>(this));
850	34x	op.object_ref_ = detail::object_ref(this);
851
852	34x	this->register_op(
853	34x	op, this->desc_state_.write_op, this->desc_state_.write_ready, true);
854	34x	return std::noop_coroutine();
855		}
856
857		// do_recv (connected mode)
858
859		template<
860		class Derived,
861		class Service,
862		class ConnOp,
863		class SendToOp,
864		class RecvFromOp,
865		class SendOp,
866		class RecvOp,
867		class WaitOp,
868		class DescState,
869		class ImplBase,
870		class Endpoint>
871		std::coroutine_handle<>
872	165x	reactor_datagram_socket<
873		Derived,
874		Service,
875		ConnOp,
876		SendToOp,
877		RecvFromOp,
878		SendOp,
879		RecvOp,
880		WaitOp,
881		DescState,
882		ImplBase,
883		Endpoint>::
884		do_recv(
885		capy::continuation& cont,
886		capy::executor_ref ex,
887		buffer_param param,
888		int flags,
889		std::stop_token const& token,
890		std::error_code* ec,
891		std::size_t* bytes_out)
892		{
893	165x	auto& op = recv_rd_;
894	165x	op.reset();
895
896	165x	capy::mutable_buffer bufs[RecvOp::max_buffers];
897	165x	op.iovec_count = static_cast<int>(param.copy_to(bufs, RecvOp::max_buffers));
898
899	165x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
900		{
901	4x	op.cont = &cont;
902	4x	op.ex = ex;
903	4x	op.ec_out = ec;
904	4x	op.bytes_out = bytes_out;
905	4x	op.start(token, static_cast<Derived*>(this));
906	4x	op.object_ref_ = detail::object_ref(this);
907	4x	op.complete(0, 0);
908	4x	this->svc_.post(&op);
909	4x	return std::noop_coroutine();
910		}
911
912	322x	for (int i = 0; i < op.iovec_count; ++i)
913		{
914	161x	op.iovecs[i].iov_base = bufs[i].data();
915	161x	op.iovecs[i].iov_len = bufs[i].size();
916		}
917
918	161x	op.fd = this->fd_;
919	161x	op.msg_flags = to_native_msg_flags(flags);
920
921		// Speculative recvmsg with no source (connected mode)
922	161x	msghdr msg{};
923	161x	msg.msg_iov = op.iovecs;
924	161x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
925
926		ssize_t n;
927		do
928		{
929	161x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
930		}
931	161x	while (n < 0 && errno == EINTR);
932
933	161x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
934		{
935	67x	int err = (n < 0) ? errno : 0;
936	67x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
937
938	67x	if (this->svc_.scheduler().try_consume_inline_budget())
939		{
940	51x	*ec = err ? make_err(err) : std::error_code{};
941	51x	*bytes_out = bytes;
942	51x	return dispatch_coro(ex, cont);
943		}
944	16x	op.cont = &cont;
945	16x	op.ex = ex;
946	16x	op.ec_out = ec;
947	16x	op.bytes_out = bytes_out;
948	16x	op.start(token, static_cast<Derived*>(this));
949	16x	op.object_ref_ = detail::object_ref(this);
950	16x	op.complete(err, bytes);
951	16x	this->svc_.post(&op);
952	16x	return std::noop_coroutine();
953		}
954
955		// EAGAIN — register with reactor
956	94x	op.cont = &cont;
957	94x	op.ex = ex;
958	94x	op.ec_out = ec;
959	94x	op.bytes_out = bytes_out;
960	94x	op.start(token, static_cast<Derived*>(this));
961	94x	op.object_ref_ = detail::object_ref(this);
962
963	94x	this->register_op(
964	94x	op, this->desc_state_.read_op, this->desc_state_.read_ready);
965	94x	return std::noop_coroutine();
966		}
967
968		// do_wait
969
970		template<
971		class Derived,
972		class Service,
973		class ConnOp,
974		class SendToOp,
975		class RecvFromOp,
976		class SendOp,
977		class RecvOp,
978		class WaitOp,
979		class DescState,
980		class ImplBase,
981		class Endpoint>
982		std::coroutine_handle<>
983	44x	reactor_datagram_socket<
984		Derived,
985		Service,
986		ConnOp,
987		SendToOp,
988		RecvFromOp,
989		SendOp,
990		RecvOp,
991		WaitOp,
992		DescState,
993		ImplBase,
994		Endpoint>::
995		do_wait(
996		capy::continuation& cont,
997		capy::executor_ref ex,
998		wait_type w,
999		std::stop_token const& token,
1000		std::error_code* ec)
1001		{
1002		WaitOp* op_ptr;
1003		reactor_op_base** desc_slot_ptr;
1004		std::uint32_t event;
1005
1006	44x	if (w == wait_type::read)
1007		{
1008	28x	op_ptr = &wait_rd_;
1009	28x	desc_slot_ptr = &this->desc_state_.wait_read_op;
1010	28x	event = reactor_event_read;
1011		}
1012	16x	else if (w == wait_type::write)
1013		{
1014	8x	op_ptr = &wait_wr_;
1015	8x	desc_slot_ptr = &this->desc_state_.wait_write_op;
1016	8x	event = reactor_event_write;
1017		}
1018		else // wait_type::error
1019		{
1020	8x	op_ptr = &wait_er_;
1021	8x	desc_slot_ptr = &this->desc_state_.wait_error_op;
1022	8x	event = reactor_event_error;
1023		}
1024
1025	44x	auto& op = *op_ptr;
1026
1027		// Speculative probe, mirroring the speculative read: an
1028		// edge-triggered reactor cannot report a condition that already
1029		// holds, so a wait initiated on an already-ready socket would
1030		// otherwise park forever.
1031	44x	int perr = 0;
1032	44x	if (WaitOp::probe(this->fd_, event, perr))
1033		{
1034	18x	if (this->svc_.scheduler().try_consume_inline_budget())
1035		{
1036	4x	*ec = perr ? make_err(perr) : std::error_code{};
1037	4x	return dispatch_coro(ex, cont);
1038		}
1039	14x	op.reset();
1040	14x	op.wait_event = event;
1041	14x	op.cont = &cont;
1042	14x	op.ex = ex;
1043	14x	op.ec_out = ec;
1044	14x	op.fd = this->fd_;
1045	14x	op.start(token, static_cast<Derived*>(this));
1046	14x	op.object_ref_ = detail::object_ref(this);
1047	14x	op.complete(perr, 0);
1048	14x	this->svc_.post(&op);
1049	14x	return std::noop_coroutine();
1050		}
1051
1052	26x	op.reset();
1053	26x	op.wait_event = event;
1054	26x	op.cont = &cont;
1055	26x	op.ex = ex;
1056	26x	op.ec_out = ec;
1057	26x	op.fd = this->fd_;
1058	26x	op.start(token, static_cast<Derived*>(this));
1059	26x	op.object_ref_ = detail::object_ref(this);
1060
1061		// Force register_op's ready path so the wait op re-probes under
1062		// the descriptor mutex before parking. An edge consumed between
1063		// the speculative probe above and the park (a concurrent short
1064		// read, or an error event dispatched to an empty slot) would
1065		// otherwise leave the wait parked on a ready socket.
1066	26x	bool force_probe = true;
1067	26x	this->register_op(
1068		op, *desc_slot_ptr, force_probe, event == reactor_event_write);
1069	26x	return std::noop_coroutine();
1070		}
1071
1072		} // namespace boost::corosio::detail
1073
1074		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
1075