TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Michael Vandeberg
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_BACKEND_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BACKEND_HPP
12 :
13 : /* Reactor backend: acceptor accept() implementation.
14 :
15 : Contains the accept() method body for reactor_acceptor_impl,
16 : which needs all socket/service types to be complete. Included
17 : by per-backend type files (epoll_types.hpp, etc.) after all
18 : named types are defined.
19 : */
20 :
21 : #include <boost/corosio/native/detail/reactor/reactor_service_finals.hpp>
22 : #include <boost/corosio/native/detail/reactor/reactor_op_complete.hpp>
23 : #include <boost/corosio/native/detail/endpoint_convert.hpp>
24 : #include <boost/corosio/detail/dispatch_coro.hpp>
25 :
26 : #include <mutex>
27 :
28 : namespace boost::corosio::detail {
29 :
30 : // ============================================================
31 : // Acceptor accept() implementation
32 : // ============================================================
33 :
34 : template<
35 : class Derived,
36 : class Traits,
37 : class Service,
38 : class SocketFinal,
39 : class AccImplBase,
40 : class Endpoint>
41 : std::coroutine_handle<>
42 HIT 7336 : reactor_acceptor_impl<
43 : Derived,
44 : Traits,
45 : Service,
46 : SocketFinal,
47 : AccImplBase,
48 : Endpoint>::
49 : accept(
50 : capy::continuation& cont,
51 : capy::executor_ref ex,
52 : std::stop_token token,
53 : std::error_code* ec,
54 : io_object::implementation** impl_out)
55 : {
56 7336 : auto& op = this->acc_;
57 7336 : op.reset();
58 7336 : op.cont = &cont;
59 7336 : op.ex = ex;
60 7336 : op.ec_out = ec;
61 7336 : op.impl_out = impl_out;
62 7336 : op.fd = this->fd_;
63 7336 : op.start(token, static_cast<Derived*>(this));
64 :
65 7336 : sockaddr_storage peer_storage{};
66 7336 : socklen_t peer_addrlen = 0;
67 :
68 : int accepted =
69 7336 : Traits::accept_policy::do_accept(this->fd_, peer_storage, peer_addrlen);
70 :
71 7336 : if (accepted >= 0)
72 : {
73 : {
74 626 : std::lock_guard lock(this->desc_state_.mutex);
75 626 : this->desc_state_.read_ready = false;
76 626 : }
77 :
78 626 : if (this->svc_.scheduler().try_consume_inline_budget())
79 : {
80 506 : auto* socket_svc = this->svc_.stream_service();
81 506 : if (socket_svc)
82 : {
83 : auto& impl =
84 506 : static_cast<SocketFinal&>(*socket_svc->construct());
85 506 : impl.set_socket(
86 : accepted,
87 506 : peer_addrlen > 0 ? peer_storage.ss_family : AF_UNSPEC);
88 :
89 506 : impl.desc_state_.fd = accepted;
90 : {
91 506 : std::lock_guard lock(impl.desc_state_.mutex);
92 506 : impl.desc_state_.read_op = nullptr;
93 506 : impl.desc_state_.write_op = nullptr;
94 506 : impl.desc_state_.connect_op = nullptr;
95 506 : }
96 506 : auto reg_ec = socket_svc->scheduler().register_descriptor(
97 : accepted, &impl.desc_state_);
98 506 : if (reg_ec)
99 : {
100 : // destroy() closes the fd the impl already owns.
101 1 : socket_svc->destroy(&impl);
102 1 : *ec = reg_ec;
103 1 : if (impl_out)
104 1 : *impl_out = nullptr;
105 : }
106 : else
107 : {
108 505 : impl.set_endpoints(
109 : this->local_endpoint_,
110 505 : from_sockaddr_as(
111 : peer_storage, peer_addrlen, Endpoint{}));
112 :
113 505 : *ec = {};
114 505 : if (impl_out)
115 505 : *impl_out = &impl;
116 : }
117 : }
118 : else
119 : {
120 MIS 0 : ::close(accepted);
121 0 : *ec = make_err(ENOENT);
122 0 : if (impl_out)
123 0 : *impl_out = nullptr;
124 : }
125 : // Completed without parking: nothing left for a stop to cancel.
126 HIT 506 : op.stop_cb.reset();
127 506 : return dispatch_coro(ex, cont);
128 : }
129 :
130 120 : op.accepted_fd = accepted;
131 120 : op.peer_storage = peer_storage;
132 120 : op.peer_addrlen = peer_addrlen;
133 120 : op.complete(0, 0);
134 120 : op.object_ref_ = detail::object_ref(this);
135 120 : this->svc_.post(&op);
136 120 : return std::noop_coroutine();
137 : }
138 :
139 6710 : if (errno == EAGAIN || errno == EWOULDBLOCK)
140 : {
141 6700 : op.object_ref_ = detail::object_ref(this);
142 6700 : this->svc_.work_started();
143 :
144 : // Posted after the descriptor lock is released: the scheduler
145 : // takes its own lock before descriptor locks.
146 6700 : bool post_now = false;
147 : {
148 6700 : std::lock_guard lock(this->desc_state_.mutex);
149 6700 : bool io_done = false;
150 6700 : if (this->desc_state_.read_ready)
151 : {
152 2 : this->desc_state_.read_ready = false;
153 2 : op.perform_io();
154 2 : io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
155 2 : if (!io_done)
156 2 : op.errn = 0;
157 : }
158 :
159 6700 : if (io_done || op.cancelled.load(std::memory_order_acquire))
160 : {
161 MIS 0 : post_now = true;
162 : }
163 : else
164 : {
165 HIT 6700 : this->desc_state_.read_op = &op;
166 : // Select watches reads only for parked ops; see
167 : // park_op.
168 : if constexpr (Traits::needs_park_notification)
169 3019 : this->svc_.scheduler().notify_reactor();
170 : }
171 6700 : }
172 6700 : if (post_now)
173 MIS 0 : this->svc_.scheduler().post_counted(&op);
174 HIT 6700 : return std::noop_coroutine();
175 : }
176 :
177 10 : op.complete(errno, 0);
178 10 : op.object_ref_ = detail::object_ref(this);
179 10 : this->svc_.post(&op);
180 10 : return std::noop_coroutine();
181 : }
182 :
183 : } // namespace boost::corosio::detail
184 :
185 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BACKEND_HPP
|
