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 137478 : complete_io_op(Op& op)
41 : {
42 137478 : op.stop_cb.reset();
43 137478 : 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 274949 : decode_io_result(
50 : op.ec_out,
51 137478 : op.cancelled.load(std::memory_order_acquire),
52 137478 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
53 137478 : op.is_read_operation(), op.bytes_transferred, /*empty_buffer=*/false);
54 :
55 137478 : *op.bytes_out = op.bytes_transferred;
56 :
57 137478 : coro_resume(&op);
58 137478 : }
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 88 : complete_wait_op(Op& op)
99 : {
100 88 : 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 88 : if (op.socket_impl_)
105 : {
106 71 : if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
107 69 : sched->reset_inline_budget();
108 : }
109 17 : else if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
110 : {
111 17 : sched->reset_inline_budget();
112 : }
113 :
114 : // Wait reports only success/cancel/error — no bytes, no EOF.
115 174 : decode_io_result(
116 : op.ec_out,
117 88 : op.cancelled.load(std::memory_order_acquire),
118 88 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
119 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
120 :
121 88 : coro_resume(&op);
122 88 : }
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 8607 : complete_connect_op(Op& op)
136 : {
137 8607 : op.stop_cb.reset();
138 8607 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
139 :
140 8607 : bool success =
141 8607 : (op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
142 :
143 8607 : if (success && op.socket_impl_)
144 : {
145 : using ep_type = decltype(op.target_endpoint);
146 8564 : ep_type local_ep;
147 8564 : sockaddr_storage local_storage{};
148 8564 : socklen_t local_len = sizeof(local_storage);
149 8564 : if (::getsockname(
150 : op.fd, reinterpret_cast<sockaddr*>(&local_storage),
151 8564 : &local_len) == 0)
152 8543 : local_ep =
153 8564 : from_sockaddr_as(local_storage, local_len, ep_type{});
154 8564 : op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
155 : }
156 :
157 17182 : decode_io_result(
158 : op.ec_out,
159 8607 : op.cancelled.load(std::memory_order_acquire),
160 8607 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
161 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
162 :
163 8607 : coro_resume(&op);
164 8607 : }
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 (set to -1 on success).
176 : @param peer_storage The peer address from accept().
177 : @param impl_out Output pointer for the new socket impl.
178 : @param ec_out Output pointer for any error.
179 : @return True on success, false on failure.
180 : */
181 : template<typename SocketImpl, typename AcceptorImpl>
182 : bool
183 8532 : setup_accepted_socket(
184 : AcceptorImpl* acceptor_impl,
185 : int& accepted_fd,
186 : sockaddr_storage const& peer_storage,
187 : socklen_t peer_addrlen,
188 : io_object::implementation** impl_out,
189 : std::error_code* ec_out)
190 : {
191 8532 : auto* socket_svc = acceptor_impl->service().stream_service();
192 8532 : if (!socket_svc)
193 : {
194 MIS 0 : *ec_out = make_err(ENOENT);
195 0 : return false;
196 : }
197 :
198 HIT 8532 : auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
199 8532 : impl.set_socket(accepted_fd);
200 :
201 8532 : impl.desc_state_.fd = accepted_fd;
202 : {
203 8532 : std::lock_guard lock(impl.desc_state_.mutex);
204 8532 : impl.desc_state_.read_op = nullptr;
205 8532 : impl.desc_state_.write_op = nullptr;
206 8532 : impl.desc_state_.connect_op = nullptr;
207 8532 : }
208 8532 : socket_svc->scheduler().register_descriptor(accepted_fd, &impl.desc_state_);
209 :
210 : using ep_type = decltype(acceptor_impl->local_endpoint());
211 8532 : impl.set_endpoints(
212 : acceptor_impl->local_endpoint(),
213 8532 : from_sockaddr_as(
214 : peer_storage,
215 : peer_addrlen,
216 : ep_type{}));
217 :
218 8532 : if (impl_out)
219 8532 : *impl_out = &impl;
220 8532 : accepted_fd = -1;
221 8532 : return true;
222 : }
223 :
224 : /** Complete an accept operation.
225 :
226 : Sets up the peer socket on success, or closes the accepted
227 : fd on failure. Then resumes the caller via symmetric transfer.
228 :
229 : @tparam SocketImpl The concrete socket implementation type.
230 : @tparam Op The concrete accept operation type.
231 : @param op The operation to complete.
232 : */
233 : template<typename SocketImpl, typename Op>
234 : void
235 8576 : complete_accept_op(Op& op)
236 : {
237 8576 : op.stop_cb.reset();
238 8576 : if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
239 8572 : sched->reset_inline_budget();
240 :
241 8576 : bool success =
242 8576 : (op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
243 :
244 17148 : decode_io_result(
245 : op.ec_out,
246 8576 : op.cancelled.load(std::memory_order_acquire),
247 8576 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
248 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
249 :
250 8576 : if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
251 : {
252 8532 : if (!setup_accepted_socket<SocketImpl>(
253 8532 : op.acceptor_impl_, op.accepted_fd, op.peer_storage,
254 : op.peer_addrlen, op.impl_out, op.ec_out))
255 MIS 0 : success = false;
256 : }
257 :
258 HIT 8576 : if (!success || !op.acceptor_impl_)
259 : {
260 44 : if (op.accepted_fd >= 0)
261 : {
262 2 : ::close(op.accepted_fd);
263 2 : op.accepted_fd = -1;
264 : }
265 44 : if (op.impl_out)
266 44 : *op.impl_out = nullptr;
267 : }
268 :
269 8576 : coro_resume(&op);
270 8576 : }
271 :
272 : /** Complete a datagram operation (send_to or recv_from).
273 :
274 : For recv_from operations, writes the source endpoint from the
275 : recorded sockaddr_storage into the caller's endpoint pointer.
276 : Then resumes the caller via symmetric transfer.
277 :
278 : @tparam Op The concrete datagram operation type.
279 : @param op The operation to complete.
280 : */
281 : template<typename Op>
282 : void
283 58 : complete_datagram_op(Op& op)
284 : {
285 58 : op.stop_cb.reset();
286 58 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
287 :
288 : // No EOF: a zero-length datagram is valid (success with 0 bytes).
289 116 : decode_io_result(
290 : op.ec_out,
291 58 : op.cancelled.load(std::memory_order_acquire),
292 58 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
293 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
294 :
295 58 : *op.bytes_out = op.bytes_transferred;
296 :
297 58 : coro_resume(&op);
298 58 : }
299 :
300 : /** Complete a datagram operation with source endpoint capture.
301 :
302 : For recv_from operations, writes the source endpoint from the
303 : recorded sockaddr_storage into the caller's endpoint pointer.
304 : Then resumes the caller via symmetric transfer.
305 :
306 : @tparam Op The concrete datagram operation type.
307 : @param op The operation to complete.
308 : @param source_out Optional pointer to store source endpoint
309 : (non-null for recv_from, null for send_to).
310 : */
311 : template<typename Op, typename Endpoint>
312 : void
313 74 : complete_datagram_op(Op& op, Endpoint* source_out)
314 : {
315 74 : op.stop_cb.reset();
316 74 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
317 :
318 : // No EOF: a zero-length datagram is valid (success with 0 bytes).
319 148 : decode_io_result(
320 : op.ec_out,
321 74 : op.cancelled.load(std::memory_order_acquire),
322 74 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
323 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
324 :
325 74 : *op.bytes_out = op.bytes_transferred;
326 :
327 124 : if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
328 50 : op.errn == 0)
329 100 : *source_out = from_sockaddr_as(
330 50 : op.source_storage,
331 : op.source_addrlen,
332 : Endpoint{});
333 :
334 74 : coro_resume(&op);
335 74 : }
336 :
337 : } // namespace boost::corosio::detail
338 :
339 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
|
