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 92752 : reactor_op() = default;
66 :
67 : /// Reset operation state for reuse.
68 415512 : void reset() noexcept
69 : {
70 415512 : fd = -1;
71 415512 : errn = 0;
72 415512 : bytes_transferred = 0;
73 415512 : cancelled.store(false, std::memory_order_relaxed);
74 415512 : impl_ptr.reset();
75 415512 : socket_impl_ = nullptr;
76 415512 : acceptor_impl_ = nullptr;
77 415512 : }
78 :
79 : /// Return true if this is a read-direction operation.
80 40819 : virtual bool is_read_operation() const noexcept
81 : {
82 40819 : 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 387 : void on_cancel() noexcept override
91 : {
92 387 : cancel();
93 387 : }
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 87008 : void start(std::stop_token const& token, Socket* impl)
104 : {
105 87008 : socket_impl_ = impl;
106 87008 : acceptor_impl_ = nullptr;
107 87008 : coro_op::start(token);
108 87008 : }
109 :
110 : /// Arm the stop-token callback for an acceptor operation.
111 4682 : void start(std::stop_token const& token, Acceptor* impl)
112 : {
113 4682 : socket_impl_ = nullptr;
114 4682 : acceptor_impl_ = impl;
115 4682 : coro_op::start(token);
116 4682 : }
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 4615 : void reset() noexcept
135 : {
136 4615 : Base::reset();
137 4615 : target_endpoint = Endpoint{};
138 4615 : }
139 :
140 4513 : 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 4513 : pollfd pfd{};
149 4513 : pfd.fd = this->fd;
150 4513 : pfd.events = POLLOUT;
151 : int r;
152 : do
153 : {
154 4513 : r = ::poll(&pfd, 1, 0);
155 : }
156 4513 : while (r < 0 && errno == EINTR);
157 :
158 4513 : if (r == 0)
159 : {
160 1 : this->complete(EAGAIN, 0);
161 3 : return;
162 : }
163 4512 : 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 4510 : int err = 0;
172 4510 : socklen_t len = sizeof(err);
173 4510 : if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
174 2 : err = errno;
175 4510 : 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 180 : void reset() noexcept
198 : {
199 180 : Base::reset();
200 180 : wait_event = 0;
201 180 : }
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 330 : 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 330 : if (fd < 0)
228 : {
229 16 : err = EBADF;
230 16 : return true;
231 : }
232 :
233 314 : pollfd pfd{};
234 314 : pfd.fd = fd;
235 314 : if (event == reactor_event_read)
236 209 : pfd.events = POLLIN;
237 105 : else if (event == reactor_event_write)
238 45 : pfd.events = POLLOUT;
239 : else
240 60 : pfd.events = POLLPRI;
241 :
242 : int r;
243 : do
244 : {
245 314 : r = ::poll(&pfd, 1, 0);
246 : }
247 314 : while (r < 0 && errno == EINTR);
248 :
249 314 : 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 312 : return r != 0;
259 : }
260 :
261 127 : void perform_io() noexcept override
262 : {
263 127 : int err = 0;
264 127 : if (probe(this->fd, wait_event, err))
265 27 : this->complete(err, 0);
266 : else
267 100 : this->complete(EAGAIN, 0);
268 127 : }
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 41199 : bool is_read_operation() const noexcept override
294 : {
295 41199 : return !empty_buffer_read;
296 : }
297 :
298 203065 : void reset() noexcept
299 : {
300 203065 : Base::reset();
301 203065 : iovec_count = 0;
302 203065 : empty_buffer_read = false;
303 203065 : }
304 :
305 666 : void perform_io() noexcept override
306 : {
307 : ssize_t n;
308 : do
309 : {
310 666 : n = ::readv(this->fd, iovecs, iovec_count);
311 : }
312 666 : while (n < 0 && errno == EINTR);
313 :
314 666 : if (n >= 0)
315 424 : this->complete(0, static_cast<std::size_t>(n));
316 : else
317 242 : this->complete(errno, 0);
318 666 : }
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 202371 : void reset() noexcept
345 : {
346 202371 : Base::reset();
347 202371 : iovec_count = 0;
348 202371 : }
349 :
350 135 : void perform_io() noexcept override
351 : {
352 135 : ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
353 135 : if (n >= 0)
354 130 : this->complete(0, static_cast<std::size_t>(n));
355 : else
356 5 : this->complete(errno, 0);
357 135 : }
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 4638 : void reset() noexcept
387 : {
388 4638 : Base::reset();
389 4638 : accepted_fd = -1;
390 4638 : peer_impl = nullptr;
391 4638 : impl_out = nullptr;
392 4638 : peer_storage = {};
393 4638 : peer_addrlen = 0;
394 4638 : }
395 :
396 4478 : void perform_io() noexcept override
397 : {
398 : int new_fd =
399 4478 : AcceptPolicy::do_accept(this->fd, peer_storage, peer_addrlen);
400 4478 : if (new_fd >= 0)
401 : {
402 4476 : accepted_fd = new_fd;
403 4476 : this->complete(0, 0);
404 : }
405 : else
406 : {
407 2 : this->complete(errno, 0);
408 : }
409 4478 : }
410 : };
411 :
412 : /** Shared connected send operation for datagram sockets.
413 :
414 : Uses sendmsg() with msg_name=nullptr (connected mode).
415 :
416 : @tparam Base The backend's base op type.
417 : */
418 : template<class Base>
419 : struct reactor_send_op : Base
420 : {
421 : /// Maximum scatter-gather buffer count.
422 : static constexpr std::size_t max_buffers = 16;
423 :
424 : /// Scatter-gather I/O vectors.
425 : iovec iovecs[max_buffers];
426 :
427 : /// Number of active I/O vectors.
428 : int iovec_count = 0;
429 :
430 : /// User-supplied message flags.
431 : int msg_flags = 0;
432 :
433 125 : void reset() noexcept
434 : {
435 125 : Base::reset();
436 125 : iovec_count = 0;
437 125 : msg_flags = 0;
438 125 : }
439 :
440 34 : void perform_io() noexcept override
441 : {
442 34 : msghdr msg{};
443 34 : msg.msg_iov = iovecs;
444 34 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
445 :
446 : #ifdef MSG_NOSIGNAL
447 34 : int send_flags = msg_flags | MSG_NOSIGNAL;
448 : #else
449 : int send_flags = msg_flags;
450 : #endif
451 :
452 : ssize_t n;
453 : do
454 : {
455 34 : n = ::sendmsg(this->fd, &msg, send_flags);
456 : }
457 34 : while (n < 0 && errno == EINTR);
458 :
459 34 : if (n >= 0)
460 30 : this->complete(0, static_cast<std::size_t>(n));
461 : else
462 4 : this->complete(errno, 0);
463 34 : }
464 : };
465 :
466 : /** Shared connected recv operation for datagram sockets.
467 :
468 : Uses recvmsg() with msg_name=nullptr (connected mode).
469 : Unlike reactor_read_op, does not map n==0 to EOF
470 : (zero-length datagrams are valid).
471 :
472 : @tparam Base The backend's base op type.
473 : */
474 : template<class Base>
475 : struct reactor_recv_op : Base
476 : {
477 : /// Maximum scatter-gather buffer count.
478 : static constexpr std::size_t max_buffers = 16;
479 :
480 : /// Scatter-gather I/O vectors.
481 : iovec iovecs[max_buffers];
482 :
483 : /// Number of active I/O vectors.
484 : int iovec_count = 0;
485 :
486 : /// User-supplied message flags.
487 : int msg_flags = 0;
488 :
489 : /// Return true (this is a read-direction operation).
490 : // LCOV_EXCL_START: devirtualized and inlined at the templated
491 : // completion call site; the out-of-line body is never entered.
492 : bool is_read_operation() const noexcept override
493 : {
494 : return true;
495 : }
496 : // LCOV_EXCL_STOP
497 :
498 162 : void reset() noexcept
499 : {
500 162 : Base::reset();
501 162 : iovec_count = 0;
502 162 : msg_flags = 0;
503 162 : }
504 :
505 39 : void perform_io() noexcept override
506 : {
507 39 : msghdr msg{};
508 39 : msg.msg_iov = iovecs;
509 39 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
510 :
511 : ssize_t n;
512 : do
513 : {
514 39 : n = ::recvmsg(this->fd, &msg, msg_flags);
515 : }
516 39 : while (n < 0 && errno == EINTR);
517 :
518 39 : if (n >= 0)
519 34 : this->complete(0, static_cast<std::size_t>(n));
520 : else
521 5 : this->complete(errno, 0);
522 39 : }
523 : };
524 :
525 : /** Shared send_to operation for datagram sockets.
526 :
527 : Uses sendmsg() with the destination endpoint in msg_name.
528 :
529 : @tparam Base The backend's base op type.
530 : */
531 : template<class Base>
532 : struct reactor_send_to_op : Base
533 : {
534 : /// Maximum scatter-gather buffer count.
535 : static constexpr std::size_t max_buffers = 16;
536 :
537 : /// Scatter-gather I/O vectors.
538 : iovec iovecs[max_buffers];
539 :
540 : /// Number of active I/O vectors.
541 : int iovec_count = 0;
542 :
543 : /// Destination address storage.
544 : sockaddr_storage dest_storage{};
545 :
546 : /// Destination address length.
547 : socklen_t dest_len = 0;
548 :
549 : /// User-supplied message flags.
550 : int msg_flags = 0;
551 :
552 167 : void reset() noexcept
553 : {
554 167 : Base::reset();
555 167 : iovec_count = 0;
556 167 : dest_storage = {};
557 167 : dest_len = 0;
558 167 : msg_flags = 0;
559 167 : }
560 :
561 34 : void perform_io() noexcept override
562 : {
563 34 : msghdr msg{};
564 34 : msg.msg_name = &dest_storage;
565 34 : msg.msg_namelen = dest_len;
566 34 : msg.msg_iov = iovecs;
567 34 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
568 :
569 : #ifdef MSG_NOSIGNAL
570 34 : int send_flags = msg_flags | MSG_NOSIGNAL;
571 : #else
572 : int send_flags = msg_flags;
573 : #endif
574 :
575 : ssize_t n;
576 : do
577 : {
578 34 : n = ::sendmsg(this->fd, &msg, send_flags);
579 : }
580 34 : while (n < 0 && errno == EINTR);
581 :
582 34 : if (n >= 0)
583 30 : this->complete(0, static_cast<std::size_t>(n));
584 : else
585 4 : this->complete(errno, 0);
586 34 : }
587 : };
588 :
589 : /** Shared recv_from operation for datagram sockets.
590 :
591 : Uses recvmsg() with msg_name to capture the source endpoint.
592 :
593 : @tparam Base The backend's base op type.
594 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
595 : */
596 : template<class Base, class Endpoint = endpoint>
597 : struct reactor_recv_from_op : Base
598 : {
599 : /// Maximum scatter-gather buffer count.
600 : static constexpr std::size_t max_buffers = 16;
601 :
602 : /// Scatter-gather I/O vectors.
603 : iovec iovecs[max_buffers];
604 :
605 : /// Number of active I/O vectors.
606 : int iovec_count = 0;
607 :
608 : /// Source address storage filled by recvmsg.
609 : sockaddr_storage source_storage{};
610 :
611 : /// Actual source address length returned by recvmsg.
612 : socklen_t source_addrlen = 0;
613 :
614 : /// Output pointer for the source endpoint (set by do_recv_from).
615 : Endpoint* source_out = nullptr;
616 :
617 : /// User-supplied message flags.
618 : int msg_flags = 0;
619 :
620 : /// Return true (this is a read-direction operation).
621 : // LCOV_EXCL_START: devirtualized and inlined at the templated
622 : // completion call site; the out-of-line body is never entered.
623 : bool is_read_operation() const noexcept override
624 : {
625 : return true;
626 : }
627 : // LCOV_EXCL_STOP
628 :
629 189 : void reset() noexcept
630 : {
631 189 : Base::reset();
632 189 : iovec_count = 0;
633 189 : source_storage = {};
634 189 : source_addrlen = 0;
635 189 : source_out = nullptr;
636 189 : msg_flags = 0;
637 189 : }
638 :
639 49 : void perform_io() noexcept override
640 : {
641 49 : msghdr msg{};
642 49 : msg.msg_name = &source_storage;
643 49 : msg.msg_namelen = sizeof(source_storage);
644 49 : msg.msg_iov = iovecs;
645 49 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
646 :
647 : ssize_t n;
648 : do
649 : {
650 49 : n = ::recvmsg(this->fd, &msg, msg_flags);
651 : }
652 49 : while (n < 0 && errno == EINTR);
653 :
654 49 : if (n >= 0)
655 : {
656 44 : source_addrlen = msg.msg_namelen;
657 44 : this->complete(0, static_cast<std::size_t>(n));
658 : }
659 : else
660 5 : this->complete(errno, 0);
661 49 : }
662 : };
663 :
664 : } // namespace boost::corosio::detail
665 :
666 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
|
