include/boost/corosio/native/detail/select/select_tcp_service.hpp

71.2% Lines (57/80) 85.7% List of functions (12/14)

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

f(x) Functions (14)

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_SELECT_SELECT_TCP_SERVICE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_TCP_SERVICE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_SELECT
16
17		#include <boost/corosio/detail/config.hpp>
18		#include <boost/corosio/detail/tcp_service.hpp>
19
20		#include <boost/corosio/native/detail/select/select_tcp_socket.hpp>
21		#include <boost/corosio/native/detail/select/select_scheduler.hpp>
22		#include <boost/corosio/native/detail/reactor/reactor_socket_service.hpp>
23
24		#include <boost/corosio/native/detail/reactor/reactor_op_complete.hpp>
25
26		#include <coroutine>
27		#include <mutex>
28
29		#include <errno.h>
30		#include <fcntl.h>
31		#include <netinet/in.h>
32		#include <netinet/tcp.h>
33		#include <sys/select.h>
34		#include <sys/socket.h>
35		#include <unistd.h>
36
37		/*
38		Each I/O op tries the syscall speculatively; only registers with
39		the reactor on EAGAIN. Fd is registered once at open time and
40		stays registered until close. The reactor only marks ready_events_;
41		actual I/O happens in invoke_deferred_io(). cancel() captures
42		shared_from_this() into op.impl_ptr to keep the impl alive.
43		*/
44
45		namespace boost::corosio::detail {
46
47		/** select TCP service implementation.
48
49		Inherits from tcp_service to enable runtime polymorphism.
50		Uses key_type = tcp_service for service lookup.
51		*/
52		class BOOST_COROSIO_DECL select_tcp_service final
53		: public reactor_socket_service<
54		select_tcp_service,
55		tcp_service,
56		select_scheduler,
57		select_tcp_socket>
58		{
59		public:
60	195x	explicit select_tcp_service(capy::execution_context& ctx)
61	195x	: reactor_socket_service(ctx)
62		{
63	195x	}
64
65		std::error_code open_socket(
66		tcp_socket::implementation& impl,
67		int family,
68		int type,
69		int protocol) override;
70		};
71
72		inline void
73	✗	select_connect_op::cancel() noexcept
74		{
75	✗	if (socket_impl_)
76	✗	socket_impl_->cancel_single_op(*this);
77		else
78	✗	request_cancel();
79	✗	}
80
81		inline void
82	94x	select_read_op::cancel() noexcept
83		{
84	94x	if (socket_impl_)
85	94x	socket_impl_->cancel_single_op(*this);
86		else
87	✗	request_cancel();
88	94x	}
89
90		inline void
91	✗	select_write_op::cancel() noexcept
92		{
93	✗	if (socket_impl_)
94	✗	socket_impl_->cancel_single_op(*this);
95		else
96	✗	request_cancel();
97	✗	}
98
99		inline void
100	35592x	select_op::operator()()
101		{
102	35592x	complete_io_op(*this);
103	35592x	}
104
105		inline void
106	3727x	select_connect_op::operator()()
107		{
108	3727x	complete_connect_op(*this);
109	3727x	}
110
111	11201x	inline select_tcp_socket::select_tcp_socket(select_tcp_service& svc) noexcept
112	11201x	: reactor_stream_socket(svc)
113		{
114	11201x	}
115
116	11201x	inline select_tcp_socket::~select_tcp_socket() = default;
117
118		inline std::coroutine_handle<>
119	3727x	select_tcp_socket::connect(
120		std::coroutine_handle<> h,
121		capy::executor_ref ex,
122		endpoint ep,
123		std::stop_token token,
124		std::error_code* ec)
125		{
126	3727x	auto result = do_connect(h, ex, ep, token, ec);
127		// Rebuild fd_sets so select() watches for writability
128	3727x	if (result == std::noop_coroutine())
129	3727x	svc_.scheduler().notify_reactor();
130	3727x	return result;
131		}
132
133		inline std::coroutine_handle<>
134	88729x	select_tcp_socket::read_some(
135		std::coroutine_handle<> h,
136		capy::executor_ref ex,
137		buffer_param param,
138		std::stop_token token,
139		std::error_code* ec,
140		std::size_t* bytes_out)
141		{
142	88729x	return do_read_some(h, ex, param, token, ec, bytes_out);
143		}
144
145		inline std::coroutine_handle<>
146	88581x	select_tcp_socket::write_some(
147		std::coroutine_handle<> h,
148		capy::executor_ref ex,
149		buffer_param param,
150		std::stop_token token,
151		std::error_code* ec,
152		std::size_t* bytes_out)
153		{
154	88581x	auto result = do_write_some(h, ex, param, token, ec, bytes_out);
155		// Rebuild fd_sets so select() watches for writability
156	88581x	if (result == std::noop_coroutine())
157	17708x	svc_.scheduler().notify_reactor();
158	88581x	return result;
159		}
160
161		inline void
162	94x	select_tcp_socket::cancel() noexcept
163		{
164	94x	do_cancel();
165	94x	}
166
167		inline void
168	33611x	select_tcp_socket::close_socket() noexcept
169		{
170	33611x	do_close_socket();
171	33611x	}
172
173		inline std::error_code
174	3742x	select_tcp_service::open_socket(
175		tcp_socket::implementation& impl, int family, int type, int protocol)
176		{
177	3742x	auto* select_impl = static_cast<select_tcp_socket*>(&impl);
178	3742x	select_impl->close_socket();
179
180	3742x	int fd = ::socket(family, type, protocol);
181	3742x	if (fd < 0)
182	✗	return make_err(errno);
183
184	3742x	if (family == AF_INET6)
185		{
186	5x	int one = 1;
187	5x	::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &one, sizeof(one));
188		}
189
190	3742x	int flags = ::fcntl(fd, F_GETFL, 0);
191	3742x	if (flags == -1)
192		{
193	✗	int errn = errno;
194	✗	::close(fd);
195	✗	return make_err(errn);
196		}
197	3742x	if (::fcntl(fd, F_SETFL, flags \| O_NONBLOCK) == -1)
198		{
199	✗	int errn = errno;
200	✗	::close(fd);
201	✗	return make_err(errn);
202		}
203	3742x	if (::fcntl(fd, F_SETFD, FD_CLOEXEC) == -1)
204		{
205	✗	int errn = errno;
206	✗	::close(fd);
207	✗	return make_err(errn);
208		}
209
210	3742x	if (fd >= FD_SETSIZE)
211		{
212	✗	::close(fd);
213	✗	return make_err(EMFILE);
214		}
215
216		#ifdef SO_NOSIGPIPE
217		{
218		int one = 1;
219		::setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one));
220		}
221		#endif
222
223	3742x	select_impl->fd_ = fd;
224
225	3742x	select_impl->desc_state_.fd = fd;
226		{
227	3742x	std::lock_guard lock(select_impl->desc_state_.mutex);
228	3742x	select_impl->desc_state_.read_op = nullptr;
229	3742x	select_impl->desc_state_.write_op = nullptr;
230	3742x	select_impl->desc_state_.connect_op = nullptr;
231	3742x	}
232	3742x	scheduler().register_descriptor(fd, &select_impl->desc_state_);
233
234	3742x	return {};
235		}
236
237		} // namespace boost::corosio::detail
238
239		#endif // BOOST_COROSIO_HAS_SELECT
240
241		#endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_TCP_SERVICE_HPP
242