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

98.7% Lines (369 / 374) 100.0% Functions (116 / 116)

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

reactor_datagram_socket.hpp

f(x) Functions (116)

Function Calls Lines Blocks

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