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