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

64.7% Lines (277/428) 72.2% List of functions (26/36)

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

reactor_stream_socket.hpp

f(x) Functions (36)

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/native/detail/reactor/reactor_basic_socket.hpp>
15		#include <boost/corosio/detail/dispatch_coro.hpp>
16		#include <boost/capy/buffers.hpp>
17
18		#include <coroutine>
19
20		#include <errno.h>
21		#include <sys/socket.h>
22		#include <sys/uio.h>
23
24		namespace boost::corosio::detail {
25
26		/** CRTP base for reactor-backed stream socket implementations.
27
28		Inherits shared data members and cancel/close/register logic
29		from reactor_basic_socket. Adds the TCP-specific remote
30		endpoint, shutdown, and I/O dispatch (connect, read, write).
31
32		@tparam Derived The concrete socket type (CRTP).
33		@tparam Service The backend's socket service type.
34		@tparam ConnOp The backend's connect op type.
35		@tparam ReadOp The backend's read op type.
36		@tparam WriteOp The backend's write op type.
37		@tparam DescState The backend's descriptor_state type.
38		*/
39		template<
40		class Derived,
41		class Service,
42		class ConnOp,
43		class ReadOp,
44		class WriteOp,
45		class DescState>
46		class reactor_stream_socket
47		: public reactor_basic_socket<
48		Derived,
49		tcp_socket::implementation,
50		Service,
51		DescState>
52		{
53		using base_type = reactor_basic_socket<
54		Derived,
55		tcp_socket::implementation,
56		Service,
57		DescState>;
58		friend base_type;
59		friend Derived;
60
61	23838x	explicit reactor_stream_socket(Service& svc) noexcept : base_type(svc) {}
62
63		protected:
64		endpoint remote_endpoint_;
65
66		public:
67		/// Pending connect operation slot.
68		ConnOp conn_;
69
70		/// Pending read operation slot.
71		ReadOp rd_;
72
73		/// Pending write operation slot.
74		WriteOp wr_;
75
76	23838x	~reactor_stream_socket() override = default;
77
78		/// Return the cached remote endpoint.
79	42x	endpoint remote_endpoint() const noexcept override
80		{
81	42x	return remote_endpoint_;
82		}
83
84		/// Shut down part or all of the full-duplex connection.
85	6x	std::error_code shutdown(tcp_socket::shutdown_type what) noexcept override
86		{
87		int how;
88	6x	switch (what)
89		{
90	2x	case tcp_socket::shutdown_receive:
91	2x	how = SHUT_RD;
92	2x	break;
93	2x	case tcp_socket::shutdown_send:
94	2x	how = SHUT_WR;
95	2x	break;
96	2x	case tcp_socket::shutdown_both:
97	2x	how = SHUT_RDWR;
98	2x	break;
99	✗	default:
100	✗	return make_err(EINVAL);
101		}
102	6x	if (::shutdown(this->fd_, how) != 0)
103	✗	return make_err(errno);
104	6x	return {};
105		}
106
107		/// Cache local and remote endpoints.
108	15832x	void set_endpoints(endpoint local, endpoint remote) noexcept
109		{
110	15832x	this->local_endpoint_ = local;
111	15832x	remote_endpoint_ = remote;
112	15832x	}
113
114		/** Shared connect dispatch.
115
116		Tries the connect syscall speculatively. On synchronous
117		completion, returns via inline budget or posts through queue.
118		On EINPROGRESS, registers with the reactor.
119		*/
120		std::coroutine_handle<> do_connect(
121		std::coroutine_handle<>,
122		capy::executor_ref,
123		endpoint,
124		std::stop_token const&,
125		std::error_code*);
126
127		/** Shared scatter-read dispatch.
128
129		Tries readv() speculatively. On success or hard error,
130		returns via inline budget or posts through queue.
131		On EAGAIN, registers with the reactor.
132		*/
133		std::coroutine_handle<> do_read_some(
134		std::coroutine_handle<>,
135		capy::executor_ref,
136		buffer_param,
137		std::stop_token const&,
138		std::error_code*,
139		std::size_t*);
140
141		/** Shared gather-write dispatch.
142
143		Tries the write via WriteOp::write_policy speculatively.
144		On success or hard error, returns via inline budget or
145		posts through queue. On EAGAIN, registers with the reactor.
146		*/
147		std::coroutine_handle<> do_write_some(
148		std::coroutine_handle<>,
149		capy::executor_ref,
150		buffer_param,
151		std::stop_token const&,
152		std::error_code*,
153		std::size_t*);
154
155		/** Close the socket and cancel pending operations.
156
157		Extends the base do_close_socket() to also reset
158		the remote endpoint.
159		*/
160	71494x	void do_close_socket() noexcept
161		{
162	71494x	base_type::do_close_socket();
163	71494x	remote_endpoint_ = endpoint{};
164	71494x	}
165
166		private:
167		// CRTP callbacks for reactor_basic_socket cancel/close
168
169		template<class Op>
170	192x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
171		{
172	192x	if (&op == static_cast<void*>(&conn_))
173	✗	return &this->desc_state_.connect_op;
174	192x	if (&op == static_cast<void*>(&rd_))
175	192x	return &this->desc_state_.read_op;
176	✗	if (&op == static_cast<void*>(&wr_))
177	✗	return &this->desc_state_.write_op;
178	✗	return nullptr;
179		}
180
181		template<class Op>
182	✗	bool* op_to_cancel_flag(Op& op) noexcept
183		{
184	✗	if (&op == static_cast<void*>(&conn_))
185	✗	return &this->desc_state_.connect_cancel_pending;
186	✗	if (&op == static_cast<void*>(&rd_))
187	✗	return &this->desc_state_.read_cancel_pending;
188	✗	if (&op == static_cast<void*>(&wr_))
189	✗	return &this->desc_state_.write_cancel_pending;
190	✗	return nullptr;
191		}
192
193		template<class Fn>
194	71679x	void for_each_op(Fn fn) noexcept
195		{
196	71679x	fn(conn_);
197	71679x	fn(rd_);
198	71679x	fn(wr_);
199	71679x	}
200
201		template<class Fn>
202	71679x	void for_each_desc_entry(Fn fn) noexcept
203		{
204	71679x	fn(conn_, this->desc_state_.connect_op);
205	71679x	fn(rd_, this->desc_state_.read_op);
206	71679x	fn(wr_, this->desc_state_.write_op);
207	71679x	}
208		};
209
210		template<
211		class Derived,
212		class Service,
213		class ConnOp,
214		class ReadOp,
215		class WriteOp,
216		class DescState>
217		std::coroutine_handle<>
218	7921x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState>::
219		do_connect(
220		std::coroutine_handle<> h,
221		capy::executor_ref ex,
222		endpoint ep,
223		std::stop_token const& token,
224		std::error_code* ec)
225		{
226	7921x	auto& op = conn_;
227
228	7921x	sockaddr_storage storage{};
229	7921x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
230		int result =
231	7921x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
232
233	7921x	if (result == 0)
234		{
235	✗	sockaddr_storage local_storage{};
236	✗	socklen_t local_len = sizeof(local_storage);
237	✗	if (::getsockname(
238		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
239	✗	&local_len) == 0)
240	✗	this->local_endpoint_ = from_sockaddr(local_storage);
241	✗	remote_endpoint_ = ep;
242		}
243
244	7921x	if (result == 0 \|\| errno != EINPROGRESS)
245		{
246	✗	int err = (result < 0) ? errno : 0;
247	✗	if (this->svc_.scheduler().try_consume_inline_budget())
248		{
249	✗	*ec = err ? make_err(err) : std::error_code{};
250	✗	op.cont_op.cont.h = h;
251	✗	return dispatch_coro(ex, op.cont_op.cont);
252		}
253	✗	op.reset();
254	✗	op.h = h;
255	✗	op.ex = ex;
256	✗	op.ec_out = ec;
257	✗	op.fd = this->fd_;
258	✗	op.target_endpoint = ep;
259	✗	op.start(token, static_cast<Derived*>(this));
260	✗	op.impl_ptr = this->shared_from_this();
261	✗	op.complete(err, 0);
262	✗	this->svc_.post(&op);
263	✗	return std::noop_coroutine();
264		}
265
266		// EINPROGRESS — register with reactor
267	7921x	op.reset();
268	7921x	op.h = h;
269	7921x	op.ex = ex;
270	7921x	op.ec_out = ec;
271	7921x	op.fd = this->fd_;
272	7921x	op.target_endpoint = ep;
273	7921x	op.start(token, static_cast<Derived*>(this));
274	7921x	op.impl_ptr = this->shared_from_this();
275
276	7921x	this->register_op(
277	7921x	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
278	7921x	this->desc_state_.connect_cancel_pending);
279	7921x	return std::noop_coroutine();
280		}
281
282		template<
283		class Derived,
284		class Service,
285		class ConnOp,
286		class ReadOp,
287		class WriteOp,
288		class DescState>
289		std::coroutine_handle<>
290	207017x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState>::
291		do_read_some(
292		std::coroutine_handle<> h,
293		capy::executor_ref ex,
294		buffer_param param,
295		std::stop_token const& token,
296		std::error_code* ec,
297		std::size_t* bytes_out)
298		{
299	207017x	auto& op = rd_;
300	207017x	op.reset();
301
302	207017x	capy::mutable_buffer bufs[ReadOp::max_buffers];
303	207017x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
304
305	207017x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
306		{
307	2x	op.empty_buffer_read = true;
308	2x	op.h = h;
309	2x	op.ex = ex;
310	2x	op.ec_out = ec;
311	2x	op.bytes_out = bytes_out;
312	2x	op.start(token, static_cast<Derived*>(this));
313	2x	op.impl_ptr = this->shared_from_this();
314	2x	op.complete(0, 0);
315	2x	this->svc_.post(&op);
316	2x	return std::noop_coroutine();
317		}
318
319	414030x	for (int i = 0; i < op.iovec_count; ++i)
320		{
321	207015x	op.iovecs[i].iov_base = bufs[i].data();
322	207015x	op.iovecs[i].iov_len = bufs[i].size();
323		}
324
325		// Speculative read
326		ssize_t n;
327		do
328		{
329	207015x	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
330		}
331	207015x	while (n < 0 && errno == EINTR);
332
333	207015x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
334		{
335	206627x	int err = (n < 0) ? errno : 0;
336	206627x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
337
338	206627x	if (this->svc_.scheduler().try_consume_inline_budget())
339		{
340	165336x	if (err)
341	✗	*ec = make_err(err);
342	165336x	else if (n == 0)
343	10x	*ec = capy::error::eof;
344		else
345	165326x	*ec = {};
346	165336x	*bytes_out = bytes;
347	165336x	op.cont_op.cont.h = h;
348	165336x	return dispatch_coro(ex, op.cont_op.cont);
349		}
350	41291x	op.h = h;
351	41291x	op.ex = ex;
352	41291x	op.ec_out = ec;
353	41291x	op.bytes_out = bytes_out;
354	41291x	op.start(token, static_cast<Derived*>(this));
355	41291x	op.impl_ptr = this->shared_from_this();
356	41291x	op.complete(err, bytes);
357	41291x	this->svc_.post(&op);
358	41291x	return std::noop_coroutine();
359		}
360
361		// EAGAIN — register with reactor
362	388x	op.h = h;
363	388x	op.ex = ex;
364	388x	op.ec_out = ec;
365	388x	op.bytes_out = bytes_out;
366	388x	op.fd = this->fd_;
367	388x	op.start(token, static_cast<Derived*>(this));
368	388x	op.impl_ptr = this->shared_from_this();
369
370	388x	this->register_op(
371	388x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
372	388x	this->desc_state_.read_cancel_pending);
373	388x	return std::noop_coroutine();
374		}
375
376		template<
377		class Derived,
378		class Service,
379		class ConnOp,
380		class ReadOp,
381		class WriteOp,
382		class DescState>
383		std::coroutine_handle<>
384	206719x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState>::
385		do_write_some(
386		std::coroutine_handle<> h,
387		capy::executor_ref ex,
388		buffer_param param,
389		std::stop_token const& token,
390		std::error_code* ec,
391		std::size_t* bytes_out)
392		{
393	206719x	auto& op = wr_;
394	206719x	op.reset();
395
396	206719x	capy::mutable_buffer bufs[WriteOp::max_buffers];
397	206719x	op.iovec_count =
398	206719x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
399
400	206719x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
401		{
402	2x	op.h = h;
403	2x	op.ex = ex;
404	2x	op.ec_out = ec;
405	2x	op.bytes_out = bytes_out;
406	2x	op.start(token, static_cast<Derived*>(this));
407	2x	op.impl_ptr = this->shared_from_this();
408	2x	op.complete(0, 0);
409	2x	this->svc_.post(&op);
410	2x	return std::noop_coroutine();
411		}
412
413	413434x	for (int i = 0; i < op.iovec_count; ++i)
414		{
415	206717x	op.iovecs[i].iov_base = bufs[i].data();
416	206717x	op.iovecs[i].iov_len = bufs[i].size();
417		}
418
419		// Speculative write via backend-specific write policy
420		ssize_t n =
421	206717x	WriteOp::write_policy::write(this->fd_, op.iovecs, op.iovec_count);
422
423	206717x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
424		{
425	206717x	int err = (n < 0) ? errno : 0;
426	206717x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
427
428	206717x	if (this->svc_.scheduler().try_consume_inline_budget())
429		{
430	165391x	*ec = err ? make_err(err) : std::error_code{};
431	165391x	*bytes_out = bytes;
432	165391x	op.cont_op.cont.h = h;
433	165391x	return dispatch_coro(ex, op.cont_op.cont);
434		}
435	41326x	op.h = h;
436	41326x	op.ex = ex;
437	41326x	op.ec_out = ec;
438	41326x	op.bytes_out = bytes_out;
439	41326x	op.start(token, static_cast<Derived*>(this));
440	41326x	op.impl_ptr = this->shared_from_this();
441	41326x	op.complete(err, bytes);
442	41326x	this->svc_.post(&op);
443	41326x	return std::noop_coroutine();
444		}
445
446		// EAGAIN — register with reactor
447	✗	op.h = h;
448	✗	op.ex = ex;
449	✗	op.ec_out = ec;
450	✗	op.bytes_out = bytes_out;
451	✗	op.fd = this->fd_;
452	✗	op.start(token, static_cast<Derived*>(this));
453	✗	op.impl_ptr = this->shared_from_this();
454
455	✗	this->register_op(
456	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
457	✗	this->desc_state_.write_cancel_pending);
458	✗	return std::noop_coroutine();
459		}
460
461		} // namespace boost::corosio::detail
462
463		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
464