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

55.6% Lines (408/734) 60.9% List of functions (28/46)

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

f(x) Functions (46)

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/native/detail/reactor/reactor_basic_socket.hpp>
15		#include <boost/corosio/detail/dispatch_coro.hpp>
16		#include <boost/capy/buffers.hpp>
17
18		#include <coroutine>
19
20		#include <errno.h>
21		#include <sys/socket.h>
22		#include <sys/uio.h>
23
24		namespace boost::corosio::detail {
25
26		/** CRTP base for reactor-backed datagram socket implementations.
27
28		Inherits shared data members and cancel/close/register logic
29		from reactor_basic_socket. Adds datagram-specific I/O dispatch
30		for both connectionless (send_to, recv_from) and connected
31		(connect, send, recv) modes.
32
33		@tparam Derived The concrete socket type (CRTP).
34		@tparam Service The backend's datagram service type.
35		@tparam ConnOp The backend's connect op type.
36		@tparam SendToOp The backend's send_to op type.
37		@tparam RecvFromOp The backend's recv_from op type.
38		@tparam SendOp The backend's connected send op type.
39		@tparam RecvOp The backend's connected recv op type.
40		@tparam DescState The backend's descriptor_state type.
41		*/
42		template<
43		class Derived,
44		class Service,
45		class ConnOp,
46		class SendToOp,
47		class RecvFromOp,
48		class SendOp,
49		class RecvOp,
50		class DescState>
51		class reactor_datagram_socket
52		: public reactor_basic_socket<
53		Derived,
54		udp_socket::implementation,
55		Service,
56		DescState>
57		{
58		using base_type = reactor_basic_socket<
59		Derived,
60		udp_socket::implementation,
61		Service,
62		DescState>;
63		friend base_type;
64		friend Derived;
65
66	78x	explicit reactor_datagram_socket(Service& svc) noexcept : base_type(svc) {}
67
68		protected:
69		endpoint remote_endpoint_;
70
71		public:
72		/// Pending connect operation slot.
73		ConnOp conn_;
74
75		/// Pending send_to operation slot.
76		SendToOp wr_;
77
78		/// Pending recv_from operation slot.
79		RecvFromOp rd_;
80
81		/// Pending connected send operation slot.
82		SendOp send_wr_;
83
84		/// Pending connected recv operation slot.
85		RecvOp recv_rd_;
86
87	78x	~reactor_datagram_socket() override = default;
88
89		/// Return the cached remote endpoint.
90	4x	endpoint remote_endpoint() const noexcept override
91		{
92	4x	return remote_endpoint_;
93		}
94
95		/// Cache local and remote endpoints.
96	10x	void set_endpoints(endpoint local, endpoint remote) noexcept
97		{
98	10x	this->local_endpoint_ = local;
99	10x	remote_endpoint_ = remote;
100	10x	}
101
102		/** Shared send_to dispatch.
103
104		Tries sendmsg() speculatively. On success or hard error,
105		returns via inline budget or posts through queue.
106		On EAGAIN, registers with the reactor.
107		*/
108		std::coroutine_handle<> do_send_to(
109		std::coroutine_handle<>,
110		capy::executor_ref,
111		buffer_param,
112		endpoint,
113		std::stop_token const&,
114		std::error_code*,
115		std::size_t*);
116
117		/** Shared recv_from dispatch.
118
119		Tries recvmsg() speculatively. On success or hard error,
120		returns via inline budget or posts through queue.
121		On EAGAIN, registers with the reactor.
122		*/
123		std::coroutine_handle<> do_recv_from(
124		std::coroutine_handle<>,
125		capy::executor_ref,
126		buffer_param,
127		endpoint*,
128		std::stop_token const&,
129		std::error_code*,
130		std::size_t*);
131
132		/** Shared connect dispatch.
133
134		Tries connect() speculatively. On synchronous completion,
135		returns via inline budget or posts through queue.
136		On EINPROGRESS, registers with the reactor.
137		*/
138		std::coroutine_handle<> do_connect(
139		std::coroutine_handle<>,
140		capy::executor_ref,
141		endpoint,
142		std::stop_token const&,
143		std::error_code*);
144
145		/** Shared connected send dispatch.
146
147		Like do_send_to but uses send_wr_ slot and sendmsg()
148		with msg_name=nullptr.
149		*/
150		std::coroutine_handle<> do_send(
151		std::coroutine_handle<>,
152		capy::executor_ref,
153		buffer_param,
154		std::stop_token const&,
155		std::error_code*,
156		std::size_t*);
157
158		/** Shared connected recv dispatch.
159
160		Like do_recv_from but uses recv_rd_ slot and recvmsg()
161		with msg_name=nullptr.
162		*/
163		std::coroutine_handle<> do_recv(
164		std::coroutine_handle<>,
165		capy::executor_ref,
166		buffer_param,
167		std::stop_token const&,
168		std::error_code*,
169		std::size_t*);
170
171		/** Close the socket and cancel pending operations.
172
173		Extends the base do_close_socket() to also reset
174		the remote endpoint.
175		*/
176	304x	void do_close_socket() noexcept
177		{
178	304x	base_type::do_close_socket();
179	304x	remote_endpoint_ = endpoint{};
180	304x	}
181
182		private:
183		// CRTP callbacks for reactor_basic_socket cancel/close
184
185		template<class Op>
186	2x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
187		{
188	2x	if (&op == static_cast<void*>(&conn_))
189	✗	return &this->desc_state_.connect_op;
190	2x	if (&op == static_cast<void*>(&rd_))
191	2x	return &this->desc_state_.read_op;
192	✗	if (&op == static_cast<void*>(&wr_))
193	✗	return &this->desc_state_.write_op;
194	✗	if (&op == static_cast<void*>(&recv_rd_))
195	✗	return &this->desc_state_.read_op;
196	✗	if (&op == static_cast<void*>(&send_wr_))
197	✗	return &this->desc_state_.write_op;
198	✗	return nullptr;
199		}
200
201		template<class Op>
202	✗	bool* op_to_cancel_flag(Op& op) noexcept
203		{
204	✗	if (&op == static_cast<void*>(&conn_))
205	✗	return &this->desc_state_.connect_cancel_pending;
206	✗	if (&op == static_cast<void*>(&rd_))
207	✗	return &this->desc_state_.read_cancel_pending;
208	✗	if (&op == static_cast<void*>(&wr_))
209	✗	return &this->desc_state_.write_cancel_pending;
210	✗	if (&op == static_cast<void*>(&recv_rd_))
211	✗	return &this->desc_state_.read_cancel_pending;
212	✗	if (&op == static_cast<void*>(&send_wr_))
213	✗	return &this->desc_state_.write_cancel_pending;
214	✗	return nullptr;
215		}
216
217		template<class Fn>
218	308x	void for_each_op(Fn fn) noexcept
219		{
220	308x	fn(conn_);
221	308x	fn(rd_);
222	308x	fn(wr_);
223	308x	fn(recv_rd_);
224	308x	fn(send_wr_);
225	308x	}
226
227		template<class Fn>
228	308x	void for_each_desc_entry(Fn fn) noexcept
229		{
230	308x	fn(conn_, this->desc_state_.connect_op);
231	308x	fn(rd_, this->desc_state_.read_op);
232	308x	fn(wr_, this->desc_state_.write_op);
233	308x	fn(recv_rd_, this->desc_state_.read_op);
234	308x	fn(send_wr_, this->desc_state_.write_op);
235	308x	}
236		};
237
238		// do_send_to
239
240		template<
241		class Derived,
242		class Service,
243		class ConnOp,
244		class SendToOp,
245		class RecvFromOp,
246		class SendOp,
247		class RecvOp,
248		class DescState>
249		std::coroutine_handle<>
250	22x	reactor_datagram_socket<
251		Derived,
252		Service,
253		ConnOp,
254		SendToOp,
255		RecvFromOp,
256		SendOp,
257		RecvOp,
258		DescState>::
259		do_send_to(
260		std::coroutine_handle<> h,
261		capy::executor_ref ex,
262		buffer_param param,
263		endpoint dest,
264		std::stop_token const& token,
265		std::error_code* ec,
266		std::size_t* bytes_out)
267		{
268	22x	auto& op = wr_;
269	22x	op.reset();
270
271	22x	capy::mutable_buffer bufs[SendToOp::max_buffers];
272	22x	op.iovec_count =
273	22x	static_cast<int>(param.copy_to(bufs, SendToOp::max_buffers));
274
275	42x	for (int i = 0; i < op.iovec_count; ++i)
276		{
277	20x	op.iovecs[i].iov_base = bufs[i].data();
278	20x	op.iovecs[i].iov_len = bufs[i].size();
279		}
280
281		// Set up destination address
282	22x	op.dest_len = to_sockaddr(dest, socket_family(this->fd_), op.dest_storage);
283	22x	op.fd = this->fd_;
284
285		// Speculative sendmsg
286	22x	msghdr msg{};
287	22x	msg.msg_name = &op.dest_storage;
288	22x	msg.msg_namelen = op.dest_len;
289	22x	msg.msg_iov = op.iovecs;
290	22x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
291
292		#ifdef MSG_NOSIGNAL
293	22x	constexpr int send_flags = MSG_NOSIGNAL;
294		#else
295		constexpr int send_flags = 0;
296		#endif
297
298		ssize_t n;
299		do
300		{
301	22x	n = ::sendmsg(this->fd_, &msg, send_flags);
302		}
303	22x	while (n < 0 && errno == EINTR);
304
305	22x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
306		{
307	22x	int err = (n < 0) ? errno : 0;
308	22x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
309
310	22x	if (this->svc_.scheduler().try_consume_inline_budget())
311		{
312	6x	*ec = err ? make_err(err) : std::error_code{};
313	6x	*bytes_out = bytes;
314	6x	return dispatch_coro(ex, h);
315		}
316	16x	op.h = h;
317	16x	op.ex = ex;
318	16x	op.ec_out = ec;
319	16x	op.bytes_out = bytes_out;
320	16x	op.start(token, static_cast<Derived*>(this));
321	16x	op.impl_ptr = this->shared_from_this();
322	16x	op.complete(err, bytes);
323	16x	this->svc_.post(&op);
324	16x	return std::noop_coroutine();
325		}
326
327		// EAGAIN — register with reactor
328	✗	op.h = h;
329	✗	op.ex = ex;
330	✗	op.ec_out = ec;
331	✗	op.bytes_out = bytes_out;
332	✗	op.start(token, static_cast<Derived*>(this));
333	✗	op.impl_ptr = this->shared_from_this();
334
335	✗	this->register_op(
336	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
337	✗	this->desc_state_.write_cancel_pending);
338	✗	return std::noop_coroutine();
339		}
340
341		// do_recv_from
342
343		template<
344		class Derived,
345		class Service,
346		class ConnOp,
347		class SendToOp,
348		class RecvFromOp,
349		class SendOp,
350		class RecvOp,
351		class DescState>
352		std::coroutine_handle<>
353	32x	reactor_datagram_socket<
354		Derived,
355		Service,
356		ConnOp,
357		SendToOp,
358		RecvFromOp,
359		SendOp,
360		RecvOp,
361		DescState>::
362		do_recv_from(
363		std::coroutine_handle<> h,
364		capy::executor_ref ex,
365		buffer_param param,
366		endpoint* source,
367		std::stop_token const& token,
368		std::error_code* ec,
369		std::size_t* bytes_out)
370		{
371	32x	auto& op = rd_;
372	32x	op.reset();
373
374	32x	capy::mutable_buffer bufs[RecvFromOp::max_buffers];
375	32x	op.iovec_count =
376	32x	static_cast<int>(param.copy_to(bufs, RecvFromOp::max_buffers));
377
378	32x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
379		{
380	2x	op.h = h;
381	2x	op.ex = ex;
382	2x	op.ec_out = ec;
383	2x	op.bytes_out = bytes_out;
384	2x	op.start(token, static_cast<Derived*>(this));
385	2x	op.impl_ptr = this->shared_from_this();
386	2x	op.complete(0, 0);
387	2x	this->svc_.post(&op);
388	2x	return std::noop_coroutine();
389		}
390
391	60x	for (int i = 0; i < op.iovec_count; ++i)
392		{
393	30x	op.iovecs[i].iov_base = bufs[i].data();
394	30x	op.iovecs[i].iov_len = bufs[i].size();
395		}
396
397	30x	op.fd = this->fd_;
398	30x	op.source_out = source;
399
400		// Speculative recvmsg
401	30x	msghdr msg{};
402	30x	msg.msg_name = &op.source_storage;
403	30x	msg.msg_namelen = sizeof(op.source_storage);
404	30x	msg.msg_iov = op.iovecs;
405	30x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
406
407		ssize_t n;
408		do
409		{
410	30x	n = ::recvmsg(this->fd_, &msg, 0);
411		}
412	30x	while (n < 0 && errno == EINTR);
413
414	30x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
415		{
416	22x	int err = (n < 0) ? errno : 0;
417	22x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
418
419	22x	if (this->svc_.scheduler().try_consume_inline_budget())
420		{
421	18x	*ec = err ? make_err(err) : std::error_code{};
422	18x	*bytes_out = bytes;
423	18x	if (source && !err && n >= 0)
424	18x	*source = from_sockaddr(op.source_storage);
425	18x	return dispatch_coro(ex, h);
426		}
427	4x	op.h = h;
428	4x	op.ex = ex;
429	4x	op.ec_out = ec;
430	4x	op.bytes_out = bytes_out;
431	4x	op.start(token, static_cast<Derived*>(this));
432	4x	op.impl_ptr = this->shared_from_this();
433	4x	op.complete(err, bytes);
434	4x	this->svc_.post(&op);
435	4x	return std::noop_coroutine();
436		}
437
438		// EAGAIN — register with reactor
439	8x	op.h = h;
440	8x	op.ex = ex;
441	8x	op.ec_out = ec;
442	8x	op.bytes_out = bytes_out;
443	8x	op.start(token, static_cast<Derived*>(this));
444	8x	op.impl_ptr = this->shared_from_this();
445
446	8x	this->register_op(
447	8x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
448	8x	this->desc_state_.read_cancel_pending);
449	8x	return std::noop_coroutine();
450		}
451
452		// do_connect
453
454		template<
455		class Derived,
456		class Service,
457		class ConnOp,
458		class SendToOp,
459		class RecvFromOp,
460		class SendOp,
461		class RecvOp,
462		class DescState>
463		std::coroutine_handle<>
464	12x	reactor_datagram_socket<
465		Derived,
466		Service,
467		ConnOp,
468		SendToOp,
469		RecvFromOp,
470		SendOp,
471		RecvOp,
472		DescState>::
473		do_connect(
474		std::coroutine_handle<> h,
475		capy::executor_ref ex,
476		endpoint ep,
477		std::stop_token const& token,
478		std::error_code* ec)
479		{
480	12x	auto& op = conn_;
481
482	12x	sockaddr_storage storage{};
483	12x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
484		int result =
485	12x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
486
487	12x	if (result == 0)
488		{
489	12x	sockaddr_storage local_storage{};
490	12x	socklen_t local_len = sizeof(local_storage);
491	12x	if (::getsockname(
492		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
493	12x	&local_len) == 0)
494	12x	this->local_endpoint_ = from_sockaddr(local_storage);
495	12x	remote_endpoint_ = ep;
496		}
497
498	12x	if (result == 0 \|\| errno != EINPROGRESS)
499		{
500	12x	int err = (result < 0) ? errno : 0;
501	12x	if (this->svc_.scheduler().try_consume_inline_budget())
502		{
503	2x	*ec = err ? make_err(err) : std::error_code{};
504	2x	return dispatch_coro(ex, h);
505		}
506	10x	op.reset();
507	10x	op.h = h;
508	10x	op.ex = ex;
509	10x	op.ec_out = ec;
510	10x	op.fd = this->fd_;
511	10x	op.target_endpoint = ep;
512	10x	op.start(token, static_cast<Derived*>(this));
513	10x	op.impl_ptr = this->shared_from_this();
514	10x	op.complete(err, 0);
515	10x	this->svc_.post(&op);
516	10x	return std::noop_coroutine();
517		}
518
519		// EINPROGRESS — register with reactor
520	✗	op.reset();
521	✗	op.h = h;
522	✗	op.ex = ex;
523	✗	op.ec_out = ec;
524	✗	op.fd = this->fd_;
525	✗	op.target_endpoint = ep;
526	✗	op.start(token, static_cast<Derived*>(this));
527	✗	op.impl_ptr = this->shared_from_this();
528
529	✗	this->register_op(
530	✗	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
531	✗	this->desc_state_.connect_cancel_pending);
532	✗	return std::noop_coroutine();
533		}
534
535		// do_send (connected mode)
536
537		template<
538		class Derived,
539		class Service,
540		class ConnOp,
541		class SendToOp,
542		class RecvFromOp,
543		class SendOp,
544		class RecvOp,
545		class DescState>
546		std::coroutine_handle<>
547	6x	reactor_datagram_socket<
548		Derived,
549		Service,
550		ConnOp,
551		SendToOp,
552		RecvFromOp,
553		SendOp,
554		RecvOp,
555		DescState>::
556		do_send(
557		std::coroutine_handle<> h,
558		capy::executor_ref ex,
559		buffer_param param,
560		std::stop_token const& token,
561		std::error_code* ec,
562		std::size_t* bytes_out)
563		{
564	6x	auto& op = send_wr_;
565	6x	op.reset();
566
567	6x	capy::mutable_buffer bufs[SendOp::max_buffers];
568	6x	op.iovec_count = static_cast<int>(param.copy_to(bufs, SendOp::max_buffers));
569
570	12x	for (int i = 0; i < op.iovec_count; ++i)
571		{
572	6x	op.iovecs[i].iov_base = bufs[i].data();
573	6x	op.iovecs[i].iov_len = bufs[i].size();
574		}
575
576	6x	op.fd = this->fd_;
577
578		// Speculative sendmsg with no destination (connected mode)
579	6x	msghdr msg{};
580	6x	msg.msg_iov = op.iovecs;
581	6x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
582
583		#ifdef MSG_NOSIGNAL
584	6x	constexpr int send_flags = MSG_NOSIGNAL;
585		#else
586		constexpr int send_flags = 0;
587		#endif
588
589		ssize_t n;
590		do
591		{
592	6x	n = ::sendmsg(this->fd_, &msg, send_flags);
593		}
594	6x	while (n < 0 && errno == EINTR);
595
596	6x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
597		{
598	6x	int err = (n < 0) ? errno : 0;
599	6x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
600
601	6x	if (this->svc_.scheduler().try_consume_inline_budget())
602		{
603	6x	*ec = err ? make_err(err) : std::error_code{};
604	6x	*bytes_out = bytes;
605	6x	return dispatch_coro(ex, h);
606		}
607	✗	op.h = h;
608	✗	op.ex = ex;
609	✗	op.ec_out = ec;
610	✗	op.bytes_out = bytes_out;
611	✗	op.start(token, static_cast<Derived*>(this));
612	✗	op.impl_ptr = this->shared_from_this();
613	✗	op.complete(err, bytes);
614	✗	this->svc_.post(&op);
615	✗	return std::noop_coroutine();
616		}
617
618		// EAGAIN — register with reactor
619	✗	op.h = h;
620	✗	op.ex = ex;
621	✗	op.ec_out = ec;
622	✗	op.bytes_out = bytes_out;
623	✗	op.start(token, static_cast<Derived*>(this));
624	✗	op.impl_ptr = this->shared_from_this();
625
626	✗	this->register_op(
627	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
628	✗	this->desc_state_.write_cancel_pending);
629	✗	return std::noop_coroutine();
630		}
631
632		// do_recv (connected mode)
633
634		template<
635		class Derived,
636		class Service,
637		class ConnOp,
638		class SendToOp,
639		class RecvFromOp,
640		class SendOp,
641		class RecvOp,
642		class DescState>
643		std::coroutine_handle<>
644	4x	reactor_datagram_socket<
645		Derived,
646		Service,
647		ConnOp,
648		SendToOp,
649		RecvFromOp,
650		SendOp,
651		RecvOp,
652		DescState>::
653		do_recv(
654		std::coroutine_handle<> h,
655		capy::executor_ref ex,
656		buffer_param param,
657		std::stop_token const& token,
658		std::error_code* ec,
659		std::size_t* bytes_out)
660		{
661	4x	auto& op = recv_rd_;
662	4x	op.reset();
663
664	4x	capy::mutable_buffer bufs[RecvOp::max_buffers];
665	4x	op.iovec_count = static_cast<int>(param.copy_to(bufs, RecvOp::max_buffers));
666
667	4x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
668		{
669	✗	op.h = h;
670	✗	op.ex = ex;
671	✗	op.ec_out = ec;
672	✗	op.bytes_out = bytes_out;
673	✗	op.start(token, static_cast<Derived*>(this));
674	✗	op.impl_ptr = this->shared_from_this();
675	✗	op.complete(0, 0);
676	✗	this->svc_.post(&op);
677	✗	return std::noop_coroutine();
678		}
679
680	8x	for (int i = 0; i < op.iovec_count; ++i)
681		{
682	4x	op.iovecs[i].iov_base = bufs[i].data();
683	4x	op.iovecs[i].iov_len = bufs[i].size();
684		}
685
686	4x	op.fd = this->fd_;
687
688		// Speculative recvmsg with no source (connected mode)
689	4x	msghdr msg{};
690	4x	msg.msg_iov = op.iovecs;
691	4x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
692
693		ssize_t n;
694		do
695		{
696	4x	n = ::recvmsg(this->fd_, &msg, 0);
697		}
698	4x	while (n < 0 && errno == EINTR);
699
700	4x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
701		{
702	2x	int err = (n < 0) ? errno : 0;
703	2x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
704
705	2x	if (this->svc_.scheduler().try_consume_inline_budget())
706		{
707	✗	*ec = err ? make_err(err) : std::error_code{};
708	✗	*bytes_out = bytes;
709	✗	return dispatch_coro(ex, h);
710		}
711	2x	op.h = h;
712	2x	op.ex = ex;
713	2x	op.ec_out = ec;
714	2x	op.bytes_out = bytes_out;
715	2x	op.start(token, static_cast<Derived*>(this));
716	2x	op.impl_ptr = this->shared_from_this();
717	2x	op.complete(err, bytes);
718	2x	this->svc_.post(&op);
719	2x	return std::noop_coroutine();
720		}
721
722		// EAGAIN — register with reactor
723	2x	op.h = h;
724	2x	op.ex = ex;
725	2x	op.ec_out = ec;
726	2x	op.bytes_out = bytes_out;
727	2x	op.start(token, static_cast<Derived*>(this));
728	2x	op.impl_ptr = this->shared_from_this();
729
730	2x	this->register_op(
731	2x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
732	2x	this->desc_state_.read_cancel_pending);
733	2x	return std::noop_coroutine();
734		}
735
736		} // namespace boost::corosio::detail
737
738		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
739