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

85.1% Lines (177/208) 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	22189x	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	22189x	~reactor_stream_socket() override = default;
88
89		/// Return the cached remote endpoint.
90	44x	Endpoint remote_endpoint() const noexcept override
91		{
92	44x	return remote_endpoint_;
93		}
94
95		// --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
96
97	7360x	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	7360x	return do_connect(h, ex, ep, token, ec);
105		}
106
107	217409x	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	217409x	return do_read_some(h, ex, param, token, ec, bytes_out);
116		}
117
118	217112x	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	217112x	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	186x	void cancel() noexcept override
136		{
137	186x	this->do_cancel();
138	186x	}
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	14726x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
176		{
177	14726x	this->local_endpoint_ = std::move(local);
178	14726x	remote_endpoint_ = std::move(remote);
179	14726x	}
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	66571x	void do_close_socket() noexcept
228		{
229	66571x	base_type::do_close_socket();
230	66571x	remote_endpoint_ = Endpoint{};
231	66571x	}
232
233		private:
234		// CRTP callbacks for reactor_basic_socket cancel/close
235
236		template<class Op>
237	192x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
238		{
239	192x	if (&op == static_cast<void*>(&conn_))
240	✗	return &this->desc_state_.connect_op;
241	192x	if (&op == static_cast<void*>(&rd_))
242	192x	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	66759x	void for_each_op(Fn fn) noexcept
262		{
263	66759x	fn(conn_);
264	66759x	fn(rd_);
265	66759x	fn(wr_);
266	66759x	}
267
268		template<class Fn>
269	66759x	void for_each_desc_entry(Fn fn) noexcept
270		{
271	66759x	fn(conn_, this->desc_state_.connect_op);
272	66759x	fn(rd_, this->desc_state_.read_op);
273	66759x	fn(wr_, this->desc_state_.write_op);
274	66759x	}
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	7360x	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	7360x	auto& op = conn_;
296
297	7360x	sockaddr_storage storage{};
298	7360x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
299		int result =
300	7360x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
301
302	7360x	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	7360x	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	7356x	op.reset();
338	7356x	op.h = h;
339	7356x	op.ex = ex;
340	7356x	op.ec_out = ec;
341	7356x	op.fd = this->fd_;
342	7356x	op.target_endpoint = ep;
343	7356x	op.start(token, static_cast<Derived*>(this));
344	7356x	op.impl_ptr = this->shared_from_this();
345
346	7356x	this->register_op(
347	7356x	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
348	7356x	this->desc_state_.connect_cancel_pending, true);
349	7356x	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	217409x	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	217409x	auto& op = rd_;
372	217409x	op.reset();
373
374	217409x	capy::mutable_buffer bufs[ReadOp::max_buffers];
375	217409x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
376
377	217409x	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	434814x	for (int i = 0; i < op.iovec_count; ++i)
392		{
393	217407x	op.iovecs[i].iov_base = bufs[i].data();
394	217407x	op.iovecs[i].iov_len = bufs[i].size();
395		}
396
397		// Speculative read
398		ssize_t n;
399		do
400		{
401	217407x	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
402		}
403	217407x	while (n < 0 && errno == EINTR);
404
405	217407x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
406		{
407	217018x	int err = (n < 0) ? errno : 0;
408	217018x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
409
410	217018x	if (this->svc_.scheduler().try_consume_inline_budget())
411		{
412	173648x	if (err)
413	✗	*ec = make_err(err);
414	173648x	else if (n == 0)
415	10x	*ec = capy::error::eof;
416		else
417	173638x	*ec = {};
418	173648x	*bytes_out = bytes;
419	173648x	op.cont_op.cont.h = h;
420	173648x	return dispatch_coro(ex, op.cont_op.cont);
421		}
422	43370x	op.h = h;
423	43370x	op.ex = ex;
424	43370x	op.ec_out = ec;
425	43370x	op.bytes_out = bytes_out;
426	43370x	op.start(token, static_cast<Derived*>(this));
427	43370x	op.impl_ptr = this->shared_from_this();
428	43370x	op.complete(err, bytes);
429	43370x	this->svc_.post(&op);
430	43370x	return std::noop_coroutine();
431		}
432
433		// EAGAIN — register with reactor
434	389x	op.h = h;
435	389x	op.ex = ex;
436	389x	op.ec_out = ec;
437	389x	op.bytes_out = bytes_out;
438	389x	op.fd = this->fd_;
439	389x	op.start(token, static_cast<Derived*>(this));
440	389x	op.impl_ptr = this->shared_from_this();
441
442	389x	this->register_op(
443	389x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
444	389x	this->desc_state_.read_cancel_pending);
445	389x	return std::noop_coroutine();
446		}
447
448		template<
449		class Derived,
450		class Service,
451		class ConnOp,
452		class ReadOp,
453		class WriteOp,
454		class DescState,
455		class ImplBase,
456		class Endpoint>
457		std::coroutine_handle<>
458	217112x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState, ImplBase, Endpoint>::
459		do_write_some(
460		std::coroutine_handle<> h,
461		capy::executor_ref ex,
462		buffer_param param,
463		std::stop_token const& token,
464		std::error_code* ec,
465		std::size_t* bytes_out)
466		{
467	217112x	auto& op = wr_;
468	217112x	op.reset();
469
470	217112x	capy::mutable_buffer bufs[WriteOp::max_buffers];
471	217112x	op.iovec_count =
472	217112x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
473
474	217112x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
475		{
476	2x	op.h = h;
477	2x	op.ex = ex;
478	2x	op.ec_out = ec;
479	2x	op.bytes_out = bytes_out;
480	2x	op.start(token, static_cast<Derived*>(this));
481	2x	op.impl_ptr = this->shared_from_this();
482	2x	op.complete(0, 0);
483	2x	this->svc_.post(&op);
484	2x	return std::noop_coroutine();
485		}
486
487	434220x	for (int i = 0; i < op.iovec_count; ++i)
488		{
489	217110x	op.iovecs[i].iov_base = bufs[i].data();
490	217110x	op.iovecs[i].iov_len = bufs[i].size();
491		}
492
493		// Speculative write via backend-specific write policy
494		ssize_t n =
495	217110x	WriteOp::write_policy::write(this->fd_, op.iovecs, op.iovec_count);
496
497	217110x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
498		{
499	217110x	int err = (n < 0) ? errno : 0;
500	217110x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
501
502	217110x	if (this->svc_.scheduler().try_consume_inline_budget())
503		{
504	173702x	*ec = err ? make_err(err) : std::error_code{};
505	173702x	*bytes_out = bytes;
506	173702x	op.cont_op.cont.h = h;
507	173702x	return dispatch_coro(ex, op.cont_op.cont);
508		}
509	43408x	op.h = h;
510	43408x	op.ex = ex;
511	43408x	op.ec_out = ec;
512	43408x	op.bytes_out = bytes_out;
513	43408x	op.start(token, static_cast<Derived*>(this));
514	43408x	op.impl_ptr = this->shared_from_this();
515	43408x	op.complete(err, bytes);
516	43408x	this->svc_.post(&op);
517	43408x	return std::noop_coroutine();
518		}
519
520		// EAGAIN — register with reactor
521	✗	op.h = h;
522	✗	op.ex = ex;
523	✗	op.ec_out = ec;
524	✗	op.bytes_out = bytes_out;
525	✗	op.fd = this->fd_;
526	✗	op.start(token, static_cast<Derived*>(this));
527	✗	op.impl_ptr = this->shared_from_this();
528
529	✗	this->register_op(
530	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
531	✗	this->desc_state_.write_cancel_pending, true);
532	✗	return std::noop_coroutine();
533		}
534
535		} // namespace boost::corosio::detail
536
537		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
538