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

89.4% Lines (186/208) 78.8% List of functions (63/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	35295x	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	35295x	~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	11723x	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	11723x	return do_connect(h, ex, ep, token, ec);
105		}
106
107	286060x	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	286060x	return do_read_some(h, ex, param, token, ec, bytes_out);
116		}
117
118	329952x	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	329952x	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	219x	void cancel() noexcept override
136		{
137	219x	this->do_cancel();
138	219x	}
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	23430x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
176		{
177	23430x	this->local_endpoint_ = std::move(local);
178	23430x	remote_endpoint_ = std::move(remote);
179	23430x	}
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	105861x	void do_close_socket() noexcept
228		{
229	105861x	base_type::do_close_socket();
230	105861x	remote_endpoint_ = Endpoint{};
231	105861x	}
232
233		private:
234		// CRTP callbacks for reactor_basic_socket cancel/close
235
236		template<class Op>
237	920x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
238		{
239	920x	if (&op == static_cast<void*>(&conn_))
240	✗	return &this->desc_state_.connect_op;
241	920x	if (&op == static_cast<void*>(&rd_))
242	919x	return &this->desc_state_.read_op;
243	1x	if (&op == static_cast<void*>(&wr_))
244	1x	return &this->desc_state_.write_op;
245	✗	return nullptr;
246		}
247
248		template<class Op>
249	2x	bool* op_to_cancel_flag(Op& op) noexcept
250		{
251	2x	if (&op == static_cast<void*>(&conn_))
252	✗	return &this->desc_state_.connect_cancel_pending;
253	2x	if (&op == static_cast<void*>(&rd_))
254	1x	return &this->desc_state_.read_cancel_pending;
255	1x	if (&op == static_cast<void*>(&wr_))
256	1x	return &this->desc_state_.write_cancel_pending;
257	✗	return nullptr;
258		}
259
260		template<class Fn>
261	106082x	void for_each_op(Fn fn) noexcept
262		{
263	106082x	fn(conn_);
264	106082x	fn(rd_);
265	106082x	fn(wr_);
266	106082x	}
267
268		template<class Fn>
269	106082x	void for_each_desc_entry(Fn fn) noexcept
270		{
271	106082x	fn(conn_, this->desc_state_.connect_op);
272	106082x	fn(rd_, this->desc_state_.read_op);
273	106082x	fn(wr_, this->desc_state_.write_op);
274	106082x	}
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	11723x	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	11723x	auto& op = conn_;
296
297	11723x	sockaddr_storage storage{};
298	11723x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
299		int result =
300	11723x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
301
302	11723x	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	11723x	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	11719x	op.reset();
338	11719x	op.h = h;
339	11719x	op.ex = ex;
340	11719x	op.ec_out = ec;
341	11719x	op.fd = this->fd_;
342	11719x	op.target_endpoint = ep;
343	11719x	op.start(token, static_cast<Derived*>(this));
344	11719x	op.impl_ptr = this->shared_from_this();
345
346	11719x	this->register_op(
347	11719x	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
348	11719x	this->desc_state_.connect_cancel_pending, true);
349	11719x	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	286060x	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	286060x	auto& op = rd_;
372	286060x	op.reset();
373
374	286060x	capy::mutable_buffer bufs[ReadOp::max_buffers];
375	286060x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
376
377	286060x	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	572116x	for (int i = 0; i < op.iovec_count; ++i)
392		{
393	286058x	op.iovecs[i].iov_base = bufs[i].data();
394	286058x	op.iovecs[i].iov_len = bufs[i].size();
395		}
396
397		// Speculative read
398		ssize_t n;
399		do
400		{
401	286058x	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
402		}
403	286058x	while (n < 0 && errno == EINTR);
404
405	286058x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
406		{
407	283578x	int err = (n < 0) ? errno : 0;
408	283578x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
409
410	283578x	if (this->svc_.scheduler().try_consume_inline_budget())
411		{
412	226201x	if (err)
413	1x	*ec = make_err(err);
414	226200x	else if (n == 0)
415	16x	*ec = capy::error::eof;
416		else
417	226184x	*ec = {};
418	226201x	*bytes_out = bytes;
419	226201x	op.cont_op.cont.h = h;
420	226201x	return dispatch_coro(ex, op.cont_op.cont);
421		}
422	57377x	op.h = h;
423	57377x	op.ex = ex;
424	57377x	op.ec_out = ec;
425	57377x	op.bytes_out = bytes_out;
426	57377x	op.start(token, static_cast<Derived*>(this));
427	57377x	op.impl_ptr = this->shared_from_this();
428	57377x	op.complete(err, bytes);
429	57377x	this->svc_.post(&op);
430	57377x	return std::noop_coroutine();
431		}
432
433		// EAGAIN — register with reactor
434	2480x	op.h = h;
435	2480x	op.ex = ex;
436	2480x	op.ec_out = ec;
437	2480x	op.bytes_out = bytes_out;
438	2480x	op.fd = this->fd_;
439	2480x	op.start(token, static_cast<Derived*>(this));
440	2480x	op.impl_ptr = this->shared_from_this();
441
442	2480x	this->register_op(
443	2480x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
444	2480x	this->desc_state_.read_cancel_pending);
445	2480x	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	329952x	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	329952x	auto& op = wr_;
468	329952x	op.reset();
469
470	329952x	capy::mutable_buffer bufs[WriteOp::max_buffers];
471	329952x	op.iovec_count =
472	329952x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
473
474	329952x	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	659900x	for (int i = 0; i < op.iovec_count; ++i)
488		{
489	329950x	op.iovecs[i].iov_base = bufs[i].data();
490	329950x	op.iovecs[i].iov_len = bufs[i].size();
491		}
492
493		// Speculative write via backend-specific write policy
494		ssize_t n =
495	329950x	WriteOp::write_policy::write(this->fd_, op.iovecs, op.iovec_count);
496
497	329950x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
498		{
499	329950x	int err = (n < 0) ? errno : 0;
500	329950x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
501
502	329950x	if (this->svc_.scheduler().try_consume_inline_budget())
503		{
504	263032x	*ec = err ? make_err(err) : std::error_code{};
505	263032x	*bytes_out = bytes;
506	263032x	op.cont_op.cont.h = h;
507	263032x	return dispatch_coro(ex, op.cont_op.cont);
508		}
509	66918x	op.h = h;
510	66918x	op.ex = ex;
511	66918x	op.ec_out = ec;
512	66918x	op.bytes_out = bytes_out;
513	66918x	op.start(token, static_cast<Derived*>(this));
514	66918x	op.impl_ptr = this->shared_from_this();
515	66918x	op.complete(err, bytes);
516	66918x	this->svc_.post(&op);
517	66918x	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