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 82018 : complete_io_op(Op& op)
41 : {
42 82018 : 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 82018 : if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
47 82002 : 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 163993 : decode_io_result(
54 82018 : op.ec_out, op.cancelled.load(std::memory_order_acquire),
55 82018 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
56 82018 : op.is_read_operation(), op.bytes_transferred, /*empty_buffer=*/false);
57 :
58 82018 : *op.bytes_out = op.bytes_transferred;
59 :
60 82018 : coro_resume(&op);
61 82018 : }
62 :
63 : /** Complete a wait operation.
64 :
65 : Wait operations report only an error_code — no bytes_transferred,
66 : no EOF translation. Used for socket and acceptor wait() awaitables;
67 : picks the impl pointer set by start() to reach the scheduler.
68 :
69 : @tparam Op The concrete wait operation type.
70 : @param op The operation to complete.
71 : */
72 : template<typename Op>
73 : void
74 171 : complete_wait_op(Op& op)
75 : {
76 171 : op.stop_cb.reset();
77 : // scheduler_ is null until the descriptor is registered; a wait
78 : // completed by the initiation probe (e.g. EBADF on a never-opened
79 : // socket) has no registration to reset a budget for.
80 171 : if (op.socket_impl_)
81 : {
82 130 : if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
83 118 : sched->reset_inline_budget();
84 : }
85 41 : else if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
86 : {
87 37 : sched->reset_inline_budget();
88 : }
89 :
90 : // Wait reports only success/cancel/error — no bytes, no EOF.
91 311 : decode_io_result(
92 171 : op.ec_out, op.cancelled.load(std::memory_order_acquire),
93 171 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
94 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
95 :
96 171 : coro_resume(&op);
97 171 : }
98 :
99 : /** Complete a connect operation with endpoint caching.
100 :
101 : On success, queries the local endpoint via getsockname and
102 : caches both endpoints in the socket impl. Then resumes the
103 : caller via symmetric transfer.
104 :
105 : @tparam Op The concrete connect operation type.
106 : @param op The operation to complete.
107 : */
108 : template<typename Op>
109 : void
110 4611 : complete_connect_op(Op& op)
111 : {
112 4611 : op.stop_cb.reset();
113 4611 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
114 :
115 4611 : bool success =
116 4611 : (op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
117 :
118 4611 : if (success && op.socket_impl_)
119 : {
120 : using ep_type = decltype(op.target_endpoint);
121 4554 : ep_type local_ep;
122 4554 : sockaddr_storage local_storage{};
123 4554 : socklen_t local_len = sizeof(local_storage);
124 4554 : if (::getsockname(
125 : op.fd, reinterpret_cast<sockaddr*>(&local_storage),
126 4554 : &local_len) == 0)
127 4554 : local_ep = from_sockaddr_as(local_storage, local_len, ep_type{});
128 4554 : op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
129 : }
130 :
131 9176 : decode_io_result(
132 4611 : op.ec_out, op.cancelled.load(std::memory_order_acquire),
133 4611 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
134 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
135 :
136 4611 : coro_resume(&op);
137 4611 : }
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 4500 : 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 4500 : auto* socket_svc = acceptor_impl->service().stream_service();
167 4500 : if (!socket_svc)
168 : {
169 MIS 0 : *ec_out = make_err(ENOENT);
170 0 : return false;
171 : }
172 :
173 HIT 4500 : auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
174 4500 : impl.set_socket(accepted_fd);
175 :
176 4500 : impl.desc_state_.fd = accepted_fd;
177 : {
178 4500 : std::lock_guard lock(impl.desc_state_.mutex);
179 4500 : impl.desc_state_.read_op = nullptr;
180 4500 : impl.desc_state_.write_op = nullptr;
181 4500 : impl.desc_state_.connect_op = nullptr;
182 4500 : }
183 4500 : 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 4499 : impl.set_endpoints(
195 : acceptor_impl->local_endpoint(),
196 4499 : from_sockaddr_as(peer_storage, peer_addrlen, ep_type{}));
197 :
198 4499 : if (impl_out)
199 4499 : *impl_out = &impl;
200 4499 : accepted_fd = -1;
201 4499 : 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 4611 : complete_accept_op(Op& op)
216 : {
217 4611 : op.stop_cb.reset();
218 4611 : if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
219 4607 : sched->reset_inline_budget();
220 :
221 4611 : bool success =
222 4611 : (op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
223 :
224 9211 : decode_io_result(
225 4611 : op.ec_out, op.cancelled.load(std::memory_order_acquire),
226 4611 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
227 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
228 :
229 4611 : if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
230 : {
231 4500 : if (!setup_accepted_socket<SocketImpl>(
232 4500 : op.acceptor_impl_, op.accepted_fd, op.peer_storage,
233 : op.peer_addrlen, op.impl_out, op.ec_out))
234 1 : success = false;
235 : }
236 :
237 4611 : if (!success || !op.acceptor_impl_)
238 : {
239 112 : if (op.accepted_fd >= 0)
240 : {
241 2 : ::close(op.accepted_fd);
242 2 : op.accepted_fd = -1;
243 : }
244 112 : if (op.impl_out)
245 112 : *op.impl_out = nullptr;
246 : }
247 :
248 4611 : coro_resume(&op);
249 4611 : }
250 :
251 : /** Complete a datagram operation (send_to or recv_from).
252 :
253 : For recv_from operations, writes the source endpoint from the
254 : recorded sockaddr_storage into the caller's endpoint pointer.
255 : Then resumes the caller via symmetric transfer.
256 :
257 : @tparam Op The concrete datagram operation type.
258 : @param op The operation to complete.
259 : */
260 : template<typename Op>
261 : void
262 113 : complete_datagram_op(Op& op)
263 : {
264 113 : op.stop_cb.reset();
265 113 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
266 :
267 : // No EOF: a zero-length datagram is valid (success with 0 bytes).
268 222 : decode_io_result(
269 113 : op.ec_out, op.cancelled.load(std::memory_order_acquire),
270 113 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
271 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
272 :
273 113 : *op.bytes_out = op.bytes_transferred;
274 :
275 113 : coro_resume(&op);
276 113 : }
277 :
278 : /** Complete a datagram operation with source endpoint capture.
279 :
280 : For recv_from operations, writes the source endpoint from the
281 : recorded sockaddr_storage into the caller's endpoint pointer.
282 : Then resumes the caller via symmetric transfer.
283 :
284 : @tparam Op The concrete datagram operation type.
285 : @param op The operation to complete.
286 : @param source_out Optional pointer to store source endpoint
287 : (non-null for recv_from, null for send_to).
288 : */
289 : template<typename Op, typename Endpoint>
290 : void
291 95 : complete_datagram_op(Op& op, Endpoint* source_out)
292 : {
293 95 : op.stop_cb.reset();
294 95 : op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
295 :
296 : // No EOF: a zero-length datagram is valid (success with 0 bytes).
297 188 : decode_io_result(
298 95 : op.ec_out, op.cancelled.load(std::memory_order_acquire),
299 95 : op.errn != 0 ? make_err(op.errn) : std::error_code{},
300 : /*is_read=*/false, /*bytes=*/0, /*empty_buffer=*/false);
301 :
302 95 : *op.bytes_out = op.bytes_transferred;
303 :
304 156 : if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
305 61 : op.errn == 0)
306 20 : *source_out =
307 59 : from_sockaddr_as(op.source_storage, op.source_addrlen, Endpoint{});
308 :
309 95 : coro_resume(&op);
310 95 : }
311 :
312 : } // namespace boost::corosio::detail
313 :
314 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
|
