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

98.6% Lines (352 / 357) 100.0% Functions (104 / 104)

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

reactor_datagram_socket.hpp

f(x) Functions (104)

Function Calls Lines Blocks

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