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

97.3% Lines (285 / 293) 100.0% Functions (97 / 97)

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

reactor_stream_socket.hpp

f(x) Functions (97)

Function Calls Lines Blocks

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