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

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