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