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_events.hpp>
14 : #include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
15 : #include <boost/corosio/io/io_object.hpp>
16 : #include <boost/corosio/endpoint.hpp>
17 : #include <boost/capy/ex/executor_ref.hpp>
18 :
19 : #include <atomic>
20 : #include <cstddef>
21 : #include <optional>
22 : #include <stop_token>
23 :
24 : #include <errno.h>
25 : #include <poll.h>
26 :
27 : #include <netinet/in.h>
28 : #include <sys/socket.h>
29 : #include <sys/uio.h>
30 : #include <unistd.h>
31 :
32 : namespace boost::corosio::detail {
33 :
34 : /** Base operation for reactor-based backends.
35 :
36 : Holds per-operation state that depends on the concrete backend
37 : socket/acceptor types: coroutine handle, executor, output
38 : pointers, file descriptor, stop_callback, and type-specific
39 : impl pointers.
40 :
41 : Fields shared across all backends (errn, bytes_transferred,
42 : cancelled, object_ref_, perform_io, complete) live in
43 : reactor_op_base so the scheduler and descriptor_state can
44 : access them without template instantiation.
45 :
46 : @tparam Socket The backend socket impl type (forward-declared).
47 : @tparam Acceptor The backend acceptor impl type (forward-declared).
48 : */
49 : template<class Socket, class Acceptor>
50 : struct reactor_op : reactor_op_base
51 : {
52 : // The op envelope — coroutine handle h, cont, executor ex, ec_out,
53 : // bytes_out, cancelled, stop_cb (+ its canceller), object_ref_ — lives in
54 : // coro_op (via reactor_op_base) and is shared with io_uring/IOCP.
55 : // reactor_op adds only the reactor-specific routing state below.
56 :
57 : /// File descriptor this operation targets.
58 : int fd = -1;
59 :
60 : /// Owning socket impl (for stop_token cancellation routing).
61 : Socket* socket_impl_ = nullptr;
62 :
63 : /// Owning acceptor impl (for stop_token cancellation routing).
64 : Acceptor* acceptor_impl_ = nullptr;
65 :
66 HIT 43012 : reactor_op() = default;
67 :
68 : /// Reset operation state for reuse.
69 490564 : void reset() noexcept
70 : {
71 490564 : fd = -1;
72 490564 : errn = 0;
73 490564 : bytes_transferred = 0;
74 490564 : cancelled.store(false, std::memory_order_relaxed);
75 490564 : object_ref_.reset();
76 490564 : socket_impl_ = nullptr;
77 490564 : acceptor_impl_ = nullptr;
78 490564 : }
79 :
80 : /// Return true if this is a read-direction operation.
81 48016 : virtual bool is_read_operation() const noexcept
82 : {
83 48016 : return false;
84 : }
85 :
86 : /// Cancel this operation via the owning impl.
87 : virtual void cancel() noexcept = 0;
88 :
89 : /// coro_op cancellation hook (fired by the shared canceller when the
90 : /// stop_token requests cancellation): route to the impl-specific cancel().
91 676 : void on_cancel() noexcept override
92 : {
93 676 : cancel();
94 676 : }
95 :
96 : /// Destroy without invoking.
97 1290 : void destroy() override
98 : {
99 1290 : reactor_op_base::destroy();
100 1290 : }
101 :
102 : /// Arm the stop-token callback for a socket operation.
103 105715 : void start(std::stop_token const& token, Socket* impl)
104 : {
105 105715 : socket_impl_ = impl;
106 105715 : acceptor_impl_ = nullptr;
107 105715 : coro_op::start(token);
108 105715 : }
109 :
110 : /// Arm the stop-token callback for an acceptor operation.
111 7531 : void start(std::stop_token const& token, Acceptor* impl)
112 : {
113 7531 : socket_impl_ = nullptr;
114 7531 : acceptor_impl_ = impl;
115 7531 : coro_op::start(token);
116 7531 : }
117 : };
118 :
119 : /** Shared connect operation.
120 :
121 : Checks SO_ERROR for connect completion status. The operator()()
122 : and cancel() are provided by the concrete backend type.
123 :
124 : @tparam Base The backend's base op type.
125 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
126 : */
127 : template<class Base, class Endpoint = endpoint>
128 : struct reactor_connect_op : Base
129 : {
130 : /// Endpoint to connect to.
131 : Endpoint target_endpoint;
132 :
133 : /// Reset operation state for reuse.
134 8229 : void reset() noexcept
135 : {
136 8229 : Base::reset();
137 8229 : target_endpoint = Endpoint{};
138 8229 : }
139 :
140 8101 : void perform_io() noexcept override
141 : {
142 : // A readiness notification does not prove the handshake
143 : // finished: fresh sockets raise a spurious writable event,
144 : // and a cached edge can trigger this check while the connect
145 : // is still in flight — where SO_ERROR also reads 0. Probe
146 : // writability first and report EAGAIN to stay parked;
147 : // SO_ERROR decides only once the socket is actually writable.
148 8101 : pollfd pfd{};
149 8101 : pfd.fd = this->fd;
150 8101 : pfd.events = POLLOUT;
151 : int r;
152 : do
153 : {
154 8101 : r = ::poll(&pfd, 1, 0);
155 : }
156 8101 : while (r < 0 && errno == EINTR);
157 :
158 8101 : if (r == 0)
159 : {
160 1 : this->complete(EAGAIN, 0);
161 3 : return;
162 : }
163 8100 : if (r < 0)
164 : {
165 : // EAGAIN must not escape: it is the stay-parked sentinel.
166 2 : this->complete(
167 2 : (errno == EAGAIN || errno == EWOULDBLOCK) ? ENOMEM : errno, 0);
168 2 : return;
169 : }
170 :
171 8098 : int err = 0;
172 8098 : socklen_t len = sizeof(err);
173 8098 : if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
174 2 : err = errno;
175 8098 : this->complete(err, 0);
176 : }
177 : };
178 :
179 : /** Readiness-only wait operation.
180 :
181 : Completion is decided by probing the descriptor with a
182 : zero-timeout `poll()`, never by the reactor's cached edge
183 : events: a speculative read can drain the socket without
184 : touching the reactor (stale edge), and a short read can
185 : consume the edge while data remains buffered (missing edge).
186 : `perform_io()` runs the probe and reports `EAGAIN` when the
187 : condition does not currently hold, which keeps the op parked.
188 :
189 : @tparam Base The backend's base op type.
190 : */
191 : template<class Base>
192 : struct reactor_wait_op : Base
193 : {
194 : /// Which event bit this wait targets (reactor_event_read/write/error).
195 : std::uint32_t wait_event = 0;
196 :
197 206 : void reset() noexcept
198 : {
199 206 : Base::reset();
200 206 : wait_event = 0;
201 206 : }
202 :
203 MIS 0 : bool is_read_operation() const noexcept override
204 : {
205 0 : return wait_event == reactor_event_read;
206 : }
207 :
208 : /** Check whether the waited-for condition currently holds.
209 :
210 : Zero-timeout `poll()` probe. `POLLERR`/`POLLHUP` count as
211 : ready for every wait type: the wait must not park on a
212 : socket whose next I/O would fail immediately. The probe is
213 : side-effect free — in particular it never reads `SO_ERROR`,
214 : which is consume-on-read and belongs to whichever operation
215 : observes the failure next.
216 :
217 : @param fd The descriptor to probe.
218 : @param event The event bit to probe for (read/write/error).
219 : @param err Receives the probe failure, if any.
220 :
221 : @return `true` if the condition holds or the probe failed.
222 : */
223 HIT 371 : static bool probe(int fd, std::uint32_t event, int& err) noexcept
224 : {
225 : // poll() silently ignores negative fds; without this guard a
226 : // wait on a never-opened or closed socket parks forever.
227 371 : if (fd < 0)
228 : {
229 18 : err = EBADF;
230 18 : return true;
231 : }
232 :
233 353 : pollfd pfd{};
234 353 : pfd.fd = fd;
235 353 : if (event == reactor_event_read)
236 228 : pfd.events = POLLIN;
237 125 : else if (event == reactor_event_write)
238 65 : pfd.events = POLLOUT;
239 : else
240 60 : pfd.events = POLLPRI;
241 :
242 : int r;
243 : do
244 : {
245 353 : r = ::poll(&pfd, 1, 0);
246 : }
247 353 : while (r < 0 && errno == EINTR);
248 :
249 353 : if (r < 0)
250 : {
251 : // Complete with the probe failure rather than park forever.
252 : // EAGAIN must not escape here: callers treat it as the
253 : // stay-parked sentinel, and poll() can fail with it on
254 : // BSD/macOS under transient resource pressure.
255 2 : err = (errno == EAGAIN || errno == EWOULDBLOCK) ? ENOMEM : errno;
256 2 : return true;
257 : }
258 351 : return r != 0;
259 : }
260 :
261 123 : void perform_io() noexcept override
262 : {
263 123 : int err = 0;
264 123 : if (probe(this->fd, wait_event, err))
265 37 : this->complete(err, 0);
266 : else
267 86 : this->complete(EAGAIN, 0);
268 123 : }
269 : };
270 :
271 : /** Shared scatter-read operation.
272 :
273 : Uses readv() with an EINTR retry loop.
274 :
275 : @tparam Base The backend's base op type.
276 : */
277 : template<class Base>
278 : struct reactor_read_op : Base
279 : {
280 : /// Maximum scatter-gather buffer count.
281 : static constexpr std::size_t max_buffers = 16;
282 :
283 : /// Scatter-gather I/O vectors.
284 : iovec iovecs[max_buffers];
285 :
286 : /// Number of active I/O vectors.
287 : int iovec_count = 0;
288 :
289 : /// True for zero-length reads (completed immediately).
290 : bool empty_buffer_read = false;
291 :
292 : /// Return true (this is a read-direction operation).
293 48651 : bool is_read_operation() const noexcept override
294 : {
295 48651 : return !empty_buffer_read;
296 : }
297 :
298 237619 : void reset() noexcept
299 : {
300 237619 : Base::reset();
301 237619 : iovec_count = 0;
302 237619 : empty_buffer_read = false;
303 237619 : }
304 :
305 1003 : void perform_io() noexcept override
306 : {
307 : ssize_t n;
308 : do
309 : {
310 1003 : n = ::readv(this->fd, iovecs, iovec_count);
311 : }
312 1003 : while (n < 0 && errno == EINTR);
313 :
314 1003 : if (n >= 0)
315 893 : this->complete(0, static_cast<std::size_t>(n));
316 : else
317 110 : this->complete(errno, 0);
318 1003 : }
319 : };
320 :
321 : /** Shared gather-write operation.
322 :
323 : Delegates the actual syscall to WritePolicy::write(fd, iovecs, count),
324 : which returns ssize_t (bytes written or -1 with errno set).
325 :
326 : @tparam Base The backend's base op type.
327 : @tparam WritePolicy Provides `static ssize_t write(int, iovec*, int)`.
328 : */
329 : template<class Base, class WritePolicy>
330 : struct reactor_write_op : Base
331 : {
332 : /// The write syscall policy type.
333 : using write_policy = WritePolicy;
334 :
335 : /// Maximum scatter-gather buffer count.
336 : static constexpr std::size_t max_buffers = 16;
337 :
338 : /// Scatter-gather I/O vectors.
339 : iovec iovecs[max_buffers];
340 :
341 : /// Number of active I/O vectors.
342 : int iovec_count = 0;
343 :
344 236378 : void reset() noexcept
345 : {
346 236378 : Base::reset();
347 236378 : iovec_count = 0;
348 236378 : }
349 :
350 140 : void perform_io() noexcept override
351 : {
352 140 : ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
353 140 : if (n >= 0)
354 132 : this->complete(0, static_cast<std::size_t>(n));
355 : else
356 8 : this->complete(errno, 0);
357 140 : }
358 : };
359 :
360 : /** Shared accept operation.
361 :
362 : Delegates the actual syscall to AcceptPolicy::do_accept(fd, peer_storage),
363 : which returns the accepted fd or -1 with errno set.
364 :
365 : @tparam Base The backend's base op type.
366 : @tparam AcceptPolicy Provides `static int do_accept(int, sockaddr_storage&)`.
367 : */
368 : template<class Base, class AcceptPolicy>
369 : struct reactor_accept_op : Base
370 : {
371 : /// File descriptor of the accepted connection.
372 : int accepted_fd = -1;
373 :
374 : /// Pointer to the peer socket implementation.
375 : io_object::implementation* peer_impl = nullptr;
376 :
377 : /// Output pointer for the accepted implementation.
378 : io_object::implementation** impl_out = nullptr;
379 :
380 : /// Peer address storage filled by accept.
381 : sockaddr_storage peer_storage{};
382 :
383 : /// Peer address length returned by accept.
384 : socklen_t peer_addrlen = 0;
385 :
386 7490 : void reset() noexcept
387 : {
388 7490 : Base::reset();
389 7490 : accepted_fd = -1;
390 7490 : peer_impl = nullptr;
391 7490 : impl_out = nullptr;
392 7490 : peer_storage = {};
393 7490 : peer_addrlen = 0;
394 7490 : }
395 :
396 : /// Destroy without invoking, closing a connection already taken.
397 816 : void destroy() override
398 : {
399 816 : if (accepted_fd >= 0)
400 : {
401 5 : ::close(accepted_fd);
402 5 : accepted_fd = -1;
403 : }
404 816 : Base::destroy();
405 816 : }
406 :
407 5768 : void perform_io() noexcept override
408 : {
409 : int new_fd =
410 5768 : AcceptPolicy::do_accept(this->fd, peer_storage, peer_addrlen);
411 5768 : if (new_fd >= 0)
412 : {
413 5765 : accepted_fd = new_fd;
414 5765 : this->complete(0, 0);
415 : }
416 : else
417 : {
418 3 : this->complete(errno, 0);
419 : }
420 5768 : }
421 : };
422 :
423 : /** Shared connected send operation for datagram sockets.
424 :
425 : Uses sendmsg() with msg_name=nullptr (connected mode).
426 :
427 : @tparam Base The backend's base op type.
428 : */
429 : template<class Base>
430 : struct reactor_send_op : Base
431 : {
432 : /// Maximum scatter-gather buffer count.
433 : static constexpr std::size_t max_buffers = 16;
434 :
435 : /// Scatter-gather I/O vectors.
436 : iovec iovecs[max_buffers];
437 :
438 : /// Number of active I/O vectors.
439 : int iovec_count = 0;
440 :
441 : /// User-supplied message flags.
442 : int msg_flags = 0;
443 :
444 123 : void reset() noexcept
445 : {
446 123 : Base::reset();
447 123 : iovec_count = 0;
448 123 : msg_flags = 0;
449 123 : }
450 :
451 34 : void perform_io() noexcept override
452 : {
453 34 : msghdr msg{};
454 34 : msg.msg_iov = iovecs;
455 34 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
456 :
457 : #ifdef MSG_NOSIGNAL
458 34 : int send_flags = msg_flags | MSG_NOSIGNAL;
459 : #else
460 : int send_flags = msg_flags;
461 : #endif
462 :
463 : ssize_t n;
464 : do
465 : {
466 34 : n = ::sendmsg(this->fd, &msg, send_flags);
467 : }
468 34 : while (n < 0 && errno == EINTR);
469 :
470 34 : if (n >= 0)
471 30 : this->complete(0, static_cast<std::size_t>(n));
472 : else
473 4 : this->complete(errno, 0);
474 34 : }
475 : };
476 :
477 : /** Shared connected recv operation for datagram sockets.
478 :
479 : Uses recvmsg() with msg_name=nullptr (connected mode).
480 : Unlike reactor_read_op, does not map n==0 to EOF
481 : (zero-length datagrams are valid).
482 :
483 : @tparam Base The backend's base op type.
484 : */
485 : template<class Base>
486 : struct reactor_recv_op : Base
487 : {
488 : /// Maximum scatter-gather buffer count.
489 : static constexpr std::size_t max_buffers = 16;
490 :
491 : /// Scatter-gather I/O vectors.
492 : iovec iovecs[max_buffers];
493 :
494 : /// Number of active I/O vectors.
495 : int iovec_count = 0;
496 :
497 : /// User-supplied message flags.
498 : int msg_flags = 0;
499 :
500 : /// Return true (this is a read-direction operation).
501 : // LCOV_EXCL_START: devirtualized and inlined at the templated
502 : // completion call site; the out-of-line body is never entered.
503 : bool is_read_operation() const noexcept override
504 : {
505 : return true;
506 : }
507 : // LCOV_EXCL_STOP
508 :
509 165 : void reset() noexcept
510 : {
511 165 : Base::reset();
512 165 : iovec_count = 0;
513 165 : msg_flags = 0;
514 165 : }
515 :
516 39 : void perform_io() noexcept override
517 : {
518 39 : msghdr msg{};
519 39 : msg.msg_iov = iovecs;
520 39 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
521 :
522 : ssize_t n;
523 : do
524 : {
525 39 : n = ::recvmsg(this->fd, &msg, msg_flags);
526 : }
527 39 : while (n < 0 && errno == EINTR);
528 :
529 39 : if (n >= 0)
530 36 : this->complete(0, static_cast<std::size_t>(n));
531 : else
532 3 : this->complete(errno, 0);
533 39 : }
534 : };
535 :
536 : /** Shared send_to operation for datagram sockets.
537 :
538 : Uses sendmsg() with the destination endpoint in msg_name.
539 :
540 : @tparam Base The backend's base op type.
541 : */
542 : template<class Base>
543 : struct reactor_send_to_op : Base
544 : {
545 : /// Maximum scatter-gather buffer count.
546 : static constexpr std::size_t max_buffers = 16;
547 :
548 : /// Scatter-gather I/O vectors.
549 : iovec iovecs[max_buffers];
550 :
551 : /// Number of active I/O vectors.
552 : int iovec_count = 0;
553 :
554 : /// Destination address storage.
555 : sockaddr_storage dest_storage{};
556 :
557 : /// Destination address length.
558 : socklen_t dest_len = 0;
559 :
560 : /// User-supplied message flags.
561 : int msg_flags = 0;
562 :
563 167 : void reset() noexcept
564 : {
565 167 : Base::reset();
566 167 : iovec_count = 0;
567 167 : dest_storage = {};
568 167 : dest_len = 0;
569 167 : msg_flags = 0;
570 167 : }
571 :
572 34 : void perform_io() noexcept override
573 : {
574 34 : msghdr msg{};
575 34 : msg.msg_name = &dest_storage;
576 34 : msg.msg_namelen = dest_len;
577 34 : msg.msg_iov = iovecs;
578 34 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
579 :
580 : #ifdef MSG_NOSIGNAL
581 34 : int send_flags = msg_flags | MSG_NOSIGNAL;
582 : #else
583 : int send_flags = msg_flags;
584 : #endif
585 :
586 : ssize_t n;
587 : do
588 : {
589 34 : n = ::sendmsg(this->fd, &msg, send_flags);
590 : }
591 34 : while (n < 0 && errno == EINTR);
592 :
593 34 : if (n >= 0)
594 30 : this->complete(0, static_cast<std::size_t>(n));
595 : else
596 4 : this->complete(errno, 0);
597 34 : }
598 : };
599 :
600 : /** Shared recv_from operation for datagram sockets.
601 :
602 : Uses recvmsg() with msg_name to capture the source endpoint.
603 :
604 : @tparam Base The backend's base op type.
605 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
606 : */
607 : template<class Base, class Endpoint = endpoint>
608 : struct reactor_recv_from_op : Base
609 : {
610 : /// Maximum scatter-gather buffer count.
611 : static constexpr std::size_t max_buffers = 16;
612 :
613 : /// Scatter-gather I/O vectors.
614 : iovec iovecs[max_buffers];
615 :
616 : /// Number of active I/O vectors.
617 : int iovec_count = 0;
618 :
619 : /// Source address storage filled by recvmsg.
620 : sockaddr_storage source_storage{};
621 :
622 : /// Actual source address length returned by recvmsg.
623 : socklen_t source_addrlen = 0;
624 :
625 : /// Output pointer for the source endpoint (set by do_recv_from).
626 : Endpoint* source_out = nullptr;
627 :
628 : /// User-supplied message flags.
629 : int msg_flags = 0;
630 :
631 : /// Return true (this is a read-direction operation).
632 : // LCOV_EXCL_START: devirtualized and inlined at the templated
633 : // completion call site; the out-of-line body is never entered.
634 : bool is_read_operation() const noexcept override
635 : {
636 : return true;
637 : }
638 : // LCOV_EXCL_STOP
639 :
640 187 : void reset() noexcept
641 : {
642 187 : Base::reset();
643 187 : iovec_count = 0;
644 187 : source_storage = {};
645 187 : source_addrlen = 0;
646 187 : source_out = nullptr;
647 187 : msg_flags = 0;
648 187 : }
649 :
650 51 : void perform_io() noexcept override
651 : {
652 51 : msghdr msg{};
653 51 : msg.msg_name = &source_storage;
654 51 : msg.msg_namelen = sizeof(source_storage);
655 51 : msg.msg_iov = iovecs;
656 51 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
657 :
658 : ssize_t n;
659 : do
660 : {
661 51 : n = ::recvmsg(this->fd, &msg, msg_flags);
662 : }
663 51 : while (n < 0 && errno == EINTR);
664 :
665 51 : if (n >= 0)
666 : {
667 48 : source_addrlen = msg.msg_namelen;
668 48 : this->complete(0, static_cast<std::size_t>(n));
669 : }
670 : else
671 3 : this->complete(errno, 0);
672 51 : }
673 : };
674 :
675 : } // namespace boost::corosio::detail
676 :
677 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
|
