include/boost/corosio/native/detail/epoll/epoll_tcp_acceptor_service.hpp

69.4% Lines (77/111) 100.0% List of functions (11/11)

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

epoll_tcp_acceptor_service.hpp

f(x) Functions (11)

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_EPOLL_EPOLL_TCP_ACCEPTOR_SERVICE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_TCP_ACCEPTOR_SERVICE_HPP
12
13		#include <boost/corosio/detail/platform.hpp>
14
15		#if BOOST_COROSIO_HAS_EPOLL
16
17		#include <boost/corosio/detail/config.hpp>
18		#include <boost/capy/ex/execution_context.hpp>
19		#include <boost/corosio/detail/tcp_acceptor_service.hpp>
20
21		#include <boost/corosio/native/detail/epoll/epoll_tcp_acceptor.hpp>
22		#include <boost/corosio/native/detail/epoll/epoll_tcp_service.hpp>
23		#include <boost/corosio/native/detail/epoll/epoll_scheduler.hpp>
24		#include <boost/corosio/native/detail/reactor/reactor_acceptor_service.hpp>
25		#include <boost/corosio/native/detail/reactor/reactor_op_complete.hpp>
26
27		#include <memory>
28		#include <mutex>
29		#include <utility>
30
31		#include <errno.h>
32		#include <netinet/in.h>
33		#include <sys/epoll.h>
34		#include <sys/socket.h>
35		#include <unistd.h>
36
37		namespace boost::corosio::detail {
38
39		/** epoll acceptor service implementation.
40
41		Derives from reactor_acceptor_service for shared construct/
42		destroy/shutdown/close logic. Provides epoll-specific socket
43		creation (SOCK_NONBLOCK \| SOCK_CLOEXEC) and dual-stack defaults.
44		Uses key_type = tcp_acceptor_service for service lookup.
45		*/
46		class BOOST_COROSIO_DECL epoll_tcp_acceptor_service final
47		: public reactor_acceptor_service<
48		epoll_tcp_acceptor_service,
49		tcp_acceptor_service,
50		epoll_scheduler,
51		epoll_tcp_acceptor,
52		epoll_tcp_service>
53		{
54		using base_type = reactor_acceptor_service<
55		epoll_tcp_acceptor_service,
56		tcp_acceptor_service,
57		epoll_scheduler,
58		epoll_tcp_acceptor,
59		epoll_tcp_service>;
60		friend base_type;
61
62		public:
63		explicit epoll_tcp_acceptor_service(capy::execution_context& ctx);
64		~epoll_tcp_acceptor_service() override;
65
66		std::error_code open_acceptor_socket(
67		tcp_acceptor::implementation& impl,
68		int family,
69		int type,
70		int protocol) override;
71		std::error_code
72		bind_acceptor(tcp_acceptor::implementation& impl, endpoint ep) override;
73		std::error_code
74		listen_acceptor(tcp_acceptor::implementation& impl, int backlog) override;
75		};
76
77		inline void
78	6x	epoll_accept_op::cancel() noexcept
79		{
80	6x	if (acceptor_impl_)
81	6x	acceptor_impl_->cancel_single_op(*this);
82		else
83	✗	request_cancel();
84	6x	}
85
86		inline void
87	4010x	epoll_accept_op::operator()()
88		{
89	4010x	complete_accept_op<epoll_tcp_socket>(*this);
90	4010x	}
91
92	84x	inline epoll_tcp_acceptor::epoll_tcp_acceptor(
93	84x	epoll_tcp_acceptor_service& svc) noexcept
94	84x	: reactor_acceptor(svc)
95		{
96	84x	}
97
98		inline std::coroutine_handle<>
99	4010x	epoll_tcp_acceptor::accept(
100		std::coroutine_handle<> h,
101		capy::executor_ref ex,
102		std::stop_token token,
103		std::error_code* ec,
104		io_object::implementation** impl_out)
105		{
106	4010x	auto& op = acc_;
107	4010x	op.reset();
108	4010x	op.h = h;
109	4010x	op.ex = ex;
110	4010x	op.ec_out = ec;
111	4010x	op.impl_out = impl_out;
112	4010x	op.fd = fd_;
113	4010x	op.start(token, this);
114
115	4010x	sockaddr_storage peer_storage{};
116	4010x	socklen_t addrlen = sizeof(peer_storage);
117		int accepted;
118		do
119		{
120	4010x	accepted = ::accept4(
121		fd_, reinterpret_cast<sockaddr*>(&peer_storage), &addrlen,
122		SOCK_NONBLOCK \| SOCK_CLOEXEC);
123		}
124	4010x	while (accepted < 0 && errno == EINTR);
125
126	4010x	if (accepted >= 0)
127		{
128		{
129	3x	std::lock_guard lock(desc_state_.mutex);
130	3x	desc_state_.read_ready = false;
131	3x	}
132
133	3x	if (svc_.scheduler().try_consume_inline_budget())
134		{
135	✗	auto* socket_svc = svc_.stream_service();
136	✗	if (socket_svc)
137		{
138		auto& impl =
139	✗	static_cast<epoll_tcp_socket&>(*socket_svc->construct());
140	✗	impl.set_socket(accepted);
141
142	✗	impl.desc_state_.fd = accepted;
143		{
144	✗	std::lock_guard lock(impl.desc_state_.mutex);
145	✗	impl.desc_state_.read_op = nullptr;
146	✗	impl.desc_state_.write_op = nullptr;
147	✗	impl.desc_state_.connect_op = nullptr;
148	✗	}
149	✗	socket_svc->scheduler().register_descriptor(
150		accepted, &impl.desc_state_);
151
152	✗	impl.set_endpoints(
153		local_endpoint_, from_sockaddr(peer_storage));
154
155	✗	*ec = {};
156	✗	if (impl_out)
157	✗	*impl_out = &impl;
158		}
159		else
160		{
161	✗	::close(accepted);
162	✗	*ec = make_err(ENOENT);
163	✗	if (impl_out)
164	✗	*impl_out = nullptr;
165		}
166	✗	op.cont_op.cont.h = h;
167	✗	return dispatch_coro(ex, op.cont_op.cont);
168		}
169
170	3x	op.accepted_fd = accepted;
171	3x	op.peer_storage = peer_storage;
172	3x	op.peer_addrlen = addrlen;
173	3x	op.complete(0, 0);
174	3x	op.impl_ptr = shared_from_this();
175	3x	svc_.post(&op);
176	3x	return std::noop_coroutine();
177		}
178
179	4007x	if (errno == EAGAIN \|\| errno == EWOULDBLOCK)
180		{
181	4007x	op.impl_ptr = shared_from_this();
182	4007x	svc_.work_started();
183
184	4007x	std::lock_guard lock(desc_state_.mutex);
185	4007x	bool io_done = false;
186	4007x	if (desc_state_.read_ready)
187		{
188	✗	desc_state_.read_ready = false;
189	✗	op.perform_io();
190	✗	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
191	✗	if (!io_done)
192	✗	op.errn = 0;
193		}
194
195	4007x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
196		{
197	✗	svc_.post(&op);
198	✗	svc_.work_finished();
199		}
200		else
201		{
202	4007x	desc_state_.read_op = &op;
203		}
204	4007x	return std::noop_coroutine();
205	4007x	}
206
207	✗	op.complete(errno, 0);
208	✗	op.impl_ptr = shared_from_this();
209	✗	svc_.post(&op);
210		// completion is always posted to scheduler queue, never inline.
211	✗	return std::noop_coroutine();
212		}
213
214		inline void
215	2x	epoll_tcp_acceptor::cancel() noexcept
216		{
217	2x	do_cancel();
218	2x	}
219
220		inline void
221	332x	epoll_tcp_acceptor::close_socket() noexcept
222		{
223	332x	do_close_socket();
224	332x	}
225
226	356x	inline epoll_tcp_acceptor_service::epoll_tcp_acceptor_service(
227	356x	capy::execution_context& ctx)
228	356x	: base_type(ctx)
229		{
230	356x	auto* svc = ctx_.find_service<detail::tcp_service>();
231	356x	stream_svc_ = svc ? dynamic_cast<epoll_tcp_service*>(svc) : nullptr;
232	356x	}
233
234	712x	inline epoll_tcp_acceptor_service::~epoll_tcp_acceptor_service() {}
235
236		inline std::error_code
237	82x	epoll_tcp_acceptor_service::open_acceptor_socket(
238		tcp_acceptor::implementation& impl, int family, int type, int protocol)
239		{
240	82x	auto* epoll_impl = static_cast<epoll_tcp_acceptor*>(&impl);
241	82x	epoll_impl->close_socket();
242
243	82x	int fd = ::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
244	82x	if (fd < 0)
245	✗	return make_err(errno);
246
247	82x	if (family == AF_INET6)
248		{
249	8x	int val = 0; // dual-stack default
250	8x	::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &val, sizeof(val));
251		}
252
253	82x	epoll_impl->fd_ = fd;
254
255		// Set up descriptor state but do NOT register with epoll yet
256	82x	epoll_impl->desc_state_.fd = fd;
257		{
258	82x	std::lock_guard lock(epoll_impl->desc_state_.mutex);
259	82x	epoll_impl->desc_state_.read_op = nullptr;
260	82x	}
261
262	82x	return {};
263		}
264
265		inline std::error_code
266	81x	epoll_tcp_acceptor_service::bind_acceptor(
267		tcp_acceptor::implementation& impl, endpoint ep)
268		{
269	81x	return static_cast<epoll_tcp_acceptor*>(&impl)->do_bind(ep);
270		}
271
272		inline std::error_code
273	76x	epoll_tcp_acceptor_service::listen_acceptor(
274		tcp_acceptor::implementation& impl, int backlog)
275		{
276	76x	return static_cast<epoll_tcp_acceptor*>(&impl)->do_listen(backlog);
277		}
278
279		} // namespace boost::corosio::detail
280
281		#endif // BOOST_COROSIO_HAS_EPOLL
282
283		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_TCP_ACCEPTOR_SERVICE_HPP
284