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_STREAM_SOCKET_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
12 :
13 : #include <boost/corosio/tcp_socket.hpp>
14 : #include <boost/corosio/native/detail/reactor/reactor_basic_socket.hpp>
15 : #include <boost/corosio/detail/dispatch_coro.hpp>
16 : #include <boost/capy/buffers.hpp>
17 :
18 : #include <coroutine>
19 :
20 : #include <errno.h>
21 : #include <sys/socket.h>
22 : #include <sys/uio.h>
23 :
24 : namespace boost::corosio::detail {
25 :
26 : /** CRTP base for reactor-backed stream socket implementations.
27 :
28 : Inherits shared data members and cancel/close/register logic
29 : from reactor_basic_socket. Adds the TCP-specific remote
30 : endpoint, shutdown, and I/O dispatch (connect, read, write).
31 :
32 : @tparam Derived The concrete socket type (CRTP).
33 : @tparam Service The backend's socket service type.
34 : @tparam ConnOp The backend's connect op type.
35 : @tparam ReadOp The backend's read op type.
36 : @tparam WriteOp The backend's write op type.
37 : @tparam DescState The backend's descriptor_state type.
38 : */
39 : template<
40 : class Derived,
41 : class Service,
42 : class ConnOp,
43 : class ReadOp,
44 : class WriteOp,
45 : class DescState>
46 : class reactor_stream_socket
47 : : public reactor_basic_socket<
48 : Derived,
49 : tcp_socket::implementation,
50 : Service,
51 : DescState>
52 : {
53 : using base_type = reactor_basic_socket<
54 : Derived,
55 : tcp_socket::implementation,
56 : Service,
57 : DescState>;
58 : friend base_type;
59 : friend Derived;
60 :
61 HIT 24026 : explicit reactor_stream_socket(Service& svc) noexcept : base_type(svc) {}
62 :
63 : protected:
64 : endpoint remote_endpoint_;
65 :
66 : public:
67 : /// Pending connect operation slot.
68 : ConnOp conn_;
69 :
70 : /// Pending read operation slot.
71 : ReadOp rd_;
72 :
73 : /// Pending write operation slot.
74 : WriteOp wr_;
75 :
76 24026 : ~reactor_stream_socket() override = default;
77 :
78 : /// Return the cached remote endpoint.
79 42 : endpoint remote_endpoint() const noexcept override
80 : {
81 42 : return remote_endpoint_;
82 : }
83 :
84 : /// Shut down part or all of the full-duplex connection.
85 6 : std::error_code shutdown(tcp_socket::shutdown_type what) noexcept override
86 : {
87 : int how;
88 6 : switch (what)
89 : {
90 2 : case tcp_socket::shutdown_receive:
91 2 : how = SHUT_RD;
92 2 : break;
93 2 : case tcp_socket::shutdown_send:
94 2 : how = SHUT_WR;
95 2 : break;
96 2 : case tcp_socket::shutdown_both:
97 2 : how = SHUT_RDWR;
98 2 : break;
99 MIS 0 : default:
100 0 : return make_err(EINVAL);
101 : }
102 HIT 6 : if (::shutdown(this->fd_, how) != 0)
103 MIS 0 : return make_err(errno);
104 HIT 6 : return {};
105 : }
106 :
107 : /// Cache local and remote endpoints.
108 15958 : void set_endpoints(endpoint local, endpoint remote) noexcept
109 : {
110 15958 : this->local_endpoint_ = local;
111 15958 : remote_endpoint_ = remote;
112 15958 : }
113 :
114 : /** Shared connect dispatch.
115 :
116 : Tries the connect syscall speculatively. On synchronous
117 : completion, returns via inline budget or posts through queue.
118 : On EINPROGRESS, registers with the reactor.
119 : */
120 : std::coroutine_handle<> do_connect(
121 : std::coroutine_handle<>,
122 : capy::executor_ref,
123 : endpoint,
124 : std::stop_token const&,
125 : std::error_code*);
126 :
127 : /** Shared scatter-read dispatch.
128 :
129 : Tries readv() speculatively. On success or hard error,
130 : returns via inline budget or posts through queue.
131 : On EAGAIN, registers with the reactor.
132 : */
133 : std::coroutine_handle<> do_read_some(
134 : std::coroutine_handle<>,
135 : capy::executor_ref,
136 : buffer_param,
137 : std::stop_token const&,
138 : std::error_code*,
139 : std::size_t*);
140 :
141 : /** Shared gather-write dispatch.
142 :
143 : Tries the write via WriteOp::write_policy speculatively.
144 : On success or hard error, returns via inline budget or
145 : posts through queue. On EAGAIN, registers with the reactor.
146 : */
147 : std::coroutine_handle<> do_write_some(
148 : std::coroutine_handle<>,
149 : capy::executor_ref,
150 : buffer_param,
151 : std::stop_token const&,
152 : std::error_code*,
153 : std::size_t*);
154 :
155 : /** Close the socket and cancel pending operations.
156 :
157 : Extends the base do_close_socket() to also reset
158 : the remote endpoint.
159 : */
160 72057 : void do_close_socket() noexcept
161 : {
162 72057 : base_type::do_close_socket();
163 72057 : remote_endpoint_ = endpoint{};
164 72057 : }
165 :
166 : private:
167 : // CRTP callbacks for reactor_basic_socket cancel/close
168 :
169 : template<class Op>
170 190 : reactor_op_base** op_to_desc_slot(Op& op) noexcept
171 : {
172 190 : if (&op == static_cast<void*>(&conn_))
173 MIS 0 : return &this->desc_state_.connect_op;
174 HIT 190 : if (&op == static_cast<void*>(&rd_))
175 190 : return &this->desc_state_.read_op;
176 MIS 0 : if (&op == static_cast<void*>(&wr_))
177 0 : return &this->desc_state_.write_op;
178 0 : return nullptr;
179 : }
180 :
181 : template<class Op>
182 0 : bool* op_to_cancel_flag(Op& op) noexcept
183 : {
184 0 : if (&op == static_cast<void*>(&conn_))
185 0 : return &this->desc_state_.connect_cancel_pending;
186 0 : if (&op == static_cast<void*>(&rd_))
187 0 : return &this->desc_state_.read_cancel_pending;
188 0 : if (&op == static_cast<void*>(&wr_))
189 0 : return &this->desc_state_.write_cancel_pending;
190 0 : return nullptr;
191 : }
192 :
193 : template<class Fn>
194 HIT 72241 : void for_each_op(Fn fn) noexcept
195 : {
196 72241 : fn(conn_);
197 72241 : fn(rd_);
198 72241 : fn(wr_);
199 72241 : }
200 :
201 : template<class Fn>
202 72241 : void for_each_desc_entry(Fn fn) noexcept
203 : {
204 72241 : fn(conn_, this->desc_state_.connect_op);
205 72241 : fn(rd_, this->desc_state_.read_op);
206 72241 : fn(wr_, this->desc_state_.write_op);
207 72241 : }
208 : };
209 :
210 : template<
211 : class Derived,
212 : class Service,
213 : class ConnOp,
214 : class ReadOp,
215 : class WriteOp,
216 : class DescState>
217 : std::coroutine_handle<>
218 7983 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState>::
219 : do_connect(
220 : std::coroutine_handle<> h,
221 : capy::executor_ref ex,
222 : endpoint ep,
223 : std::stop_token const& token,
224 : std::error_code* ec)
225 : {
226 7983 : auto& op = conn_;
227 :
228 7983 : sockaddr_storage storage{};
229 7983 : socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
230 : int result =
231 7983 : ::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
232 :
233 7983 : if (result == 0)
234 : {
235 MIS 0 : sockaddr_storage local_storage{};
236 0 : socklen_t local_len = sizeof(local_storage);
237 0 : if (::getsockname(
238 : this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
239 0 : &local_len) == 0)
240 0 : this->local_endpoint_ = from_sockaddr(local_storage);
241 0 : remote_endpoint_ = ep;
242 : }
243 :
244 HIT 7983 : if (result == 0 || errno != EINPROGRESS)
245 : {
246 MIS 0 : int err = (result < 0) ? errno : 0;
247 0 : if (this->svc_.scheduler().try_consume_inline_budget())
248 : {
249 0 : *ec = err ? make_err(err) : std::error_code{};
250 0 : return dispatch_coro(ex, h);
251 : }
252 0 : op.reset();
253 0 : op.h = h;
254 0 : op.ex = ex;
255 0 : op.ec_out = ec;
256 0 : op.fd = this->fd_;
257 0 : op.target_endpoint = ep;
258 0 : op.start(token, static_cast<Derived*>(this));
259 0 : op.impl_ptr = this->shared_from_this();
260 0 : op.complete(err, 0);
261 0 : this->svc_.post(&op);
262 0 : return std::noop_coroutine();
263 : }
264 :
265 : // EINPROGRESS — register with reactor
266 HIT 7983 : op.reset();
267 7983 : op.h = h;
268 7983 : op.ex = ex;
269 7983 : op.ec_out = ec;
270 7983 : op.fd = this->fd_;
271 7983 : op.target_endpoint = ep;
272 7983 : op.start(token, static_cast<Derived*>(this));
273 7983 : op.impl_ptr = this->shared_from_this();
274 :
275 7983 : this->register_op(
276 7983 : op, this->desc_state_.connect_op, this->desc_state_.write_ready,
277 7983 : this->desc_state_.connect_cancel_pending);
278 7983 : return std::noop_coroutine();
279 : }
280 :
281 : template<
282 : class Derived,
283 : class Service,
284 : class ConnOp,
285 : class ReadOp,
286 : class WriteOp,
287 : class DescState>
288 : std::coroutine_handle<>
289 221043 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState>::
290 : do_read_some(
291 : std::coroutine_handle<> h,
292 : capy::executor_ref ex,
293 : buffer_param param,
294 : std::stop_token const& token,
295 : std::error_code* ec,
296 : std::size_t* bytes_out)
297 : {
298 221043 : auto& op = rd_;
299 221043 : op.reset();
300 :
301 221043 : capy::mutable_buffer bufs[ReadOp::max_buffers];
302 221043 : op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
303 :
304 221043 : if (op.iovec_count == 0 || (op.iovec_count == 1 && bufs[0].size() == 0))
305 : {
306 2 : op.empty_buffer_read = true;
307 2 : op.h = h;
308 2 : op.ex = ex;
309 2 : op.ec_out = ec;
310 2 : op.bytes_out = bytes_out;
311 2 : op.start(token, static_cast<Derived*>(this));
312 2 : op.impl_ptr = this->shared_from_this();
313 2 : op.complete(0, 0);
314 2 : this->svc_.post(&op);
315 2 : return std::noop_coroutine();
316 : }
317 :
318 442082 : for (int i = 0; i < op.iovec_count; ++i)
319 : {
320 221041 : op.iovecs[i].iov_base = bufs[i].data();
321 221041 : op.iovecs[i].iov_len = bufs[i].size();
322 : }
323 :
324 : // Speculative read
325 : ssize_t n;
326 : do
327 : {
328 221041 : n = ::readv(this->fd_, op.iovecs, op.iovec_count);
329 : }
330 221041 : while (n < 0 && errno == EINTR);
331 :
332 221041 : if (n >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK))
333 : {
334 220656 : int err = (n < 0) ? errno : 0;
335 220656 : auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
336 :
337 220656 : if (this->svc_.scheduler().try_consume_inline_budget())
338 : {
339 176558 : if (err)
340 MIS 0 : *ec = make_err(err);
341 HIT 176558 : else if (n == 0)
342 10 : *ec = capy::error::eof;
343 : else
344 176548 : *ec = {};
345 176558 : *bytes_out = bytes;
346 176558 : return dispatch_coro(ex, h);
347 : }
348 44098 : op.h = h;
349 44098 : op.ex = ex;
350 44098 : op.ec_out = ec;
351 44098 : op.bytes_out = bytes_out;
352 44098 : op.start(token, static_cast<Derived*>(this));
353 44098 : op.impl_ptr = this->shared_from_this();
354 44098 : op.complete(err, bytes);
355 44098 : this->svc_.post(&op);
356 44098 : return std::noop_coroutine();
357 : }
358 :
359 : // EAGAIN — register with reactor
360 385 : op.h = h;
361 385 : op.ex = ex;
362 385 : op.ec_out = ec;
363 385 : op.bytes_out = bytes_out;
364 385 : op.fd = this->fd_;
365 385 : op.start(token, static_cast<Derived*>(this));
366 385 : op.impl_ptr = this->shared_from_this();
367 :
368 385 : this->register_op(
369 385 : op, this->desc_state_.read_op, this->desc_state_.read_ready,
370 385 : this->desc_state_.read_cancel_pending);
371 385 : return std::noop_coroutine();
372 : }
373 :
374 : template<
375 : class Derived,
376 : class Service,
377 : class ConnOp,
378 : class ReadOp,
379 : class WriteOp,
380 : class DescState>
381 : std::coroutine_handle<>
382 220750 : reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState>::
383 : do_write_some(
384 : std::coroutine_handle<> h,
385 : capy::executor_ref ex,
386 : buffer_param param,
387 : std::stop_token const& token,
388 : std::error_code* ec,
389 : std::size_t* bytes_out)
390 : {
391 220750 : auto& op = wr_;
392 220750 : op.reset();
393 :
394 220750 : capy::mutable_buffer bufs[WriteOp::max_buffers];
395 220750 : op.iovec_count =
396 220750 : static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
397 :
398 220750 : if (op.iovec_count == 0 || (op.iovec_count == 1 && bufs[0].size() == 0))
399 : {
400 2 : op.h = h;
401 2 : op.ex = ex;
402 2 : op.ec_out = ec;
403 2 : op.bytes_out = bytes_out;
404 2 : op.start(token, static_cast<Derived*>(this));
405 2 : op.impl_ptr = this->shared_from_this();
406 2 : op.complete(0, 0);
407 2 : this->svc_.post(&op);
408 2 : return std::noop_coroutine();
409 : }
410 :
411 441496 : for (int i = 0; i < op.iovec_count; ++i)
412 : {
413 220748 : op.iovecs[i].iov_base = bufs[i].data();
414 220748 : op.iovecs[i].iov_len = bufs[i].size();
415 : }
416 :
417 : // Speculative write via backend-specific write policy
418 : ssize_t n =
419 220748 : WriteOp::write_policy::write(this->fd_, op.iovecs, op.iovec_count);
420 :
421 220748 : if (n >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK))
422 : {
423 220748 : int err = (n < 0) ? errno : 0;
424 220748 : auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
425 :
426 220748 : if (this->svc_.scheduler().try_consume_inline_budget())
427 : {
428 176616 : *ec = err ? make_err(err) : std::error_code{};
429 176616 : *bytes_out = bytes;
430 176616 : return dispatch_coro(ex, h);
431 : }
432 44132 : op.h = h;
433 44132 : op.ex = ex;
434 44132 : op.ec_out = ec;
435 44132 : op.bytes_out = bytes_out;
436 44132 : op.start(token, static_cast<Derived*>(this));
437 44132 : op.impl_ptr = this->shared_from_this();
438 44132 : op.complete(err, bytes);
439 44132 : this->svc_.post(&op);
440 44132 : return std::noop_coroutine();
441 : }
442 :
443 : // EAGAIN — register with reactor
444 MIS 0 : op.h = h;
445 0 : op.ex = ex;
446 0 : op.ec_out = ec;
447 0 : op.bytes_out = bytes_out;
448 0 : op.fd = this->fd_;
449 0 : op.start(token, static_cast<Derived*>(this));
450 0 : op.impl_ptr = this->shared_from_this();
451 :
452 0 : this->register_op(
453 0 : op, this->desc_state_.write_op, this->desc_state_.write_ready,
454 0 : this->desc_state_.write_cancel_pending);
455 0 : return std::noop_coroutine();
456 : }
457 :
458 : } // namespace boost::corosio::detail
459 :
460 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
|
