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

80.1% Lines (221/276) 79.5% List of functions (70/88)

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

reactor_stream_socket.hpp

f(x) Functions (88)

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	25399x	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	25399x	~reactor_stream_socket() override = default;
104
105		/// Return the cached remote endpoint.
106	46x	Endpoint remote_endpoint() const noexcept override
107		{
108	46x	return remote_endpoint_;
109		}
110
111		// --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
112
113	8438x	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	8438x	return do_connect(h, ex, ep, token, ec);
121		}
122
123	233659x	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	233659x	return do_read_some(h, ex, param, token, ec, bytes_out);
132		}
133
134	233368x	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	233368x	return do_write_some(h, ex, param, token, ec, bytes_out);
143		}
144
145	8x	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	8x	return do_wait(h, ex, w, token, ec);
153		}
154
155		std::error_code
156	6x	shutdown(corosio::shutdown_type what) noexcept override
157		{
158	6x	return do_shutdown(static_cast<int>(what));
159		}
160
161	188x	void cancel() noexcept override
162		{
163	188x	this->do_cancel();
164	188x	}
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	6x	std::error_code do_shutdown(int what) noexcept
179		{
180		int how;
181	6x	switch (what)
182		{
183	2x	case 0: // shutdown_receive
184	2x	how = SHUT_RD;
185	2x	break;
186	2x	case 1: // shutdown_send
187	2x	how = SHUT_WR;
188	2x	break;
189	2x	case 2: // shutdown_both
190	2x	how = SHUT_RDWR;
191	2x	break;
192	✗	default:
193	✗	return make_err(EINVAL);
194		}
195	6x	if (::shutdown(this->fd_, how) != 0)
196	✗	return make_err(errno);
197	6x	return {};
198		}
199
200		/// Cache local and remote endpoints.
201	16860x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
202		{
203	16860x	this->local_endpoint_ = std::move(local);
204	16860x	remote_endpoint_ = std::move(remote);
205	16860x	}
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		Registers a wait op for the requested direction. Does not
251		perform any I/O syscall — completion is signalled when the
252		reactor delivers the matching edge event.
253		*/
254		std::coroutine_handle<> do_wait(
255		std::coroutine_handle<>,
256		capy::executor_ref,
257		wait_type,
258		std::stop_token const&,
259		std::error_code*);
260
261		/** Close the socket and cancel pending operations.
262
263		Extends the base do_close_socket() to also reset
264		the remote endpoint.
265		*/
266	76214x	void do_close_socket() noexcept
267		{
268	76214x	base_type::do_close_socket();
269	76214x	remote_endpoint_ = Endpoint{};
270	76214x	}
271
272		private:
273		// CRTP callbacks for reactor_basic_socket cancel/close
274
275		template<class Op>
276	191x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
277		{
278	191x	if (&op == static_cast<void*>(&conn_))
279	✗	return &this->desc_state_.connect_op;
280	191x	if (&op == static_cast<void*>(&rd_))
281	191x	return &this->desc_state_.read_op;
282	✗	if (&op == static_cast<void*>(&wr_))
283	✗	return &this->desc_state_.write_op;
284	✗	if (&op == static_cast<void*>(&wait_rd_))
285	✗	return &this->desc_state_.wait_read_op;
286	✗	if (&op == static_cast<void*>(&wait_wr_))
287	✗	return &this->desc_state_.wait_write_op;
288	✗	if (&op == static_cast<void*>(&wait_er_))
289	✗	return &this->desc_state_.wait_error_op;
290	✗	return nullptr;
291		}
292
293		template<class Op>
294	✗	bool* op_to_cancel_flag(Op& op) noexcept
295		{
296	✗	if (&op == static_cast<void*>(&conn_))
297	✗	return &this->desc_state_.connect_cancel_pending;
298	✗	if (&op == static_cast<void*>(&rd_))
299	✗	return &this->desc_state_.read_cancel_pending;
300	✗	if (&op == static_cast<void*>(&wr_))
301	✗	return &this->desc_state_.write_cancel_pending;
302	✗	if (&op == static_cast<void*>(&wait_rd_))
303	✗	return &this->desc_state_.wait_read_cancel_pending;
304	✗	if (&op == static_cast<void*>(&wait_wr_))
305	✗	return &this->desc_state_.wait_write_cancel_pending;
306	✗	if (&op == static_cast<void*>(&wait_er_))
307	✗	return &this->desc_state_.wait_error_cancel_pending;
308	✗	return nullptr;
309		}
310
311		template<class Fn>
312	76404x	void for_each_op(Fn fn) noexcept
313		{
314	76404x	fn(conn_);
315	76404x	fn(rd_);
316	76404x	fn(wr_);
317	76404x	fn(wait_rd_);
318	76404x	fn(wait_wr_);
319	76404x	fn(wait_er_);
320	76404x	}
321
322		template<class Fn>
323	76404x	void for_each_desc_entry(Fn fn) noexcept
324		{
325	76404x	fn(conn_, this->desc_state_.connect_op);
326	76404x	fn(rd_, this->desc_state_.read_op);
327	76404x	fn(wr_, this->desc_state_.write_op);
328	76404x	fn(wait_rd_, this->desc_state_.wait_read_op);
329	76404x	fn(wait_wr_, this->desc_state_.wait_write_op);
330	76404x	fn(wait_er_, this->desc_state_.wait_error_op);
331	76404x	}
332		};
333
334		template<
335		class Derived,
336		class Service,
337		class ConnOp,
338		class ReadOp,
339		class WriteOp,
340		class WaitOp,
341		class DescState,
342		class ImplBase,
343		class Endpoint>
344		std::coroutine_handle<>
345	8438x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
346		do_connect(
347		std::coroutine_handle<> h,
348		capy::executor_ref ex,
349		Endpoint const& ep,
350		std::stop_token const& token,
351		std::error_code* ec)
352		{
353	8438x	auto& op = conn_;
354
355	8438x	sockaddr_storage storage{};
356	8438x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
357		int result =
358	8438x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
359
360	8438x	if (result == 0)
361		{
362	6x	sockaddr_storage local_storage{};
363	6x	socklen_t local_len = sizeof(local_storage);
364	6x	if (::getsockname(
365		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
366	6x	&local_len) == 0)
367	✗	this->local_endpoint_ =
368	6x	from_sockaddr_as(local_storage, local_len, Endpoint{});
369	6x	remote_endpoint_ = ep;
370		}
371
372	8438x	if (result == 0 \|\| errno != EINPROGRESS)
373		{
374	6x	int err = (result < 0) ? errno : 0;
375	6x	if (this->svc_.scheduler().try_consume_inline_budget())
376		{
377	✗	*ec = err ? make_err(err) : std::error_code{};
378	✗	op.cont_op.cont.h = h;
379	✗	return dispatch_coro(ex, op.cont_op.cont);
380		}
381	6x	op.reset();
382	6x	op.h = h;
383	6x	op.ex = ex;
384	6x	op.ec_out = ec;
385	6x	op.fd = this->fd_;
386	6x	op.target_endpoint = ep;
387	6x	op.start(token, static_cast<Derived*>(this));
388	6x	op.impl_ptr = this->shared_from_this();
389	6x	op.complete(err, 0);
390	6x	this->svc_.post(&op);
391	6x	return std::noop_coroutine();
392		}
393
394		// EINPROGRESS — register with reactor
395	8432x	op.reset();
396	8432x	op.h = h;
397	8432x	op.ex = ex;
398	8432x	op.ec_out = ec;
399	8432x	op.fd = this->fd_;
400	8432x	op.target_endpoint = ep;
401	8432x	op.start(token, static_cast<Derived*>(this));
402	8432x	op.impl_ptr = this->shared_from_this();
403
404	8432x	this->register_op(
405	8432x	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
406	8432x	this->desc_state_.connect_cancel_pending, true);
407	8432x	return std::noop_coroutine();
408		}
409
410		template<
411		class Derived,
412		class Service,
413		class ConnOp,
414		class ReadOp,
415		class WriteOp,
416		class WaitOp,
417		class DescState,
418		class ImplBase,
419		class Endpoint>
420		std::coroutine_handle<>
421	233659x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
422		do_read_some(
423		std::coroutine_handle<> h,
424		capy::executor_ref ex,
425		buffer_param param,
426		std::stop_token const& token,
427		std::error_code* ec,
428		std::size_t* bytes_out)
429		{
430	233659x	auto& op = rd_;
431	233659x	op.reset();
432
433	233659x	capy::mutable_buffer bufs[ReadOp::max_buffers];
434	233659x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
435
436	233659x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
437		{
438	2x	op.empty_buffer_read = true;
439	2x	op.h = h;
440	2x	op.ex = ex;
441	2x	op.ec_out = ec;
442	2x	op.bytes_out = bytes_out;
443	2x	op.start(token, static_cast<Derived*>(this));
444	2x	op.impl_ptr = this->shared_from_this();
445	2x	op.complete(0, 0);
446	2x	this->svc_.post(&op);
447	2x	return std::noop_coroutine();
448		}
449
450	467314x	for (int i = 0; i < op.iovec_count; ++i)
451		{
452	233657x	op.iovecs[i].iov_base = bufs[i].data();
453	233657x	op.iovecs[i].iov_len = bufs[i].size();
454		}
455
456		// Speculative read; for the single-buffer case use recv() so the
457		// kernel skips the readv iov_iter setup.
458		ssize_t n;
459	233657x	if (op.iovec_count == 1)
460		{
461		do
462		{
463	233657x	n = ::recv(this->fd_, bufs[0].data(), bufs[0].size(), 0);
464		}
465	233657x	while (n < 0 && errno == EINTR);
466		}
467		else
468		{
469		do
470		{
471	✗	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
472		}
473	✗	while (n < 0 && errno == EINTR);
474		}
475
476	233657x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
477		{
478	233271x	int err = (n < 0) ? errno : 0;
479	233271x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
480
481	233271x	if (this->svc_.scheduler().try_consume_inline_budget())
482		{
483	186649x	if (err)
484	✗	*ec = make_err(err);
485	186649x	else if (n == 0)
486	10x	*ec = capy::error::eof;
487		else
488	186639x	*ec = {};
489	186649x	*bytes_out = bytes;
490	186649x	op.cont_op.cont.h = h;
491	186649x	return dispatch_coro(ex, op.cont_op.cont);
492		}
493	46622x	op.h = h;
494	46622x	op.ex = ex;
495	46622x	op.ec_out = ec;
496	46622x	op.bytes_out = bytes_out;
497	46622x	op.start(token, static_cast<Derived*>(this));
498	46622x	op.impl_ptr = this->shared_from_this();
499	46622x	op.complete(err, bytes);
500	46622x	this->svc_.post(&op);
501	46622x	return std::noop_coroutine();
502		}
503
504		// EAGAIN — register with reactor
505	386x	op.h = h;
506	386x	op.ex = ex;
507	386x	op.ec_out = ec;
508	386x	op.bytes_out = bytes_out;
509	386x	op.fd = this->fd_;
510	386x	op.start(token, static_cast<Derived*>(this));
511	386x	op.impl_ptr = this->shared_from_this();
512
513	386x	this->register_op(
514	386x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
515	386x	this->desc_state_.read_cancel_pending);
516	386x	return std::noop_coroutine();
517		}
518
519		template<
520		class Derived,
521		class Service,
522		class ConnOp,
523		class ReadOp,
524		class WriteOp,
525		class WaitOp,
526		class DescState,
527		class ImplBase,
528		class Endpoint>
529		std::coroutine_handle<>
530	233368x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
531		do_write_some(
532		std::coroutine_handle<> h,
533		capy::executor_ref ex,
534		buffer_param param,
535		std::stop_token const& token,
536		std::error_code* ec,
537		std::size_t* bytes_out)
538		{
539	233368x	auto& op = wr_;
540	233368x	op.reset();
541
542	233368x	capy::mutable_buffer bufs[WriteOp::max_buffers];
543	233368x	op.iovec_count =
544	233368x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
545
546	233368x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
547		{
548	2x	op.h = h;
549	2x	op.ex = ex;
550	2x	op.ec_out = ec;
551	2x	op.bytes_out = bytes_out;
552	2x	op.start(token, static_cast<Derived*>(this));
553	2x	op.impl_ptr = this->shared_from_this();
554	2x	op.complete(0, 0);
555	2x	this->svc_.post(&op);
556	2x	return std::noop_coroutine();
557		}
558
559	466732x	for (int i = 0; i < op.iovec_count; ++i)
560		{
561	233366x	op.iovecs[i].iov_base = bufs[i].data();
562	233366x	op.iovecs[i].iov_len = bufs[i].size();
563		}
564
565		// Speculative write; the single-buffer case dispatches to a
566		// backend-specific fast path so the kernel skips msghdr/iov_iter
567		// setup (and so each backend can pick the right SIGPIPE strategy).
568		ssize_t n;
569	233366x	if (op.iovec_count == 1)
570		{
571	466732x	n = WriteOp::write_policy::write_one(
572	233366x	this->fd_, bufs[0].data(), bufs[0].size());
573		}
574		else
575		{
576	✗	n = WriteOp::write_policy::write(
577	✗	this->fd_, op.iovecs, op.iovec_count);
578		}
579
580	233366x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
581		{
582	233366x	int err = (n < 0) ? errno : 0;
583	233366x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
584
585	233366x	if (this->svc_.scheduler().try_consume_inline_budget())
586		{
587	186704x	*ec = err ? make_err(err) : std::error_code{};
588	186704x	*bytes_out = bytes;
589	186704x	op.cont_op.cont.h = h;
590	186704x	return dispatch_coro(ex, op.cont_op.cont);
591		}
592	46662x	op.h = h;
593	46662x	op.ex = ex;
594	46662x	op.ec_out = ec;
595	46662x	op.bytes_out = bytes_out;
596	46662x	op.start(token, static_cast<Derived*>(this));
597	46662x	op.impl_ptr = this->shared_from_this();
598	46662x	op.complete(err, bytes);
599	46662x	this->svc_.post(&op);
600	46662x	return std::noop_coroutine();
601		}
602
603		// EAGAIN — register with reactor
604	✗	op.h = h;
605	✗	op.ex = ex;
606	✗	op.ec_out = ec;
607	✗	op.bytes_out = bytes_out;
608	✗	op.fd = this->fd_;
609	✗	op.start(token, static_cast<Derived*>(this));
610	✗	op.impl_ptr = this->shared_from_this();
611
612	✗	this->register_op(
613	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
614	✗	this->desc_state_.write_cancel_pending, true);
615	✗	return std::noop_coroutine();
616		}
617
618		template<
619		class Derived,
620		class Service,
621		class ConnOp,
622		class ReadOp,
623		class WriteOp,
624		class WaitOp,
625		class DescState,
626		class ImplBase,
627		class Endpoint>
628		std::coroutine_handle<>
629	8x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, WaitOp, DescState, ImplBase, Endpoint>::
630		do_wait(
631		std::coroutine_handle<> h,
632		capy::executor_ref ex,
633		wait_type w,
634		std::stop_token const& token,
635		std::error_code* ec)
636		{
637		// wait_type::write completes immediately on a connected socket,
638		// matching asio's behavior on IOCP. Corosio's reactor backends use
639		// edge-triggered EPOLLOUT, which would never fire on an already-
640		// writable socket; an immediate completion is also a more useful
641		// contract than parking until a non-writable -> writable transition.
642	8x	if (w == wait_type::write)
643		{
644	2x	auto& op = wait_wr_;
645	2x	if (this->svc_.scheduler().try_consume_inline_budget())
646		{
647	✗	*ec = std::error_code{};
648	✗	op.cont_op.cont.h = h;
649	✗	return dispatch_coro(ex, op.cont_op.cont);
650		}
651	2x	op.reset();
652	2x	op.wait_event = reactor_event_write;
653	2x	op.h = h;
654	2x	op.ex = ex;
655	2x	op.ec_out = ec;
656	2x	op.fd = this->fd_;
657	2x	op.start(token, static_cast<Derived*>(this));
658	2x	op.impl_ptr = this->shared_from_this();
659	2x	op.complete(0, 0);
660	2x	this->svc_.post(&op);
661	2x	return std::noop_coroutine();
662		}
663
664		// Pick refs up-front to avoid duplicating the register_op call.
665		WaitOp* op_ptr;
666		reactor_op_base** desc_slot_ptr;
667		bool* ready_flag_ptr;
668		bool* cancel_flag_ptr;
669		std::uint32_t event;
670
671	6x	bool dummy_ready = false; // placeholder for error waits (no cached edge)
672
673	6x	if (w == wait_type::read)
674		{
675	6x	op_ptr = &wait_rd_;
676	6x	desc_slot_ptr = &this->desc_state_.wait_read_op;
677	6x	ready_flag_ptr = &this->desc_state_.read_ready;
678	6x	cancel_flag_ptr = &this->desc_state_.wait_read_cancel_pending;
679	6x	event = reactor_event_read;
680		}
681		else // wait_type::error
682		{
683	✗	op_ptr = &wait_er_;
684	✗	desc_slot_ptr = &this->desc_state_.wait_error_op;
685	✗	ready_flag_ptr = &dummy_ready;
686	✗	cancel_flag_ptr = &this->desc_state_.wait_error_cancel_pending;
687	✗	event = reactor_event_error;
688		}
689
690	6x	auto& op = *op_ptr;
691	6x	op.reset();
692	6x	op.wait_event = event;
693	6x	op.h = h;
694	6x	op.ex = ex;
695	6x	op.ec_out = ec;
696	6x	op.fd = this->fd_;
697	6x	op.start(token, static_cast<Derived*>(this));
698	6x	op.impl_ptr = this->shared_from_this();
699
700	6x	this->register_op(op, desc_slot_ptr, ready_flag_ptr, *cancel_flag_ptr,
701		false);
702	6x	return std::noop_coroutine();
703		}
704
705		} // namespace boost::corosio::detail
706
707		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
708