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

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