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

52.7% Lines (437/830) 60.5% List of functions (46/76)

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

reactor_stream_socket.hpp

f(x) Functions (76)

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