TLA Line data 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 HIT 6 : epoll_accept_op::cancel() noexcept
79 : {
80 6 : if (acceptor_impl_)
81 6 : acceptor_impl_->cancel_single_op(*this);
82 : else
83 MIS 0 : request_cancel();
84 HIT 6 : }
85 :
86 : inline void
87 4825 : epoll_accept_op::operator()()
88 : {
89 4825 : complete_accept_op<epoll_tcp_socket>(*this);
90 4825 : }
91 :
92 84 : inline epoll_tcp_acceptor::epoll_tcp_acceptor(
93 84 : epoll_tcp_acceptor_service& svc) noexcept
94 84 : : reactor_acceptor(svc)
95 : {
96 84 : }
97 :
98 : inline std::coroutine_handle<>
99 4825 : 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 4825 : auto& op = acc_;
107 4825 : op.reset();
108 4825 : op.h = h;
109 4825 : op.ex = ex;
110 4825 : op.ec_out = ec;
111 4825 : op.impl_out = impl_out;
112 4825 : op.fd = fd_;
113 4825 : op.start(token, this);
114 :
115 4825 : sockaddr_storage peer_storage{};
116 4825 : socklen_t addrlen = sizeof(peer_storage);
117 : int accepted;
118 : do
119 : {
120 4825 : accepted = ::accept4(
121 : fd_, reinterpret_cast<sockaddr*>(&peer_storage), &addrlen,
122 : SOCK_NONBLOCK | SOCK_CLOEXEC);
123 : }
124 4825 : while (accepted < 0 && errno == EINTR);
125 :
126 4825 : if (accepted >= 0)
127 : {
128 : {
129 3 : std::lock_guard lock(desc_state_.mutex);
130 3 : desc_state_.read_ready = false;
131 3 : }
132 :
133 3 : if (svc_.scheduler().try_consume_inline_budget())
134 : {
135 MIS 0 : auto* socket_svc = svc_.stream_service();
136 0 : if (socket_svc)
137 : {
138 : auto& impl =
139 0 : static_cast<epoll_tcp_socket&>(*socket_svc->construct());
140 0 : impl.set_socket(accepted);
141 :
142 0 : impl.desc_state_.fd = accepted;
143 : {
144 0 : std::lock_guard lock(impl.desc_state_.mutex);
145 0 : impl.desc_state_.read_op = nullptr;
146 0 : impl.desc_state_.write_op = nullptr;
147 0 : impl.desc_state_.connect_op = nullptr;
148 0 : }
149 0 : socket_svc->scheduler().register_descriptor(
150 : accepted, &impl.desc_state_);
151 :
152 0 : impl.set_endpoints(
153 : local_endpoint_, from_sockaddr(peer_storage));
154 :
155 0 : *ec = {};
156 0 : if (impl_out)
157 0 : *impl_out = &impl;
158 : }
159 : else
160 : {
161 0 : ::close(accepted);
162 0 : *ec = make_err(ENOENT);
163 0 : if (impl_out)
164 0 : *impl_out = nullptr;
165 : }
166 0 : op.cont_op.cont.h = h;
167 0 : return dispatch_coro(ex, op.cont_op.cont);
168 : }
169 :
170 HIT 3 : op.accepted_fd = accepted;
171 3 : op.peer_storage = peer_storage;
172 3 : op.complete(0, 0);
173 3 : op.impl_ptr = shared_from_this();
174 3 : svc_.post(&op);
175 3 : return std::noop_coroutine();
176 : }
177 :
178 4822 : if (errno == EAGAIN || errno == EWOULDBLOCK)
179 : {
180 4822 : op.impl_ptr = shared_from_this();
181 4822 : svc_.work_started();
182 :
183 4822 : std::lock_guard lock(desc_state_.mutex);
184 4822 : bool io_done = false;
185 4822 : if (desc_state_.read_ready)
186 : {
187 MIS 0 : desc_state_.read_ready = false;
188 0 : op.perform_io();
189 0 : io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
190 0 : if (!io_done)
191 0 : op.errn = 0;
192 : }
193 :
194 HIT 4822 : if (io_done || op.cancelled.load(std::memory_order_acquire))
195 : {
196 MIS 0 : svc_.post(&op);
197 0 : svc_.work_finished();
198 : }
199 : else
200 : {
201 HIT 4822 : desc_state_.read_op = &op;
202 : }
203 4822 : return std::noop_coroutine();
204 4822 : }
205 :
206 MIS 0 : op.complete(errno, 0);
207 0 : op.impl_ptr = shared_from_this();
208 0 : svc_.post(&op);
209 : // completion is always posted to scheduler queue, never inline.
210 0 : return std::noop_coroutine();
211 : }
212 :
213 : inline void
214 HIT 2 : epoll_tcp_acceptor::cancel() noexcept
215 : {
216 2 : do_cancel();
217 2 : }
218 :
219 : inline void
220 332 : epoll_tcp_acceptor::close_socket() noexcept
221 : {
222 332 : do_close_socket();
223 332 : }
224 :
225 355 : inline epoll_tcp_acceptor_service::epoll_tcp_acceptor_service(
226 355 : capy::execution_context& ctx)
227 355 : : base_type(ctx)
228 : {
229 355 : auto* svc = ctx_.find_service<detail::tcp_service>();
230 355 : stream_svc_ = svc ? dynamic_cast<epoll_tcp_service*>(svc) : nullptr;
231 355 : }
232 :
233 710 : inline epoll_tcp_acceptor_service::~epoll_tcp_acceptor_service() {}
234 :
235 : inline std::error_code
236 82 : epoll_tcp_acceptor_service::open_acceptor_socket(
237 : tcp_acceptor::implementation& impl, int family, int type, int protocol)
238 : {
239 82 : auto* epoll_impl = static_cast<epoll_tcp_acceptor*>(&impl);
240 82 : epoll_impl->close_socket();
241 :
242 82 : int fd = ::socket(family, type | SOCK_NONBLOCK | SOCK_CLOEXEC, protocol);
243 82 : if (fd < 0)
244 MIS 0 : return make_err(errno);
245 :
246 HIT 82 : if (family == AF_INET6)
247 : {
248 8 : int val = 0; // dual-stack default
249 8 : ::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &val, sizeof(val));
250 : }
251 :
252 82 : epoll_impl->fd_ = fd;
253 :
254 : // Set up descriptor state but do NOT register with epoll yet
255 82 : epoll_impl->desc_state_.fd = fd;
256 : {
257 82 : std::lock_guard lock(epoll_impl->desc_state_.mutex);
258 82 : epoll_impl->desc_state_.read_op = nullptr;
259 82 : }
260 :
261 82 : return {};
262 : }
263 :
264 : inline std::error_code
265 81 : epoll_tcp_acceptor_service::bind_acceptor(
266 : tcp_acceptor::implementation& impl, endpoint ep)
267 : {
268 81 : return static_cast<epoll_tcp_acceptor*>(&impl)->do_bind(ep);
269 : }
270 :
271 : inline std::error_code
272 76 : epoll_tcp_acceptor_service::listen_acceptor(
273 : tcp_acceptor::implementation& impl, int backlog)
274 : {
275 76 : return static_cast<epoll_tcp_acceptor*>(&impl)->do_listen(backlog);
276 : }
277 :
278 : } // namespace boost::corosio::detail
279 :
280 : #endif // BOOST_COROSIO_HAS_EPOLL
281 :
282 : #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_TCP_ACCEPTOR_SERVICE_HPP
|
