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

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