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

62.4% Lines (759/1216) 73.2% List of functions (71/97)

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

reactor_datagram_socket.hpp

f(x) Functions (97)

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