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

83.2% Lines (331/398) 94.3% List of functions (99/105)

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

reactor_datagram_socket.hpp

f(x) Functions (105)

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