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

97.0% Lines (288 / 297) 100.0% Functions (101 / 101)

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

reactor_stream_socket.hpp

f(x) Functions (101)

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