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

56.0% Lines (416/743) 60.9% List of functions (28/46)

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

reactor_datagram_socket.hpp

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	op.cont_op.cont.h = h;
315	6x	return dispatch_coro(ex, op.cont_op.cont);
316		}
317	16x	op.h = h;
318	16x	op.ex = ex;
319	16x	op.ec_out = ec;
320	16x	op.bytes_out = bytes_out;
321	16x	op.start(token, static_cast<Derived*>(this));
322	16x	op.impl_ptr = this->shared_from_this();
323	16x	op.complete(err, bytes);
324	16x	this->svc_.post(&op);
325	16x	return std::noop_coroutine();
326		}
327
328		// EAGAIN — register with reactor
329	✗	op.h = h;
330	✗	op.ex = ex;
331	✗	op.ec_out = ec;
332	✗	op.bytes_out = bytes_out;
333	✗	op.start(token, static_cast<Derived*>(this));
334	✗	op.impl_ptr = this->shared_from_this();
335
336	✗	this->register_op(
337	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
338	✗	this->desc_state_.write_cancel_pending);
339	✗	return std::noop_coroutine();
340		}
341
342		// do_recv_from
343
344		template<
345		class Derived,
346		class Service,
347		class ConnOp,
348		class SendToOp,
349		class RecvFromOp,
350		class SendOp,
351		class RecvOp,
352		class DescState>
353		std::coroutine_handle<>
354	32x	reactor_datagram_socket<
355		Derived,
356		Service,
357		ConnOp,
358		SendToOp,
359		RecvFromOp,
360		SendOp,
361		RecvOp,
362		DescState>::
363		do_recv_from(
364		std::coroutine_handle<> h,
365		capy::executor_ref ex,
366		buffer_param param,
367		endpoint* source,
368		std::stop_token const& token,
369		std::error_code* ec,
370		std::size_t* bytes_out)
371		{
372	32x	auto& op = rd_;
373	32x	op.reset();
374
375	32x	capy::mutable_buffer bufs[RecvFromOp::max_buffers];
376	32x	op.iovec_count =
377	32x	static_cast<int>(param.copy_to(bufs, RecvFromOp::max_buffers));
378
379	32x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
380		{
381	2x	op.h = h;
382	2x	op.ex = ex;
383	2x	op.ec_out = ec;
384	2x	op.bytes_out = bytes_out;
385	2x	op.start(token, static_cast<Derived*>(this));
386	2x	op.impl_ptr = this->shared_from_this();
387	2x	op.complete(0, 0);
388	2x	this->svc_.post(&op);
389	2x	return std::noop_coroutine();
390		}
391
392	60x	for (int i = 0; i < op.iovec_count; ++i)
393		{
394	30x	op.iovecs[i].iov_base = bufs[i].data();
395	30x	op.iovecs[i].iov_len = bufs[i].size();
396		}
397
398	30x	op.fd = this->fd_;
399	30x	op.source_out = source;
400
401		// Speculative recvmsg
402	30x	msghdr msg{};
403	30x	msg.msg_name = &op.source_storage;
404	30x	msg.msg_namelen = sizeof(op.source_storage);
405	30x	msg.msg_iov = op.iovecs;
406	30x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
407
408		ssize_t n;
409		do
410		{
411	30x	n = ::recvmsg(this->fd_, &msg, 0);
412		}
413	30x	while (n < 0 && errno == EINTR);
414
415	30x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
416		{
417	22x	int err = (n < 0) ? errno : 0;
418	22x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
419
420	22x	if (this->svc_.scheduler().try_consume_inline_budget())
421		{
422	18x	*ec = err ? make_err(err) : std::error_code{};
423	18x	*bytes_out = bytes;
424	18x	if (source && !err && n >= 0)
425	18x	*source = from_sockaddr(op.source_storage);
426	18x	op.cont_op.cont.h = h;
427	18x	return dispatch_coro(ex, op.cont_op.cont);
428		}
429	4x	op.h = h;
430	4x	op.ex = ex;
431	4x	op.ec_out = ec;
432	4x	op.bytes_out = bytes_out;
433	4x	op.start(token, static_cast<Derived*>(this));
434	4x	op.impl_ptr = this->shared_from_this();
435	4x	op.complete(err, bytes);
436	4x	this->svc_.post(&op);
437	4x	return std::noop_coroutine();
438		}
439
440		// EAGAIN — register with reactor
441	8x	op.h = h;
442	8x	op.ex = ex;
443	8x	op.ec_out = ec;
444	8x	op.bytes_out = bytes_out;
445	8x	op.start(token, static_cast<Derived*>(this));
446	8x	op.impl_ptr = this->shared_from_this();
447
448	8x	this->register_op(
449	8x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
450	8x	this->desc_state_.read_cancel_pending);
451	8x	return std::noop_coroutine();
452		}
453
454		// do_connect
455
456		template<
457		class Derived,
458		class Service,
459		class ConnOp,
460		class SendToOp,
461		class RecvFromOp,
462		class SendOp,
463		class RecvOp,
464		class DescState>
465		std::coroutine_handle<>
466	12x	reactor_datagram_socket<
467		Derived,
468		Service,
469		ConnOp,
470		SendToOp,
471		RecvFromOp,
472		SendOp,
473		RecvOp,
474		DescState>::
475		do_connect(
476		std::coroutine_handle<> h,
477		capy::executor_ref ex,
478		endpoint ep,
479		std::stop_token const& token,
480		std::error_code* ec)
481		{
482	12x	auto& op = conn_;
483
484	12x	sockaddr_storage storage{};
485	12x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
486		int result =
487	12x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
488
489	12x	if (result == 0)
490		{
491	12x	sockaddr_storage local_storage{};
492	12x	socklen_t local_len = sizeof(local_storage);
493	12x	if (::getsockname(
494		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
495	12x	&local_len) == 0)
496	12x	this->local_endpoint_ = from_sockaddr(local_storage);
497	12x	remote_endpoint_ = ep;
498		}
499
500	12x	if (result == 0 \|\| errno != EINPROGRESS)
501		{
502	12x	int err = (result < 0) ? errno : 0;
503	12x	if (this->svc_.scheduler().try_consume_inline_budget())
504		{
505	2x	*ec = err ? make_err(err) : std::error_code{};
506	2x	op.cont_op.cont.h = h;
507	2x	return dispatch_coro(ex, op.cont_op.cont);
508		}
509	10x	op.reset();
510	10x	op.h = h;
511	10x	op.ex = ex;
512	10x	op.ec_out = ec;
513	10x	op.fd = this->fd_;
514	10x	op.target_endpoint = ep;
515	10x	op.start(token, static_cast<Derived*>(this));
516	10x	op.impl_ptr = this->shared_from_this();
517	10x	op.complete(err, 0);
518	10x	this->svc_.post(&op);
519	10x	return std::noop_coroutine();
520		}
521
522		// EINPROGRESS — register with reactor
523	✗	op.reset();
524	✗	op.h = h;
525	✗	op.ex = ex;
526	✗	op.ec_out = ec;
527	✗	op.fd = this->fd_;
528	✗	op.target_endpoint = ep;
529	✗	op.start(token, static_cast<Derived*>(this));
530	✗	op.impl_ptr = this->shared_from_this();
531
532	✗	this->register_op(
533	✗	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
534	✗	this->desc_state_.connect_cancel_pending);
535	✗	return std::noop_coroutine();
536		}
537
538		// do_send (connected mode)
539
540		template<
541		class Derived,
542		class Service,
543		class ConnOp,
544		class SendToOp,
545		class RecvFromOp,
546		class SendOp,
547		class RecvOp,
548		class DescState>
549		std::coroutine_handle<>
550	6x	reactor_datagram_socket<
551		Derived,
552		Service,
553		ConnOp,
554		SendToOp,
555		RecvFromOp,
556		SendOp,
557		RecvOp,
558		DescState>::
559		do_send(
560		std::coroutine_handle<> h,
561		capy::executor_ref ex,
562		buffer_param param,
563		std::stop_token const& token,
564		std::error_code* ec,
565		std::size_t* bytes_out)
566		{
567	6x	auto& op = send_wr_;
568	6x	op.reset();
569
570	6x	capy::mutable_buffer bufs[SendOp::max_buffers];
571	6x	op.iovec_count = static_cast<int>(param.copy_to(bufs, SendOp::max_buffers));
572
573	12x	for (int i = 0; i < op.iovec_count; ++i)
574		{
575	6x	op.iovecs[i].iov_base = bufs[i].data();
576	6x	op.iovecs[i].iov_len = bufs[i].size();
577		}
578
579	6x	op.fd = this->fd_;
580
581		// Speculative sendmsg with no destination (connected mode)
582	6x	msghdr msg{};
583	6x	msg.msg_iov = op.iovecs;
584	6x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
585
586		#ifdef MSG_NOSIGNAL
587	6x	constexpr int send_flags = MSG_NOSIGNAL;
588		#else
589		constexpr int send_flags = 0;
590		#endif
591
592		ssize_t n;
593		do
594		{
595	6x	n = ::sendmsg(this->fd_, &msg, send_flags);
596		}
597	6x	while (n < 0 && errno == EINTR);
598
599	6x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
600		{
601	6x	int err = (n < 0) ? errno : 0;
602	6x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
603
604	6x	if (this->svc_.scheduler().try_consume_inline_budget())
605		{
606	6x	*ec = err ? make_err(err) : std::error_code{};
607	6x	*bytes_out = bytes;
608	6x	op.cont_op.cont.h = h;
609	6x	return dispatch_coro(ex, op.cont_op.cont);
610		}
611	✗	op.h = h;
612	✗	op.ex = ex;
613	✗	op.ec_out = ec;
614	✗	op.bytes_out = bytes_out;
615	✗	op.start(token, static_cast<Derived*>(this));
616	✗	op.impl_ptr = this->shared_from_this();
617	✗	op.complete(err, bytes);
618	✗	this->svc_.post(&op);
619	✗	return std::noop_coroutine();
620		}
621
622		// EAGAIN — register with reactor
623	✗	op.h = h;
624	✗	op.ex = ex;
625	✗	op.ec_out = ec;
626	✗	op.bytes_out = bytes_out;
627	✗	op.start(token, static_cast<Derived*>(this));
628	✗	op.impl_ptr = this->shared_from_this();
629
630	✗	this->register_op(
631	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
632	✗	this->desc_state_.write_cancel_pending);
633	✗	return std::noop_coroutine();
634		}
635
636		// do_recv (connected mode)
637
638		template<
639		class Derived,
640		class Service,
641		class ConnOp,
642		class SendToOp,
643		class RecvFromOp,
644		class SendOp,
645		class RecvOp,
646		class DescState>
647		std::coroutine_handle<>
648	4x	reactor_datagram_socket<
649		Derived,
650		Service,
651		ConnOp,
652		SendToOp,
653		RecvFromOp,
654		SendOp,
655		RecvOp,
656		DescState>::
657		do_recv(
658		std::coroutine_handle<> h,
659		capy::executor_ref ex,
660		buffer_param param,
661		std::stop_token const& token,
662		std::error_code* ec,
663		std::size_t* bytes_out)
664		{
665	4x	auto& op = recv_rd_;
666	4x	op.reset();
667
668	4x	capy::mutable_buffer bufs[RecvOp::max_buffers];
669	4x	op.iovec_count = static_cast<int>(param.copy_to(bufs, RecvOp::max_buffers));
670
671	4x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
672		{
673	✗	op.h = h;
674	✗	op.ex = ex;
675	✗	op.ec_out = ec;
676	✗	op.bytes_out = bytes_out;
677	✗	op.start(token, static_cast<Derived*>(this));
678	✗	op.impl_ptr = this->shared_from_this();
679	✗	op.complete(0, 0);
680	✗	this->svc_.post(&op);
681	✗	return std::noop_coroutine();
682		}
683
684	8x	for (int i = 0; i < op.iovec_count; ++i)
685		{
686	4x	op.iovecs[i].iov_base = bufs[i].data();
687	4x	op.iovecs[i].iov_len = bufs[i].size();
688		}
689
690	4x	op.fd = this->fd_;
691
692		// Speculative recvmsg with no source (connected mode)
693	4x	msghdr msg{};
694	4x	msg.msg_iov = op.iovecs;
695	4x	msg.msg_iovlen = static_cast<std::size_t>(op.iovec_count);
696
697		ssize_t n;
698		do
699		{
700	4x	n = ::recvmsg(this->fd_, &msg, 0);
701		}
702	4x	while (n < 0 && errno == EINTR);
703
704	4x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
705		{
706	2x	int err = (n < 0) ? errno : 0;
707	2x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
708
709	2x	if (this->svc_.scheduler().try_consume_inline_budget())
710		{
711	✗	*ec = err ? make_err(err) : std::error_code{};
712	✗	*bytes_out = bytes;
713	✗	op.cont_op.cont.h = h;
714	✗	return dispatch_coro(ex, op.cont_op.cont);
715		}
716	2x	op.h = h;
717	2x	op.ex = ex;
718	2x	op.ec_out = ec;
719	2x	op.bytes_out = bytes_out;
720	2x	op.start(token, static_cast<Derived*>(this));
721	2x	op.impl_ptr = this->shared_from_this();
722	2x	op.complete(err, bytes);
723	2x	this->svc_.post(&op);
724	2x	return std::noop_coroutine();
725		}
726
727		// EAGAIN — register with reactor
728	2x	op.h = h;
729	2x	op.ex = ex;
730	2x	op.ec_out = ec;
731	2x	op.bytes_out = bytes_out;
732	2x	op.start(token, static_cast<Derived*>(this));
733	2x	op.impl_ptr = this->shared_from_this();
734
735	2x	this->register_op(
736	2x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
737	2x	this->desc_state_.read_cancel_pending);
738	2x	return std::noop_coroutine();
739		}
740
741		} // namespace boost::corosio::detail
742
743		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DATAGRAM_SOCKET_HPP
744