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

96.9% Lines (280/0/289) 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	21512x	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	21512x	~reactor_stream_socket() override = default;
104
105		/// Return the cached remote endpoint.
106	60x	Endpoint remote_endpoint() const noexcept override
107		{
108	60x	return remote_endpoint_;
109		}
110
111		// --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
112
113	7052x	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	7052x	return do_connect(h, ex, ep, token, ec);
121		}
122
123	337754x	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	337754x	return do_read_some(h, ex, param, token, ec, bytes_out);
132		}
133
134	337096x	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	337096x	return do_write_some(h, ex, param, token, ec, bytes_out);
143		}
144
145	57x	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	57x	return do_wait(h, ex, w, token, ec);
153		}
154
155		std::error_code
156	23x	shutdown(corosio::shutdown_type what) noexcept override
157		{
158	23x	return do_shutdown(static_cast<int>(what));
159		}
160
161	216x	void cancel() noexcept override
162		{
163	216x	this->do_cancel();
164	216x	}
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	23x	std::error_code do_shutdown(int what) noexcept
179		{
180		int how;
181	23x	switch (what)
182		{
183	4x	case 0: // shutdown_receive
184	4x	how = SHUT_RD;
185	4x	break;
186	15x	case 1: // shutdown_send
187	15x	how = SHUT_WR;
188	15x	break;
189	4x	case 2: // shutdown_both
190	4x	how = SHUT_RDWR;
191	4x	break;
192	✗	default:
193	✗	return make_err(EINVAL);
194		}
195	23x	if (::shutdown(this->fd_, how) != 0)
196	✗	return make_err(errno);
197	23x	return {};
198		}
199
200		/// Cache local and remote endpoints.
201	14147x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
202		{
203	14147x	this->local_endpoint_ = std::move(local);
204	14147x	remote_endpoint_ = std::move(remote);
205	14147x	}
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	64516x	void do_close_socket() noexcept
270		{
271	64516x	base_type::do_close_socket();
272	64516x	remote_endpoint_ = Endpoint{};
273	64516x	}
274
275		/// Release ownership of the descriptor and drop the cached peer.
276	6x	native_handle_type do_release_socket() noexcept
277		{
278	6x	auto fd = base_type::do_release_socket();
279	6x	remote_endpoint_ = Endpoint{};
280	6x	return fd;
281		}
282
283		private:
284		// CRTP callbacks for reactor_basic_socket cancel/close
285
286		template<class Op>
287	232x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
288		{
289	232x	if (&op == static_cast<void*>(&conn_))
290	5x	return &this->desc_state_.connect_op;
291	227x	if (&op == static_cast<void*>(&rd_))
292	212x	return &this->desc_state_.read_op;
293	15x	if (&op == static_cast<void*>(&wr_))
294	4x	return &this->desc_state_.write_op;
295	11x	if (&op == static_cast<void*>(&wait_rd_))
296	7x	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		template<class Op>
305		bool* op_to_cancel_flag(Op& op) noexcept
306		{
307		if (&op == static_cast<void*>(&conn_))
308		return &this->desc_state_.connect_cancel_pending;
309		if (&op == static_cast<void*>(&rd_))
310		return &this->desc_state_.read_cancel_pending;
311		if (&op == static_cast<void*>(&wr_))
312		return &this->desc_state_.write_cancel_pending;
313		if (&op == static_cast<void*>(&wait_rd_))
314		return &this->desc_state_.wait_read_cancel_pending;
315		if (&op == static_cast<void*>(&wait_wr_))
316		return &this->desc_state_.wait_write_cancel_pending;
317		if (&op == static_cast<void*>(&wait_er_))
318		return &this->desc_state_.wait_error_cancel_pending;
319		return nullptr;
320		}
321
322		template<class Fn>
323	64738x	void for_each_op(Fn fn) noexcept
324		{
325	64738x	fn(conn_);
326	64738x	fn(rd_);
327	64738x	fn(wr_);
328	64738x	fn(wait_rd_);
329	64738x	fn(wait_wr_);
330	64738x	fn(wait_er_);
331	64738x	}
332
333		template<class Fn>
334	64738x	void for_each_desc_entry(Fn fn) noexcept
335		{
336	64738x	fn(conn_, this->desc_state_.connect_op);
337	64738x	fn(rd_, this->desc_state_.read_op);
338	64738x	fn(wr_, this->desc_state_.write_op);
339	64738x	fn(wait_rd_, this->desc_state_.wait_read_op);
340	64738x	fn(wait_wr_, this->desc_state_.wait_write_op);
341	64738x	fn(wait_er_, this->desc_state_.wait_error_op);
342	64738x	}
343		};
344
345		template<
346		class Derived,
347		class Service,
348		class ConnOp,
349		class ReadOp,
350		class WriteOp,
351		class WaitOp,
352		class DescState,
353		class ImplBase,
354		class Endpoint>
355		std::coroutine_handle<>
356	7052x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
357		do_connect(
358		std::coroutine_handle<> h,
359		capy::executor_ref ex,
360		Endpoint const& ep,
361		std::stop_token const& token,
362		std::error_code* ec)
363		{
364	7052x	auto& op = conn_;
365
366	7052x	sockaddr_storage storage{};
367	7052x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
368		int result =
369	7052x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
370
371	7052x	if (result == 0)
372		{
373	27x	sockaddr_storage local_storage{};
374	27x	socklen_t local_len = sizeof(local_storage);
375	27x	if (::getsockname(
376		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
377	27x	&local_len) == 0)
378	✗	this->local_endpoint_ =
379	27x	from_sockaddr_as(local_storage, local_len, Endpoint{});
380	27x	remote_endpoint_ = ep;
381		}
382
383	7052x	if (result == 0 \|\| errno != EINPROGRESS)
384		{
385	31x	int err = (result < 0) ? errno : 0;
386	31x	if (this->svc_.scheduler().try_consume_inline_budget())
387		{
388	✗	*ec = err ? make_err(err) : std::error_code{};
389	✗	op.cont.h = h;
390	✗	return dispatch_coro(ex, op.cont);
391		}
392	31x	op.reset();
393	31x	op.h = h;
394	31x	op.ex = ex;
395	31x	op.ec_out = ec;
396	31x	op.fd = this->fd_;
397	31x	op.target_endpoint = ep;
398	31x	op.start(token, static_cast<Derived*>(this));
399	31x	op.impl_ptr = this->shared_from_this();
400	31x	op.complete(err, 0);
401	31x	this->svc_.post(&op);
402	31x	return std::noop_coroutine();
403		}
404
405		// EINPROGRESS — register with reactor
406	7021x	op.reset();
407	7021x	op.h = h;
408	7021x	op.ex = ex;
409	7021x	op.ec_out = ec;
410	7021x	op.fd = this->fd_;
411	7021x	op.target_endpoint = ep;
412	7021x	op.start(token, static_cast<Derived*>(this));
413	7021x	op.impl_ptr = this->shared_from_this();
414
415	7021x	this->register_op(
416	7021x	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
417	7021x	this->desc_state_.connect_cancel_pending, true);
418	7021x	return std::noop_coroutine();
419		}
420
421		template<
422		class Derived,
423		class Service,
424		class ConnOp,
425		class ReadOp,
426		class WriteOp,
427		class WaitOp,
428		class DescState,
429		class ImplBase,
430		class Endpoint>
431		std::coroutine_handle<>
432	337754x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
433		do_read_some(
434		std::coroutine_handle<> h,
435		capy::executor_ref ex,
436		buffer_param param,
437		std::stop_token const& token,
438		std::error_code* ec,
439		std::size_t* bytes_out)
440		{
441	337754x	auto& op = rd_;
442	337754x	op.reset();
443
444		// Closed-object contract: complete with bad_file_descriptor without
445		// touching the kernel or the unregistered descriptor state.
446	337754x	if (this->fd_ < 0)
447		{
448	8x	op.h = h;
449	8x	op.ex = ex;
450	8x	op.ec_out = ec;
451	8x	op.bytes_out = bytes_out;
452	8x	op.start(token, static_cast<Derived*>(this));
453	8x	op.impl_ptr = this->shared_from_this();
454	8x	op.complete(EBADF, 0);
455	8x	this->svc_.post(&op);
456	8x	return std::noop_coroutine();
457		}
458
459	337746x	capy::mutable_buffer bufs[ReadOp::max_buffers];
460	337746x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
461
462	337746x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
463		{
464	4x	op.empty_buffer_read = true;
465	4x	op.h = h;
466	4x	op.ex = ex;
467	4x	op.ec_out = ec;
468	4x	op.bytes_out = bytes_out;
469	4x	op.start(token, static_cast<Derived*>(this));
470	4x	op.impl_ptr = this->shared_from_this();
471	4x	op.complete(0, 0);
472	4x	this->svc_.post(&op);
473	4x	return std::noop_coroutine();
474		}
475
476	675496x	for (int i = 0; i < op.iovec_count; ++i)
477		{
478	337754x	op.iovecs[i].iov_base = bufs[i].data();
479	337754x	op.iovecs[i].iov_len = bufs[i].size();
480		}
481
482		// Speculative read; for the single-buffer case use recv() so the
483		// kernel skips the readv iov_iter setup.
484		ssize_t n;
485	337742x	if (op.iovec_count == 1)
486		{
487		do
488		{
489	337734x	n = ::recv(this->fd_, bufs[0].data(), bufs[0].size(), 0);
490		}
491	337734x	while (n < 0 && errno == EINTR);
492		}
493		else
494		{
495		do
496		{
497	8x	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
498		}
499	8x	while (n < 0 && errno == EINTR);
500		}
501
502	337742x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
503		{
504	336983x	int err = (n < 0) ? errno : 0;
505	336983x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
506
507	336983x	if (this->svc_.scheduler().try_consume_inline_budget())
508		{
509	269626x	if (err)
510	✗	*ec = make_err(err);
511	269626x	else if (n == 0)
512	15x	*ec = capy::error::eof;
513		else
514	269611x	*ec = {};
515	269626x	*bytes_out = bytes;
516	269626x	op.cont.h = h;
517	269626x	return dispatch_coro(ex, op.cont);
518		}
519	67357x	op.h = h;
520	67357x	op.ex = ex;
521	67357x	op.ec_out = ec;
522	67357x	op.bytes_out = bytes_out;
523	67357x	op.start(token, static_cast<Derived*>(this));
524	67357x	op.impl_ptr = this->shared_from_this();
525	67357x	op.complete(err, bytes);
526	67357x	this->svc_.post(&op);
527	67357x	return std::noop_coroutine();
528		}
529
530		// EAGAIN — register with reactor
531	759x	op.h = h;
532	759x	op.ex = ex;
533	759x	op.ec_out = ec;
534	759x	op.bytes_out = bytes_out;
535	759x	op.fd = this->fd_;
536	759x	op.start(token, static_cast<Derived*>(this));
537	759x	op.impl_ptr = this->shared_from_this();
538
539	759x	this->register_op(
540	759x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
541	759x	this->desc_state_.read_cancel_pending);
542	759x	return std::noop_coroutine();
543		}
544
545		template<
546		class Derived,
547		class Service,
548		class ConnOp,
549		class ReadOp,
550		class WriteOp,
551		class WaitOp,
552		class DescState,
553		class ImplBase,
554		class Endpoint>
555		std::coroutine_handle<>
556	337096x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
557		do_write_some(
558		std::coroutine_handle<> h,
559		capy::executor_ref ex,
560		buffer_param param,
561		std::stop_token const& token,
562		std::error_code* ec,
563		std::size_t* bytes_out)
564		{
565	337096x	auto& op = wr_;
566	337096x	op.reset();
567
568		// Closed-object contract: complete with bad_file_descriptor without
569		// touching the kernel or the unregistered descriptor state.
570	337096x	if (this->fd_ < 0)
571		{
572	8x	op.h = h;
573	8x	op.ex = ex;
574	8x	op.ec_out = ec;
575	8x	op.bytes_out = bytes_out;
576	8x	op.start(token, static_cast<Derived*>(this));
577	8x	op.impl_ptr = this->shared_from_this();
578	8x	op.complete(EBADF, 0);
579	8x	this->svc_.post(&op);
580	8x	return std::noop_coroutine();
581		}
582
583	337088x	capy::mutable_buffer bufs[WriteOp::max_buffers];
584	337088x	op.iovec_count =
585	337088x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
586
587	337088x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
588		{
589	4x	op.h = h;
590	4x	op.ex = ex;
591	4x	op.ec_out = ec;
592	4x	op.bytes_out = bytes_out;
593	4x	op.start(token, static_cast<Derived*>(this));
594	4x	op.impl_ptr = this->shared_from_this();
595	4x	op.complete(0, 0);
596	4x	this->svc_.post(&op);
597	4x	return std::noop_coroutine();
598		}
599
600	674178x	for (int i = 0; i < op.iovec_count; ++i)
601		{
602	337094x	op.iovecs[i].iov_base = bufs[i].data();
603	337094x	op.iovecs[i].iov_len = bufs[i].size();
604		}
605
606		// Speculative write; the single-buffer case dispatches to a
607		// backend-specific fast path so the kernel skips msghdr/iov_iter
608		// setup (and so each backend can pick the right SIGPIPE strategy).
609		ssize_t n;
610	337084x	if (op.iovec_count == 1)
611		{
612	674156x	n = WriteOp::write_policy::write_one(
613	337078x	this->fd_, bufs[0].data(), bufs[0].size());
614		}
615		else
616		{
617	6x	n = WriteOp::write_policy::write(
618	6x	this->fd_, op.iovecs, op.iovec_count);
619		}
620
621	337084x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
622		{
623	336955x	int err = (n < 0) ? errno : 0;
624	336955x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
625
626	336955x	if (this->svc_.scheduler().try_consume_inline_budget())
627		{
628	269533x	*ec = err ? make_err(err) : std::error_code{};
629	269533x	*bytes_out = bytes;
630	269533x	op.cont.h = h;
631	269533x	return dispatch_coro(ex, op.cont);
632		}
633	67422x	op.h = h;
634	67422x	op.ex = ex;
635	67422x	op.ec_out = ec;
636	67422x	op.bytes_out = bytes_out;
637	67422x	op.start(token, static_cast<Derived*>(this));
638	67422x	op.impl_ptr = this->shared_from_this();
639	67422x	op.complete(err, bytes);
640	67422x	this->svc_.post(&op);
641	67422x	return std::noop_coroutine();
642		}
643
644		// EAGAIN — register with reactor
645	129x	op.h = h;
646	129x	op.ex = ex;
647	129x	op.ec_out = ec;
648	129x	op.bytes_out = bytes_out;
649	129x	op.fd = this->fd_;
650	129x	op.start(token, static_cast<Derived*>(this));
651	129x	op.impl_ptr = this->shared_from_this();
652
653	129x	this->register_op(
654	129x	op, this->desc_state_.write_op, this->desc_state_.write_ready,
655	129x	this->desc_state_.write_cancel_pending, true);
656	129x	return std::noop_coroutine();
657		}
658
659		template<
660		class Derived,
661		class Service,
662		class ConnOp,
663		class ReadOp,
664		class WriteOp,
665		class WaitOp,
666		class DescState,
667		class ImplBase,
668		class Endpoint>
669		std::coroutine_handle<>
670	57x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
671		do_wait(
672		std::coroutine_handle<> h,
673		capy::executor_ref ex,
674		wait_type w,
675		std::stop_token const& token,
676		std::error_code* ec)
677		{
678		// Pick refs up-front to avoid duplicating the register_op call.
679		WaitOp* op_ptr;
680		reactor_op_base** desc_slot_ptr;
681		bool* cancel_flag_ptr;
682		std::uint32_t event;
683
684	57x	if (w == wait_type::read)
685		{
686	27x	op_ptr = &wait_rd_;
687	27x	desc_slot_ptr = &this->desc_state_.wait_read_op;
688	27x	cancel_flag_ptr = &this->desc_state_.wait_read_cancel_pending;
689	27x	event = reactor_event_read;
690		}
691	30x	else if (w == wait_type::write)
692		{
693	16x	op_ptr = &wait_wr_;
694	16x	desc_slot_ptr = &this->desc_state_.wait_write_op;
695	16x	cancel_flag_ptr = &this->desc_state_.wait_write_cancel_pending;
696	16x	event = reactor_event_write;
697		}
698		else // wait_type::error
699		{
700	14x	op_ptr = &wait_er_;
701	14x	desc_slot_ptr = &this->desc_state_.wait_error_op;
702	14x	cancel_flag_ptr = &this->desc_state_.wait_error_cancel_pending;
703	14x	event = reactor_event_error;
704		}
705
706	57x	auto& op = *op_ptr;
707
708		// Speculative probe, mirroring the speculative read: an
709		// edge-triggered reactor cannot report a condition that already
710		// holds, so a wait initiated on an already-ready socket would
711		// otherwise park forever.
712	57x	int perr = 0;
713	57x	if (WaitOp::probe(this->fd_, event, perr))
714		{
715	24x	if (this->svc_.scheduler().try_consume_inline_budget())
716		{
717	4x	*ec = perr ? make_err(perr) : std::error_code{};
718	4x	op.cont.h = h;
719	4x	return dispatch_coro(ex, op.cont);
720		}
721	20x	op.reset();
722	20x	op.wait_event = event;
723	20x	op.h = h;
724	20x	op.ex = ex;
725	20x	op.ec_out = ec;
726	20x	op.fd = this->fd_;
727	20x	op.start(token, static_cast<Derived*>(this));
728	20x	op.impl_ptr = this->shared_from_this();
729	20x	op.complete(perr, 0);
730	20x	this->svc_.post(&op);
731	20x	return std::noop_coroutine();
732		}
733
734	33x	op.reset();
735	33x	op.wait_event = event;
736	33x	op.h = h;
737	33x	op.ex = ex;
738	33x	op.ec_out = ec;
739	33x	op.fd = this->fd_;
740	33x	op.start(token, static_cast<Derived*>(this));
741	33x	op.impl_ptr = this->shared_from_this();
742
743		// Force register_op's ready path so the wait op re-probes under
744		// the descriptor mutex before parking. An edge consumed between
745		// the speculative probe above and the park (a concurrent short
746		// read, or an error event dispatched to an empty slot) would
747		// otherwise leave the wait parked on a ready socket.
748	33x	bool force_probe = true;
749	33x	this->register_op(op, desc_slot_ptr, force_probe, cancel_flag_ptr,
750		event == reactor_event_write);
751	33x	return std::noop_coroutine();
752		}
753
754		} // namespace boost::corosio::detail
755
756		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
757