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

71.4% Lines (284/398) 73.3% List of functions (77/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	96x	to_native_msg_flags(int flags) noexcept
32		{
33	96x	int native = 0;
34	96x	if (flags & 1) native \|= MSG_PEEK;
35	96x	if (flags & 2) native \|= MSG_OOB;
36	96x	if (flags & 4) native \|= MSG_DONTROUTE;
37	96x	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	126x	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	126x	~reactor_datagram_socket() override = default;
126
127		/// Return the cached remote endpoint.
128	4x	Endpoint remote_endpoint() const noexcept override
129		{
130	4x	return remote_endpoint_;
131		}
132
133		// --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
134
135	30x	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	30x	return do_send_to(h, ex, buf, dest, flags, token, ec, bytes_out);
146		}
147
148	40x	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	40x	return do_recv_from(h, ex, buf, source, flags, token, ec, bytes_out);
159		}
160
161	12x	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	12x	return do_connect(h, ex, ep, token, ec);
169		}
170
171	14x	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	14x	return do_send(h, ex, buf, flags, token, ec, bytes_out);
181		}
182
183	14x	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	14x	return do_recv(h, ex, buf, flags, token, ec, bytes_out);
193		}
194
195	6x	void cancel() noexcept override
196		{
197	6x	this->do_cancel();
198	6x	}
199
200	4x	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	4x	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	488x	void close_socket() noexcept
214		{
215	488x	do_close_socket();
216	488x	}
217
218		/// Cache local and remote endpoints.
219	22x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
220		{
221	22x	this->local_endpoint_ = std::move(local);
222	22x	remote_endpoint_ = std::move(remote);
223	22x	}
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	488x	void do_close_socket() noexcept
317		{
318	488x	base_type::do_close_socket();
319	488x	remote_endpoint_ = Endpoint{};
320	488x	}
321
322	✗	native_handle_type do_release_socket() noexcept
323		{
324	✗	auto fd = base_type::do_release_socket();
325	✗	remote_endpoint_ = Endpoint{};
326	✗	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	✗	std::error_code do_shutdown(int what) noexcept
336		{
337		int how;
338	✗	switch (what)
339		{
340	✗	case 0:
341	✗	how = SHUT_RD;
342	✗	break;
343	✗	case 1:
344	✗	how = SHUT_WR;
345	✗	break;
346	✗	case 2:
347	✗	how = SHUT_RDWR;
348	✗	break;
349	✗	default:
350	✗	return make_err(EINVAL);
351		}
352	✗	if (::shutdown(this->fd_, how) != 0)
353	✗	return make_err(errno);
354	✗	return {};
355		}
356
357		private:
358		// CRTP callbacks for reactor_basic_socket cancel/close
359
360		template<class Op>
361	2x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
362		{
363	2x	if (&op == static_cast<void*>(&conn_))
364	✗	return &this->desc_state_.connect_op;
365	2x	if (&op == static_cast<void*>(&rd_))
366	2x	return &this->desc_state_.read_op;
367	✗	if (&op == static_cast<void*>(&wr_))
368	✗	return &this->desc_state_.write_op;
369	✗	if (&op == static_cast<void*>(&recv_rd_))
370	✗	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	494x	void for_each_op(Fn fn) noexcept
406		{
407	494x	fn(conn_);
408	494x	fn(rd_);
409	494x	fn(wr_);
410	494x	fn(recv_rd_);
411	494x	fn(send_wr_);
412	494x	fn(wait_rd_);
413	494x	fn(wait_wr_);
414	494x	fn(wait_er_);
415	494x	}
416
417		template<class Fn>
418	494x	void for_each_desc_entry(Fn fn) noexcept
419		{
420	494x	fn(conn_, this->desc_state_.connect_op);
421	494x	fn(rd_, this->desc_state_.read_op);
422	494x	fn(wr_, this->desc_state_.write_op);
423	494x	fn(recv_rd_, this->desc_state_.read_op);
424	494x	fn(send_wr_, this->desc_state_.write_op);
425	494x	fn(wait_rd_, this->desc_state_.wait_read_op);
426	494x	fn(wait_wr_, this->desc_state_.wait_write_op);
427	494x	fn(wait_er_, this->desc_state_.wait_error_op);
428	494x	}
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	30x	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	30x	auto& op = wr_;
469	30x	op.reset();
470
471	30x	capy::mutable_buffer bufs[SendToOp::max_buffers];
472	30x	op.iovec_count =
473	30x	static_cast<int>(param.copy_to(bufs, SendToOp::max_buffers));
474
475	58x	for (int i = 0; i < op.iovec_count; ++i)
476		{
477	28x	op.iovecs[i].iov_base = bufs[i].data();
478	28x	op.iovecs[i].iov_len = bufs[i].size();
479		}
480
481		// Set up destination address
482	30x	op.dest_len = to_sockaddr(dest, socket_family(this->fd_), op.dest_storage);
483	30x	op.fd = this->fd_;
484	30x	op.msg_flags = to_native_msg_flags(flags);
485
486		// Speculative sendmsg
487	30x	msghdr msg{};
488	30x	msg.msg_name = &op.dest_storage;
489	30x	msg.msg_namelen = op.dest_len;
490	30x	msg.msg_iov = op.iovecs;
491	30x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
492
493		#ifdef MSG_NOSIGNAL
494	30x	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	30x	n = ::sendmsg(this->fd_, &msg, send_flags);
503		}
504	30x	while (n < 0 && errno == EINTR);
505
506	30x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
507		{
508	30x	int err = (n < 0) ? errno : 0;
509	30x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
510
511	30x	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	24x	op.h = h;
519	24x	op.ex = ex;
520	24x	op.ec_out = ec;
521	24x	op.bytes_out = bytes_out;
522	24x	op.start(token, static_cast<Derived*>(this));
523	24x	op.impl_ptr = this->shared_from_this();
524	24x	op.complete(err, bytes);
525	24x	this->svc_.post(&op);
526	24x	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	40x	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	40x	auto& op = rd_;
581	40x	op.reset();
582
583	40x	capy::mutable_buffer bufs[RecvFromOp::max_buffers];
584	40x	op.iovec_count =
585	40x	static_cast<int>(param.copy_to(bufs, RecvFromOp::max_buffers));
586
587	40x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
588		{
589	2x	op.h = h;
590	2x	op.ex = ex;
591	2x	op.ec_out = ec;
592	2x	op.bytes_out = bytes_out;
593	2x	op.start(token, static_cast<Derived*>(this));
594	2x	op.impl_ptr = this->shared_from_this();
595	2x	op.complete(0, 0);
596	2x	this->svc_.post(&op);
597	2x	return std::noop_coroutine();
598		}
599
600	76x	for (int i = 0; i < op.iovec_count; ++i)
601		{
602	38x	op.iovecs[i].iov_base = bufs[i].data();
603	38x	op.iovecs[i].iov_len = bufs[i].size();
604		}
605
606	38x	op.fd = this->fd_;
607	38x	op.source_out = source;
608	38x	op.msg_flags = to_native_msg_flags(flags);
609
610		// Speculative recvmsg
611	38x	msghdr msg{};
612	38x	msg.msg_name = &op.source_storage;
613	38x	msg.msg_namelen = sizeof(op.source_storage);
614	38x	msg.msg_iov = op.iovecs;
615	38x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
616
617		ssize_t n;
618		do
619		{
620	38x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
621		}
622	38x	while (n < 0 && errno == EINTR);
623
624	38x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
625		{
626	30x	int err = (n < 0) ? errno : 0;
627	30x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
628	30x	if (n >= 0)
629	30x	op.source_addrlen = msg.msg_namelen;
630
631	30x	if (this->svc_.scheduler().try_consume_inline_budget())
632		{
633	22x	*ec = err ? make_err(err) : std::error_code{};
634	22x	*bytes_out = bytes;
635	22x	if (source && !err && n >= 0)
636	44x	*source = from_sockaddr_as(
637	22x	op.source_storage,
638		op.source_addrlen,
639		Endpoint{});
640	22x	op.cont_op.cont.h = h;
641	22x	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	8x	op.h = h;
656	8x	op.ex = ex;
657	8x	op.ec_out = ec;
658	8x	op.bytes_out = bytes_out;
659	8x	op.start(token, static_cast<Derived*>(this));
660	8x	op.impl_ptr = this->shared_from_this();
661
662	8x	this->register_op(
663	8x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
664	8x	this->desc_state_.read_cancel_pending);
665	8x	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	12x	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	12x	auto& op = conn_;
703
704	12x	sockaddr_storage storage{};
705	12x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
706		int result =
707	12x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
708
709	12x	if (result == 0)
710		{
711	12x	sockaddr_storage local_storage{};
712	12x	socklen_t local_len = sizeof(local_storage);
713	12x	if (::getsockname(
714		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
715	12x	&local_len) == 0)
716	12x	this->local_endpoint_ =
717	12x	from_sockaddr_as(local_storage, local_len, Endpoint{});
718	12x	remote_endpoint_ = ep;
719		}
720
721	12x	if (result == 0 \|\| errno != EINPROGRESS)
722		{
723	12x	int err = (result < 0) ? errno : 0;
724	12x	if (this->svc_.scheduler().try_consume_inline_budget())
725		{
726	2x	*ec = err ? make_err(err) : std::error_code{};
727	2x	op.cont_op.cont.h = h;
728	2x	return dispatch_coro(ex, op.cont_op.cont);
729		}
730	10x	op.reset();
731	10x	op.h = h;
732	10x	op.ex = ex;
733	10x	op.ec_out = ec;
734	10x	op.fd = this->fd_;
735	10x	op.target_endpoint = ep;
736	10x	op.start(token, static_cast<Derived*>(this));
737	10x	op.impl_ptr = this->shared_from_this();
738	10x	op.complete(err, 0);
739	10x	this->svc_.post(&op);
740	10x	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	14x	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	14x	auto& op = send_wr_;
796	14x	op.reset();
797
798	14x	capy::mutable_buffer bufs[SendOp::max_buffers];
799	14x	op.iovec_count = static_cast<int>(param.copy_to(bufs, SendOp::max_buffers));
800
801	28x	for (int i = 0; i < op.iovec_count; ++i)
802		{
803	14x	op.iovecs[i].iov_base = bufs[i].data();
804	14x	op.iovecs[i].iov_len = bufs[i].size();
805		}
806
807	14x	op.fd = this->fd_;
808	14x	op.msg_flags = to_native_msg_flags(flags);
809
810		// Speculative sendmsg with no destination (connected mode)
811	14x	msghdr msg{};
812	14x	msg.msg_iov = op.iovecs;
813	14x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
814
815		#ifdef MSG_NOSIGNAL
816	14x	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	14x	n = ::sendmsg(this->fd_, &msg, send_flags);
825		}
826	14x	while (n < 0 && errno == EINTR);
827
828	14x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
829		{
830	14x	int err = (n < 0) ? errno : 0;
831	14x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
832
833	14x	if (this->svc_.scheduler().try_consume_inline_budget())
834		{
835	8x	*ec = err ? make_err(err) : std::error_code{};
836	8x	*bytes_out = bytes;
837	8x	op.cont_op.cont.h = h;
838	8x	return dispatch_coro(ex, op.cont_op.cont);
839		}
840	6x	op.h = h;
841	6x	op.ex = ex;
842	6x	op.ec_out = ec;
843	6x	op.bytes_out = bytes_out;
844	6x	op.start(token, static_cast<Derived*>(this));
845	6x	op.impl_ptr = this->shared_from_this();
846	6x	op.complete(err, bytes);
847	6x	this->svc_.post(&op);
848	6x	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	14x	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	14x	auto& op = recv_rd_;
902	14x	op.reset();
903
904	14x	capy::mutable_buffer bufs[RecvOp::max_buffers];
905	14x	op.iovec_count = static_cast<int>(param.copy_to(bufs, RecvOp::max_buffers));
906
907	14x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
908		{
909	✗	op.h = h;
910	✗	op.ex = ex;
911	✗	op.ec_out = ec;
912	✗	op.bytes_out = bytes_out;
913	✗	op.start(token, static_cast<Derived*>(this));
914	✗	op.impl_ptr = this->shared_from_this();
915	✗	op.complete(0, 0);
916	✗	this->svc_.post(&op);
917	✗	return std::noop_coroutine();
918		}
919
920	28x	for (int i = 0; i < op.iovec_count; ++i)
921		{
922	14x	op.iovecs[i].iov_base = bufs[i].data();
923	14x	op.iovecs[i].iov_len = bufs[i].size();
924		}
925
926	14x	op.fd = this->fd_;
927	14x	op.msg_flags = to_native_msg_flags(flags);
928
929		// Speculative recvmsg with no source (connected mode)
930	14x	msghdr msg{};
931	14x	msg.msg_iov = op.iovecs;
932	14x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
933
934		ssize_t n;
935		do
936		{
937	14x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
938		}
939	14x	while (n < 0 && errno == EINTR);
940
941	14x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
942		{
943	12x	int err = (n < 0) ? errno : 0;
944	12x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
945
946	12x	if (this->svc_.scheduler().try_consume_inline_budget())
947		{
948	8x	*ec = err ? make_err(err) : std::error_code{};
949	8x	*bytes_out = bytes;
950	8x	op.cont_op.cont.h = h;
951	8x	return dispatch_coro(ex, op.cont_op.cont);
952		}
953	4x	op.h = h;
954	4x	op.ex = ex;
955	4x	op.ec_out = ec;
956	4x	op.bytes_out = bytes_out;
957	4x	op.start(token, static_cast<Derived*>(this));
958	4x	op.impl_ptr = this->shared_from_this();
959	4x	op.complete(err, bytes);
960	4x	this->svc_.post(&op);
961	4x	return std::noop_coroutine();
962		}
963
964		// EAGAIN — register with reactor
965	2x	op.h = h;
966	2x	op.ex = ex;
967	2x	op.ec_out = ec;
968	2x	op.bytes_out = bytes_out;
969	2x	op.start(token, static_cast<Derived*>(this));
970	2x	op.impl_ptr = this->shared_from_this();
971
972	2x	this->register_op(
973	2x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
974	2x	this->desc_state_.read_cancel_pending);
975	2x	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	4x	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	4x	if (w == wait_type::write)
1014		{
1015	✗	auto& op = wait_wr_;
1016	✗	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	✗	op.reset();
1023	✗	op.wait_event = reactor_event_write;
1024	✗	op.h = h;
1025	✗	op.ex = ex;
1026	✗	op.ec_out = ec;
1027	✗	op.fd = this->fd_;
1028	✗	op.start(token, static_cast<Derived*>(this));
1029	✗	op.impl_ptr = this->shared_from_this();
1030	✗	op.complete(0, 0);
1031	✗	this->svc_.post(&op);
1032	✗	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	4x	bool dummy_ready = false; // no cached edge for error waits
1042
1043	4x	if (w == wait_type::read)
1044		{
1045	4x	op_ptr = &wait_rd_;
1046	4x	desc_slot_ptr = &this->desc_state_.wait_read_op;
1047	4x	ready_flag_ptr = &this->desc_state_.read_ready;
1048	4x	cancel_flag_ptr = &this->desc_state_.wait_read_cancel_pending;
1049	4x	event = reactor_event_read;
1050		}
1051		else // wait_type::error
1052		{
1053	✗	op_ptr = &wait_er_;
1054	✗	desc_slot_ptr = &this->desc_state_.wait_error_op;
1055	✗	ready_flag_ptr = &dummy_ready;
1056	✗	cancel_flag_ptr = &this->desc_state_.wait_error_cancel_pending;
1057	✗	event = reactor_event_error;
1058		}
1059
1060	4x	auto& op = *op_ptr;
1061	4x	op.reset();
1062	4x	op.wait_event = event;
1063	4x	op.h = h;
1064	4x	op.ex = ex;
1065	4x	op.ec_out = ec;
1066	4x	op.fd = this->fd_;
1067	4x	op.start(token, static_cast<Derived*>(this));
1068	4x	op.impl_ptr = this->shared_from_this();
1069
1070	4x	this->register_op(
1071		op, desc_slot_ptr, ready_flag_ptr, *cancel_flag_ptr,
1072		false);
1073	4x	return std::noop_coroutine();
1074		}
1075
1076		} // namespace boost::corosio::detail
1077
1078		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
1079