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_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
12 :
13 : #include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
14 : #include <boost/corosio/io/io_object.hpp>
15 : #include <boost/corosio/endpoint.hpp>
16 : #include <boost/capy/ex/executor_ref.hpp>
17 :
18 : #include <atomic>
19 : #include <cstddef>
20 : #include <optional>
21 : #include <stop_token>
22 :
23 : #include <errno.h>
24 :
25 : #include <netinet/in.h>
26 : #include <sys/socket.h>
27 : #include <sys/uio.h>
28 :
29 : namespace boost::corosio::detail {
30 :
31 : /** Base operation for reactor-based backends.
32 :
33 : Holds per-operation state that depends on the concrete backend
34 : socket/acceptor types: coroutine handle, executor, output
35 : pointers, file descriptor, stop_callback, and type-specific
36 : impl pointers.
37 :
38 : Fields shared across all backends (errn, bytes_transferred,
39 : cancelled, impl_ptr, perform_io, complete) live in
40 : reactor_op_base so the scheduler and descriptor_state can
41 : access them without template instantiation.
42 :
43 : @tparam Socket The backend socket impl type (forward-declared).
44 : @tparam Acceptor The backend acceptor impl type (forward-declared).
45 : */
46 : template<class Socket, class Acceptor>
47 : struct reactor_op : reactor_op_base
48 : {
49 : // The op envelope — coroutine handle h, cont, executor ex, ec_out,
50 : // bytes_out, cancelled, stop_cb (+ its canceller), impl_ptr — lives in
51 : // coro_op (via reactor_op_base) and is shared with io_uring/IOCP.
52 : // reactor_op adds only the reactor-specific routing state below.
53 :
54 : /// File descriptor this operation targets.
55 : int fd = -1;
56 :
57 : /// Owning socket impl (for stop_token cancellation routing).
58 : Socket* socket_impl_ = nullptr;
59 :
60 : /// Owning acceptor impl (for stop_token cancellation routing).
61 : Acceptor* acceptor_impl_ = nullptr;
62 :
63 HIT 110278 : reactor_op() = default;
64 :
65 : /// Reset operation state for reuse.
66 431498 : void reset() noexcept
67 : {
68 431498 : fd = -1;
69 431498 : errn = 0;
70 431498 : bytes_transferred = 0;
71 431498 : cancelled.store(false, std::memory_order_relaxed);
72 431498 : impl_ptr.reset();
73 431498 : socket_impl_ = nullptr;
74 431498 : acceptor_impl_ = nullptr;
75 431498 : }
76 :
77 : /// Return true if this is a read-direction operation.
78 42119 : virtual bool is_read_operation() const noexcept
79 : {
80 42119 : return false;
81 : }
82 :
83 : /// Cancel this operation via the owning impl.
84 : virtual void cancel() noexcept = 0;
85 :
86 : /// coro_op cancellation hook (fired by the shared canceller when the
87 : /// stop_token requests cancellation): route to the impl-specific cancel().
88 289 : void on_cancel() noexcept override
89 : {
90 289 : cancel();
91 289 : }
92 :
93 : /// Destroy without invoking.
94 12 : void destroy() override
95 : {
96 12 : stop_cb.reset();
97 12 : reactor_op_base::destroy();
98 12 : }
99 :
100 : /// Arm the stop-token callback for a socket operation.
101 90637 : void start(std::stop_token const& token, Socket* impl)
102 : {
103 90637 : socket_impl_ = impl;
104 90637 : acceptor_impl_ = nullptr;
105 90637 : coro_op::start(token);
106 90637 : }
107 :
108 : /// Arm the stop-token callback for an acceptor operation.
109 5821 : void start(std::stop_token const& token, Acceptor* impl)
110 : {
111 5821 : socket_impl_ = nullptr;
112 5821 : acceptor_impl_ = impl;
113 5821 : coro_op::start(token);
114 5821 : }
115 : };
116 :
117 : /** Shared connect operation.
118 :
119 : Checks SO_ERROR for connect completion status. The operator()()
120 : and cancel() are provided by the concrete backend type.
121 :
122 : @tparam Base The backend's base op type.
123 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
124 : */
125 : template<class Base, class Endpoint = endpoint>
126 : struct reactor_connect_op : Base
127 : {
128 : /// Endpoint to connect to.
129 : Endpoint target_endpoint;
130 :
131 : /// Reset operation state for reuse.
132 5825 : void reset() noexcept
133 : {
134 5825 : Base::reset();
135 5825 : target_endpoint = Endpoint{};
136 5825 : }
137 :
138 5760 : void perform_io() noexcept override
139 : {
140 5760 : int err = 0;
141 5760 : socklen_t len = sizeof(err);
142 5760 : if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
143 MIS 0 : err = errno;
144 HIT 5760 : this->complete(err, 0);
145 5760 : }
146 : };
147 :
148 : /** Readiness-only wait operation.
149 :
150 : Does not perform any I/O syscall. Completion is signalled by
151 : the reactor delivering the requested edge event; reactor_descriptor_state
152 : calls complete() directly and never invokes perform_io().
153 :
154 : @tparam Base The backend's base op type.
155 : */
156 : template<class Base>
157 : struct reactor_wait_op : Base
158 : {
159 : /* Mirror of reactor_event_read from reactor_descriptor_state.hpp.
160 : Including that header from here would create an include cycle
161 : (descriptor_state -> reactor_op_base; reactor_op -> reactor_op_base),
162 : so we carry the value locally. Both must stay in sync. */
163 : static constexpr std::uint32_t read_event = 0x001;
164 :
165 : /// Which event bit this wait targets (reactor_event_read/write/error).
166 : std::uint32_t wait_event = 0;
167 :
168 72 : void reset() noexcept
169 : {
170 72 : Base::reset();
171 72 : wait_event = 0;
172 72 : }
173 :
174 MIS 0 : bool is_read_operation() const noexcept override
175 : {
176 0 : return wait_event == read_event;
177 : }
178 :
179 : /* perform_io() should never be called for a wait op — readiness
180 : IS the completion. Overridden here to satisfy the virtual and
181 : produce a safe result if called defensively. */
182 0 : void perform_io() noexcept override
183 : {
184 0 : this->complete(0, 0);
185 0 : }
186 : };
187 :
188 : /** Shared scatter-read operation.
189 :
190 : Uses readv() with an EINTR retry loop.
191 :
192 : @tparam Base The backend's base op type.
193 : */
194 : template<class Base>
195 : struct reactor_read_op : Base
196 : {
197 : /// Maximum scatter-gather buffer count.
198 : static constexpr std::size_t max_buffers = 16;
199 :
200 : /// Scatter-gather I/O vectors.
201 : iovec iovecs[max_buffers];
202 :
203 : /// Number of active I/O vectors.
204 : int iovec_count = 0;
205 :
206 : /// True for zero-length reads (completed immediately).
207 : bool empty_buffer_read = false;
208 :
209 : /// Return true (this is a read-direction operation).
210 HIT 42501 : bool is_read_operation() const noexcept override
211 : {
212 42501 : return !empty_buffer_read;
213 : }
214 :
215 209965 : void reset() noexcept
216 : {
217 209965 : Base::reset();
218 209965 : iovec_count = 0;
219 209965 : empty_buffer_read = false;
220 209965 : }
221 :
222 640 : void perform_io() noexcept override
223 : {
224 : ssize_t n;
225 : do
226 : {
227 640 : n = ::readv(this->fd, iovecs, iovec_count);
228 : }
229 640 : while (n < 0 && errno == EINTR);
230 :
231 640 : if (n >= 0)
232 404 : this->complete(0, static_cast<std::size_t>(n));
233 : else
234 236 : this->complete(errno, 0);
235 640 : }
236 : };
237 :
238 : /** Shared gather-write operation.
239 :
240 : Delegates the actual syscall to WritePolicy::write(fd, iovecs, count),
241 : which returns ssize_t (bytes written or -1 with errno set).
242 :
243 : @tparam Base The backend's base op type.
244 : @tparam WritePolicy Provides `static ssize_t write(int, iovec*, int)`.
245 : */
246 : template<class Base, class WritePolicy>
247 : struct reactor_write_op : Base
248 : {
249 : /// The write syscall policy type.
250 : using write_policy = WritePolicy;
251 :
252 : /// Maximum scatter-gather buffer count.
253 : static constexpr std::size_t max_buffers = 16;
254 :
255 : /// Scatter-gather I/O vectors.
256 : iovec iovecs[max_buffers];
257 :
258 : /// Number of active I/O vectors.
259 : int iovec_count = 0;
260 :
261 209361 : void reset() noexcept
262 : {
263 209361 : Base::reset();
264 209361 : iovec_count = 0;
265 209361 : }
266 :
267 132 : void perform_io() noexcept override
268 : {
269 132 : ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
270 132 : if (n >= 0)
271 129 : this->complete(0, static_cast<std::size_t>(n));
272 : else
273 3 : this->complete(errno, 0);
274 132 : }
275 : };
276 :
277 : /** Shared accept operation.
278 :
279 : Delegates the actual syscall to AcceptPolicy::do_accept(fd, peer_storage),
280 : which returns the accepted fd or -1 with errno set.
281 :
282 : @tparam Base The backend's base op type.
283 : @tparam AcceptPolicy Provides `static int do_accept(int, sockaddr_storage&)`.
284 : */
285 : template<class Base, class AcceptPolicy>
286 : struct reactor_accept_op : Base
287 : {
288 : /// File descriptor of the accepted connection.
289 : int accepted_fd = -1;
290 :
291 : /// Pointer to the peer socket implementation.
292 : io_object::implementation* peer_impl = nullptr;
293 :
294 : /// Output pointer for the accepted implementation.
295 : io_object::implementation** impl_out = nullptr;
296 :
297 : /// Peer address storage filled by accept.
298 : sockaddr_storage peer_storage{};
299 :
300 : /// Peer address length returned by accept.
301 : socklen_t peer_addrlen = 0;
302 :
303 5809 : void reset() noexcept
304 : {
305 5809 : Base::reset();
306 5809 : accepted_fd = -1;
307 5809 : peer_impl = nullptr;
308 5809 : impl_out = nullptr;
309 5809 : peer_storage = {};
310 5809 : peer_addrlen = 0;
311 5809 : }
312 :
313 5748 : void perform_io() noexcept override
314 : {
315 5748 : int new_fd = AcceptPolicy::do_accept(
316 5748 : this->fd, peer_storage, peer_addrlen);
317 5748 : if (new_fd >= 0)
318 : {
319 5748 : accepted_fd = new_fd;
320 5748 : this->complete(0, 0);
321 : }
322 : else
323 : {
324 MIS 0 : this->complete(errno, 0);
325 : }
326 HIT 5748 : }
327 : };
328 :
329 : /** Shared connected send operation for datagram sockets.
330 :
331 : Uses sendmsg() with msg_name=nullptr (connected mode).
332 :
333 : @tparam Base The backend's base op type.
334 : */
335 : template<class Base>
336 : struct reactor_send_op : Base
337 : {
338 : /// Maximum scatter-gather buffer count.
339 : static constexpr std::size_t max_buffers = 16;
340 :
341 : /// Scatter-gather I/O vectors.
342 : iovec iovecs[max_buffers];
343 :
344 : /// Number of active I/O vectors.
345 : int iovec_count = 0;
346 :
347 : /// User-supplied message flags.
348 : int msg_flags = 0;
349 :
350 106 : void reset() noexcept
351 : {
352 106 : Base::reset();
353 106 : iovec_count = 0;
354 106 : msg_flags = 0;
355 106 : }
356 :
357 32 : void perform_io() noexcept override
358 : {
359 32 : msghdr msg{};
360 32 : msg.msg_iov = iovecs;
361 32 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
362 :
363 : #ifdef MSG_NOSIGNAL
364 32 : int send_flags = msg_flags | MSG_NOSIGNAL;
365 : #else
366 : int send_flags = msg_flags;
367 : #endif
368 :
369 : ssize_t n;
370 : do
371 : {
372 32 : n = ::sendmsg(this->fd, &msg, send_flags);
373 : }
374 32 : while (n < 0 && errno == EINTR);
375 :
376 32 : if (n >= 0)
377 30 : this->complete(0, static_cast<std::size_t>(n));
378 : else
379 2 : this->complete(errno, 0);
380 32 : }
381 : };
382 :
383 : /** Shared connected recv operation for datagram sockets.
384 :
385 : Uses recvmsg() with msg_name=nullptr (connected mode).
386 : Unlike reactor_read_op, does not map n==0 to EOF
387 : (zero-length datagrams are valid).
388 :
389 : @tparam Base The backend's base op type.
390 : */
391 : template<class Base>
392 : struct reactor_recv_op : Base
393 : {
394 : /// Maximum scatter-gather buffer count.
395 : static constexpr std::size_t max_buffers = 16;
396 :
397 : /// Scatter-gather I/O vectors.
398 : iovec iovecs[max_buffers];
399 :
400 : /// Number of active I/O vectors.
401 : int iovec_count = 0;
402 :
403 : /// User-supplied message flags.
404 : int msg_flags = 0;
405 :
406 : /// Return true (this is a read-direction operation).
407 MIS 0 : bool is_read_operation() const noexcept override
408 : {
409 0 : return true;
410 : }
411 :
412 HIT 104 : void reset() noexcept
413 : {
414 104 : Base::reset();
415 104 : iovec_count = 0;
416 104 : msg_flags = 0;
417 104 : }
418 :
419 37 : void perform_io() noexcept override
420 : {
421 37 : msghdr msg{};
422 37 : msg.msg_iov = iovecs;
423 37 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
424 :
425 : ssize_t n;
426 : do
427 : {
428 37 : n = ::recvmsg(this->fd, &msg, msg_flags);
429 : }
430 37 : while (n < 0 && errno == EINTR);
431 :
432 37 : if (n >= 0)
433 34 : this->complete(0, static_cast<std::size_t>(n));
434 : else
435 3 : this->complete(errno, 0);
436 37 : }
437 : };
438 :
439 : /** Shared send_to operation for datagram sockets.
440 :
441 : Uses sendmsg() with the destination endpoint in msg_name.
442 :
443 : @tparam Base The backend's base op type.
444 : */
445 : template<class Base>
446 : struct reactor_send_to_op : Base
447 : {
448 : /// Maximum scatter-gather buffer count.
449 : static constexpr std::size_t max_buffers = 16;
450 :
451 : /// Scatter-gather I/O vectors.
452 : iovec iovecs[max_buffers];
453 :
454 : /// Number of active I/O vectors.
455 : int iovec_count = 0;
456 :
457 : /// Destination address storage.
458 : sockaddr_storage dest_storage{};
459 :
460 : /// Destination address length.
461 : socklen_t dest_len = 0;
462 :
463 : /// User-supplied message flags.
464 : int msg_flags = 0;
465 :
466 122 : void reset() noexcept
467 : {
468 122 : Base::reset();
469 122 : iovec_count = 0;
470 122 : dest_storage = {};
471 122 : dest_len = 0;
472 122 : msg_flags = 0;
473 122 : }
474 :
475 32 : void perform_io() noexcept override
476 : {
477 32 : msghdr msg{};
478 32 : msg.msg_name = &dest_storage;
479 32 : msg.msg_namelen = dest_len;
480 32 : msg.msg_iov = iovecs;
481 32 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
482 :
483 : #ifdef MSG_NOSIGNAL
484 32 : int send_flags = msg_flags | MSG_NOSIGNAL;
485 : #else
486 : int send_flags = msg_flags;
487 : #endif
488 :
489 : ssize_t n;
490 : do
491 : {
492 32 : n = ::sendmsg(this->fd, &msg, send_flags);
493 : }
494 32 : while (n < 0 && errno == EINTR);
495 :
496 32 : if (n >= 0)
497 30 : this->complete(0, static_cast<std::size_t>(n));
498 : else
499 2 : this->complete(errno, 0);
500 32 : }
501 : };
502 :
503 : /** Shared recv_from operation for datagram sockets.
504 :
505 : Uses recvmsg() with msg_name to capture the source endpoint.
506 :
507 : @tparam Base The backend's base op type.
508 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
509 : */
510 : template<class Base, class Endpoint = endpoint>
511 : struct reactor_recv_from_op : Base
512 : {
513 : /// Maximum scatter-gather buffer count.
514 : static constexpr std::size_t max_buffers = 16;
515 :
516 : /// Scatter-gather I/O vectors.
517 : iovec iovecs[max_buffers];
518 :
519 : /// Number of active I/O vectors.
520 : int iovec_count = 0;
521 :
522 : /// Source address storage filled by recvmsg.
523 : sockaddr_storage source_storage{};
524 :
525 : /// Actual source address length returned by recvmsg.
526 : socklen_t source_addrlen = 0;
527 :
528 : /// Output pointer for the source endpoint (set by do_recv_from).
529 : Endpoint* source_out = nullptr;
530 :
531 : /// User-supplied message flags.
532 : int msg_flags = 0;
533 :
534 : /// Return true (this is a read-direction operation).
535 MIS 0 : bool is_read_operation() const noexcept override
536 : {
537 0 : return true;
538 : }
539 :
540 HIT 134 : void reset() noexcept
541 : {
542 134 : Base::reset();
543 134 : iovec_count = 0;
544 134 : source_storage = {};
545 134 : source_addrlen = 0;
546 134 : source_out = nullptr;
547 134 : msg_flags = 0;
548 134 : }
549 :
550 39 : void perform_io() noexcept override
551 : {
552 39 : msghdr msg{};
553 39 : msg.msg_name = &source_storage;
554 39 : msg.msg_namelen = sizeof(source_storage);
555 39 : msg.msg_iov = iovecs;
556 39 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
557 :
558 : ssize_t n;
559 : do
560 : {
561 39 : n = ::recvmsg(this->fd, &msg, msg_flags);
562 : }
563 39 : while (n < 0 && errno == EINTR);
564 :
565 39 : if (n >= 0)
566 : {
567 36 : source_addrlen = msg.msg_namelen;
568 36 : this->complete(0, static_cast<std::size_t>(n));
569 : }
570 : else
571 3 : this->complete(errno, 0);
572 39 : }
573 : };
574 :
575 : } // namespace boost::corosio::detail
576 :
577 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
|
