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_REACTOR_REACTOR_OP_COMPLETE_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
12 :
13 : #include <boost/corosio/detail/dispatch_coro.hpp>
14 : #include <boost/corosio/native/detail/coro_op_complete.hpp>
15 : #include <boost/corosio/native/detail/endpoint_convert.hpp>
16 : #include <boost/corosio/native/detail/make_err.hpp>
17 : #include <boost/corosio/io/io_object.hpp>
18 :
19 : #include <coroutine>
20 : #include <mutex>
21 : #include <utility>
22 :
23 : #include <netinet/in.h>
24 : #include <sys/socket.h>
25 : #include <unistd.h>
26 :
27 : namespace boost::corosio::detail {
28 :
29 : /** Complete a base read/write operation.
30 :
31 : Translates the recorded errno and cancellation state into
32 : an error_code, stores the byte count, then resumes the
33 : caller via symmetric transfer.
34 :
35 : @tparam Op The concrete operation type.
36 : @param op The operation to complete.
37 : */
38 : template<typename Op>
39 : void
40 HIT 86504 : complete_io_op(Op& op)
41 : {
42 86504 : op.stop_cb.reset();
43 : // scheduler_ is null until the descriptor is registered; an op
44 : // completed by the closed-object entry check never registered and
45 : // has no budget to reset.
46 86504 : if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
47 86488 : sched->reset_inline_budget();
48 :
49 : // is_read_operation() already folds in the empty-buffer case (it
50 : // returns false for a zero-length read), so empty_buffer stays false
51 : // here and the shared EOF test reduces to the reactor's original
52 : // `is_read && bytes == 0`.
53 172965 : decode_io_result(
54 : op.ec_out,
55 86504 : op.cancelled.load(std::memory_order_acquire),
56 86504 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
57 86504 : op.is_read_operation(), op.bytes_transferred, /*empty_buffer=*/false);
58 :
59 86504 : *op.bytes_out = op.bytes_transferred;
60 :
61 86504 : coro_resume(&op);
62 86504 : }
63 :
64 : /** Complete a wait operation.
65 :
66 : Wait operations report only an error_code — no bytes_transferred,
67 : no EOF translation. Used for socket and acceptor wait() awaitables;
68 : picks the impl pointer set by start() to reach the scheduler.
69 :
70 : @tparam Op The concrete wait operation type.
71 : @param op The operation to complete.
72 : */
73 : template<typename Op>
74 : void
75 171 : complete_wait_op(Op& op)
76 : {
77 171 : op.stop_cb.reset();
78 : // scheduler_ is null until the descriptor is registered; a wait
79 : // completed by the initiation probe (e.g. EBADF on a never-opened
80 : // socket) has no registration to reset a budget for.
81 171 : if (op.socket_impl_)
82 : {
83 130 : if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
84 118 : sched->reset_inline_budget();
85 : }
86 41 : else if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
87 : {
88 37 : sched->reset_inline_budget();
89 : }
90 :
91 : // Wait reports only success/cancel/error — no bytes, no EOF.
92 311 : decode_io_result(
93 : op.ec_out,
94 171 : op.cancelled.load(std::memory_order_acquire),
95 171 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
96 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
97 :
98 171 : coro_resume(&op);
99 171 : }
100 :
101 : /** Complete a connect operation with endpoint caching.
102 :
103 : On success, queries the local endpoint via getsockname and
104 : caches both endpoints in the socket impl. Then resumes the
105 : caller via symmetric transfer.
106 :
107 : @tparam Op The concrete connect operation type.
108 : @param op The operation to complete.
109 : */
110 : template<typename Op>
111 : void
112 4404 : complete_connect_op(Op& op)
113 : {
114 4404 : op.stop_cb.reset();
115 4404 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
116 :
117 4404 : bool success =
118 4404 : (op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
119 :
120 4404 : if (success && op.socket_impl_)
121 : {
122 : using ep_type = decltype(op.target_endpoint);
123 4347 : ep_type local_ep;
124 4347 : sockaddr_storage local_storage{};
125 4347 : socklen_t local_len = sizeof(local_storage);
126 4347 : if (::getsockname(
127 : op.fd, reinterpret_cast<sockaddr*>(&local_storage),
128 4347 : &local_len) == 0)
129 4314 : local_ep =
130 4347 : from_sockaddr_as(local_storage, local_len, ep_type{});
131 4347 : op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
132 : }
133 :
134 8762 : decode_io_result(
135 : op.ec_out,
136 4404 : op.cancelled.load(std::memory_order_acquire),
137 4404 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
138 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
139 :
140 4404 : coro_resume(&op);
141 4404 : }
142 :
143 : /** Construct and register a peer socket from an accepted fd.
144 :
145 : Creates a new socket impl via the acceptor's associated
146 : socket service, registers it with the scheduler, and caches
147 : the local and remote endpoints.
148 :
149 : @tparam SocketImpl The concrete socket implementation type.
150 : @tparam AcceptorImpl The concrete acceptor implementation type.
151 : @param acceptor_impl The acceptor that accepted the connection.
152 : @param accepted_fd The accepted file descriptor. Cleared to -1
153 : once the socket impl owns it, which includes the registration
154 : failure that destroys the impl and closes the fd with it.
155 : @param peer_storage The peer address from accept().
156 : @param impl_out Output pointer for the new socket impl.
157 : @param ec_out Output pointer for any error.
158 : @return True on success, false on failure.
159 : */
160 : template<typename SocketImpl, typename AcceptorImpl>
161 : bool
162 4293 : setup_accepted_socket(
163 : AcceptorImpl* acceptor_impl,
164 : int& accepted_fd,
165 : sockaddr_storage const& peer_storage,
166 : socklen_t peer_addrlen,
167 : io_object::implementation** impl_out,
168 : std::error_code* ec_out)
169 : {
170 4293 : auto* socket_svc = acceptor_impl->service().stream_service();
171 4293 : if (!socket_svc)
172 : {
173 MIS 0 : *ec_out = make_err(ENOENT);
174 0 : return false;
175 : }
176 :
177 HIT 4293 : auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
178 4293 : impl.set_socket(accepted_fd);
179 :
180 4293 : impl.desc_state_.fd = accepted_fd;
181 : {
182 4293 : std::lock_guard lock(impl.desc_state_.mutex);
183 4293 : impl.desc_state_.read_op = nullptr;
184 4293 : impl.desc_state_.write_op = nullptr;
185 4293 : impl.desc_state_.connect_op = nullptr;
186 4293 : }
187 4293 : if (auto ec = socket_svc->scheduler().register_descriptor(
188 : accepted_fd, &impl.desc_state_))
189 : {
190 : // destroy() closes the fd the impl already owns.
191 1 : accepted_fd = -1;
192 1 : socket_svc->destroy(&impl);
193 1 : *ec_out = ec;
194 1 : return false;
195 : }
196 :
197 : using ep_type = decltype(acceptor_impl->local_endpoint());
198 4292 : impl.set_endpoints(
199 : acceptor_impl->local_endpoint(),
200 4292 : from_sockaddr_as(
201 : peer_storage,
202 : peer_addrlen,
203 : ep_type{}));
204 :
205 4292 : if (impl_out)
206 4292 : *impl_out = &impl;
207 4292 : accepted_fd = -1;
208 4292 : return true;
209 : }
210 :
211 : /** Complete an accept operation.
212 :
213 : Sets up the peer socket on success, or closes the accepted
214 : fd on failure. Then resumes the caller via symmetric transfer.
215 :
216 : @tparam SocketImpl The concrete socket implementation type.
217 : @tparam Op The concrete accept operation type.
218 : @param op The operation to complete.
219 : */
220 : template<typename SocketImpl, typename Op>
221 : void
222 4404 : complete_accept_op(Op& op)
223 : {
224 4404 : op.stop_cb.reset();
225 4404 : if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
226 4400 : sched->reset_inline_budget();
227 :
228 4404 : bool success =
229 4404 : (op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
230 :
231 8797 : decode_io_result(
232 : op.ec_out,
233 4404 : op.cancelled.load(std::memory_order_acquire),
234 4404 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
235 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
236 :
237 4404 : if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
238 : {
239 4293 : if (!setup_accepted_socket<SocketImpl>(
240 4293 : op.acceptor_impl_, op.accepted_fd, op.peer_storage,
241 : op.peer_addrlen, op.impl_out, op.ec_out))
242 1 : success = false;
243 : }
244 :
245 4404 : if (!success || !op.acceptor_impl_)
246 : {
247 112 : if (op.accepted_fd >= 0)
248 : {
249 2 : ::close(op.accepted_fd);
250 2 : op.accepted_fd = -1;
251 : }
252 112 : if (op.impl_out)
253 112 : *op.impl_out = nullptr;
254 : }
255 :
256 4404 : coro_resume(&op);
257 4404 : }
258 :
259 : /** Complete a datagram operation (send_to or recv_from).
260 :
261 : For recv_from operations, writes the source endpoint from the
262 : recorded sockaddr_storage into the caller's endpoint pointer.
263 : Then resumes the caller via symmetric transfer.
264 :
265 : @tparam Op The concrete datagram operation type.
266 : @param op The operation to complete.
267 : */
268 : template<typename Op>
269 : void
270 113 : complete_datagram_op(Op& op)
271 : {
272 113 : op.stop_cb.reset();
273 113 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
274 :
275 : // No EOF: a zero-length datagram is valid (success with 0 bytes).
276 222 : decode_io_result(
277 : op.ec_out,
278 113 : op.cancelled.load(std::memory_order_acquire),
279 113 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
280 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
281 :
282 113 : *op.bytes_out = op.bytes_transferred;
283 :
284 113 : coro_resume(&op);
285 113 : }
286 :
287 : /** Complete a datagram operation with source endpoint capture.
288 :
289 : For recv_from operations, writes the source endpoint from the
290 : recorded sockaddr_storage into the caller's endpoint pointer.
291 : Then resumes the caller via symmetric transfer.
292 :
293 : @tparam Op The concrete datagram operation type.
294 : @param op The operation to complete.
295 : @param source_out Optional pointer to store source endpoint
296 : (non-null for recv_from, null for send_to).
297 : */
298 : template<typename Op, typename Endpoint>
299 : void
300 95 : complete_datagram_op(Op& op, Endpoint* source_out)
301 : {
302 95 : op.stop_cb.reset();
303 95 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
304 :
305 : // No EOF: a zero-length datagram is valid (success with 0 bytes).
306 188 : decode_io_result(
307 : op.ec_out,
308 95 : op.cancelled.load(std::memory_order_acquire),
309 95 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
310 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
311 :
312 95 : *op.bytes_out = op.bytes_transferred;
313 :
314 156 : if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
315 61 : op.errn == 0)
316 118 : *source_out = from_sockaddr_as(
317 59 : op.source_storage,
318 : op.source_addrlen,
319 : Endpoint{});
320 :
321 95 : coro_resume(&op);
322 95 : }
323 :
324 : } // namespace boost::corosio::detail
325 :
326 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
|
