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