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

95.8% Lines (361/0/377) 100.0% List of functions (108/0/108)

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

reactor_datagram_socket.hpp

f(x) Functions (108)

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	522x	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	522x	~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	157x	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	157x	return do_send_to(h, ex, buf, dest, flags, token, ec, bytes_out);
136		}
137
138	177x	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	177x	return do_recv_from(h, ex, buf, source, flags, token, ec, bytes_out);
149		}
150
151	44x	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	44x	return do_connect(h, ex, ep, token, ec);
159		}
160
161	115x	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	115x	return do_send(h, ex, buf, flags, token, ec, bytes_out);
171		}
172
173	115x	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	115x	return do_recv(h, ex, buf, flags, token, ec, bytes_out);
183		}
184
185	17x	void cancel() noexcept override
186		{
187	17x	this->do_cancel();
188	17x	}
189
190	40x	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	40x	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	1952x	void close_socket() noexcept
204		{
205	1952x	do_close_socket();
206	1952x	}
207
208		/// Cache local and remote endpoints.
209	160x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
210		{
211	160x	this->local_endpoint_ = std::move(local);
212	160x	remote_endpoint_ = std::move(remote);
213	160x	}
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		Every wait type probes the descriptor with a zero-timeout
291		`poll()` and completes at once if the condition already
292		holds; otherwise the op re-probes under the descriptor mutex
293		and parks, completing when a reactor event arrives and a
294		fresh probe confirms the condition. A write wait therefore
295		completes only while a non-blocking write can make progress.
296		*/
297		std::coroutine_handle<> do_wait(
298		std::coroutine_handle<>,
299		capy::executor_ref,
300		wait_type,
301		std::stop_token const&,
302		std::error_code*);
303
304		/** Close the socket and cancel pending operations.
305
306		Extends the base do_close_socket() to also reset
307		the remote endpoint.
308		*/
309	1952x	void do_close_socket() noexcept
310		{
311	1952x	base_type::do_close_socket();
312	1952x	remote_endpoint_ = Endpoint{};
313	1952x	}
314
315	4x	native_handle_type do_release_socket() noexcept
316		{
317	4x	auto fd = base_type::do_release_socket();
318	4x	remote_endpoint_ = Endpoint{};
319	4x	return fd;
320		}
321
322		/** Shut down part or all of the full-duplex connection.
323
324		Not an override — concrete backends forward here.
325
326		@param what 0 = receive, 1 = send, 2 = both.
327		*/
328	12x	std::error_code do_shutdown(int what) noexcept
329		{
330		int how;
331	12x	switch (what)
332		{
333	2x	case 0:
334	2x	how = SHUT_RD;
335	2x	break;
336	6x	case 1:
337	6x	how = SHUT_WR;
338	6x	break;
339	4x	case 2:
340	4x	how = SHUT_RDWR;
341	4x	break;
342	✗	default:
343	✗	return make_err(EINVAL);
344		}
345	12x	if (::shutdown(this->fd_, how) != 0)
346	4x	return make_err(errno);
347	8x	return {};
348		}
349
350		private:
351		// CRTP callbacks for reactor_basic_socket cancel/close
352
353		template<class Op>
354	46x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
355		{
356	46x	if (&op == static_cast<void*>(&conn_))
357	2x	return &this->desc_state_.connect_op;
358	44x	if (&op == static_cast<void*>(&rd_))
359	12x	return &this->desc_state_.read_op;
360	32x	if (&op == static_cast<void*>(&wr_))
361	6x	return &this->desc_state_.write_op;
362	26x	if (&op == static_cast<void*>(&recv_rd_))
363	8x	return &this->desc_state_.read_op;
364	18x	if (&op == static_cast<void*>(&send_wr_))
365	6x	return &this->desc_state_.write_op;
366	12x	if (&op == static_cast<void*>(&wait_rd_))
367	8x	return &this->desc_state_.wait_read_op;
368	4x	if (&op == static_cast<void*>(&wait_wr_))
369	✗	return &this->desc_state_.wait_write_op;
370	4x	if (&op == static_cast<void*>(&wait_er_))
371	4x	return &this->desc_state_.wait_error_op;
372	✗	return nullptr;
373		}
374
375		template<class Op>
376		bool* op_to_cancel_flag(Op& op) noexcept
377		{
378		if (&op == static_cast<void*>(&conn_))
379		return &this->desc_state_.connect_cancel_pending;
380		if (&op == static_cast<void*>(&rd_))
381		return &this->desc_state_.read_cancel_pending;
382		if (&op == static_cast<void*>(&wr_))
383		return &this->desc_state_.write_cancel_pending;
384		if (&op == static_cast<void*>(&recv_rd_))
385		return &this->desc_state_.read_cancel_pending;
386		if (&op == static_cast<void*>(&send_wr_))
387		return &this->desc_state_.write_cancel_pending;
388		if (&op == static_cast<void*>(&wait_rd_))
389		return &this->desc_state_.wait_read_cancel_pending;
390		if (&op == static_cast<void*>(&wait_wr_))
391		return &this->desc_state_.wait_write_cancel_pending;
392		if (&op == static_cast<void*>(&wait_er_))
393		return &this->desc_state_.wait_error_cancel_pending;
394		return nullptr;
395		}
396
397		template<class Fn>
398	1973x	void for_each_op(Fn fn) noexcept
399		{
400	1973x	fn(conn_);
401	1973x	fn(rd_);
402	1973x	fn(wr_);
403	1973x	fn(recv_rd_);
404	1973x	fn(send_wr_);
405	1973x	fn(wait_rd_);
406	1973x	fn(wait_wr_);
407	1973x	fn(wait_er_);
408	1973x	}
409
410		template<class Fn>
411	1973x	void for_each_desc_entry(Fn fn) noexcept
412		{
413	1973x	fn(conn_, this->desc_state_.connect_op);
414	1973x	fn(rd_, this->desc_state_.read_op);
415	1973x	fn(wr_, this->desc_state_.write_op);
416	1973x	fn(recv_rd_, this->desc_state_.read_op);
417	1973x	fn(send_wr_, this->desc_state_.write_op);
418	1973x	fn(wait_rd_, this->desc_state_.wait_read_op);
419	1973x	fn(wait_wr_, this->desc_state_.wait_write_op);
420	1973x	fn(wait_er_, this->desc_state_.wait_error_op);
421	1973x	}
422		};
423
424		// do_send_to
425
426		template<
427		class Derived,
428		class Service,
429		class ConnOp,
430		class SendToOp,
431		class RecvFromOp,
432		class SendOp,
433		class RecvOp,
434		class WaitOp,
435		class DescState,
436		class ImplBase,
437		class Endpoint>
438		std::coroutine_handle<>
439	157x	reactor_datagram_socket<
440		Derived,
441		Service,
442		ConnOp,
443		SendToOp,
444		RecvFromOp,
445		SendOp,
446		RecvOp,
447		WaitOp,
448		DescState,
449		ImplBase,
450		Endpoint>::
451		do_send_to(
452		std::coroutine_handle<> h,
453		capy::executor_ref ex,
454		buffer_param param,
455		Endpoint const& dest,
456		int flags,
457		std::stop_token const& token,
458		std::error_code* ec,
459		std::size_t* bytes_out)
460		{
461	157x	auto& op = wr_;
462	157x	op.reset();
463
464	157x	capy::mutable_buffer bufs[SendToOp::max_buffers];
465	157x	op.iovec_count =
466	157x	static_cast<int>(param.copy_to(bufs, SendToOp::max_buffers));
467
468	308x	for (int i = 0; i < op.iovec_count; ++i)
469		{
470	151x	op.iovecs[i].iov_base = bufs[i].data();
471	151x	op.iovecs[i].iov_len = bufs[i].size();
472		}
473
474		// Set up destination address
475	157x	op.dest_len = to_sockaddr(dest, socket_family(this->fd_), op.dest_storage);
476	157x	op.fd = this->fd_;
477	157x	op.msg_flags = to_native_msg_flags(flags);
478
479		// Speculative sendmsg
480	157x	msghdr msg{};
481	157x	msg.msg_name = &op.dest_storage;
482	157x	msg.msg_namelen = op.dest_len;
483	157x	msg.msg_iov = op.iovecs;
484	157x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
485
486		#ifdef MSG_NOSIGNAL
487	157x	int send_flags = op.msg_flags \| MSG_NOSIGNAL;
488		#else
489		int send_flags = op.msg_flags;
490		#endif
491
492		ssize_t n;
493		do
494		{
495	157x	n = ::sendmsg(this->fd_, &msg, send_flags);
496		}
497	157x	while (n < 0 && errno == EINTR);
498
499	157x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
500		{
501	123x	int err = (n < 0) ? errno : 0;
502	123x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
503
504	123x	if (this->svc_.scheduler().try_consume_inline_budget())
505		{
506	49x	*ec = err ? make_err(err) : std::error_code{};
507	49x	*bytes_out = bytes;
508	49x	op.cont.h = h;
509	49x	return dispatch_coro(ex, op.cont);
510		}
511	74x	op.h = h;
512	74x	op.ex = ex;
513	74x	op.ec_out = ec;
514	74x	op.bytes_out = bytes_out;
515	74x	op.start(token, static_cast<Derived*>(this));
516	74x	op.impl_ptr = this->shared_from_this();
517	74x	op.complete(err, bytes);
518	74x	this->svc_.post(&op);
519	74x	return std::noop_coroutine();
520		}
521
522		// EAGAIN — register with reactor
523	34x	op.h = h;
524	34x	op.ex = ex;
525	34x	op.ec_out = ec;
526	34x	op.bytes_out = bytes_out;
527	34x	op.start(token, static_cast<Derived*>(this));
528	34x	op.impl_ptr = this->shared_from_this();
529
530	34x	this->register_op(
531	34x	op, this->desc_state_.write_op, this->desc_state_.write_ready,
532	34x	this->desc_state_.write_cancel_pending, true);
533	34x	return std::noop_coroutine();
534		}
535
536		// do_recv_from
537
538		template<
539		class Derived,
540		class Service,
541		class ConnOp,
542		class SendToOp,
543		class RecvFromOp,
544		class SendOp,
545		class RecvOp,
546		class WaitOp,
547		class DescState,
548		class ImplBase,
549		class Endpoint>
550		std::coroutine_handle<>
551	177x	reactor_datagram_socket<
552		Derived,
553		Service,
554		ConnOp,
555		SendToOp,
556		RecvFromOp,
557		SendOp,
558		RecvOp,
559		WaitOp,
560		DescState,
561		ImplBase,
562		Endpoint>::
563		do_recv_from(
564		std::coroutine_handle<> h,
565		capy::executor_ref ex,
566		buffer_param param,
567		Endpoint* source,
568		int flags,
569		std::stop_token const& token,
570		std::error_code* ec,
571		std::size_t* bytes_out)
572		{
573	177x	auto& op = rd_;
574	177x	op.reset();
575
576	177x	capy::mutable_buffer bufs[RecvFromOp::max_buffers];
577	177x	op.iovec_count =
578	177x	static_cast<int>(param.copy_to(bufs, RecvFromOp::max_buffers));
579
580	177x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
581		{
582	4x	op.h = h;
583	4x	op.ex = ex;
584	4x	op.ec_out = ec;
585	4x	op.bytes_out = bytes_out;
586	4x	op.start(token, static_cast<Derived*>(this));
587	4x	op.impl_ptr = this->shared_from_this();
588	4x	op.complete(0, 0);
589	4x	this->svc_.post(&op);
590	4x	return std::noop_coroutine();
591		}
592
593	346x	for (int i = 0; i < op.iovec_count; ++i)
594		{
595	173x	op.iovecs[i].iov_base = bufs[i].data();
596	173x	op.iovecs[i].iov_len = bufs[i].size();
597		}
598
599	173x	op.fd = this->fd_;
600	173x	op.source_out = source;
601	173x	op.msg_flags = to_native_msg_flags(flags);
602
603		// Speculative recvmsg
604	173x	msghdr msg{};
605	173x	msg.msg_name = &op.source_storage;
606	173x	msg.msg_namelen = sizeof(op.source_storage);
607	173x	msg.msg_iov = op.iovecs;
608	173x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
609
610		ssize_t n;
611		do
612		{
613	173x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
614		}
615	173x	while (n < 0 && errno == EINTR);
616
617	173x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
618		{
619	96x	int err = (n < 0) ? errno : 0;
620	96x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
621	96x	if (n >= 0)
622	94x	op.source_addrlen = msg.msg_namelen;
623
624	96x	if (this->svc_.scheduler().try_consume_inline_budget())
625		{
626	85x	*ec = err ? make_err(err) : std::error_code{};
627	85x	*bytes_out = bytes;
628	85x	if (source && !err && n >= 0)
629	166x	*source = from_sockaddr_as(
630	83x	op.source_storage,
631		op.source_addrlen,
632		Endpoint{});
633	85x	op.cont.h = h;
634	85x	return dispatch_coro(ex, op.cont);
635		}
636	11x	op.h = h;
637	11x	op.ex = ex;
638	11x	op.ec_out = ec;
639	11x	op.bytes_out = bytes_out;
640	11x	op.start(token, static_cast<Derived*>(this));
641	11x	op.impl_ptr = this->shared_from_this();
642	11x	op.complete(err, bytes);
643	11x	this->svc_.post(&op);
644	11x	return std::noop_coroutine();
645		}
646
647		// EAGAIN — register with reactor
648	77x	op.h = h;
649	77x	op.ex = ex;
650	77x	op.ec_out = ec;
651	77x	op.bytes_out = bytes_out;
652	77x	op.start(token, static_cast<Derived*>(this));
653	77x	op.impl_ptr = this->shared_from_this();
654
655	77x	this->register_op(
656	77x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
657	77x	this->desc_state_.read_cancel_pending);
658	77x	return std::noop_coroutine();
659		}
660
661		// do_connect
662
663		template<
664		class Derived,
665		class Service,
666		class ConnOp,
667		class SendToOp,
668		class RecvFromOp,
669		class SendOp,
670		class RecvOp,
671		class WaitOp,
672		class DescState,
673		class ImplBase,
674		class Endpoint>
675		std::coroutine_handle<>
676	44x	reactor_datagram_socket<
677		Derived,
678		Service,
679		ConnOp,
680		SendToOp,
681		RecvFromOp,
682		SendOp,
683		RecvOp,
684		WaitOp,
685		DescState,
686		ImplBase,
687		Endpoint>::
688		do_connect(
689		std::coroutine_handle<> h,
690		capy::executor_ref ex,
691		Endpoint const& ep,
692		std::stop_token const& token,
693		std::error_code* ec)
694		{
695	44x	auto& op = conn_;
696
697	44x	sockaddr_storage storage{};
698	44x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
699		int result =
700	44x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
701
702	44x	if (result == 0)
703		{
704	40x	sockaddr_storage local_storage{};
705	40x	socklen_t local_len = sizeof(local_storage);
706	40x	if (::getsockname(
707		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
708	40x	&local_len) == 0)
709	36x	this->local_endpoint_ =
710	40x	from_sockaddr_as(local_storage, local_len, Endpoint{});
711	40x	remote_endpoint_ = ep;
712		}
713
714	44x	if (result == 0 \|\| errno != EINPROGRESS)
715		{
716	44x	int err = (result < 0) ? errno : 0;
717	44x	if (this->svc_.scheduler().try_consume_inline_budget())
718		{
719	12x	*ec = err ? make_err(err) : std::error_code{};
720	12x	op.cont.h = h;
721	12x	return dispatch_coro(ex, op.cont);
722		}
723	32x	op.reset();
724	32x	op.h = h;
725	32x	op.ex = ex;
726	32x	op.ec_out = ec;
727	32x	op.fd = this->fd_;
728	32x	op.target_endpoint = ep;
729	32x	op.start(token, static_cast<Derived*>(this));
730	32x	op.impl_ptr = this->shared_from_this();
731	32x	op.complete(err, 0);
732	32x	this->svc_.post(&op);
733	32x	return std::noop_coroutine();
734		}
735
736		// EINPROGRESS — register with reactor
737	✗	op.reset();
738	✗	op.h = h;
739	✗	op.ex = ex;
740	✗	op.ec_out = ec;
741	✗	op.fd = this->fd_;
742	✗	op.target_endpoint = ep;
743	✗	op.start(token, static_cast<Derived*>(this));
744	✗	op.impl_ptr = this->shared_from_this();
745
746	✗	this->register_op(
747	✗	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
748	✗	this->desc_state_.connect_cancel_pending);
749	✗	return std::noop_coroutine();
750		}
751
752		// do_send (connected mode)
753
754		template<
755		class Derived,
756		class Service,
757		class ConnOp,
758		class SendToOp,
759		class RecvFromOp,
760		class SendOp,
761		class RecvOp,
762		class WaitOp,
763		class DescState,
764		class ImplBase,
765		class Endpoint>
766		std::coroutine_handle<>
767	115x	reactor_datagram_socket<
768		Derived,
769		Service,
770		ConnOp,
771		SendToOp,
772		RecvFromOp,
773		SendOp,
774		RecvOp,
775		WaitOp,
776		DescState,
777		ImplBase,
778		Endpoint>::
779		do_send(
780		std::coroutine_handle<> h,
781		capy::executor_ref ex,
782		buffer_param param,
783		int flags,
784		std::stop_token const& token,
785		std::error_code* ec,
786		std::size_t* bytes_out)
787		{
788	115x	auto& op = send_wr_;
789	115x	op.reset();
790
791	115x	capy::mutable_buffer bufs[SendOp::max_buffers];
792	115x	op.iovec_count = static_cast<int>(param.copy_to(bufs, SendOp::max_buffers));
793
794	226x	for (int i = 0; i < op.iovec_count; ++i)
795		{
796	111x	op.iovecs[i].iov_base = bufs[i].data();
797	111x	op.iovecs[i].iov_len = bufs[i].size();
798		}
799
800	115x	op.fd = this->fd_;
801	115x	op.msg_flags = to_native_msg_flags(flags);
802
803		// Speculative sendmsg with no destination (connected mode)
804	115x	msghdr msg{};
805	115x	msg.msg_iov = op.iovecs;
806	115x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
807
808		#ifdef MSG_NOSIGNAL
809	115x	int send_flags = op.msg_flags \| MSG_NOSIGNAL;
810		#else
811		int send_flags = op.msg_flags;
812		#endif
813
814		ssize_t n;
815		do
816		{
817	115x	n = ::sendmsg(this->fd_, &msg, send_flags);
818		}
819	115x	while (n < 0 && errno == EINTR);
820
821	115x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
822		{
823	81x	int err = (n < 0) ? errno : 0;
824	81x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
825
826	81x	if (this->svc_.scheduler().try_consume_inline_budget())
827		{
828	54x	*ec = err ? make_err(err) : std::error_code{};
829	54x	*bytes_out = bytes;
830	54x	op.cont.h = h;
831	54x	return dispatch_coro(ex, op.cont);
832		}
833	27x	op.h = h;
834	27x	op.ex = ex;
835	27x	op.ec_out = ec;
836	27x	op.bytes_out = bytes_out;
837	27x	op.start(token, static_cast<Derived*>(this));
838	27x	op.impl_ptr = this->shared_from_this();
839	27x	op.complete(err, bytes);
840	27x	this->svc_.post(&op);
841	27x	return std::noop_coroutine();
842		}
843
844		// EAGAIN — register with reactor
845	34x	op.h = h;
846	34x	op.ex = ex;
847	34x	op.ec_out = ec;
848	34x	op.bytes_out = bytes_out;
849	34x	op.start(token, static_cast<Derived*>(this));
850	34x	op.impl_ptr = this->shared_from_this();
851
852	34x	this->register_op(
853	34x	op, this->desc_state_.write_op, this->desc_state_.write_ready,
854	34x	this->desc_state_.write_cancel_pending, true);
855	34x	return std::noop_coroutine();
856		}
857
858		// do_recv (connected mode)
859
860		template<
861		class Derived,
862		class Service,
863		class ConnOp,
864		class SendToOp,
865		class RecvFromOp,
866		class SendOp,
867		class RecvOp,
868		class WaitOp,
869		class DescState,
870		class ImplBase,
871		class Endpoint>
872		std::coroutine_handle<>
873	115x	reactor_datagram_socket<
874		Derived,
875		Service,
876		ConnOp,
877		SendToOp,
878		RecvFromOp,
879		SendOp,
880		RecvOp,
881		WaitOp,
882		DescState,
883		ImplBase,
884		Endpoint>::
885		do_recv(
886		std::coroutine_handle<> h,
887		capy::executor_ref ex,
888		buffer_param param,
889		int flags,
890		std::stop_token const& token,
891		std::error_code* ec,
892		std::size_t* bytes_out)
893		{
894	115x	auto& op = recv_rd_;
895	115x	op.reset();
896
897	115x	capy::mutable_buffer bufs[RecvOp::max_buffers];
898	115x	op.iovec_count = static_cast<int>(param.copy_to(bufs, RecvOp::max_buffers));
899
900	115x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
901		{
902	4x	op.h = h;
903	4x	op.ex = ex;
904	4x	op.ec_out = ec;
905	4x	op.bytes_out = bytes_out;
906	4x	op.start(token, static_cast<Derived*>(this));
907	4x	op.impl_ptr = this->shared_from_this();
908	4x	op.complete(0, 0);
909	4x	this->svc_.post(&op);
910	4x	return std::noop_coroutine();
911		}
912
913	222x	for (int i = 0; i < op.iovec_count; ++i)
914		{
915	111x	op.iovecs[i].iov_base = bufs[i].data();
916	111x	op.iovecs[i].iov_len = bufs[i].size();
917		}
918
919	111x	op.fd = this->fd_;
920	111x	op.msg_flags = to_native_msg_flags(flags);
921
922		// Speculative recvmsg with no source (connected mode)
923	111x	msghdr msg{};
924	111x	msg.msg_iov = op.iovecs;
925	111x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
926
927		ssize_t n;
928		do
929		{
930	111x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
931		}
932	111x	while (n < 0 && errno == EINTR);
933
934	111x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
935		{
936	59x	int err = (n < 0) ? errno : 0;
937	59x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
938
939	59x	if (this->svc_.scheduler().try_consume_inline_budget())
940		{
941	47x	*ec = err ? make_err(err) : std::error_code{};
942	47x	*bytes_out = bytes;
943	47x	op.cont.h = h;
944	47x	return dispatch_coro(ex, op.cont);
945		}
946	12x	op.h = h;
947	12x	op.ex = ex;
948	12x	op.ec_out = ec;
949	12x	op.bytes_out = bytes_out;
950	12x	op.start(token, static_cast<Derived*>(this));
951	12x	op.impl_ptr = this->shared_from_this();
952	12x	op.complete(err, bytes);
953	12x	this->svc_.post(&op);
954	12x	return std::noop_coroutine();
955		}
956
957		// EAGAIN — register with reactor
958	52x	op.h = h;
959	52x	op.ex = ex;
960	52x	op.ec_out = ec;
961	52x	op.bytes_out = bytes_out;
962	52x	op.start(token, static_cast<Derived*>(this));
963	52x	op.impl_ptr = this->shared_from_this();
964
965	52x	this->register_op(
966	52x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
967	52x	this->desc_state_.read_cancel_pending);
968	52x	return std::noop_coroutine();
969		}
970
971		// do_wait
972
973		template<
974		class Derived,
975		class Service,
976		class ConnOp,
977		class SendToOp,
978		class RecvFromOp,
979		class SendOp,
980		class RecvOp,
981		class WaitOp,
982		class DescState,
983		class ImplBase,
984		class Endpoint>
985		std::coroutine_handle<>
986	40x	reactor_datagram_socket<
987		Derived,
988		Service,
989		ConnOp,
990		SendToOp,
991		RecvFromOp,
992		SendOp,
993		RecvOp,
994		WaitOp,
995		DescState,
996		ImplBase,
997		Endpoint>::
998		do_wait(
999		std::coroutine_handle<> h,
1000		capy::executor_ref ex,
1001		wait_type w,
1002		std::stop_token const& token,
1003		std::error_code* ec)
1004		{
1005		WaitOp* op_ptr;
1006		reactor_op_base** desc_slot_ptr;
1007		bool* cancel_flag_ptr;
1008		std::uint32_t event;
1009
1010	40x	if (w == wait_type::read)
1011		{
1012	28x	op_ptr = &wait_rd_;
1013	28x	desc_slot_ptr = &this->desc_state_.wait_read_op;
1014	28x	cancel_flag_ptr = &this->desc_state_.wait_read_cancel_pending;
1015	28x	event = reactor_event_read;
1016		}
1017	12x	else if (w == wait_type::write)
1018		{
1019	4x	op_ptr = &wait_wr_;
1020	4x	desc_slot_ptr = &this->desc_state_.wait_write_op;
1021	4x	cancel_flag_ptr = &this->desc_state_.wait_write_cancel_pending;
1022	4x	event = reactor_event_write;
1023		}
1024		else // wait_type::error
1025		{
1026	8x	op_ptr = &wait_er_;
1027	8x	desc_slot_ptr = &this->desc_state_.wait_error_op;
1028	8x	cancel_flag_ptr = &this->desc_state_.wait_error_cancel_pending;
1029	8x	event = reactor_event_error;
1030		}
1031
1032	40x	auto& op = *op_ptr;
1033
1034		// Speculative probe, mirroring the speculative read: an
1035		// edge-triggered reactor cannot report a condition that already
1036		// holds, so a wait initiated on an already-ready socket would
1037		// otherwise park forever.
1038	40x	int perr = 0;
1039	40x	if (WaitOp::probe(this->fd_, event, perr))
1040		{
1041	12x	if (this->svc_.scheduler().try_consume_inline_budget())
1042		{
1043	4x	*ec = perr ? make_err(perr) : std::error_code{};
1044	4x	op.cont.h = h;
1045	4x	return dispatch_coro(ex, op.cont);
1046		}
1047	8x	op.reset();
1048	8x	op.wait_event = event;
1049	8x	op.h = h;
1050	8x	op.ex = ex;
1051	8x	op.ec_out = ec;
1052	8x	op.fd = this->fd_;
1053	8x	op.start(token, static_cast<Derived*>(this));
1054	8x	op.impl_ptr = this->shared_from_this();
1055	8x	op.complete(perr, 0);
1056	8x	this->svc_.post(&op);
1057	8x	return std::noop_coroutine();
1058		}
1059
1060	28x	op.reset();
1061	28x	op.wait_event = event;
1062	28x	op.h = h;
1063	28x	op.ex = ex;
1064	28x	op.ec_out = ec;
1065	28x	op.fd = this->fd_;
1066	28x	op.start(token, static_cast<Derived*>(this));
1067	28x	op.impl_ptr = this->shared_from_this();
1068
1069		// Force register_op's ready path so the wait op re-probes under
1070		// the descriptor mutex before parking. An edge consumed between
1071		// the speculative probe above and the park (a concurrent short
1072		// read, or an error event dispatched to an empty slot) would
1073		// otherwise leave the wait parked on a ready socket.
1074	28x	bool force_probe = true;
1075	28x	this->register_op(
1076		op, desc_slot_ptr, force_probe, cancel_flag_ptr,
1077		event == reactor_event_write);
1078	28x	return std::noop_coroutine();
1079		}
1080
1081		} // namespace boost::corosio::detail
1082
1083		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
1084