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 102876 : reactor_op() = default;
66 :
67 : /// Reset operation state for reuse.
68 419801 : void reset() noexcept
69 : {
70 419801 : fd = -1;
71 419801 : errn = 0;
72 419801 : bytes_transferred = 0;
73 419801 : cancelled.store(false, std::memory_order_relaxed);
74 419801 : impl_ptr.reset();
75 419801 : socket_impl_ = nullptr;
76 419801 : acceptor_impl_ = nullptr;
77 419801 : }
78 :
79 : /// Return true if this is a read-direction operation.
80 41119 : virtual bool is_read_operation() const noexcept
81 : {
82 41119 : 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 302 : void on_cancel() noexcept override
91 : {
92 302 : cancel();
93 302 : }
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 88136 : void start(std::stop_token const& token, Socket* impl)
104 : {
105 88136 : socket_impl_ = impl;
106 88136 : acceptor_impl_ = nullptr;
107 88136 : coro_op::start(token);
108 88136 : }
109 :
110 : /// Arm the stop-token callback for an acceptor operation.
111 5242 : void start(std::stop_token const& token, Acceptor* impl)
112 : {
113 5242 : socket_impl_ = nullptr;
114 5242 : acceptor_impl_ = impl;
115 5242 : coro_op::start(token);
116 5242 : }
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 5233 : void reset() noexcept
135 : {
136 5233 : Base::reset();
137 5233 : target_endpoint = Endpoint{};
138 5233 : }
139 :
140 5146 : 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 5146 : pollfd pfd{};
149 5146 : pfd.fd = this->fd;
150 5146 : pfd.events = POLLOUT;
151 : int r;
152 : do
153 : {
154 5146 : r = ::poll(&pfd, 1, 0);
155 : }
156 5146 : while (r < 0 && errno == EINTR);
157 :
158 5146 : if (r == 0)
159 : {
160 1 : this->complete(EAGAIN, 0);
161 3 : return;
162 : }
163 5145 : 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 5143 : int err = 0;
174 5143 : socklen_t len = sizeof(err);
175 5143 : if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
176 2 : err = errno;
177 5143 : 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 144 : void reset() noexcept
200 : {
201 144 : Base::reset();
202 144 : wait_event = 0;
203 144 : }
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 268 : 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 268 : if (fd < 0)
230 : {
231 16 : err = EBADF;
232 16 : return true;
233 : }
234 :
235 252 : pollfd pfd{};
236 252 : pfd.fd = fd;
237 252 : if (event == reactor_event_read)
238 151 : pfd.events = POLLIN;
239 101 : else if (event == reactor_event_write)
240 41 : pfd.events = POLLOUT;
241 : else
242 60 : pfd.events = POLLPRI;
243 :
244 : int r;
245 : do
246 : {
247 252 : r = ::poll(&pfd, 1, 0);
248 : }
249 252 : while (r < 0 && errno == EINTR);
250 :
251 252 : 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 250 : return r != 0;
263 : }
264 :
265 107 : void perform_io() noexcept override
266 : {
267 107 : int err = 0;
268 107 : if (probe(this->fd, wait_event, err))
269 25 : this->complete(err, 0);
270 : else
271 82 : this->complete(EAGAIN, 0);
272 107 : }
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 41494 : bool is_read_operation() const noexcept override
298 : {
299 41494 : return !empty_buffer_read;
300 : }
301 :
302 204665 : void reset() noexcept
303 : {
304 204665 : Base::reset();
305 204665 : iovec_count = 0;
306 204665 : empty_buffer_read = false;
307 204665 : }
308 :
309 663 : void perform_io() noexcept override
310 : {
311 : ssize_t n;
312 : do
313 : {
314 663 : n = ::readv(this->fd, iovecs, iovec_count);
315 : }
316 663 : while (n < 0 && errno == EINTR);
317 :
318 663 : if (n >= 0)
319 422 : this->complete(0, static_cast<std::size_t>(n));
320 : else
321 241 : this->complete(errno, 0);
322 663 : }
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 203987 : void reset() noexcept
349 : {
350 203987 : Base::reset();
351 203987 : iovec_count = 0;
352 203987 : }
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 5208 : void reset() noexcept
391 : {
392 5208 : Base::reset();
393 5208 : accepted_fd = -1;
394 5208 : peer_impl = nullptr;
395 5208 : impl_out = nullptr;
396 5208 : peer_storage = {};
397 5208 : peer_addrlen = 0;
398 5208 : }
399 :
400 5109 : void perform_io() noexcept override
401 : {
402 5109 : int new_fd = AcceptPolicy::do_accept(
403 5109 : this->fd, peer_storage, peer_addrlen);
404 5109 : if (new_fd >= 0)
405 : {
406 5108 : accepted_fd = new_fd;
407 5108 : this->complete(0, 0);
408 : }
409 : else
410 : {
411 1 : this->complete(errno, 0);
412 : }
413 5109 : }
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 115 : void reset() noexcept
438 : {
439 115 : Base::reset();
440 115 : iovec_count = 0;
441 115 : msg_flags = 0;
442 115 : }
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 MIS 0 : bool is_read_operation() const noexcept override
495 : {
496 0 : return true;
497 : }
498 :
499 HIT 115 : void reset() noexcept
500 : {
501 115 : Base::reset();
502 115 : iovec_count = 0;
503 115 : msg_flags = 0;
504 115 : }
505 :
506 39 : void perform_io() noexcept override
507 : {
508 39 : msghdr msg{};
509 39 : msg.msg_iov = iovecs;
510 39 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
511 :
512 : ssize_t n;
513 : do
514 : {
515 39 : n = ::recvmsg(this->fd, &msg, msg_flags);
516 : }
517 39 : while (n < 0 && errno == EINTR);
518 :
519 39 : if (n >= 0)
520 34 : this->complete(0, static_cast<std::size_t>(n));
521 : else
522 5 : this->complete(errno, 0);
523 39 : }
524 : };
525 :
526 : /** Shared send_to operation for datagram sockets.
527 :
528 : Uses sendmsg() with the destination endpoint in msg_name.
529 :
530 : @tparam Base The backend's base op type.
531 : */
532 : template<class Base>
533 : struct reactor_send_to_op : Base
534 : {
535 : /// Maximum scatter-gather buffer count.
536 : static constexpr std::size_t max_buffers = 16;
537 :
538 : /// Scatter-gather I/O vectors.
539 : iovec iovecs[max_buffers];
540 :
541 : /// Number of active I/O vectors.
542 : int iovec_count = 0;
543 :
544 : /// Destination address storage.
545 : sockaddr_storage dest_storage{};
546 :
547 : /// Destination address length.
548 : socklen_t dest_len = 0;
549 :
550 : /// User-supplied message flags.
551 : int msg_flags = 0;
552 :
553 157 : void reset() noexcept
554 : {
555 157 : Base::reset();
556 157 : iovec_count = 0;
557 157 : dest_storage = {};
558 157 : dest_len = 0;
559 157 : msg_flags = 0;
560 157 : }
561 :
562 34 : void perform_io() noexcept override
563 : {
564 34 : msghdr msg{};
565 34 : msg.msg_name = &dest_storage;
566 34 : msg.msg_namelen = dest_len;
567 34 : msg.msg_iov = iovecs;
568 34 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
569 :
570 : #ifdef MSG_NOSIGNAL
571 34 : int send_flags = msg_flags | MSG_NOSIGNAL;
572 : #else
573 : int send_flags = msg_flags;
574 : #endif
575 :
576 : ssize_t n;
577 : do
578 : {
579 34 : n = ::sendmsg(this->fd, &msg, send_flags);
580 : }
581 34 : while (n < 0 && errno == EINTR);
582 :
583 34 : if (n >= 0)
584 30 : this->complete(0, static_cast<std::size_t>(n));
585 : else
586 4 : this->complete(errno, 0);
587 34 : }
588 : };
589 :
590 : /** Shared recv_from operation for datagram sockets.
591 :
592 : Uses recvmsg() with msg_name to capture the source endpoint.
593 :
594 : @tparam Base The backend's base op type.
595 : @tparam Endpoint The endpoint type (endpoint or local_endpoint).
596 : */
597 : template<class Base, class Endpoint = endpoint>
598 : struct reactor_recv_from_op : Base
599 : {
600 : /// Maximum scatter-gather buffer count.
601 : static constexpr std::size_t max_buffers = 16;
602 :
603 : /// Scatter-gather I/O vectors.
604 : iovec iovecs[max_buffers];
605 :
606 : /// Number of active I/O vectors.
607 : int iovec_count = 0;
608 :
609 : /// Source address storage filled by recvmsg.
610 : sockaddr_storage source_storage{};
611 :
612 : /// Actual source address length returned by recvmsg.
613 : socklen_t source_addrlen = 0;
614 :
615 : /// Output pointer for the source endpoint (set by do_recv_from).
616 : Endpoint* source_out = nullptr;
617 :
618 : /// User-supplied message flags.
619 : int msg_flags = 0;
620 :
621 : /// Return true (this is a read-direction operation).
622 MIS 0 : bool is_read_operation() const noexcept override
623 : {
624 0 : return true;
625 : }
626 :
627 HIT 177 : void reset() noexcept
628 : {
629 177 : Base::reset();
630 177 : iovec_count = 0;
631 177 : source_storage = {};
632 177 : source_addrlen = 0;
633 177 : source_out = nullptr;
634 177 : msg_flags = 0;
635 177 : }
636 :
637 47 : void perform_io() noexcept override
638 : {
639 47 : msghdr msg{};
640 47 : msg.msg_name = &source_storage;
641 47 : msg.msg_namelen = sizeof(source_storage);
642 47 : msg.msg_iov = iovecs;
643 47 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
644 :
645 : ssize_t n;
646 : do
647 : {
648 47 : n = ::recvmsg(this->fd, &msg, msg_flags);
649 : }
650 47 : while (n < 0 && errno == EINTR);
651 :
652 47 : if (n >= 0)
653 : {
654 42 : source_addrlen = msg.msg_namelen;
655 42 : this->complete(0, static_cast<std::size_t>(n));
656 : }
657 : else
658 5 : this->complete(errno, 0);
659 47 : }
660 : };
661 :
662 : } // namespace boost::corosio::detail
663 :
664 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
|
