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

83.7% Lines (180/215) 77.5% List of functions (62/80)

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

reactor_stream_socket.hpp

f(x) Functions (80)

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/native/detail/reactor/reactor_basic_socket.hpp>
16		#include <boost/corosio/detail/dispatch_coro.hpp>
17		#include <boost/capy/buffers.hpp>
18
19		#include <coroutine>
20
21		#include <errno.h>
22		#include <sys/socket.h>
23		#include <sys/uio.h>
24
25		namespace boost::corosio::detail {
26
27		/** CRTP base for reactor-backed stream socket implementations.
28
29		Inherits shared data members and cancel/close/register logic
30		from reactor_basic_socket. Adds the stream-specific remote
31		endpoint, shutdown, and I/O dispatch (connect, read, write).
32
33		@tparam Derived The concrete socket type (CRTP).
34		@tparam Service The backend's socket service type.
35		@tparam ConnOp The backend's connect op type.
36		@tparam ReadOp The backend's read op type.
37		@tparam WriteOp The backend's write op type.
38		@tparam DescState The backend's descriptor_state type.
39		@tparam ImplBase The public vtable base
40		(tcp_socket::implementation or
41		local_stream_socket::implementation).
42		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
43		*/
44		template<
45		class Derived,
46		class Service,
47		class ConnOp,
48		class ReadOp,
49		class WriteOp,
50		class DescState,
51		class ImplBase = tcp_socket::implementation,
52		class Endpoint = endpoint>
53		class reactor_stream_socket
54		: public reactor_basic_socket<
55		Derived,
56		ImplBase,
57		Service,
58		DescState,
59		Endpoint>
60		{
61		using base_type = reactor_basic_socket<
62		Derived,
63		ImplBase,
64		Service,
65		DescState,
66		Endpoint>;
67		friend base_type;
68		friend Derived;
69
70		protected:
71		// NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
72	11099x	explicit reactor_stream_socket(Service& svc) noexcept : base_type(svc) {}
73
74		protected:
75		Endpoint remote_endpoint_;
76
77		public:
78		/// Pending connect operation slot.
79		ConnOp conn_;
80
81		/// Pending read operation slot.
82		ReadOp rd_;
83
84		/// Pending write operation slot.
85		WriteOp wr_;
86
87	11099x	~reactor_stream_socket() override = default;
88
89		/// Return the cached remote endpoint.
90	46x	Endpoint remote_endpoint() const noexcept override
91		{
92	46x	return remote_endpoint_;
93		}
94
95		// --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
96
97	3672x	std::coroutine_handle<> connect(
98		std::coroutine_handle<> h,
99		capy::executor_ref ex,
100		Endpoint ep,
101		std::stop_token token,
102		std::error_code* ec) override
103		{
104	3672x	return do_connect(h, ex, ep, token, ec);
105		}
106
107	254529x	std::coroutine_handle<> read_some(
108		std::coroutine_handle<> h,
109		capy::executor_ref ex,
110		buffer_param param,
111		std::stop_token token,
112		std::error_code* ec,
113		std::size_t* bytes_out) override
114		{
115	254529x	return do_read_some(h, ex, param, token, ec, bytes_out);
116		}
117
118	254235x	std::coroutine_handle<> write_some(
119		std::coroutine_handle<> h,
120		capy::executor_ref ex,
121		buffer_param param,
122		std::stop_token token,
123		std::error_code* ec,
124		std::size_t* bytes_out) override
125		{
126	254235x	return do_write_some(h, ex, param, token, ec, bytes_out);
127		}
128
129		std::error_code
130	6x	shutdown(corosio::shutdown_type what) noexcept override
131		{
132	6x	return do_shutdown(static_cast<int>(what));
133		}
134
135	183x	void cancel() noexcept override
136		{
137	183x	this->do_cancel();
138	183x	}
139
140		// --- End virtual overrides ---
141
142		/// Close the socket (non-virtual, called by the service).
143		void close_socket() noexcept
144		{
145		this->do_close_socket();
146		}
147
148		/** Shut down part or all of the full-duplex connection.
149
150		@param what 0 = receive, 1 = send, 2 = both.
151		*/
152	6x	std::error_code do_shutdown(int what) noexcept
153		{
154		int how;
155	6x	switch (what)
156		{
157	2x	case 0: // shutdown_receive
158	2x	how = SHUT_RD;
159	2x	break;
160	2x	case 1: // shutdown_send
161	2x	how = SHUT_WR;
162	2x	break;
163	2x	case 2: // shutdown_both
164	2x	how = SHUT_RDWR;
165	2x	break;
166	✗	default:
167	✗	return make_err(EINVAL);
168		}
169	6x	if (::shutdown(this->fd_, how) != 0)
170	✗	return make_err(errno);
171	6x	return {};
172		}
173
174		/// Cache local and remote endpoints.
175	7326x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
176		{
177	7326x	this->local_endpoint_ = std::move(local);
178	7326x	remote_endpoint_ = std::move(remote);
179	7326x	}
180
181		/** Shared connect dispatch.
182
183		Tries the connect syscall speculatively. On synchronous
184		completion, returns via inline budget or posts through queue.
185		On EINPROGRESS, registers with the reactor.
186		*/
187		std::coroutine_handle<> do_connect(
188		std::coroutine_handle<>,
189		capy::executor_ref,
190		Endpoint const&,
191		std::stop_token const&,
192		std::error_code*);
193
194		/** Shared scatter-read dispatch.
195
196		Tries readv() speculatively. On success or hard error,
197		returns via inline budget or posts through queue.
198		On EAGAIN, registers with the reactor.
199		*/
200		std::coroutine_handle<> do_read_some(
201		std::coroutine_handle<>,
202		capy::executor_ref,
203		buffer_param,
204		std::stop_token const&,
205		std::error_code*,
206		std::size_t*);
207
208		/** Shared gather-write dispatch.
209
210		Tries the write via WriteOp::write_policy speculatively.
211		On success or hard error, returns via inline budget or
212		posts through queue. On EAGAIN, registers with the reactor.
213		*/
214		std::coroutine_handle<> do_write_some(
215		std::coroutine_handle<>,
216		capy::executor_ref,
217		buffer_param,
218		std::stop_token const&,
219		std::error_code*,
220		std::size_t*);
221
222		/** Close the socket and cancel pending operations.
223
224		Extends the base do_close_socket() to also reset
225		the remote endpoint.
226		*/
227	33315x	void do_close_socket() noexcept
228		{
229	33315x	base_type::do_close_socket();
230	33315x	remote_endpoint_ = Endpoint{};
231	33315x	}
232
233		private:
234		// CRTP callbacks for reactor_basic_socket cancel/close
235
236		template<class Op>
237	193x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
238		{
239	193x	if (&op == static_cast<void*>(&conn_))
240	✗	return &this->desc_state_.connect_op;
241	193x	if (&op == static_cast<void*>(&rd_))
242	193x	return &this->desc_state_.read_op;
243	✗	if (&op == static_cast<void*>(&wr_))
244	✗	return &this->desc_state_.write_op;
245	✗	return nullptr;
246		}
247
248		template<class Op>
249	✗	bool* op_to_cancel_flag(Op& op) noexcept
250		{
251	✗	if (&op == static_cast<void*>(&conn_))
252	✗	return &this->desc_state_.connect_cancel_pending;
253	✗	if (&op == static_cast<void*>(&rd_))
254	✗	return &this->desc_state_.read_cancel_pending;
255	✗	if (&op == static_cast<void*>(&wr_))
256	✗	return &this->desc_state_.write_cancel_pending;
257	✗	return nullptr;
258		}
259
260		template<class Fn>
261	33500x	void for_each_op(Fn fn) noexcept
262		{
263	33500x	fn(conn_);
264	33500x	fn(rd_);
265	33500x	fn(wr_);
266	33500x	}
267
268		template<class Fn>
269	33500x	void for_each_desc_entry(Fn fn) noexcept
270		{
271	33500x	fn(conn_, this->desc_state_.connect_op);
272	33500x	fn(rd_, this->desc_state_.read_op);
273	33500x	fn(wr_, this->desc_state_.write_op);
274	33500x	}
275		};
276
277		template<
278		class Derived,
279		class Service,
280		class ConnOp,
281		class ReadOp,
282		class WriteOp,
283		class DescState,
284		class ImplBase,
285		class Endpoint>
286		std::coroutine_handle<>
287	3672x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState, ImplBase, Endpoint>::
288		do_connect(
289		std::coroutine_handle<> h,
290		capy::executor_ref ex,
291		Endpoint const& ep,
292		std::stop_token const& token,
293		std::error_code* ec)
294		{
295	3672x	auto& op = conn_;
296
297	3672x	sockaddr_storage storage{};
298	3672x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
299		int result =
300	3672x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
301
302	3672x	if (result == 0)
303		{
304	4x	sockaddr_storage local_storage{};
305	4x	socklen_t local_len = sizeof(local_storage);
306	4x	if (::getsockname(
307		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
308	4x	&local_len) == 0)
309	✗	this->local_endpoint_ =
310	4x	from_sockaddr_as(local_storage, local_len, Endpoint{});
311	4x	remote_endpoint_ = ep;
312		}
313
314	3672x	if (result == 0 \|\| errno != EINPROGRESS)
315		{
316	4x	int err = (result < 0) ? errno : 0;
317	4x	if (this->svc_.scheduler().try_consume_inline_budget())
318		{
319	✗	*ec = err ? make_err(err) : std::error_code{};
320	✗	op.cont_op.cont.h = h;
321	✗	return dispatch_coro(ex, op.cont_op.cont);
322		}
323	4x	op.reset();
324	4x	op.h = h;
325	4x	op.ex = ex;
326	4x	op.ec_out = ec;
327	4x	op.fd = this->fd_;
328	4x	op.target_endpoint = ep;
329	4x	op.start(token, static_cast<Derived*>(this));
330	4x	op.impl_ptr = this->shared_from_this();
331	4x	op.complete(err, 0);
332	4x	this->svc_.post(&op);
333	4x	return std::noop_coroutine();
334		}
335
336		// EINPROGRESS — register with reactor
337	3668x	op.reset();
338	3668x	op.h = h;
339	3668x	op.ex = ex;
340	3668x	op.ec_out = ec;
341	3668x	op.fd = this->fd_;
342	3668x	op.target_endpoint = ep;
343	3668x	op.start(token, static_cast<Derived*>(this));
344	3668x	op.impl_ptr = this->shared_from_this();
345
346	3668x	this->register_op(
347	3668x	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
348	3668x	this->desc_state_.connect_cancel_pending, true);
349	3668x	return std::noop_coroutine();
350		}
351
352		template<
353		class Derived,
354		class Service,
355		class ConnOp,
356		class ReadOp,
357		class WriteOp,
358		class DescState,
359		class ImplBase,
360		class Endpoint>
361		std::coroutine_handle<>
362	254529x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState, ImplBase, Endpoint>::
363		do_read_some(
364		std::coroutine_handle<> h,
365		capy::executor_ref ex,
366		buffer_param param,
367		std::stop_token const& token,
368		std::error_code* ec,
369		std::size_t* bytes_out)
370		{
371	254529x	auto& op = rd_;
372	254529x	op.reset();
373
374	254529x	capy::mutable_buffer bufs[ReadOp::max_buffers];
375	254529x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
376
377	254529x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
378		{
379	2x	op.empty_buffer_read = true;
380	2x	op.h = h;
381	2x	op.ex = ex;
382	2x	op.ec_out = ec;
383	2x	op.bytes_out = bytes_out;
384	2x	op.start(token, static_cast<Derived*>(this));
385	2x	op.impl_ptr = this->shared_from_this();
386	2x	op.complete(0, 0);
387	2x	this->svc_.post(&op);
388	2x	return std::noop_coroutine();
389		}
390
391	509054x	for (int i = 0; i < op.iovec_count; ++i)
392		{
393	254527x	op.iovecs[i].iov_base = bufs[i].data();
394	254527x	op.iovecs[i].iov_len = bufs[i].size();
395		}
396
397		// Speculative read; for the single-buffer case use recv() so the
398		// kernel skips the readv iov_iter setup.
399		ssize_t n;
400	254527x	if (op.iovec_count == 1)
401		{
402		do
403		{
404	254527x	n = ::recv(this->fd_, bufs[0].data(), bufs[0].size(), 0);
405		}
406	254527x	while (n < 0 && errno == EINTR);
407		}
408		else
409		{
410		do
411		{
412	✗	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
413		}
414	✗	while (n < 0 && errno == EINTR);
415		}
416
417	254527x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
418		{
419	254142x	int err = (n < 0) ? errno : 0;
420	254142x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
421
422	254142x	if (this->svc_.scheduler().try_consume_inline_budget())
423		{
424	203346x	if (err)
425	✗	*ec = make_err(err);
426	203346x	else if (n == 0)
427	10x	*ec = capy::error::eof;
428		else
429	203336x	*ec = {};
430	203346x	*bytes_out = bytes;
431	203346x	op.cont_op.cont.h = h;
432	203346x	return dispatch_coro(ex, op.cont_op.cont);
433		}
434	50796x	op.h = h;
435	50796x	op.ex = ex;
436	50796x	op.ec_out = ec;
437	50796x	op.bytes_out = bytes_out;
438	50796x	op.start(token, static_cast<Derived*>(this));
439	50796x	op.impl_ptr = this->shared_from_this();
440	50796x	op.complete(err, bytes);
441	50796x	this->svc_.post(&op);
442	50796x	return std::noop_coroutine();
443		}
444
445		// EAGAIN — register with reactor
446	385x	op.h = h;
447	385x	op.ex = ex;
448	385x	op.ec_out = ec;
449	385x	op.bytes_out = bytes_out;
450	385x	op.fd = this->fd_;
451	385x	op.start(token, static_cast<Derived*>(this));
452	385x	op.impl_ptr = this->shared_from_this();
453
454	385x	this->register_op(
455	385x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
456	385x	this->desc_state_.read_cancel_pending);
457	385x	return std::noop_coroutine();
458		}
459
460		template<
461		class Derived,
462		class Service,
463		class ConnOp,
464		class ReadOp,
465		class WriteOp,
466		class DescState,
467		class ImplBase,
468		class Endpoint>
469		std::coroutine_handle<>
470	254235x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState, ImplBase, Endpoint>::
471		do_write_some(
472		std::coroutine_handle<> h,
473		capy::executor_ref ex,
474		buffer_param param,
475		std::stop_token const& token,
476		std::error_code* ec,
477		std::size_t* bytes_out)
478		{
479	254235x	auto& op = wr_;
480	254235x	op.reset();
481
482	254235x	capy::mutable_buffer bufs[WriteOp::max_buffers];
483	254235x	op.iovec_count =
484	254235x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
485
486	254235x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
487		{
488	2x	op.h = h;
489	2x	op.ex = ex;
490	2x	op.ec_out = ec;
491	2x	op.bytes_out = bytes_out;
492	2x	op.start(token, static_cast<Derived*>(this));
493	2x	op.impl_ptr = this->shared_from_this();
494	2x	op.complete(0, 0);
495	2x	this->svc_.post(&op);
496	2x	return std::noop_coroutine();
497		}
498
499	508466x	for (int i = 0; i < op.iovec_count; ++i)
500		{
501	254233x	op.iovecs[i].iov_base = bufs[i].data();
502	254233x	op.iovecs[i].iov_len = bufs[i].size();
503		}
504
505		// Speculative write; the single-buffer case dispatches to a
506		// backend-specific fast path so the kernel skips msghdr/iov_iter
507		// setup (and so each backend can pick the right SIGPIPE strategy).
508		ssize_t n;
509	254233x	if (op.iovec_count == 1)
510		{
511	508466x	n = WriteOp::write_policy::write_one(
512	254233x	this->fd_, bufs[0].data(), bufs[0].size());
513		}
514		else
515		{
516	✗	n = WriteOp::write_policy::write(
517	✗	this->fd_, op.iovecs, op.iovec_count);
518		}
519
520	254233x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
521		{
522	254233x	int err = (n < 0) ? errno : 0;
523	254233x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
524
525	254233x	if (this->svc_.scheduler().try_consume_inline_budget())
526		{
527	203400x	*ec = err ? make_err(err) : std::error_code{};
528	203400x	*bytes_out = bytes;
529	203400x	op.cont_op.cont.h = h;
530	203400x	return dispatch_coro(ex, op.cont_op.cont);
531		}
532	50833x	op.h = h;
533	50833x	op.ex = ex;
534	50833x	op.ec_out = ec;
535	50833x	op.bytes_out = bytes_out;
536	50833x	op.start(token, static_cast<Derived*>(this));
537	50833x	op.impl_ptr = this->shared_from_this();
538	50833x	op.complete(err, bytes);
539	50833x	this->svc_.post(&op);
540	50833x	return std::noop_coroutine();
541		}
542
543		// EAGAIN — register with reactor
544	✗	op.h = h;
545	✗	op.ex = ex;
546	✗	op.ec_out = ec;
547	✗	op.bytes_out = bytes_out;
548	✗	op.fd = this->fd_;
549	✗	op.start(token, static_cast<Derived*>(this));
550	✗	op.impl_ptr = this->shared_from_this();
551
552	✗	this->register_op(
553	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
554	✗	this->desc_state_.write_cancel_pending, true);
555	✗	return std::noop_coroutine();
556		}
557
558		} // namespace boost::corosio::detail
559
560		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
561