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

96.8% Lines (273 / 282) 100.0% Functions (89 / 89)

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

reactor_stream_socket.hpp

f(x) Functions (89)

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/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
24		#include <errno.h>
25		#include <sys/socket.h>
26		#include <sys/uio.h>
27
28		namespace boost::corosio::detail {
29
30		/** CRTP base for reactor-backed stream socket implementations.
31
32		Inherits shared data members and cancel/close/register logic
33		from reactor_basic_socket. Adds the stream-specific remote
34		endpoint, shutdown, and I/O dispatch (connect, read, write, wait).
35
36		@tparam Derived The concrete socket type (CRTP).
37		@tparam Service The backend's socket service type.
38		@tparam ConnOp The backend's connect op type.
39		@tparam ReadOp The backend's read op type.
40		@tparam WriteOp The backend's write op type.
41		@tparam WaitOp The backend's wait op type.
42		@tparam DescState The backend's descriptor_state type.
43		@tparam ImplBase The public vtable base
44		(tcp_socket::implementation or
45		local_stream_socket::implementation).
46		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
47		*/
48		template<
49		class Derived,
50		class Service,
51		class ConnOp,
52		class ReadOp,
53		class WriteOp,
54		class WaitOp,
55		class DescState,
56		class ImplBase = tcp_socket::implementation,
57		class Endpoint = endpoint>
58		class reactor_stream_socket
59		: public reactor_basic_socket<
60		Derived,
61		ImplBase,
62		Service,
63		DescState,
64		Endpoint>
65		{
66		using base_type =
67		reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>;
68		using self_type = reactor_stream_socket<
69		Derived,
70		Service,
71		ConnOp,
72		ReadOp,
73		WriteOp,
74		WaitOp,
75		DescState,
76		ImplBase,
77		Endpoint>;
78		friend base_type;
79		friend reactor_io_core<Derived, Service, DescState>;
80		friend Derived;
81
82		protected:
83		// NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
84	14466x	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	14466x	~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	4667x	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	4667x	return do_connect(h, ex, ep, token, ec);
126		}
127
128	210443x	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	210443x	return do_read_some(h, ex, param, token, ec, bytes_out);
137		}
138
139	209742x	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	209742x	return do_write_some(h, ex, param, token, ec, bytes_out);
148		}
149
150	96x	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	96x	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	226x	void cancel() noexcept override
166		{
167	226x	this->do_cancel();
168	226x	}
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	9392x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
206		{
207	9392x	this->local_endpoint_ = std::move(local);
208	9392x	remote_endpoint_ = std::move(remote);
209	9392x	}
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	43218x	void do_close_socket() noexcept
274		{
275	43218x	base_type::do_close_socket();
276	43218x	remote_endpoint_ = Endpoint{};
277	43218x	}
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		private:
288		// CRTP callbacks for reactor_io_core cancel/close
289
290		template<class Op>
291	219x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
292		{
293	219x	if (&op == static_cast<void*>(&conn_))
294	✗	return &this->desc_state_.connect_op;
295	219x	if (&op == static_cast<void*>(&rd_))
296	208x	return &this->desc_state_.read_op;
297	11x	if (&op == static_cast<void*>(&wr_))
298	6x	return &this->desc_state_.write_op;
299	5x	if (&op == static_cast<void*>(&wait_rd_))
300	3x	return &this->desc_state_.wait_read_op;
301	2x	if (&op == static_cast<void*>(&wait_wr_))
302	✗	return &this->desc_state_.wait_write_op;
303	2x	if (&op == static_cast<void*>(&wait_er_))
304	2x	return &this->desc_state_.wait_error_op;
305	✗	return nullptr;
306		}
307
308		template<class Fn>
309	43452x	void for_each_op(Fn fn) noexcept
310		{
311	43452x	fn(conn_);
312	43452x	fn(rd_);
313	43452x	fn(wr_);
314	43452x	fn(wait_rd_);
315	43452x	fn(wait_wr_);
316	43452x	fn(wait_er_);
317	43452x	}
318
319		template<class Fn>
320	48374x	void for_each_desc_entry(Fn fn) noexcept
321		{
322	48374x	fn(conn_, this->desc_state_.connect_op);
323	48374x	fn(rd_, this->desc_state_.read_op);
324	48374x	fn(wr_, this->desc_state_.write_op);
325	48374x	fn(wait_rd_, this->desc_state_.wait_read_op);
326	48374x	fn(wait_wr_, this->desc_state_.wait_write_op);
327	48374x	fn(wait_er_, this->desc_state_.wait_error_op);
328	48374x	}
329		};
330
331		template<
332		class Derived,
333		class Service,
334		class ConnOp,
335		class ReadOp,
336		class WriteOp,
337		class WaitOp,
338		class DescState,
339		class ImplBase,
340		class Endpoint>
341		std::coroutine_handle<>
342	4667x	reactor_stream_socket<
343		Derived,
344		Service,
345		ConnOp,
346		ReadOp,
347		WriteOp,
348		WaitOp,
349		DescState,
350		ImplBase,
351		Endpoint>::
352		do_connect(
353		std::coroutine_handle<> h,
354		capy::executor_ref ex,
355		Endpoint const& ep,
356		std::stop_token const& token,
357		std::error_code* ec)
358		{
359	4667x	auto& op = conn_;
360
361	4667x	sockaddr_storage storage{};
362	4667x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
363		int result =
364	4667x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
365
366	4667x	if (result == 0)
367		{
368	29x	sockaddr_storage local_storage{};
369	29x	socklen_t local_len = sizeof(local_storage);
370	29x	if (::getsockname(
371		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
372	29x	&local_len) == 0)
373	✗	this->local_endpoint_ =
374	29x	from_sockaddr_as(local_storage, local_len, Endpoint{});
375	29x	remote_endpoint_ = ep;
376		}
377
378	4667x	if (result == 0 \|\| errno != EINPROGRESS)
379		{
380	37x	int err = (result < 0) ? errno : 0;
381	37x	if (this->svc_.scheduler().try_consume_inline_budget())
382		{
383	✗	*ec = err ? make_err(err) : std::error_code{};
384	✗	op.cont.h = h;
385	✗	return dispatch_coro(ex, op.cont);
386		}
387	37x	op.reset();
388	37x	op.h = h;
389	37x	op.ex = ex;
390	37x	op.ec_out = ec;
391	37x	op.fd = this->fd_;
392	37x	op.target_endpoint = ep;
393	37x	op.start(token, static_cast<Derived*>(this));
394	37x	op.impl_ptr = this->shared_from_this();
395	37x	op.complete(err, 0);
396	37x	this->svc_.post(&op);
397	37x	return std::noop_coroutine();
398		}
399
400		// EINPROGRESS — register with reactor
401	4630x	op.reset();
402	4630x	op.h = h;
403	4630x	op.ex = ex;
404	4630x	op.ec_out = ec;
405	4630x	op.fd = this->fd_;
406	4630x	op.target_endpoint = ep;
407	4630x	op.start(token, static_cast<Derived*>(this));
408	4630x	op.impl_ptr = this->shared_from_this();
409
410	4630x	this->register_op(
411	4630x	op, this->desc_state_.connect_op, this->desc_state_.write_ready, true);
412	4630x	return std::noop_coroutine();
413		}
414
415		template<
416		class Derived,
417		class Service,
418		class ConnOp,
419		class ReadOp,
420		class WriteOp,
421		class WaitOp,
422		class DescState,
423		class ImplBase,
424		class Endpoint>
425		std::coroutine_handle<>
426	210443x	reactor_stream_socket<
427		Derived,
428		Service,
429		ConnOp,
430		ReadOp,
431		WriteOp,
432		WaitOp,
433		DescState,
434		ImplBase,
435		Endpoint>::
436		do_read_some(
437		std::coroutine_handle<> h,
438		capy::executor_ref ex,
439		buffer_param param,
440		std::stop_token const& token,
441		std::error_code* ec,
442		std::size_t* bytes_out)
443		{
444	210443x	auto& op = rd_;
445	210443x	op.reset();
446
447		// Closed-object contract: complete with bad_file_descriptor without
448		// touching the kernel or the unregistered descriptor state.
449	210443x	if (this->fd_ < 0)
450		{
451	8x	op.h = h;
452	8x	op.ex = ex;
453	8x	op.ec_out = ec;
454	8x	op.bytes_out = bytes_out;
455	8x	op.start(token, static_cast<Derived*>(this));
456	8x	op.impl_ptr = this->shared_from_this();
457	8x	op.complete(EBADF, 0);
458	8x	this->svc_.post(&op);
459	8x	return std::noop_coroutine();
460		}
461
462	210435x	capy::mutable_buffer bufs[ReadOp::max_buffers];
463	210435x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
464
465	210435x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
466		{
467	4x	op.empty_buffer_read = true;
468	4x	op.h = h;
469	4x	op.ex = ex;
470	4x	op.ec_out = ec;
471	4x	op.bytes_out = bytes_out;
472	4x	op.start(token, static_cast<Derived*>(this));
473	4x	op.impl_ptr = this->shared_from_this();
474	4x	op.complete(0, 0);
475	4x	this->svc_.post(&op);
476	4x	return std::noop_coroutine();
477		}
478
479	420880x	for (int i = 0; i < op.iovec_count; ++i)
480		{
481	210449x	op.iovecs[i].iov_base = bufs[i].data();
482	210449x	op.iovecs[i].iov_len = bufs[i].size();
483		}
484
485		// Speculative read; for the single-buffer case use recv() so the
486		// kernel skips the readv iov_iter setup.
487		ssize_t n;
488	210431x	if (op.iovec_count == 1)
489		{
490		do
491		{
492	210419x	n = ::recv(this->fd_, bufs[0].data(), bufs[0].size(), 0);
493		}
494	210419x	while (n < 0 && errno == EINTR);
495		}
496		else
497		{
498		do
499		{
500	16x	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
501		}
502	16x	while (n < 0 && errno == EINTR);
503		}
504
505	210431x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
506		{
507	209650x	int err = (n < 0) ? errno : 0;
508	209650x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
509
510	209650x	if (this->svc_.scheduler().try_consume_inline_budget())
511		{
512	167752x	if (err)
513	8x	*ec = make_err(err);
514	167744x	else if (n == 0)
515	17x	*ec = capy::error::eof;
516		else
517	167727x	*ec = {};
518	167752x	*bytes_out = bytes;
519	167752x	op.cont.h = h;
520	167752x	return dispatch_coro(ex, op.cont);
521		}
522	41898x	op.h = h;
523	41898x	op.ex = ex;
524	41898x	op.ec_out = ec;
525	41898x	op.bytes_out = bytes_out;
526	41898x	op.start(token, static_cast<Derived*>(this));
527	41898x	op.impl_ptr = this->shared_from_this();
528	41898x	op.complete(err, bytes);
529	41898x	this->svc_.post(&op);
530	41898x	return std::noop_coroutine();
531		}
532
533		// EAGAIN — register with reactor
534	781x	op.h = h;
535	781x	op.ex = ex;
536	781x	op.ec_out = ec;
537	781x	op.bytes_out = bytes_out;
538	781x	op.fd = this->fd_;
539	781x	op.start(token, static_cast<Derived*>(this));
540	781x	op.impl_ptr = this->shared_from_this();
541
542	781x	this->register_op(
543	781x	op, this->desc_state_.read_op, this->desc_state_.read_ready);
544	781x	return std::noop_coroutine();
545		}
546
547		template<
548		class Derived,
549		class Service,
550		class ConnOp,
551		class ReadOp,
552		class WriteOp,
553		class WaitOp,
554		class DescState,
555		class ImplBase,
556		class Endpoint>
557		std::coroutine_handle<>
558	209742x	reactor_stream_socket<
559		Derived,
560		Service,
561		ConnOp,
562		ReadOp,
563		WriteOp,
564		WaitOp,
565		DescState,
566		ImplBase,
567		Endpoint>::
568		do_write_some(
569		std::coroutine_handle<> h,
570		capy::executor_ref ex,
571		buffer_param param,
572		std::stop_token const& token,
573		std::error_code* ec,
574		std::size_t* bytes_out)
575		{
576	209742x	auto& op = wr_;
577	209742x	op.reset();
578
579		// Closed-object contract: complete with bad_file_descriptor without
580		// touching the kernel or the unregistered descriptor state.
581	209742x	if (this->fd_ < 0)
582		{
583	10x	op.h = h;
584	10x	op.ex = ex;
585	10x	op.ec_out = ec;
586	10x	op.bytes_out = bytes_out;
587	10x	op.start(token, static_cast<Derived*>(this));
588	10x	op.impl_ptr = this->shared_from_this();
589	10x	op.complete(EBADF, 0);
590	10x	this->svc_.post(&op);
591	10x	return std::noop_coroutine();
592		}
593
594	209732x	capy::mutable_buffer bufs[WriteOp::max_buffers];
595	209732x	op.iovec_count =
596	209732x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
597
598	209732x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
599		{
600	4x	op.h = h;
601	4x	op.ex = ex;
602	4x	op.ec_out = ec;
603	4x	op.bytes_out = bytes_out;
604	4x	op.start(token, static_cast<Derived*>(this));
605	4x	op.impl_ptr = this->shared_from_this();
606	4x	op.complete(0, 0);
607	4x	this->svc_.post(&op);
608	4x	return std::noop_coroutine();
609		}
610
611	419472x	for (int i = 0; i < op.iovec_count; ++i)
612		{
613	209744x	op.iovecs[i].iov_base = bufs[i].data();
614	209744x	op.iovecs[i].iov_len = bufs[i].size();
615		}
616
617		// Speculative write; the single-buffer case dispatches to a
618		// backend-specific fast path so the kernel skips msghdr/iov_iter
619		// setup (and so each backend can pick the right SIGPIPE strategy).
620		ssize_t n;
621	209728x	if (op.iovec_count == 1)
622		{
623	419432x	n = WriteOp::write_policy::write_one(
624	209716x	this->fd_, bufs[0].data(), bufs[0].size());
625		}
626		else
627		{
628	12x	n = WriteOp::write_policy::write(this->fd_, op.iovecs, op.iovec_count);
629		}
630
631	209728x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
632		{
633	209584x	int err = (n < 0) ? errno : 0;
634	209584x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
635
636	209584x	if (this->svc_.scheduler().try_consume_inline_budget())
637		{
638	167611x	*ec = err ? make_err(err) : std::error_code{};
639	167611x	*bytes_out = bytes;
640	167611x	op.cont.h = h;
641	167611x	return dispatch_coro(ex, op.cont);
642		}
643	41973x	op.h = h;
644	41973x	op.ex = ex;
645	41973x	op.ec_out = ec;
646	41973x	op.bytes_out = bytes_out;
647	41973x	op.start(token, static_cast<Derived*>(this));
648	41973x	op.impl_ptr = this->shared_from_this();
649	41973x	op.complete(err, bytes);
650	41973x	this->svc_.post(&op);
651	41973x	return std::noop_coroutine();
652		}
653
654		// EAGAIN — register with reactor
655	144x	op.h = h;
656	144x	op.ex = ex;
657	144x	op.ec_out = ec;
658	144x	op.bytes_out = bytes_out;
659	144x	op.fd = this->fd_;
660	144x	op.start(token, static_cast<Derived*>(this));
661	144x	op.impl_ptr = this->shared_from_this();
662
663	144x	this->register_op(
664	144x	op, this->desc_state_.write_op, this->desc_state_.write_ready, true);
665	144x	return std::noop_coroutine();
666		}
667
668		template<
669		class Derived,
670		class Service,
671		class ConnOp,
672		class ReadOp,
673		class WriteOp,
674		class WaitOp,
675		class DescState,
676		class ImplBase,
677		class Endpoint>
678		std::coroutine_handle<>
679	96x	reactor_stream_socket<
680		Derived,
681		Service,
682		ConnOp,
683		ReadOp,
684		WriteOp,
685		WaitOp,
686		DescState,
687		ImplBase,
688		Endpoint>::
689		do_wait(
690		std::coroutine_handle<> h,
691		capy::executor_ref ex,
692		wait_type w,
693		std::stop_token const& token,
694		std::error_code* ec)
695		{
696		// Pick refs up-front to avoid duplicating the register_op call.
697		WaitOp* op_ptr;
698		reactor_op_base** desc_slot_ptr;
699		std::uint32_t event;
700
701	96x	if (w == wait_type::read)
702		{
703	55x	op_ptr = &wait_rd_;
704	55x	desc_slot_ptr = &this->desc_state_.wait_read_op;
705	55x	event = reactor_event_read;
706		}
707	41x	else if (w == wait_type::write)
708		{
709	23x	op_ptr = &wait_wr_;
710	23x	desc_slot_ptr = &this->desc_state_.wait_write_op;
711	23x	event = reactor_event_write;
712		}
713		else // wait_type::error
714		{
715	18x	op_ptr = &wait_er_;
716	18x	desc_slot_ptr = &this->desc_state_.wait_error_op;
717	18x	event = reactor_event_error;
718		}
719
720	96x	auto& op = *op_ptr;
721
722		// Speculative probe, mirroring the speculative read: an
723		// edge-triggered reactor cannot report a condition that already
724		// holds, so a wait initiated on an already-ready socket would
725		// otherwise park forever.
726	96x	int perr = 0;
727	96x	if (WaitOp::probe(this->fd_, event, perr))
728		{
729	40x	if (this->svc_.scheduler().try_consume_inline_budget())
730		{
731	8x	*ec = perr ? make_err(perr) : std::error_code{};
732	8x	op.cont.h = h;
733	8x	return dispatch_coro(ex, op.cont);
734		}
735	32x	op.reset();
736	32x	op.wait_event = event;
737	32x	op.h = h;
738	32x	op.ex = ex;
739	32x	op.ec_out = ec;
740	32x	op.fd = this->fd_;
741	32x	op.start(token, static_cast<Derived*>(this));
742	32x	op.impl_ptr = this->shared_from_this();
743	32x	op.complete(perr, 0);
744	32x	this->svc_.post(&op);
745	32x	return std::noop_coroutine();
746		}
747
748	56x	op.reset();
749	56x	op.wait_event = event;
750	56x	op.h = h;
751	56x	op.ex = ex;
752	56x	op.ec_out = ec;
753	56x	op.fd = this->fd_;
754	56x	op.start(token, static_cast<Derived*>(this));
755	56x	op.impl_ptr = this->shared_from_this();
756
757		// Force register_op's ready path so the wait op re-probes under
758		// the descriptor mutex before parking. An edge consumed between
759		// the speculative probe above and the park (a concurrent short
760		// read, or an error event dispatched to an empty slot) would
761		// otherwise leave the wait parked on a ready socket.
762	56x	bool force_probe = true;
763	56x	this->register_op(
764		op, *desc_slot_ptr, force_probe, event == reactor_event_write);
765	56x	return std::noop_coroutine();
766		}
767
768		} // namespace boost::corosio::detail
769
770		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
771