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

91.2% Lines (371/0/407) 100.0% List of functions (104/0/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		//
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	423x	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	423x	~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	135x	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	135x	return do_send_to(h, ex, buf, dest, flags, token, ec, bytes_out);
136		}
137
138	147x	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	147x	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	105x	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	105x	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	36x	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	36x	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	1596x	void close_socket() noexcept
204		{
205	1596x	do_close_socket();
206	1596x	}
207
208		/// Cache local and remote endpoints.
209	120x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
210		{
211	120x	this->local_endpoint_ = std::move(local);
212	120x	remote_endpoint_ = std::move(remote);
213	120x	}
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		`wait_type::write` completes immediately. Read and error
291		waits probe the descriptor with a zero-timeout `poll()` and
292		complete at once if the condition already holds; otherwise
293		the op re-probes under the descriptor mutex and parks,
294		completing when a reactor event arrives and a fresh probe
295		confirms the condition.
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	1596x	void do_close_socket() noexcept
310		{
311	1596x	base_type::do_close_socket();
312	1596x	remote_endpoint_ = Endpoint{};
313	1596x	}
314
315	2x	native_handle_type do_release_socket() noexcept
316		{
317	2x	auto fd = base_type::do_release_socket();
318	2x	remote_endpoint_ = Endpoint{};
319	2x	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	8x	std::error_code do_shutdown(int what) noexcept
329		{
330		int how;
331	8x	switch (what)
332		{
333	2x	case 0:
334	2x	how = SHUT_RD;
335	2x	break;
336	4x	case 1:
337	4x	how = SHUT_WR;
338	4x	break;
339	2x	case 2:
340	2x	how = SHUT_RDWR;
341	2x	break;
342	✗	default:
343	✗	return make_err(EINVAL);
344		}
345	8x	if (::shutdown(this->fd_, how) != 0)
346	✗	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	24x	bool* op_to_cancel_flag(Op& op) noexcept
377		{
378	24x	if (&op == static_cast<void*>(&conn_))
379	2x	return &this->desc_state_.connect_cancel_pending;
380	22x	if (&op == static_cast<void*>(&rd_))
381	6x	return &this->desc_state_.read_cancel_pending;
382	16x	if (&op == static_cast<void*>(&wr_))
383	4x	return &this->desc_state_.write_cancel_pending;
384	12x	if (&op == static_cast<void*>(&recv_rd_))
385	4x	return &this->desc_state_.read_cancel_pending;
386	8x	if (&op == static_cast<void*>(&send_wr_))
387	4x	return &this->desc_state_.write_cancel_pending;
388	4x	if (&op == static_cast<void*>(&wait_rd_))
389	4x	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	1615x	void for_each_op(Fn fn) noexcept
399		{
400	1615x	fn(conn_);
401	1615x	fn(rd_);
402	1615x	fn(wr_);
403	1615x	fn(recv_rd_);
404	1615x	fn(send_wr_);
405	1615x	fn(wait_rd_);
406	1615x	fn(wait_wr_);
407	1615x	fn(wait_er_);
408	1615x	}
409
410		template<class Fn>
411	1615x	void for_each_desc_entry(Fn fn) noexcept
412		{
413	1615x	fn(conn_, this->desc_state_.connect_op);
414	1615x	fn(rd_, this->desc_state_.read_op);
415	1615x	fn(wr_, this->desc_state_.write_op);
416	1615x	fn(recv_rd_, this->desc_state_.read_op);
417	1615x	fn(send_wr_, this->desc_state_.write_op);
418	1615x	fn(wait_rd_, this->desc_state_.wait_read_op);
419	1615x	fn(wait_wr_, this->desc_state_.wait_write_op);
420	1615x	fn(wait_er_, this->desc_state_.wait_error_op);
421	1615x	}
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	135x	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	135x	auto& op = wr_;
462	135x	op.reset();
463
464	135x	capy::mutable_buffer bufs[SendToOp::max_buffers];
465	135x	op.iovec_count =
466	135x	static_cast<int>(param.copy_to(bufs, SendToOp::max_buffers));
467
468	264x	for (int i = 0; i < op.iovec_count; ++i)
469		{
470	129x	op.iovecs[i].iov_base = bufs[i].data();
471	129x	op.iovecs[i].iov_len = bufs[i].size();
472		}
473
474		// Set up destination address
475	135x	op.dest_len = to_sockaddr(dest, socket_family(this->fd_), op.dest_storage);
476	135x	op.fd = this->fd_;
477	135x	op.msg_flags = to_native_msg_flags(flags);
478
479		// Speculative sendmsg
480	135x	msghdr msg{};
481	135x	msg.msg_name = &op.dest_storage;
482	135x	msg.msg_namelen = op.dest_len;
483	135x	msg.msg_iov = op.iovecs;
484	135x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
485
486		#ifdef MSG_NOSIGNAL
487	135x	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	135x	n = ::sendmsg(this->fd_, &msg, send_flags);
496		}
497	135x	while (n < 0 && errno == EINTR);
498
499	135x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
500		{
501	103x	int err = (n < 0) ? errno : 0;
502	103x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
503
504	103x	if (this->svc_.scheduler().try_consume_inline_budget())
505		{
506	43x	*ec = err ? make_err(err) : std::error_code{};
507	43x	*bytes_out = bytes;
508	43x	op.cont.h = h;
509	43x	return dispatch_coro(ex, op.cont);
510		}
511	60x	op.h = h;
512	60x	op.ex = ex;
513	60x	op.ec_out = ec;
514	60x	op.bytes_out = bytes_out;
515	60x	op.start(token, static_cast<Derived*>(this));
516	60x	op.impl_ptr = this->shared_from_this();
517	60x	op.complete(err, bytes);
518	60x	this->svc_.post(&op);
519	60x	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	147x	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	147x	auto& op = rd_;
574	147x	op.reset();
575
576	147x	capy::mutable_buffer bufs[RecvFromOp::max_buffers];
577	147x	op.iovec_count =
578	147x	static_cast<int>(param.copy_to(bufs, RecvFromOp::max_buffers));
579
580	147x	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	286x	for (int i = 0; i < op.iovec_count; ++i)
594		{
595	143x	op.iovecs[i].iov_base = bufs[i].data();
596	143x	op.iovecs[i].iov_len = bufs[i].size();
597		}
598
599	143x	op.fd = this->fd_;
600	143x	op.source_out = source;
601	143x	op.msg_flags = to_native_msg_flags(flags);
602
603		// Speculative recvmsg
604	143x	msghdr msg{};
605	143x	msg.msg_name = &op.source_storage;
606	143x	msg.msg_namelen = sizeof(op.source_storage);
607	143x	msg.msg_iov = op.iovecs;
608	143x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
609
610		ssize_t n;
611		do
612		{
613	143x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
614		}
615	143x	while (n < 0 && errno == EINTR);
616
617	143x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
618		{
619	80x	int err = (n < 0) ? errno : 0;
620	80x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
621	80x	if (n >= 0)
622	80x	op.source_addrlen = msg.msg_namelen;
623
624	80x	if (this->svc_.scheduler().try_consume_inline_budget())
625		{
626	71x	*ec = err ? make_err(err) : std::error_code{};
627	71x	*bytes_out = bytes;
628	71x	if (source && !err && n >= 0)
629	142x	*source = from_sockaddr_as(
630	71x	op.source_storage,
631		op.source_addrlen,
632		Endpoint{});
633	71x	op.cont.h = h;
634	71x	return dispatch_coro(ex, op.cont);
635		}
636	9x	op.h = h;
637	9x	op.ex = ex;
638	9x	op.ec_out = ec;
639	9x	op.bytes_out = bytes_out;
640	9x	op.start(token, static_cast<Derived*>(this));
641	9x	op.impl_ptr = this->shared_from_this();
642	9x	op.complete(err, bytes);
643	9x	this->svc_.post(&op);
644	9x	return std::noop_coroutine();
645		}
646
647		// EAGAIN — register with reactor
648	63x	op.h = h;
649	63x	op.ex = ex;
650	63x	op.ec_out = ec;
651	63x	op.bytes_out = bytes_out;
652	63x	op.start(token, static_cast<Derived*>(this));
653	63x	op.impl_ptr = this->shared_from_this();
654
655	63x	this->register_op(
656	63x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
657	63x	this->desc_state_.read_cancel_pending);
658	63x	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	105x	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	105x	auto& op = recv_rd_;
895	105x	op.reset();
896
897	105x	capy::mutable_buffer bufs[RecvOp::max_buffers];
898	105x	op.iovec_count = static_cast<int>(param.copy_to(bufs, RecvOp::max_buffers));
899
900	105x	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	202x	for (int i = 0; i < op.iovec_count; ++i)
914		{
915	101x	op.iovecs[i].iov_base = bufs[i].data();
916	101x	op.iovecs[i].iov_len = bufs[i].size();
917		}
918
919	101x	op.fd = this->fd_;
920	101x	op.msg_flags = to_native_msg_flags(flags);
921
922		// Speculative recvmsg with no source (connected mode)
923	101x	msghdr msg{};
924	101x	msg.msg_iov = op.iovecs;
925	101x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
926
927		ssize_t n;
928		do
929		{
930	101x	n = ::recvmsg(this->fd_, &msg, op.msg_flags);
931		}
932	101x	while (n < 0 && errno == EINTR);
933
934	101x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
935		{
936	53x	int err = (n < 0) ? errno : 0;
937	53x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
938
939	53x	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	8x	op.h = h;
947	8x	op.ex = ex;
948	8x	op.ec_out = ec;
949	8x	op.bytes_out = bytes_out;
950	8x	op.start(token, static_cast<Derived*>(this));
951	8x	op.impl_ptr = this->shared_from_this();
952	8x	op.complete(err, bytes);
953	8x	this->svc_.post(&op);
954	8x	return std::noop_coroutine();
955		}
956
957		// EAGAIN — register with reactor
958	48x	op.h = h;
959	48x	op.ex = ex;
960	48x	op.ec_out = ec;
961	48x	op.bytes_out = bytes_out;
962	48x	op.start(token, static_cast<Derived*>(this));
963	48x	op.impl_ptr = this->shared_from_this();
964
965	48x	this->register_op(
966	48x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
967	48x	this->desc_state_.read_cancel_pending);
968	48x	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	36x	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		// wait_type::write completes immediately (see reactor_stream_socket::do_wait).
1006	36x	if (w == wait_type::write)
1007		{
1008	4x	auto& op = wait_wr_;
1009	4x	if (this->svc_.scheduler().try_consume_inline_budget())
1010		{
1011	✗	*ec = std::error_code{};
1012	✗	op.cont.h = h;
1013	✗	return dispatch_coro(ex, op.cont);
1014		}
1015	4x	op.reset();
1016	4x	op.wait_event = reactor_event_write;
1017	4x	op.h = h;
1018	4x	op.ex = ex;
1019	4x	op.ec_out = ec;
1020	4x	op.fd = this->fd_;
1021	4x	op.start(token, static_cast<Derived*>(this));
1022	4x	op.impl_ptr = this->shared_from_this();
1023	4x	op.complete(0, 0);
1024	4x	this->svc_.post(&op);
1025	4x	return std::noop_coroutine();
1026		}
1027
1028		WaitOp* op_ptr;
1029		reactor_op_base** desc_slot_ptr;
1030		bool* cancel_flag_ptr;
1031		std::uint32_t event;
1032
1033	32x	if (w == wait_type::read)
1034		{
1035	24x	op_ptr = &wait_rd_;
1036	24x	desc_slot_ptr = &this->desc_state_.wait_read_op;
1037	24x	cancel_flag_ptr = &this->desc_state_.wait_read_cancel_pending;
1038	24x	event = reactor_event_read;
1039		}
1040		else // wait_type::error
1041		{
1042	8x	op_ptr = &wait_er_;
1043	8x	desc_slot_ptr = &this->desc_state_.wait_error_op;
1044	8x	cancel_flag_ptr = &this->desc_state_.wait_error_cancel_pending;
1045	8x	event = reactor_event_error;
1046		}
1047
1048	32x	auto& op = *op_ptr;
1049
1050		// Speculative probe, mirroring the speculative read: an
1051		// edge-triggered reactor cannot report a condition that already
1052		// holds, so a wait initiated on an already-ready socket would
1053		// otherwise park forever.
1054	32x	int perr = 0;
1055	32x	if (WaitOp::probe(this->fd_, event, perr))
1056		{
1057	4x	if (this->svc_.scheduler().try_consume_inline_budget())
1058		{
1059	4x	*ec = perr ? make_err(perr) : std::error_code{};
1060	4x	op.cont.h = h;
1061	4x	return dispatch_coro(ex, op.cont);
1062		}
1063	✗	op.reset();
1064	✗	op.wait_event = event;
1065	✗	op.h = h;
1066	✗	op.ex = ex;
1067	✗	op.ec_out = ec;
1068	✗	op.fd = this->fd_;
1069	✗	op.start(token, static_cast<Derived*>(this));
1070	✗	op.impl_ptr = this->shared_from_this();
1071	✗	op.complete(perr, 0);
1072	✗	this->svc_.post(&op);
1073	✗	return std::noop_coroutine();
1074		}
1075
1076	28x	op.reset();
1077	28x	op.wait_event = event;
1078	28x	op.h = h;
1079	28x	op.ex = ex;
1080	28x	op.ec_out = ec;
1081	28x	op.fd = this->fd_;
1082	28x	op.start(token, static_cast<Derived*>(this));
1083	28x	op.impl_ptr = this->shared_from_this();
1084
1085		// Force register_op's ready path so the wait op re-probes under
1086		// the descriptor mutex before parking. An edge consumed between
1087		// the speculative probe above and the park (a concurrent short
1088		// read, or an error event dispatched to an empty slot) would
1089		// otherwise leave the wait parked on a ready socket.
1090	28x	bool force_probe = true;
1091	28x	this->register_op(
1092		op, desc_slot_ptr, force_probe, cancel_flag_ptr,
1093		false);
1094	28x	return std::noop_coroutine();
1095		}
1096
1097		} // namespace boost::corosio::detail
1098
1099		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
1100