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

98.6% Lines (279/0/283) 98.9% List of functions (92/0/93)

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

reactor_stream_socket.hpp

f(x) Functions (93)

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_STREAM_SOCKET_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
12
13		#include <boost/corosio/tcp_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/detail/dispatch_coro.hpp>
19		#include <boost/capy/buffers.hpp>
20
21		#include <coroutine>
22
23		#include <errno.h>
24		#include <sys/socket.h>
25		#include <sys/uio.h>
26
27		namespace boost::corosio::detail {
28
29		/** CRTP base for reactor-backed stream socket implementations.
30
31		Inherits shared data members and cancel/close/register logic
32		from reactor_basic_socket. Adds the stream-specific remote
33		endpoint, shutdown, and I/O dispatch (connect, read, write, wait).
34
35		@tparam Derived The concrete socket type (CRTP).
36		@tparam Service The backend's socket service type.
37		@tparam ConnOp The backend's connect op type.
38		@tparam ReadOp The backend's read op type.
39		@tparam WriteOp The backend's write op type.
40		@tparam WaitOp The backend's wait op type.
41		@tparam DescState The backend's descriptor_state type.
42		@tparam ImplBase The public vtable base
43		(tcp_socket::implementation or
44		local_stream_socket::implementation).
45		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
46		*/
47		template<
48		class Derived,
49		class Service,
50		class ConnOp,
51		class ReadOp,
52		class WriteOp,
53		class WaitOp,
54		class DescState,
55		class ImplBase = tcp_socket::implementation,
56		class Endpoint = endpoint>
57		class reactor_stream_socket
58		: public reactor_basic_socket<
59		Derived,
60		ImplBase,
61		Service,
62		DescState,
63		Endpoint>
64		{
65		using base_type = reactor_basic_socket<
66		Derived,
67		ImplBase,
68		Service,
69		DescState,
70		Endpoint>;
71		using self_type = reactor_stream_socket<
72		Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp,
73		DescState, ImplBase, Endpoint>;
74		friend base_type;
75		friend Derived;
76
77		protected:
78		// NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
79	20259x	explicit reactor_stream_socket(Service& svc) noexcept : base_type(svc) {}
80
81		protected:
82		Endpoint remote_endpoint_;
83
84		public:
85		/// Pending connect operation slot.
86		ConnOp conn_;
87
88		/// Pending read operation slot.
89		ReadOp rd_;
90
91		/// Pending write operation slot.
92		WriteOp wr_;
93
94		/// Pending wait-for-read operation slot.
95		WaitOp wait_rd_;
96
97		/// Pending wait-for-write operation slot.
98		WaitOp wait_wr_;
99
100		/// Pending wait-for-error operation slot.
101		WaitOp wait_er_;
102
103	20259x	~reactor_stream_socket() override = default;
104
105		/// Return the cached remote endpoint.
106	62x	Endpoint remote_endpoint() const noexcept override
107		{
108	62x	return remote_endpoint_;
109		}
110
111		// --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
112
113	6616x	std::coroutine_handle<> connect(
114		std::coroutine_handle<> h,
115		capy::executor_ref ex,
116		Endpoint ep,
117		std::stop_token token,
118		std::error_code* ec) override
119		{
120	6616x	return do_connect(h, ex, ep, token, ec);
121		}
122
123	214835x	std::coroutine_handle<> read_some(
124		std::coroutine_handle<> h,
125		capy::executor_ref ex,
126		buffer_param param,
127		std::stop_token token,
128		std::error_code* ec,
129		std::size_t* bytes_out) override
130		{
131	214835x	return do_read_some(h, ex, param, token, ec, bytes_out);
132		}
133
134	214143x	std::coroutine_handle<> write_some(
135		std::coroutine_handle<> h,
136		capy::executor_ref ex,
137		buffer_param param,
138		std::stop_token token,
139		std::error_code* ec,
140		std::size_t* bytes_out) override
141		{
142	214143x	return do_write_some(h, ex, param, token, ec, bytes_out);
143		}
144
145	92x	std::coroutine_handle<> wait(
146		std::coroutine_handle<> h,
147		capy::executor_ref ex,
148		wait_type w,
149		std::stop_token token,
150		std::error_code* ec) override
151		{
152	92x	return do_wait(h, ex, w, token, ec);
153		}
154
155		std::error_code
156	25x	shutdown(corosio::shutdown_type what) noexcept override
157		{
158	25x	return do_shutdown(static_cast<int>(what));
159		}
160
161	219x	void cancel() noexcept override
162		{
163	219x	this->do_cancel();
164	219x	}
165
166		// --- End virtual overrides ---
167
168		/// Close the socket (non-virtual, called by the service).
169		void close_socket() noexcept
170		{
171		this->do_close_socket();
172		}
173
174		/** Shut down part or all of the full-duplex connection.
175
176		@param what 0 = receive, 1 = send, 2 = both.
177		*/
178	25x	std::error_code do_shutdown(int what) noexcept
179		{
180		int how;
181	25x	switch (what)
182		{
183	4x	case 0: // shutdown_receive
184	4x	how = SHUT_RD;
185	4x	break;
186	17x	case 1: // shutdown_send
187	17x	how = SHUT_WR;
188	17x	break;
189	4x	case 2: // shutdown_both
190	4x	how = SHUT_RDWR;
191	4x	break;
192	✗	default:
193	✗	return make_err(EINVAL);
194		}
195	25x	if (::shutdown(this->fd_, how) != 0)
196	2x	return make_err(errno);
197	23x	return {};
198		}
199
200		/// Cache local and remote endpoints.
201	13273x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
202		{
203	13273x	this->local_endpoint_ = std::move(local);
204	13273x	remote_endpoint_ = std::move(remote);
205	13273x	}
206
207		/** Shared connect dispatch.
208
209		Tries the connect syscall speculatively. On synchronous
210		completion, returns via inline budget or posts through queue.
211		On EINPROGRESS, registers with the reactor.
212		*/
213		std::coroutine_handle<> do_connect(
214		std::coroutine_handle<>,
215		capy::executor_ref,
216		Endpoint const&,
217		std::stop_token const&,
218		std::error_code*);
219
220		/** Shared scatter-read dispatch.
221
222		Tries readv() speculatively. On success or hard error,
223		returns via inline budget or posts through queue.
224		On EAGAIN, registers with the reactor.
225		*/
226		std::coroutine_handle<> do_read_some(
227		std::coroutine_handle<>,
228		capy::executor_ref,
229		buffer_param,
230		std::stop_token const&,
231		std::error_code*,
232		std::size_t*);
233
234		/** Shared gather-write dispatch.
235
236		Tries the write via WriteOp::write_policy speculatively.
237		On success or hard error, returns via inline budget or
238		posts through queue. On EAGAIN, registers with the reactor.
239		*/
240		std::coroutine_handle<> do_write_some(
241		std::coroutine_handle<>,
242		capy::executor_ref,
243		buffer_param,
244		std::stop_token const&,
245		std::error_code*,
246		std::size_t*);
247
248		/** Shared readiness-wait dispatch.
249
250		Every wait type probes the descriptor with a zero-timeout
251		`poll()` and completes at once if the condition already
252		holds; otherwise the op re-probes under the descriptor mutex
253		and parks, completing when a reactor event arrives and a
254		fresh probe confirms the condition. A write wait therefore
255		completes only while a non-blocking write can make progress.
256		*/
257		std::coroutine_handle<> do_wait(
258		std::coroutine_handle<>,
259		capy::executor_ref,
260		wait_type,
261		std::stop_token const&,
262		std::error_code*);
263
264		/** Close the socket and cancel pending operations.
265
266		Extends the base do_close_socket() to also reset
267		the remote endpoint.
268		*/
269	60784x	void do_close_socket() noexcept
270		{
271	60784x	base_type::do_close_socket();
272	60784x	remote_endpoint_ = Endpoint{};
273	60784x	}
274
275		/// Release ownership of the descriptor and drop the cached peer.
276	8x	native_handle_type do_release_socket() noexcept
277		{
278	8x	auto fd = base_type::do_release_socket();
279	8x	remote_endpoint_ = Endpoint{};
280	8x	return fd;
281		}
282
283		private:
284		// CRTP callbacks for reactor_basic_socket cancel/close
285
286		template<class Op>
287	235x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
288		{
289	235x	if (&op == static_cast<void*>(&conn_))
290	5x	return &this->desc_state_.connect_op;
291	230x	if (&op == static_cast<void*>(&rd_))
292	211x	return &this->desc_state_.read_op;
293	19x	if (&op == static_cast<void*>(&wr_))
294	4x	return &this->desc_state_.write_op;
295	15x	if (&op == static_cast<void*>(&wait_rd_))
296	11x	return &this->desc_state_.wait_read_op;
297	4x	if (&op == static_cast<void*>(&wait_wr_))
298	2x	return &this->desc_state_.wait_write_op;
299	2x	if (&op == static_cast<void*>(&wait_er_))
300	2x	return &this->desc_state_.wait_error_op;
301	✗	return nullptr;
302		}
303
304
305		template<class Fn>
306	61011x	void for_each_op(Fn fn) noexcept
307		{
308	61011x	fn(conn_);
309	61011x	fn(rd_);
310	61011x	fn(wr_);
311	61011x	fn(wait_rd_);
312	61011x	fn(wait_wr_);
313	61011x	fn(wait_er_);
314	61011x	}
315
316		template<class Fn>
317	61011x	void for_each_desc_entry(Fn fn) noexcept
318		{
319	61011x	fn(conn_, this->desc_state_.connect_op);
320	61011x	fn(rd_, this->desc_state_.read_op);
321	61011x	fn(wr_, this->desc_state_.write_op);
322	61011x	fn(wait_rd_, this->desc_state_.wait_read_op);
323	61011x	fn(wait_wr_, this->desc_state_.wait_write_op);
324	61011x	fn(wait_er_, this->desc_state_.wait_error_op);
325	61011x	}
326		};
327
328		template<
329		class Derived,
330		class Service,
331		class ConnOp,
332		class ReadOp,
333		class WriteOp,
334		class WaitOp,
335		class DescState,
336		class ImplBase,
337		class Endpoint>
338		std::coroutine_handle<>
339	6616x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
340		do_connect(
341		std::coroutine_handle<> h,
342		capy::executor_ref ex,
343		Endpoint const& ep,
344		std::stop_token const& token,
345		std::error_code* ec)
346		{
347	6616x	auto& op = conn_;
348
349	6616x	sockaddr_storage storage{};
350	6616x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
351		int result =
352	6616x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
353
354	6616x	if (result == 0)
355		{
356	31x	sockaddr_storage local_storage{};
357	31x	socklen_t local_len = sizeof(local_storage);
358	31x	if (::getsockname(
359		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
360	31x	&local_len) == 0)
361	✗	this->local_endpoint_ =
362	31x	from_sockaddr_as(local_storage, local_len, Endpoint{});
363	31x	remote_endpoint_ = ep;
364		}
365
366	6616x	if (result == 0 \|\| errno != EINPROGRESS)
367		{
368	43x	int err = (result < 0) ? errno : 0;
369	43x	if (this->svc_.scheduler().try_consume_inline_budget())
370		{
371	4x	*ec = err ? make_err(err) : std::error_code{};
372	4x	op.cont.h = h;
373	4x	return dispatch_coro(ex, op.cont);
374		}
375	39x	op.reset();
376	39x	op.h = h;
377	39x	op.ex = ex;
378	39x	op.ec_out = ec;
379	39x	op.fd = this->fd_;
380	39x	op.target_endpoint = ep;
381	39x	op.start(token, static_cast<Derived*>(this));
382	39x	op.impl_ptr = this->shared_from_this();
383	39x	op.complete(err, 0);
384	39x	this->svc_.post(&op);
385	39x	return std::noop_coroutine();
386		}
387
388		// EINPROGRESS — register with reactor
389	6573x	op.reset();
390	6573x	op.h = h;
391	6573x	op.ex = ex;
392	6573x	op.ec_out = ec;
393	6573x	op.fd = this->fd_;
394	6573x	op.target_endpoint = ep;
395	6573x	op.start(token, static_cast<Derived*>(this));
396	6573x	op.impl_ptr = this->shared_from_this();
397
398	6573x	this->register_op(
399	6573x	op, this->desc_state_.connect_op, this->desc_state_.write_ready, true);
400	6573x	return std::noop_coroutine();
401		}
402
403		template<
404		class Derived,
405		class Service,
406		class ConnOp,
407		class ReadOp,
408		class WriteOp,
409		class WaitOp,
410		class DescState,
411		class ImplBase,
412		class Endpoint>
413		std::coroutine_handle<>
414	214835x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
415		do_read_some(
416		std::coroutine_handle<> h,
417		capy::executor_ref ex,
418		buffer_param param,
419		std::stop_token const& token,
420		std::error_code* ec,
421		std::size_t* bytes_out)
422		{
423	214835x	auto& op = rd_;
424	214835x	op.reset();
425
426		// Closed-object contract: complete with bad_file_descriptor without
427		// touching the kernel or the unregistered descriptor state.
428	214835x	if (this->fd_ < 0)
429		{
430	8x	op.h = h;
431	8x	op.ex = ex;
432	8x	op.ec_out = ec;
433	8x	op.bytes_out = bytes_out;
434	8x	op.start(token, static_cast<Derived*>(this));
435	8x	op.impl_ptr = this->shared_from_this();
436	8x	op.complete(EBADF, 0);
437	8x	this->svc_.post(&op);
438	8x	return std::noop_coroutine();
439		}
440
441	214827x	capy::mutable_buffer bufs[ReadOp::max_buffers];
442	214827x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
443
444	214827x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
445		{
446	4x	op.empty_buffer_read = true;
447	4x	op.h = h;
448	4x	op.ex = ex;
449	4x	op.ec_out = ec;
450	4x	op.bytes_out = bytes_out;
451	4x	op.start(token, static_cast<Derived*>(this));
452	4x	op.impl_ptr = this->shared_from_this();
453	4x	op.complete(0, 0);
454	4x	this->svc_.post(&op);
455	4x	return std::noop_coroutine();
456		}
457
458	429664x	for (int i = 0; i < op.iovec_count; ++i)
459		{
460	214841x	op.iovecs[i].iov_base = bufs[i].data();
461	214841x	op.iovecs[i].iov_len = bufs[i].size();
462		}
463
464		// Speculative read; for the single-buffer case use recv() so the
465		// kernel skips the readv iov_iter setup.
466		ssize_t n;
467	214823x	if (op.iovec_count == 1)
468		{
469		do
470		{
471	214811x	n = ::recv(this->fd_, bufs[0].data(), bufs[0].size(), 0);
472		}
473	214811x	while (n < 0 && errno == EINTR);
474		}
475		else
476		{
477		do
478		{
479	16x	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
480		}
481	16x	while (n < 0 && errno == EINTR);
482		}
483
484	214823x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
485		{
486	214048x	int err = (n < 0) ? errno : 0;
487	214048x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
488
489	214048x	if (this->svc_.scheduler().try_consume_inline_budget())
490		{
491	171272x	if (err)
492	4x	*ec = make_err(err);
493	171268x	else if (n == 0)
494	15x	*ec = capy::error::eof;
495		else
496	171253x	*ec = {};
497	171272x	*bytes_out = bytes;
498	171272x	op.cont.h = h;
499	171272x	return dispatch_coro(ex, op.cont);
500		}
501	42776x	op.h = h;
502	42776x	op.ex = ex;
503	42776x	op.ec_out = ec;
504	42776x	op.bytes_out = bytes_out;
505	42776x	op.start(token, static_cast<Derived*>(this));
506	42776x	op.impl_ptr = this->shared_from_this();
507	42776x	op.complete(err, bytes);
508	42776x	this->svc_.post(&op);
509	42776x	return std::noop_coroutine();
510		}
511
512		// EAGAIN — register with reactor
513	775x	op.h = h;
514	775x	op.ex = ex;
515	775x	op.ec_out = ec;
516	775x	op.bytes_out = bytes_out;
517	775x	op.fd = this->fd_;
518	775x	op.start(token, static_cast<Derived*>(this));
519	775x	op.impl_ptr = this->shared_from_this();
520
521	775x	this->register_op(
522	775x	op, this->desc_state_.read_op, this->desc_state_.read_ready);
523	775x	return std::noop_coroutine();
524		}
525
526		template<
527		class Derived,
528		class Service,
529		class ConnOp,
530		class ReadOp,
531		class WriteOp,
532		class WaitOp,
533		class DescState,
534		class ImplBase,
535		class Endpoint>
536		std::coroutine_handle<>
537	214143x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
538		do_write_some(
539		std::coroutine_handle<> h,
540		capy::executor_ref ex,
541		buffer_param param,
542		std::stop_token const& token,
543		std::error_code* ec,
544		std::size_t* bytes_out)
545		{
546	214143x	auto& op = wr_;
547	214143x	op.reset();
548
549		// Closed-object contract: complete with bad_file_descriptor without
550		// touching the kernel or the unregistered descriptor state.
551	214143x	if (this->fd_ < 0)
552		{
553	8x	op.h = h;
554	8x	op.ex = ex;
555	8x	op.ec_out = ec;
556	8x	op.bytes_out = bytes_out;
557	8x	op.start(token, static_cast<Derived*>(this));
558	8x	op.impl_ptr = this->shared_from_this();
559	8x	op.complete(EBADF, 0);
560	8x	this->svc_.post(&op);
561	8x	return std::noop_coroutine();
562		}
563
564	214135x	capy::mutable_buffer bufs[WriteOp::max_buffers];
565	214135x	op.iovec_count =
566	214135x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
567
568	214135x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
569		{
570	4x	op.h = h;
571	4x	op.ex = ex;
572	4x	op.ec_out = ec;
573	4x	op.bytes_out = bytes_out;
574	4x	op.start(token, static_cast<Derived*>(this));
575	4x	op.impl_ptr = this->shared_from_this();
576	4x	op.complete(0, 0);
577	4x	this->svc_.post(&op);
578	4x	return std::noop_coroutine();
579		}
580
581	428278x	for (int i = 0; i < op.iovec_count; ++i)
582		{
583	214147x	op.iovecs[i].iov_base = bufs[i].data();
584	214147x	op.iovecs[i].iov_len = bufs[i].size();
585		}
586
587		// Speculative write; the single-buffer case dispatches to a
588		// backend-specific fast path so the kernel skips msghdr/iov_iter
589		// setup (and so each backend can pick the right SIGPIPE strategy).
590		ssize_t n;
591	214131x	if (op.iovec_count == 1)
592		{
593	428238x	n = WriteOp::write_policy::write_one(
594	214119x	this->fd_, bufs[0].data(), bufs[0].size());
595		}
596		else
597		{
598	12x	n = WriteOp::write_policy::write(
599	12x	this->fd_, op.iovecs, op.iovec_count);
600		}
601
602	214131x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
603		{
604	213991x	int err = (n < 0) ? errno : 0;
605	213991x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
606
607	213991x	if (this->svc_.scheduler().try_consume_inline_budget())
608		{
609	171144x	*ec = err ? make_err(err) : std::error_code{};
610	171144x	*bytes_out = bytes;
611	171144x	op.cont.h = h;
612	171144x	return dispatch_coro(ex, op.cont);
613		}
614	42847x	op.h = h;
615	42847x	op.ex = ex;
616	42847x	op.ec_out = ec;
617	42847x	op.bytes_out = bytes_out;
618	42847x	op.start(token, static_cast<Derived*>(this));
619	42847x	op.impl_ptr = this->shared_from_this();
620	42847x	op.complete(err, bytes);
621	42847x	this->svc_.post(&op);
622	42847x	return std::noop_coroutine();
623		}
624
625		// EAGAIN — register with reactor
626	140x	op.h = h;
627	140x	op.ex = ex;
628	140x	op.ec_out = ec;
629	140x	op.bytes_out = bytes_out;
630	140x	op.fd = this->fd_;
631	140x	op.start(token, static_cast<Derived*>(this));
632	140x	op.impl_ptr = this->shared_from_this();
633
634	140x	this->register_op(
635	140x	op, this->desc_state_.write_op, this->desc_state_.write_ready, true);
636	140x	return std::noop_coroutine();
637		}
638
639		template<
640		class Derived,
641		class Service,
642		class ConnOp,
643		class ReadOp,
644		class WriteOp,
645		class WaitOp,
646		class DescState,
647		class ImplBase,
648		class Endpoint>
649		std::coroutine_handle<>
650	92x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
651		do_wait(
652		std::coroutine_handle<> h,
653		capy::executor_ref ex,
654		wait_type w,
655		std::stop_token const& token,
656		std::error_code* ec)
657		{
658		// Pick refs up-front to avoid duplicating the register_op call.
659		WaitOp* op_ptr;
660		reactor_op_base** desc_slot_ptr;
661		std::uint32_t event;
662
663	92x	if (w == wait_type::read)
664		{
665	51x	op_ptr = &wait_rd_;
666	51x	desc_slot_ptr = &this->desc_state_.wait_read_op;
667	51x	event = reactor_event_read;
668		}
669	41x	else if (w == wait_type::write)
670		{
671	23x	op_ptr = &wait_wr_;
672	23x	desc_slot_ptr = &this->desc_state_.wait_write_op;
673	23x	event = reactor_event_write;
674		}
675		else // wait_type::error
676		{
677	18x	op_ptr = &wait_er_;
678	18x	desc_slot_ptr = &this->desc_state_.wait_error_op;
679	18x	event = reactor_event_error;
680		}
681
682	92x	auto& op = *op_ptr;
683
684		// Speculative probe, mirroring the speculative read: an
685		// edge-triggered reactor cannot report a condition that already
686		// holds, so a wait initiated on an already-ready socket would
687		// otherwise park forever.
688	92x	int perr = 0;
689	92x	if (WaitOp::probe(this->fd_, event, perr))
690		{
691	30x	if (this->svc_.scheduler().try_consume_inline_budget())
692		{
693	4x	*ec = perr ? make_err(perr) : std::error_code{};
694	4x	op.cont.h = h;
695	4x	return dispatch_coro(ex, op.cont);
696		}
697	26x	op.reset();
698	26x	op.wait_event = event;
699	26x	op.h = h;
700	26x	op.ex = ex;
701	26x	op.ec_out = ec;
702	26x	op.fd = this->fd_;
703	26x	op.start(token, static_cast<Derived*>(this));
704	26x	op.impl_ptr = this->shared_from_this();
705	26x	op.complete(perr, 0);
706	26x	this->svc_.post(&op);
707	26x	return std::noop_coroutine();
708		}
709
710	62x	op.reset();
711	62x	op.wait_event = event;
712	62x	op.h = h;
713	62x	op.ex = ex;
714	62x	op.ec_out = ec;
715	62x	op.fd = this->fd_;
716	62x	op.start(token, static_cast<Derived*>(this));
717	62x	op.impl_ptr = this->shared_from_this();
718
719		// Force register_op's ready path so the wait op re-probes under
720		// the descriptor mutex before parking. An edge consumed between
721		// the speculative probe above and the park (a concurrent short
722		// read, or an error event dispatched to an empty slot) would
723		// otherwise leave the wait parked on a ready socket.
724	62x	bool force_probe = true;
725	62x	this->register_op(op, *desc_slot_ptr, force_probe,
726		event == reactor_event_write);
727	62x	return std::noop_coroutine();
728		}
729
730		} // namespace boost::corosio::detail
731
732		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
733