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