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