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	463x	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	463x	~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	145x	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	145x	return do_send_to(h, ex, buf, dest, flags, token, ec, bytes_out);
136		}
137
138	163x	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	163x	return do_recv_from(h, ex, buf, source, flags, token, ec, bytes_out);
149		}
150
151	38x	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	38x	return do_connect(h, ex, ep, token, ec);
159		}
160
161	107x	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	107x	return do_send(h, ex, buf, flags, token, ec, bytes_out);
171		}
172
173	109x	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	109x	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	1736x	void close_socket() noexcept
204		{
205	1736x	do_close_socket();
206	1736x	}
207
208		/// Cache local and remote endpoints.
209	132x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
210		{
211	132x	this->local_endpoint_ = std::move(local);
212	132x	remote_endpoint_ = std::move(remote);
213	132x	}
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	1736x	void do_close_socket() noexcept
310		{
311	1736x	base_type::do_close_socket();
312	1736x	remote_endpoint_ = Endpoint{};
313	1736x	}
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	10x	std::error_code do_shutdown(int what) noexcept
329		{
330		int how;
331	10x	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	2x	case 2:
340	2x	how = SHUT_RDWR;
341	2x	break;
342	✗	default:
343	✗	return make_err(EINVAL);
344		}
345	10x	if (::shutdown(this->fd_, how) != 0)
346	2x	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	1757x	void for_each_op(Fn fn) noexcept
399		{
400	1757x	fn(conn_);
401	1757x	fn(rd_);
402	1757x	fn(wr_);
403	1757x	fn(recv_rd_);
404	1757x	fn(send_wr_);
405	1757x	fn(wait_rd_);
406	1757x	fn(wait_wr_);
407	1757x	fn(wait_er_);
408	1757x	}
409
410		template<class Fn>
411	1757x	void for_each_desc_entry(Fn fn) noexcept
412		{
413	1757x	fn(conn_, this->desc_state_.connect_op);
414	1757x	fn(rd_, this->desc_state_.read_op);
415	1757x	fn(wr_, this->desc_state_.write_op);
416	1757x	fn(recv_rd_, this->desc_state_.read_op);
417	1757x	fn(send_wr_, this->desc_state_.write_op);
418	1757x	fn(wait_rd_, this->desc_state_.wait_read_op);
419	1757x	fn(wait_wr_, this->desc_state_.wait_write_op);
420	1757x	fn(wait_er_, this->desc_state_.wait_error_op);
421	1757x	}
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	145x	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	145x	auto& op = wr_;
462	145x	op.reset();
463
464	145x	capy::mutable_buffer bufs[SendToOp::max_buffers];
465	145x	op.iovec_count =
466	145x	static_cast<int>(param.copy_to(bufs, SendToOp::max_buffers));
467
468	284x	for (int i = 0; i < op.iovec_count; ++i)
469		{
470	139x	op.iovecs[i].iov_base = bufs[i].data();
471	139x	op.iovecs[i].iov_len = bufs[i].size();
472		}
473
474		// Set up destination address
475	145x	op.dest_len = to_sockaddr(dest, socket_family(this->fd_), op.dest_storage);
476	145x	op.fd = this->fd_;
477	145x	op.msg_flags = to_native_msg_flags(flags);
478
479		// Speculative sendmsg
480	145x	msghdr msg{};
481	145x	msg.msg_name = &op.dest_storage;
482	145x	msg.msg_namelen = op.dest_len;
483	145x	msg.msg_iov = op.iovecs;
484	145x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
485
486		#ifdef MSG_NOSIGNAL
487	145x	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	145x	n = ::sendmsg(this->fd_, &msg, send_flags);
496		}
497	145x	while (n < 0 && errno == EINTR);
498
499	145x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
500		{
501	113x	int err = (n < 0) ? errno : 0;
502	113x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
503
504	113x	if (this->svc_.scheduler().try_consume_inline_budget())
505		{
506	45x	*ec = err ? make_err(err) : std::error_code{};
507	45x	*bytes_out = bytes;
508	45x	op.cont.h = h;
509	45x	return dispatch_coro(ex, op.cont);
510		}
511	68x	op.h = h;
512	68x	op.ex = ex;
513	68x	op.ec_out = ec;
514	68x	op.bytes_out = bytes_out;
515	68x	op.start(token, static_cast<Derived*>(this));
516	68x	op.impl_ptr = this->shared_from_this();
517	68x	op.complete(err, bytes);
518	68x	this->svc_.post(&op);
519	68x	return std::noop_coroutine();
520		}
521
522		// EAGAIN — register with reactor
523	32x	op.h = h;
524	32x	op.ex = ex;
525	32x	op.ec_out = ec;
526	32x	op.bytes_out = bytes_out;
527	32x	op.start(token, static_cast<Derived*>(this));
528	32x	op.impl_ptr = this->shared_from_this();
529
530	32x	this->register_op(
531	32x	op, this->desc_state_.write_op, this->desc_state_.write_ready,
532	32x	this->desc_state_.write_cancel_pending, true);
533	32x	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	163x	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	163x	auto& op = rd_;
574	163x	op.reset();
575
576	163x	capy::mutable_buffer bufs[RecvFromOp::max_buffers];
577	163x	op.iovec_count =
578	163x	static_cast<int>(param.copy_to(bufs, RecvFromOp::max_buffers));
579
580	163x	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	318x	for (int i = 0; i < op.iovec_count; ++i)
594		{
595	159x	op.iovecs[i].iov_base = bufs[i].data();
596	159x	op.iovecs[i].iov_len = bufs[i].size();
597		}
598
599	159x	op.fd = this->fd_;
600	159x	op.source_out = source;
601	159x	op.msg_flags = to_native_msg_flags(flags);
602
603		// Speculative recvmsg
604	159x	msghdr msg{};
605	159x	msg.msg_name = &op.source_storage;
606	159x	msg.msg_namelen = sizeof(op.source_storage);
607	159x	msg.msg_iov = op.iovecs;
608	159x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
609
610		ssize_t n;
611		do
612		{
613	159x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
614		}
615	159x	while (n < 0 && errno == EINTR);
616
617	159x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
618		{
619	92x	int err = (n < 0) ? errno : 0;
620	92x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
621	92x	if (n >= 0)
622	92x	op.source_addrlen = msg.msg_namelen;
623
624	92x	if (this->svc_.scheduler().try_consume_inline_budget())
625		{
626	81x	*ec = err ? make_err(err) : std::error_code{};
627	81x	*bytes_out = bytes;
628	81x	if (source && !err && n >= 0)
629	162x	*source = from_sockaddr_as(
630	81x	op.source_storage,
631		op.source_addrlen,
632		Endpoint{});
633	81x	op.cont.h = h;
634	81x	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	67x	op.h = h;
649	67x	op.ex = ex;
650	67x	op.ec_out = ec;
651	67x	op.bytes_out = bytes_out;
652	67x	op.start(token, static_cast<Derived*>(this));
653	67x	op.impl_ptr = this->shared_from_this();
654
655	67x	this->register_op(
656	67x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
657	67x	this->desc_state_.read_cancel_pending);
658	67x	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	38x	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	38x	auto& op = conn_;
696
697	38x	sockaddr_storage storage{};
698	38x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
699		int result =
700	38x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
701
702	38x	if (result == 0)
703		{
704	36x	sockaddr_storage local_storage{};
705	36x	socklen_t local_len = sizeof(local_storage);
706	36x	if (::getsockname(
707		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
708	36x	&local_len) == 0)
709	32x	this->local_endpoint_ =
710	36x	from_sockaddr_as(local_storage, local_len, Endpoint{});
711	36x	remote_endpoint_ = ep;
712		}
713
714	38x	if (result == 0 \|\| errno != EINPROGRESS)
715		{
716	38x	int err = (result < 0) ? errno : 0;
717	38x	if (this->svc_.scheduler().try_consume_inline_budget())
718		{
719	8x	*ec = err ? make_err(err) : std::error_code{};
720	8x	op.cont.h = h;
721	8x	return dispatch_coro(ex, op.cont);
722		}
723	30x	op.reset();
724	30x	op.h = h;
725	30x	op.ex = ex;
726	30x	op.ec_out = ec;
727	30x	op.fd = this->fd_;
728	30x	op.target_endpoint = ep;
729	30x	op.start(token, static_cast<Derived*>(this));
730	30x	op.impl_ptr = this->shared_from_this();
731	30x	op.complete(err, 0);
732	30x	this->svc_.post(&op);
733	30x	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	107x	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	107x	auto& op = send_wr_;
789	107x	op.reset();
790
791	107x	capy::mutable_buffer bufs[SendOp::max_buffers];
792	107x	op.iovec_count = static_cast<int>(param.copy_to(bufs, SendOp::max_buffers));
793
794	210x	for (int i = 0; i < op.iovec_count; ++i)
795		{
796	103x	op.iovecs[i].iov_base = bufs[i].data();
797	103x	op.iovecs[i].iov_len = bufs[i].size();
798		}
799
800	107x	op.fd = this->fd_;
801	107x	op.msg_flags = to_native_msg_flags(flags);
802
803		// Speculative sendmsg with no destination (connected mode)
804	107x	msghdr msg{};
805	107x	msg.msg_iov = op.iovecs;
806	107x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
807
808		#ifdef MSG_NOSIGNAL
809	107x	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	107x	n = ::sendmsg(this->fd_, &msg, send_flags);
818		}
819	107x	while (n < 0 && errno == EINTR);
820
821	107x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
822		{
823	75x	int err = (n < 0) ? errno : 0;
824	75x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
825
826	75x	if (this->svc_.scheduler().try_consume_inline_budget())
827		{
828	48x	*ec = err ? make_err(err) : std::error_code{};
829	48x	*bytes_out = bytes;
830	48x	op.cont.h = h;
831	48x	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	32x	op.h = h;
846	32x	op.ex = ex;
847	32x	op.ec_out = ec;
848	32x	op.bytes_out = bytes_out;
849	32x	op.start(token, static_cast<Derived*>(this));
850	32x	op.impl_ptr = this->shared_from_this();
851
852	32x	this->register_op(
853	32x	op, this->desc_state_.write_op, this->desc_state_.write_ready,
854	32x	this->desc_state_.write_cancel_pending, true);
855	32x	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	109x	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	109x	auto& op = recv_rd_;
895	109x	op.reset();
896
897	109x	capy::mutable_buffer bufs[RecvOp::max_buffers];
898	109x	op.iovec_count = static_cast<int>(param.copy_to(bufs, RecvOp::max_buffers));
899
900	109x	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	210x	for (int i = 0; i < op.iovec_count; ++i)
914		{
915	105x	op.iovecs[i].iov_base = bufs[i].data();
916	105x	op.iovecs[i].iov_len = bufs[i].size();
917		}
918
919	105x	op.fd = this->fd_;
920	105x	op.msg_flags = to_native_msg_flags(flags);
921
922		// Speculative recvmsg with no source (connected mode)
923	105x	msghdr msg{};
924	105x	msg.msg_iov = op.iovecs;
925	105x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
926
927		ssize_t n;
928		do
929		{
930	105x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
931		}
932	105x	while (n < 0 && errno == EINTR);
933
934	105x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
935		{
936	55x	int err = (n < 0) ? errno : 0;
937	55x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
938
939	55x	if (this->svc_.scheduler().try_consume_inline_budget())
940		{
941	45x	*ec = err ? make_err(err) : std::error_code{};
942	45x	*bytes_out = bytes;
943	45x	op.cont.h = h;
944	45x	return dispatch_coro(ex, op.cont);
945		}
946	10x	op.h = h;
947	10x	op.ex = ex;
948	10x	op.ec_out = ec;
949	10x	op.bytes_out = bytes_out;
950	10x	op.start(token, static_cast<Derived*>(this));
951	10x	op.impl_ptr = this->shared_from_this();
952	10x	op.complete(err, bytes);
953	10x	this->svc_.post(&op);
954	10x	return std::noop_coroutine();
955		}
956
957		// EAGAIN — register with reactor
958	50x	op.h = h;
959	50x	op.ex = ex;
960	50x	op.ec_out = ec;
961	50x	op.bytes_out = bytes_out;
962	50x	op.start(token, static_cast<Derived*>(this));
963	50x	op.impl_ptr = this->shared_from_this();
964
965	50x	this->register_op(
966	50x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
967	50x	this->desc_state_.read_cancel_pending);
968	50x	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