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

82.9% Lines (325/392) 94.2% List of functions (98/104)

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

reactor_datagram_socket.hpp

f(x) Functions (104)

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