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

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