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_SELECT_SELECT_TCP_ACCEPTOR_SERVICE_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_TCP_ACCEPTOR_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/capy/ex/execution_context.hpp>
19 : #include <boost/corosio/detail/tcp_acceptor_service.hpp>
20 :
21 : #include <boost/corosio/native/detail/select/select_tcp_acceptor.hpp>
22 : #include <boost/corosio/native/detail/select/select_tcp_service.hpp>
23 : #include <boost/corosio/native/detail/select/select_scheduler.hpp>
24 : #include <boost/corosio/native/detail/reactor/reactor_service_state.hpp>
25 :
26 : #include <boost/corosio/native/detail/reactor/reactor_op_complete.hpp>
27 :
28 : #include <memory>
29 : #include <mutex>
30 : #include <utility>
31 :
32 : #include <errno.h>
33 : #include <fcntl.h>
34 : #include <netinet/in.h>
35 : #include <sys/select.h>
36 : #include <sys/socket.h>
37 : #include <unistd.h>
38 :
39 : namespace boost::corosio::detail {
40 :
41 : /// State for select acceptor service.
42 : using select_tcp_acceptor_state =
43 : reactor_service_state<select_scheduler, select_tcp_acceptor>;
44 :
45 : /** select acceptor service implementation.
46 :
47 : Inherits from tcp_acceptor_service to enable runtime polymorphism.
48 : Uses key_type = tcp_acceptor_service for service lookup.
49 : */
50 : class BOOST_COROSIO_DECL select_tcp_acceptor_service final
51 : : public tcp_acceptor_service
52 : {
53 : public:
54 : explicit select_tcp_acceptor_service(capy::execution_context& ctx);
55 : ~select_tcp_acceptor_service() override;
56 :
57 : select_tcp_acceptor_service(select_tcp_acceptor_service const&) = delete;
58 : select_tcp_acceptor_service&
59 : operator=(select_tcp_acceptor_service const&) = delete;
60 :
61 : void shutdown() override;
62 :
63 : io_object::implementation* construct() override;
64 : void destroy(io_object::implementation*) override;
65 : void close(io_object::handle&) override;
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 HIT 116 : select_scheduler& scheduler() const noexcept
77 : {
78 116 : return state_->sched_;
79 : }
80 : void post(scheduler_op* op);
81 : void work_started() noexcept;
82 : void work_finished() noexcept;
83 :
84 : /** Get the TCP service for creating peer sockets during accept. */
85 : select_tcp_service* tcp_service() const noexcept;
86 :
87 : private:
88 : capy::execution_context& ctx_;
89 : std::unique_ptr<select_tcp_acceptor_state> state_;
90 : };
91 :
92 : inline void
93 MIS 0 : select_accept_op::cancel() noexcept
94 : {
95 0 : if (acceptor_impl_)
96 0 : acceptor_impl_->cancel_single_op(*this);
97 : else
98 0 : request_cancel();
99 0 : }
100 :
101 : inline void
102 HIT 2024 : select_accept_op::operator()()
103 : {
104 2024 : complete_accept_op<select_tcp_socket>(*this);
105 2024 : }
106 :
107 61 : inline select_tcp_acceptor::select_tcp_acceptor(
108 61 : select_tcp_acceptor_service& svc) noexcept
109 61 : : reactor_acceptor(svc)
110 : {
111 61 : }
112 :
113 : inline std::coroutine_handle<>
114 2024 : select_tcp_acceptor::accept(
115 : std::coroutine_handle<> h,
116 : capy::executor_ref ex,
117 : std::stop_token token,
118 : std::error_code* ec,
119 : io_object::implementation** impl_out)
120 : {
121 2024 : auto& op = acc_;
122 2024 : op.reset();
123 2024 : op.h = h;
124 2024 : op.ex = ex;
125 2024 : op.ec_out = ec;
126 2024 : op.impl_out = impl_out;
127 2024 : op.fd = fd_;
128 2024 : op.start(token, this);
129 :
130 2024 : sockaddr_storage peer_storage{};
131 2024 : socklen_t addrlen = sizeof(peer_storage);
132 : int accepted;
133 : do
134 : {
135 : accepted =
136 2024 : ::accept(fd_, reinterpret_cast<sockaddr*>(&peer_storage), &addrlen);
137 : }
138 2024 : while (accepted < 0 && errno == EINTR);
139 :
140 2024 : if (accepted >= 0)
141 : {
142 2 : if (accepted >= FD_SETSIZE)
143 : {
144 MIS 0 : ::close(accepted);
145 0 : op.complete(EINVAL, 0);
146 0 : op.impl_ptr = shared_from_this();
147 0 : svc_.post(&op);
148 0 : return std::noop_coroutine();
149 : }
150 :
151 HIT 2 : int flags = ::fcntl(accepted, F_GETFL, 0);
152 2 : if (flags == -1)
153 : {
154 MIS 0 : int err = errno;
155 0 : ::close(accepted);
156 0 : op.complete(err, 0);
157 0 : op.impl_ptr = shared_from_this();
158 0 : svc_.post(&op);
159 0 : return std::noop_coroutine();
160 : }
161 :
162 HIT 2 : if (::fcntl(accepted, F_SETFL, flags | O_NONBLOCK) == -1)
163 : {
164 MIS 0 : int err = errno;
165 0 : ::close(accepted);
166 0 : op.complete(err, 0);
167 0 : op.impl_ptr = shared_from_this();
168 0 : svc_.post(&op);
169 0 : return std::noop_coroutine();
170 : }
171 :
172 HIT 2 : if (::fcntl(accepted, F_SETFD, FD_CLOEXEC) == -1)
173 : {
174 MIS 0 : int err = errno;
175 0 : ::close(accepted);
176 0 : op.complete(err, 0);
177 0 : op.impl_ptr = shared_from_this();
178 0 : svc_.post(&op);
179 0 : return std::noop_coroutine();
180 : }
181 :
182 : {
183 HIT 2 : std::lock_guard lock(desc_state_.mutex);
184 2 : desc_state_.read_ready = false;
185 2 : }
186 :
187 2 : if (svc_.scheduler().try_consume_inline_budget())
188 : {
189 MIS 0 : auto* socket_svc = svc_.tcp_service();
190 0 : if (socket_svc)
191 : {
192 : auto& impl =
193 0 : static_cast<select_tcp_socket&>(*socket_svc->construct());
194 0 : impl.set_socket(accepted);
195 :
196 0 : impl.desc_state_.fd = accepted;
197 : {
198 0 : std::lock_guard lock(impl.desc_state_.mutex);
199 0 : impl.desc_state_.read_op = nullptr;
200 0 : impl.desc_state_.write_op = nullptr;
201 0 : impl.desc_state_.connect_op = nullptr;
202 0 : }
203 0 : socket_svc->scheduler().register_descriptor(
204 : accepted, &impl.desc_state_);
205 :
206 0 : impl.set_endpoints(
207 : local_endpoint_, from_sockaddr(peer_storage));
208 :
209 0 : *ec = {};
210 0 : if (impl_out)
211 0 : *impl_out = &impl;
212 : }
213 : else
214 : {
215 0 : ::close(accepted);
216 0 : *ec = make_err(ENOENT);
217 0 : if (impl_out)
218 0 : *impl_out = nullptr;
219 : }
220 0 : return dispatch_coro(ex, h);
221 : }
222 :
223 HIT 2 : op.accepted_fd = accepted;
224 2 : op.peer_storage = peer_storage;
225 2 : op.complete(0, 0);
226 2 : op.impl_ptr = shared_from_this();
227 2 : svc_.post(&op);
228 2 : return std::noop_coroutine();
229 : }
230 :
231 2022 : if (errno == EAGAIN || errno == EWOULDBLOCK)
232 : {
233 2022 : op.impl_ptr = shared_from_this();
234 2022 : svc_.work_started();
235 :
236 2022 : std::lock_guard lock(desc_state_.mutex);
237 2022 : bool io_done = false;
238 2022 : if (desc_state_.read_ready)
239 : {
240 MIS 0 : desc_state_.read_ready = false;
241 0 : op.perform_io();
242 0 : io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
243 0 : if (!io_done)
244 0 : op.errn = 0;
245 : }
246 :
247 HIT 2022 : if (io_done || op.cancelled.load(std::memory_order_acquire))
248 : {
249 MIS 0 : svc_.post(&op);
250 0 : svc_.work_finished();
251 : }
252 : else
253 : {
254 HIT 2022 : desc_state_.read_op = &op;
255 : }
256 2022 : return std::noop_coroutine();
257 2022 : }
258 :
259 MIS 0 : op.complete(errno, 0);
260 0 : op.impl_ptr = shared_from_this();
261 0 : svc_.post(&op);
262 0 : return std::noop_coroutine();
263 : }
264 :
265 : inline void
266 HIT 2 : select_tcp_acceptor::cancel() noexcept
267 : {
268 2 : do_cancel();
269 2 : }
270 :
271 : inline void
272 240 : select_tcp_acceptor::close_socket() noexcept
273 : {
274 240 : do_close_socket();
275 240 : }
276 :
277 195 : inline select_tcp_acceptor_service::select_tcp_acceptor_service(
278 195 : capy::execution_context& ctx)
279 195 : : ctx_(ctx)
280 195 : , state_(
281 : std::make_unique<select_tcp_acceptor_state>(
282 195 : ctx.use_service<select_scheduler>()))
283 : {
284 195 : }
285 :
286 390 : inline select_tcp_acceptor_service::~select_tcp_acceptor_service() {}
287 :
288 : inline void
289 195 : select_tcp_acceptor_service::shutdown()
290 : {
291 195 : std::lock_guard lock(state_->mutex_);
292 :
293 195 : while (auto* impl = state_->impl_list_.pop_front())
294 MIS 0 : impl->close_socket();
295 :
296 : // Don't clear impl_ptrs_ here — same rationale as
297 : // select_tcp_service::shutdown(). Let ~state_ release ptrs
298 : // after scheduler shutdown has drained all queued ops.
299 HIT 195 : }
300 :
301 : inline io_object::implementation*
302 61 : select_tcp_acceptor_service::construct()
303 : {
304 61 : auto impl = std::make_shared<select_tcp_acceptor>(*this);
305 61 : auto* raw = impl.get();
306 :
307 61 : std::lock_guard lock(state_->mutex_);
308 61 : state_->impl_ptrs_.emplace(raw, std::move(impl));
309 61 : state_->impl_list_.push_back(raw);
310 :
311 61 : return raw;
312 61 : }
313 :
314 : inline void
315 61 : select_tcp_acceptor_service::destroy(io_object::implementation* impl)
316 : {
317 61 : auto* select_impl = static_cast<select_tcp_acceptor*>(impl);
318 61 : select_impl->close_socket();
319 61 : std::lock_guard lock(state_->mutex_);
320 61 : state_->impl_list_.remove(select_impl);
321 61 : state_->impl_ptrs_.erase(select_impl);
322 61 : }
323 :
324 : inline void
325 120 : select_tcp_acceptor_service::close(io_object::handle& h)
326 : {
327 120 : static_cast<select_tcp_acceptor*>(h.get())->close_socket();
328 120 : }
329 :
330 : inline std::error_code
331 59 : select_tcp_acceptor_service::open_acceptor_socket(
332 : tcp_acceptor::implementation& impl, int family, int type, int protocol)
333 : {
334 59 : auto* select_impl = static_cast<select_tcp_acceptor*>(&impl);
335 59 : select_impl->close_socket();
336 :
337 59 : int fd = ::socket(family, type, protocol);
338 59 : if (fd < 0)
339 MIS 0 : return make_err(errno);
340 :
341 HIT 59 : int flags = ::fcntl(fd, F_GETFL, 0);
342 59 : if (flags == -1)
343 : {
344 MIS 0 : int errn = errno;
345 0 : ::close(fd);
346 0 : return make_err(errn);
347 : }
348 HIT 59 : if (::fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1)
349 : {
350 MIS 0 : int errn = errno;
351 0 : ::close(fd);
352 0 : return make_err(errn);
353 : }
354 HIT 59 : if (::fcntl(fd, F_SETFD, FD_CLOEXEC) == -1)
355 : {
356 MIS 0 : int errn = errno;
357 0 : ::close(fd);
358 0 : return make_err(errn);
359 : }
360 :
361 HIT 59 : if (fd >= FD_SETSIZE)
362 : {
363 MIS 0 : ::close(fd);
364 0 : return make_err(EMFILE);
365 : }
366 :
367 HIT 59 : if (family == AF_INET6)
368 : {
369 8 : int val = 0; // dual-stack default
370 8 : ::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &val, sizeof(val));
371 : }
372 :
373 : #ifdef SO_NOSIGPIPE
374 : {
375 : int nosig = 1;
376 : ::setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &nosig, sizeof(nosig));
377 : }
378 : #endif
379 :
380 59 : select_impl->fd_ = fd;
381 :
382 : // Set up descriptor state but do NOT register with reactor yet
383 : // (registration happens in do_listen via reactor_acceptor base)
384 59 : select_impl->desc_state_.fd = fd;
385 : {
386 59 : std::lock_guard lock(select_impl->desc_state_.mutex);
387 59 : select_impl->desc_state_.read_op = nullptr;
388 59 : }
389 :
390 59 : return {};
391 : }
392 :
393 : inline std::error_code
394 58 : select_tcp_acceptor_service::bind_acceptor(
395 : tcp_acceptor::implementation& impl, endpoint ep)
396 : {
397 58 : return static_cast<select_tcp_acceptor*>(&impl)->do_bind(ep);
398 : }
399 :
400 : inline std::error_code
401 57 : select_tcp_acceptor_service::listen_acceptor(
402 : tcp_acceptor::implementation& impl, int backlog)
403 : {
404 57 : return static_cast<select_tcp_acceptor*>(&impl)->do_listen(backlog);
405 : }
406 :
407 : inline void
408 5 : select_tcp_acceptor_service::post(scheduler_op* op)
409 : {
410 5 : state_->sched_.post(op);
411 5 : }
412 :
413 : inline void
414 2022 : select_tcp_acceptor_service::work_started() noexcept
415 : {
416 2022 : state_->sched_.work_started();
417 2022 : }
418 :
419 : inline void
420 3 : select_tcp_acceptor_service::work_finished() noexcept
421 : {
422 3 : state_->sched_.work_finished();
423 3 : }
424 :
425 : inline select_tcp_service*
426 2021 : select_tcp_acceptor_service::tcp_service() const noexcept
427 : {
428 2021 : auto* svc = ctx_.find_service<detail::tcp_service>();
429 2021 : return svc ? dynamic_cast<select_tcp_service*>(svc) : nullptr;
430 : }
431 :
432 : } // namespace boost::corosio::detail
433 :
434 : #endif // BOOST_COROSIO_HAS_SELECT
435 :
436 : #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_TCP_ACCEPTOR_SERVICE_HPP
|
