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

45.4% Lines (697/1536) 44.6% List of functions (45/101)

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

reactor_datagram_socket.hpp

f(x) Functions (101)

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/native/detail/reactor/reactor_basic_socket.hpp>
15		#include <boost/corosio/detail/dispatch_coro.hpp>
16		#include <boost/capy/buffers.hpp>
17
18		#include <coroutine>
19
20		#include <errno.h>
21		#include <sys/socket.h>
22		#include <sys/uio.h>
23
24		namespace boost::corosio::detail {
25
26		/* Map portable message_flags values to native MSG_* constants. */
27		inline int
28	94x	to_native_msg_flags(int flags) noexcept
29		{
30	94x	int native = 0;
31	94x	if (flags & 1) native \|= MSG_PEEK;
32	94x	if (flags & 2) native \|= MSG_OOB;
33	94x	if (flags & 4) native \|= MSG_DONTROUTE;
34	94x	return native;
35		}
36
37		/** CRTP base for reactor-backed datagram socket implementations.
38
39		Inherits shared data members and cancel/close/register logic
40		from reactor_basic_socket. Adds datagram-specific I/O dispatch
41		for both connectionless (send_to, recv_from) and connected
42		(connect, send, recv) modes.
43
44		@tparam Derived The concrete socket type (CRTP).
45		@tparam Service The backend's datagram service type.
46		@tparam ConnOp The backend's connect op type.
47		@tparam SendToOp The backend's send_to op type.
48		@tparam RecvFromOp The backend's recv_from op type.
49		@tparam SendOp The backend's connected send op type.
50		@tparam RecvOp The backend's connected recv op type.
51		@tparam DescState The backend's descriptor_state type.
52		@tparam ImplBase The public vtable base
53		(udp_socket::implementation or
54		local_datagram_socket::implementation).
55		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
56		*/
57		template<
58		class Derived,
59		class Service,
60		class ConnOp,
61		class SendToOp,
62		class RecvFromOp,
63		class SendOp,
64		class RecvOp,
65		class DescState,
66		class ImplBase = udp_socket::implementation,
67		class Endpoint = endpoint>
68		class reactor_datagram_socket
69		: public reactor_basic_socket<
70		Derived,
71		ImplBase,
72		Service,
73		DescState,
74		Endpoint>
75		{
76		using base_type = reactor_basic_socket<
77		Derived,
78		ImplBase,
79		Service,
80		DescState,
81		Endpoint>;
82		friend base_type;
83		friend Derived;
84
85	120x	explicit reactor_datagram_socket(Service& svc) noexcept : base_type(svc) {}
86
87		protected:
88		Endpoint remote_endpoint_;
89
90		public:
91		/// Pending connect operation slot.
92		ConnOp conn_;
93
94		/// Pending send_to operation slot.
95		SendToOp wr_;
96
97		/// Pending recv_from operation slot.
98		RecvFromOp rd_;
99
100		/// Pending connected send operation slot.
101		SendOp send_wr_;
102
103		/// Pending connected recv operation slot.
104		RecvOp recv_rd_;
105
106	120x	~reactor_datagram_socket() override = default;
107
108		/// Return the cached remote endpoint.
109	4x	Endpoint remote_endpoint() const noexcept override
110		{
111	4x	return remote_endpoint_;
112		}
113
114		/// Cache local and remote endpoints.
115	10x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
116		{
117	10x	this->local_endpoint_ = std::move(local);
118	10x	remote_endpoint_ = std::move(remote);
119	10x	}
120
121		/** Shared send_to dispatch.
122
123		Tries sendmsg() speculatively. On success or hard error,
124		returns via inline budget or posts through queue.
125		On EAGAIN, registers with the reactor.
126		*/
127		std::coroutine_handle<> do_send_to(
128		std::coroutine_handle<>,
129		capy::executor_ref,
130		buffer_param,
131		Endpoint const&,
132		int flags,
133		std::stop_token const&,
134		std::error_code*,
135		std::size_t*);
136
137		/** Shared recv_from dispatch.
138
139		Tries recvmsg() speculatively. On success or hard error,
140		returns via inline budget or posts through queue.
141		On EAGAIN, registers with the reactor.
142		*/
143		std::coroutine_handle<> do_recv_from(
144		std::coroutine_handle<>,
145		capy::executor_ref,
146		buffer_param,
147		Endpoint*,
148		int flags,
149		std::stop_token const&,
150		std::error_code*,
151		std::size_t*);
152
153		/** Shared connect dispatch.
154
155		Tries connect() speculatively. On synchronous completion,
156		returns via inline budget or posts through queue.
157		On EINPROGRESS, registers with the reactor.
158		*/
159		std::coroutine_handle<> do_connect(
160		std::coroutine_handle<>,
161		capy::executor_ref,
162		Endpoint const&,
163		std::stop_token const&,
164		std::error_code*);
165
166		/** Shared connected send dispatch.
167
168		Like do_send_to but uses send_wr_ slot and sendmsg()
169		with msg_name=nullptr.
170		*/
171		std::coroutine_handle<> do_send(
172		std::coroutine_handle<>,
173		capy::executor_ref,
174		buffer_param,
175		int flags,
176		std::stop_token const&,
177		std::error_code*,
178		std::size_t*);
179
180		/** Shared connected recv dispatch.
181
182		Like do_recv_from but uses recv_rd_ slot and recvmsg()
183		with msg_name=nullptr.
184		*/
185		std::coroutine_handle<> do_recv(
186		std::coroutine_handle<>,
187		capy::executor_ref,
188		buffer_param,
189		int flags,
190		std::stop_token const&,
191		std::error_code*,
192		std::size_t*);
193
194		/** Close the socket and cancel pending operations.
195
196		Extends the base do_close_socket() to also reset
197		the remote endpoint.
198		*/
199	464x	void do_close_socket() noexcept
200		{
201	464x	base_type::do_close_socket();
202	464x	remote_endpoint_ = Endpoint{};
203	464x	}
204
205	✗	native_handle_type do_release_socket() noexcept
206		{
207	✗	auto fd = base_type::do_release_socket();
208	✗	remote_endpoint_ = Endpoint{};
209	✗	return fd;
210		}
211
212		/** Shut down part or all of the full-duplex connection.
213
214		Not an override — concrete backends forward here.
215
216		@param what 0 = receive, 1 = send, 2 = both.
217		*/
218	✗	std::error_code do_shutdown(int what) noexcept
219		{
220		int how;
221	✗	switch (what)
222		{
223	✗	case 0:
224	✗	how = SHUT_RD;
225	✗	break;
226	✗	case 1:
227	✗	how = SHUT_WR;
228	✗	break;
229	✗	case 2:
230	✗	how = SHUT_RDWR;
231	✗	break;
232	✗	default:
233	✗	return make_err(EINVAL);
234		}
235	✗	if (::shutdown(this->fd_, how) != 0)
236	✗	return make_err(errno);
237	✗	return {};
238		}
239
240		private:
241		// CRTP callbacks for reactor_basic_socket cancel/close
242
243		template<class Op>
244	2x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
245		{
246	2x	if (&op == static_cast<void*>(&conn_))
247	✗	return &this->desc_state_.connect_op;
248	2x	if (&op == static_cast<void*>(&rd_))
249	2x	return &this->desc_state_.read_op;
250	✗	if (&op == static_cast<void*>(&wr_))
251	✗	return &this->desc_state_.write_op;
252	✗	if (&op == static_cast<void*>(&recv_rd_))
253	✗	return &this->desc_state_.read_op;
254	✗	if (&op == static_cast<void*>(&send_wr_))
255	✗	return &this->desc_state_.write_op;
256	✗	return nullptr;
257		}
258
259		template<class Op>
260	✗	bool* op_to_cancel_flag(Op& op) noexcept
261		{
262	✗	if (&op == static_cast<void*>(&conn_))
263	✗	return &this->desc_state_.connect_cancel_pending;
264	✗	if (&op == static_cast<void*>(&rd_))
265	✗	return &this->desc_state_.read_cancel_pending;
266	✗	if (&op == static_cast<void*>(&wr_))
267	✗	return &this->desc_state_.write_cancel_pending;
268	✗	if (&op == static_cast<void*>(&recv_rd_))
269	✗	return &this->desc_state_.read_cancel_pending;
270	✗	if (&op == static_cast<void*>(&send_wr_))
271	✗	return &this->desc_state_.write_cancel_pending;
272	✗	return nullptr;
273		}
274
275		template<class Fn>
276	468x	void for_each_op(Fn fn) noexcept
277		{
278	468x	fn(conn_);
279	468x	fn(rd_);
280	468x	fn(wr_);
281	468x	fn(recv_rd_);
282	468x	fn(send_wr_);
283	468x	}
284
285		template<class Fn>
286	468x	void for_each_desc_entry(Fn fn) noexcept
287		{
288	468x	fn(conn_, this->desc_state_.connect_op);
289	468x	fn(rd_, this->desc_state_.read_op);
290	468x	fn(wr_, this->desc_state_.write_op);
291	468x	fn(recv_rd_, this->desc_state_.read_op);
292	468x	fn(send_wr_, this->desc_state_.write_op);
293	468x	}
294		};
295
296		// do_send_to
297
298		template<
299		class Derived,
300		class Service,
301		class ConnOp,
302		class SendToOp,
303		class RecvFromOp,
304		class SendOp,
305		class RecvOp,
306		class DescState,
307		class ImplBase,
308		class Endpoint>
309		std::coroutine_handle<>
310	28x	reactor_datagram_socket<
311		Derived,
312		Service,
313		ConnOp,
314		SendToOp,
315		RecvFromOp,
316		SendOp,
317		RecvOp,
318		DescState,
319		ImplBase,
320		Endpoint>::
321		do_send_to(
322		std::coroutine_handle<> h,
323		capy::executor_ref ex,
324		buffer_param param,
325		Endpoint const& dest,
326		int flags,
327		std::stop_token const& token,
328		std::error_code* ec,
329		std::size_t* bytes_out)
330		{
331	28x	auto& op = wr_;
332	28x	op.reset();
333
334	28x	capy::mutable_buffer bufs[SendToOp::max_buffers];
335	28x	op.iovec_count =
336	28x	static_cast<int>(param.copy_to(bufs, SendToOp::max_buffers));
337
338	54x	for (int i = 0; i < op.iovec_count; ++i)
339		{
340	26x	op.iovecs[i].iov_base = bufs[i].data();
341	26x	op.iovecs[i].iov_len = bufs[i].size();
342		}
343
344		// Set up destination address
345	28x	op.dest_len = to_sockaddr(dest, socket_family(this->fd_), op.dest_storage);
346	28x	op.fd = this->fd_;
347	28x	op.msg_flags = to_native_msg_flags(flags);
348
349		// Speculative sendmsg
350	28x	msghdr msg{};
351	28x	msg.msg_name = &op.dest_storage;
352	28x	msg.msg_namelen = op.dest_len;
353	28x	msg.msg_iov = op.iovecs;
354	28x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
355
356		#ifdef MSG_NOSIGNAL
357	28x	int send_flags = op.msg_flags \| MSG_NOSIGNAL;
358		#else
359		int send_flags = op.msg_flags;
360		#endif
361
362		ssize_t n;
363		do
364		{
365	28x	n = ::sendmsg(this->fd_, &msg, send_flags);
366		}
367	28x	while (n < 0 && errno == EINTR);
368
369	28x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
370		{
371	28x	int err = (n < 0) ? errno : 0;
372	28x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
373
374	28x	if (this->svc_.scheduler().try_consume_inline_budget())
375		{
376	6x	*ec = err ? make_err(err) : std::error_code{};
377	6x	*bytes_out = bytes;
378	6x	op.cont_op.cont.h = h;
379	6x	return dispatch_coro(ex, op.cont_op.cont);
380		}
381	22x	op.h = h;
382	22x	op.ex = ex;
383	22x	op.ec_out = ec;
384	22x	op.bytes_out = bytes_out;
385	22x	op.start(token, static_cast<Derived*>(this));
386	22x	op.impl_ptr = this->shared_from_this();
387	22x	op.complete(err, bytes);
388	22x	this->svc_.post(&op);
389	22x	return std::noop_coroutine();
390		}
391
392		// EAGAIN — register with reactor
393	✗	op.h = h;
394	✗	op.ex = ex;
395	✗	op.ec_out = ec;
396	✗	op.bytes_out = bytes_out;
397	✗	op.start(token, static_cast<Derived*>(this));
398	✗	op.impl_ptr = this->shared_from_this();
399
400	✗	this->register_op(
401	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
402	✗	this->desc_state_.write_cancel_pending);
403	✗	return std::noop_coroutine();
404		}
405
406		// do_recv_from
407
408		template<
409		class Derived,
410		class Service,
411		class ConnOp,
412		class SendToOp,
413		class RecvFromOp,
414		class SendOp,
415		class RecvOp,
416		class DescState,
417		class ImplBase,
418		class Endpoint>
419		std::coroutine_handle<>
420	40x	reactor_datagram_socket<
421		Derived,
422		Service,
423		ConnOp,
424		SendToOp,
425		RecvFromOp,
426		SendOp,
427		RecvOp,
428		DescState,
429		ImplBase,
430		Endpoint>::
431		do_recv_from(
432		std::coroutine_handle<> h,
433		capy::executor_ref ex,
434		buffer_param param,
435		Endpoint* source,
436		int flags,
437		std::stop_token const& token,
438		std::error_code* ec,
439		std::size_t* bytes_out)
440		{
441	40x	auto& op = rd_;
442	40x	op.reset();
443
444	40x	capy::mutable_buffer bufs[RecvFromOp::max_buffers];
445	40x	op.iovec_count =
446	40x	static_cast<int>(param.copy_to(bufs, RecvFromOp::max_buffers));
447
448	40x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
449		{
450	2x	op.h = h;
451	2x	op.ex = ex;
452	2x	op.ec_out = ec;
453	2x	op.bytes_out = bytes_out;
454	2x	op.start(token, static_cast<Derived*>(this));
455	2x	op.impl_ptr = this->shared_from_this();
456	2x	op.complete(0, 0);
457	2x	this->svc_.post(&op);
458	2x	return std::noop_coroutine();
459		}
460
461	76x	for (int i = 0; i < op.iovec_count; ++i)
462		{
463	38x	op.iovecs[i].iov_base = bufs[i].data();
464	38x	op.iovecs[i].iov_len = bufs[i].size();
465		}
466
467	38x	op.fd = this->fd_;
468	38x	op.source_out = source;
469	38x	op.msg_flags = to_native_msg_flags(flags);
470
471		// Speculative recvmsg
472	38x	msghdr msg{};
473	38x	msg.msg_name = &op.source_storage;
474	38x	msg.msg_namelen = sizeof(op.source_storage);
475	38x	msg.msg_iov = op.iovecs;
476	38x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
477
478		ssize_t n;
479		do
480		{
481	38x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
482		}
483	38x	while (n < 0 && errno == EINTR);
484
485	38x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
486		{
487	30x	int err = (n < 0) ? errno : 0;
488	30x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
489	30x	if (n >= 0)
490	30x	op.source_addrlen = msg.msg_namelen;
491
492	30x	if (this->svc_.scheduler().try_consume_inline_budget())
493		{
494	22x	*ec = err ? make_err(err) : std::error_code{};
495	22x	*bytes_out = bytes;
496	22x	if (source && !err && n >= 0)
497	44x	*source = from_sockaddr_as(
498	22x	op.source_storage,
499		op.source_addrlen,
500		Endpoint{});
501	22x	op.cont_op.cont.h = h;
502	22x	return dispatch_coro(ex, op.cont_op.cont);
503		}
504	8x	op.h = h;
505	8x	op.ex = ex;
506	8x	op.ec_out = ec;
507	8x	op.bytes_out = bytes_out;
508	8x	op.start(token, static_cast<Derived*>(this));
509	8x	op.impl_ptr = this->shared_from_this();
510	8x	op.complete(err, bytes);
511	8x	this->svc_.post(&op);
512	8x	return std::noop_coroutine();
513		}
514
515		// EAGAIN — register with reactor
516	8x	op.h = h;
517	8x	op.ex = ex;
518	8x	op.ec_out = ec;
519	8x	op.bytes_out = bytes_out;
520	8x	op.start(token, static_cast<Derived*>(this));
521	8x	op.impl_ptr = this->shared_from_this();
522
523	8x	this->register_op(
524	8x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
525	8x	this->desc_state_.read_cancel_pending);
526	8x	return std::noop_coroutine();
527		}
528
529		// do_connect
530
531		template<
532		class Derived,
533		class Service,
534		class ConnOp,
535		class SendToOp,
536		class RecvFromOp,
537		class SendOp,
538		class RecvOp,
539		class DescState,
540		class ImplBase,
541		class Endpoint>
542		std::coroutine_handle<>
543	12x	reactor_datagram_socket<
544		Derived,
545		Service,
546		ConnOp,
547		SendToOp,
548		RecvFromOp,
549		SendOp,
550		RecvOp,
551		DescState,
552		ImplBase,
553		Endpoint>::
554		do_connect(
555		std::coroutine_handle<> h,
556		capy::executor_ref ex,
557		Endpoint const& ep,
558		std::stop_token const& token,
559		std::error_code* ec)
560		{
561	12x	auto& op = conn_;
562
563	12x	sockaddr_storage storage{};
564	12x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
565		int result =
566	12x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
567
568	12x	if (result == 0)
569		{
570	12x	sockaddr_storage local_storage{};
571	12x	socklen_t local_len = sizeof(local_storage);
572	12x	if (::getsockname(
573		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
574	12x	&local_len) == 0)
575	12x	this->local_endpoint_ =
576	12x	from_sockaddr_as(local_storage, local_len, Endpoint{});
577	12x	remote_endpoint_ = ep;
578		}
579
580	12x	if (result == 0 \|\| errno != EINPROGRESS)
581		{
582	12x	int err = (result < 0) ? errno : 0;
583	12x	if (this->svc_.scheduler().try_consume_inline_budget())
584		{
585	2x	*ec = err ? make_err(err) : std::error_code{};
586	2x	op.cont_op.cont.h = h;
587	2x	return dispatch_coro(ex, op.cont_op.cont);
588		}
589	10x	op.reset();
590	10x	op.h = h;
591	10x	op.ex = ex;
592	10x	op.ec_out = ec;
593	10x	op.fd = this->fd_;
594	10x	op.target_endpoint = ep;
595	10x	op.start(token, static_cast<Derived*>(this));
596	10x	op.impl_ptr = this->shared_from_this();
597	10x	op.complete(err, 0);
598	10x	this->svc_.post(&op);
599	10x	return std::noop_coroutine();
600		}
601
602		// EINPROGRESS — register with reactor
603	✗	op.reset();
604	✗	op.h = h;
605	✗	op.ex = ex;
606	✗	op.ec_out = ec;
607	✗	op.fd = this->fd_;
608	✗	op.target_endpoint = ep;
609	✗	op.start(token, static_cast<Derived*>(this));
610	✗	op.impl_ptr = this->shared_from_this();
611
612	✗	this->register_op(
613	✗	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
614	✗	this->desc_state_.connect_cancel_pending);
615	✗	return std::noop_coroutine();
616		}
617
618		// do_send (connected mode)
619
620		template<
621		class Derived,
622		class Service,
623		class ConnOp,
624		class SendToOp,
625		class RecvFromOp,
626		class SendOp,
627		class RecvOp,
628		class DescState,
629		class ImplBase,
630		class Endpoint>
631		std::coroutine_handle<>
632	14x	reactor_datagram_socket<
633		Derived,
634		Service,
635		ConnOp,
636		SendToOp,
637		RecvFromOp,
638		SendOp,
639		RecvOp,
640		DescState,
641		ImplBase,
642		Endpoint>::
643		do_send(
644		std::coroutine_handle<> h,
645		capy::executor_ref ex,
646		buffer_param param,
647		int flags,
648		std::stop_token const& token,
649		std::error_code* ec,
650		std::size_t* bytes_out)
651		{
652	14x	auto& op = send_wr_;
653	14x	op.reset();
654
655	14x	capy::mutable_buffer bufs[SendOp::max_buffers];
656	14x	op.iovec_count = static_cast<int>(param.copy_to(bufs, SendOp::max_buffers));
657
658	28x	for (int i = 0; i < op.iovec_count; ++i)
659		{
660	14x	op.iovecs[i].iov_base = bufs[i].data();
661	14x	op.iovecs[i].iov_len = bufs[i].size();
662		}
663
664	14x	op.fd = this->fd_;
665	14x	op.msg_flags = to_native_msg_flags(flags);
666
667		// Speculative sendmsg with no destination (connected mode)
668	14x	msghdr msg{};
669	14x	msg.msg_iov = op.iovecs;
670	14x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
671
672		#ifdef MSG_NOSIGNAL
673	14x	int send_flags = op.msg_flags \| MSG_NOSIGNAL;
674		#else
675		int send_flags = op.msg_flags;
676		#endif
677
678		ssize_t n;
679		do
680		{
681	14x	n = ::sendmsg(this->fd_, &msg, send_flags);
682		}
683	14x	while (n < 0 && errno == EINTR);
684
685	14x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
686		{
687	14x	int err = (n < 0) ? errno : 0;
688	14x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
689
690	14x	if (this->svc_.scheduler().try_consume_inline_budget())
691		{
692	8x	*ec = err ? make_err(err) : std::error_code{};
693	8x	*bytes_out = bytes;
694	8x	op.cont_op.cont.h = h;
695	8x	return dispatch_coro(ex, op.cont_op.cont);
696		}
697	6x	op.h = h;
698	6x	op.ex = ex;
699	6x	op.ec_out = ec;
700	6x	op.bytes_out = bytes_out;
701	6x	op.start(token, static_cast<Derived*>(this));
702	6x	op.impl_ptr = this->shared_from_this();
703	6x	op.complete(err, bytes);
704	6x	this->svc_.post(&op);
705	6x	return std::noop_coroutine();
706		}
707
708		// EAGAIN — register with reactor
709	✗	op.h = h;
710	✗	op.ex = ex;
711	✗	op.ec_out = ec;
712	✗	op.bytes_out = bytes_out;
713	✗	op.start(token, static_cast<Derived*>(this));
714	✗	op.impl_ptr = this->shared_from_this();
715
716	✗	this->register_op(
717	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
718	✗	this->desc_state_.write_cancel_pending);
719	✗	return std::noop_coroutine();
720		}
721
722		// do_recv (connected mode)
723
724		template<
725		class Derived,
726		class Service,
727		class ConnOp,
728		class SendToOp,
729		class RecvFromOp,
730		class SendOp,
731		class RecvOp,
732		class DescState,
733		class ImplBase,
734		class Endpoint>
735		std::coroutine_handle<>
736	14x	reactor_datagram_socket<
737		Derived,
738		Service,
739		ConnOp,
740		SendToOp,
741		RecvFromOp,
742		SendOp,
743		RecvOp,
744		DescState,
745		ImplBase,
746		Endpoint>::
747		do_recv(
748		std::coroutine_handle<> h,
749		capy::executor_ref ex,
750		buffer_param param,
751		int flags,
752		std::stop_token const& token,
753		std::error_code* ec,
754		std::size_t* bytes_out)
755		{
756	14x	auto& op = recv_rd_;
757	14x	op.reset();
758
759	14x	capy::mutable_buffer bufs[RecvOp::max_buffers];
760	14x	op.iovec_count = static_cast<int>(param.copy_to(bufs, RecvOp::max_buffers));
761
762	14x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
763		{
764	✗	op.h = h;
765	✗	op.ex = ex;
766	✗	op.ec_out = ec;
767	✗	op.bytes_out = bytes_out;
768	✗	op.start(token, static_cast<Derived*>(this));
769	✗	op.impl_ptr = this->shared_from_this();
770	✗	op.complete(0, 0);
771	✗	this->svc_.post(&op);
772	✗	return std::noop_coroutine();
773		}
774
775	28x	for (int i = 0; i < op.iovec_count; ++i)
776		{
777	14x	op.iovecs[i].iov_base = bufs[i].data();
778	14x	op.iovecs[i].iov_len = bufs[i].size();
779		}
780
781	14x	op.fd = this->fd_;
782	14x	op.msg_flags = to_native_msg_flags(flags);
783
784		// Speculative recvmsg with no source (connected mode)
785	14x	msghdr msg{};
786	14x	msg.msg_iov = op.iovecs;
787	14x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
788
789		ssize_t n;
790		do
791		{
792	14x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
793		}
794	14x	while (n < 0 && errno == EINTR);
795
796	14x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
797		{
798	12x	int err = (n < 0) ? errno : 0;
799	12x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
800
801	12x	if (this->svc_.scheduler().try_consume_inline_budget())
802		{
803	8x	*ec = err ? make_err(err) : std::error_code{};
804	8x	*bytes_out = bytes;
805	8x	op.cont_op.cont.h = h;
806	8x	return dispatch_coro(ex, op.cont_op.cont);
807		}
808	4x	op.h = h;
809	4x	op.ex = ex;
810	4x	op.ec_out = ec;
811	4x	op.bytes_out = bytes_out;
812	4x	op.start(token, static_cast<Derived*>(this));
813	4x	op.impl_ptr = this->shared_from_this();
814	4x	op.complete(err, bytes);
815	4x	this->svc_.post(&op);
816	4x	return std::noop_coroutine();
817		}
818
819		// EAGAIN — register with reactor
820	2x	op.h = h;
821	2x	op.ex = ex;
822	2x	op.ec_out = ec;
823	2x	op.bytes_out = bytes_out;
824	2x	op.start(token, static_cast<Derived*>(this));
825	2x	op.impl_ptr = this->shared_from_this();
826
827	2x	this->register_op(
828	2x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
829	2x	this->desc_state_.read_cancel_pending);
830	2x	return std::noop_coroutine();
831		}
832
833		} // namespace boost::corosio::detail
834
835		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
836