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 134928 : reactor_op() = default;
66 :
67 : /// Reset operation state for reuse.
68 689649 : void reset() noexcept
69 : {
70 689649 : fd = -1;
71 689649 : errn = 0;
72 689649 : bytes_transferred = 0;
73 689649 : cancelled.store(false, std::memory_order_relaxed);
74 689649 : impl_ptr.reset();
75 689649 : socket_impl_ = nullptr;
76 689649 : acceptor_impl_ = nullptr;
77 689649 : }
78 :
79 : /// Return true if this is a read-direction operation.
80 67722 : virtual bool is_read_operation() const noexcept
81 : {
82 67722 : 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 311 : void on_cancel() noexcept override
91 : {
92 311 : cancel();
93 311 : }
94 :
95 : /// Destroy without invoking.
96 16 : void destroy() override
97 : {
98 16 : stop_cb.reset();
99 16 : reactor_op_base::destroy();
100 16 : }
101 :
102 : /// Arm the stop-token callback for a socket operation.
103 143167 : void start(std::stop_token const& token, Socket* impl)
104 : {
105 143167 : socket_impl_ = impl;
106 143167 : acceptor_impl_ = nullptr;
107 143167 : coro_op::start(token);
108 143167 : }
109 :
110 : /// Arm the stop-token callback for an acceptor operation.
111 7104 : void start(std::stop_token const& token, Acceptor* impl)
112 : {
113 7104 : socket_impl_ = nullptr;
114 7104 : acceptor_impl_ = impl;
115 7104 : coro_op::start(token);
116 7104 : }
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 7082 : void reset() noexcept
135 : {
136 7082 : Base::reset();
137 7082 : target_endpoint = Endpoint{};
138 7082 : }
139 :
140 7007 : 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 7007 : pollfd pfd{};
149 7007 : pfd.fd = this->fd;
150 7007 : pfd.events = POLLOUT;
151 : int r;
152 : do
153 : {
154 7007 : r = ::poll(&pfd, 1, 0);
155 : }
156 7007 : while (r < 0 && errno == EINTR);
157 :
158 7007 : if (r == 0)
159 : {
160 1 : this->complete(EAGAIN, 0);
161 1 : return;
162 : }
163 7006 : if (r < 0)
164 : {
165 : // EAGAIN must not escape: it is the stay-parked sentinel.
166 MIS 0 : this->complete(
167 0 : (errno == EAGAIN || errno == EWOULDBLOCK) ? ENOMEM
168 0 : : errno,
169 : 0);
170 0 : return;
171 : }
172 :
173 HIT 7006 : int err = 0;
174 7006 : socklen_t len = sizeof(err);
175 7006 : if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
176 MIS 0 : err = errno;
177 HIT 7006 : 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 120 : void reset() noexcept
200 : {
201 120 : Base::reset();
202 120 : wait_event = 0;
203 120 : }
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 224 : 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 224 : if (fd < 0)
230 : {
231 16 : err = EBADF;
232 16 : return true;
233 : }
234 :
235 208 : pollfd pfd{};
236 208 : pfd.fd = fd;
237 208 : if (event == reactor_event_read)
238 126 : pfd.events = POLLIN;
239 82 : else if (event == reactor_event_write)
240 30 : pfd.events = POLLOUT;
241 : else
242 52 : pfd.events = POLLPRI;
243 :
244 : int r;
245 : do
246 : {
247 208 : r = ::poll(&pfd, 1, 0);
248 : }
249 208 : while (r < 0 && errno == EINTR);
250 :
251 208 : 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 MIS 0 : err = (errno == EAGAIN || errno == EWOULDBLOCK)
258 0 : ? ENOMEM
259 0 : : errno;
260 0 : return true;
261 : }
262 HIT 208 : return r != 0;
263 : }
264 :
265 87 : void perform_io() noexcept override
266 : {
267 87 : int err = 0;
268 87 : if (probe(this->fd, wait_event, err))
269 22 : this->complete(err, 0);
270 : else
271 65 : this->complete(EAGAIN, 0);
272 87 : }
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 68120 : bool is_read_operation() const noexcept override
298 : {
299 68120 : return !empty_buffer_read;
300 : }
301 :
302 337754 : void reset() noexcept
303 : {
304 337754 : Base::reset();
305 337754 : iovec_count = 0;
306 337754 : empty_buffer_read = false;
307 337754 : }
308 :
309 672 : void perform_io() noexcept override
310 : {
311 : ssize_t n;
312 : do
313 : {
314 672 : n = ::readv(this->fd, iovecs, iovec_count);
315 : }
316 672 : while (n < 0 && errno == EINTR);
317 :
318 672 : if (n >= 0)
319 423 : this->complete(0, static_cast<std::size_t>(n));
320 : else
321 249 : this->complete(errno, 0);
322 672 : }
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 337096 : void reset() noexcept
349 : {
350 337096 : Base::reset();
351 337096 : iovec_count = 0;
352 337096 : }
353 :
354 132 : void perform_io() noexcept override
355 : {
356 132 : ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
357 132 : if (n >= 0)
358 129 : this->complete(0, static_cast<std::size_t>(n));
359 : else
360 3 : this->complete(errno, 0);
361 132 : }
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 7073 : void reset() noexcept
391 : {
392 7073 : Base::reset();
393 7073 : accepted_fd = -1;
394 7073 : peer_impl = nullptr;
395 7073 : impl_out = nullptr;
396 7073 : peer_storage = {};
397 7073 : peer_addrlen = 0;
398 7073 : }
399 :
400 6996 : void perform_io() noexcept override
401 : {
402 6996 : int new_fd = AcceptPolicy::do_accept(
403 6996 : this->fd, peer_storage, peer_addrlen);
404 6996 : if (new_fd >= 0)
405 : {
406 6996 : accepted_fd = new_fd;
407 6996 : this->complete(0, 0);
408 : }
409 : else
410 : {
411 MIS 0 : this->complete(errno, 0);
412 : }
413 HIT 6996 : }
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 107 : void reset() noexcept
438 : {
439 107 : Base::reset();
440 107 : iovec_count = 0;
441 107 : msg_flags = 0;
442 107 : }
443 :
444 32 : void perform_io() noexcept override
445 : {
446 32 : msghdr msg{};
447 32 : msg.msg_iov = iovecs;
448 32 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
449 :
450 : #ifdef MSG_NOSIGNAL
451 32 : 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 32 : n = ::sendmsg(this->fd, &msg, send_flags);
460 : }
461 32 : while (n < 0 && errno == EINTR);
462 :
463 32 : if (n >= 0)
464 30 : this->complete(0, static_cast<std::size_t>(n));
465 : else
466 2 : this->complete(errno, 0);
467 32 : }
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 109 : void reset() noexcept
500 : {
501 109 : Base::reset();
502 109 : iovec_count = 0;
503 109 : msg_flags = 0;
504 109 : }
505 :
506 37 : void perform_io() noexcept override
507 : {
508 37 : msghdr msg{};
509 37 : msg.msg_iov = iovecs;
510 37 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
511 :
512 : ssize_t n;
513 : do
514 : {
515 37 : n = ::recvmsg(this->fd, &msg, msg_flags);
516 : }
517 37 : while (n < 0 && errno == EINTR);
518 :
519 37 : if (n >= 0)
520 34 : this->complete(0, static_cast<std::size_t>(n));
521 : else
522 3 : this->complete(errno, 0);
523 37 : }
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 145 : void reset() noexcept
554 : {
555 145 : Base::reset();
556 145 : iovec_count = 0;
557 145 : dest_storage = {};
558 145 : dest_len = 0;
559 145 : msg_flags = 0;
560 145 : }
561 :
562 32 : void perform_io() noexcept override
563 : {
564 32 : msghdr msg{};
565 32 : msg.msg_name = &dest_storage;
566 32 : msg.msg_namelen = dest_len;
567 32 : msg.msg_iov = iovecs;
568 32 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
569 :
570 : #ifdef MSG_NOSIGNAL
571 32 : 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 32 : n = ::sendmsg(this->fd, &msg, send_flags);
580 : }
581 32 : while (n < 0 && errno == EINTR);
582 :
583 32 : if (n >= 0)
584 30 : this->complete(0, static_cast<std::size_t>(n));
585 : else
586 2 : this->complete(errno, 0);
587 32 : }
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 163 : void reset() noexcept
628 : {
629 163 : Base::reset();
630 163 : iovec_count = 0;
631 163 : source_storage = {};
632 163 : source_addrlen = 0;
633 163 : source_out = nullptr;
634 163 : msg_flags = 0;
635 163 : }
636 :
637 44 : void perform_io() noexcept override
638 : {
639 44 : msghdr msg{};
640 44 : msg.msg_name = &source_storage;
641 44 : msg.msg_namelen = sizeof(source_storage);
642 44 : msg.msg_iov = iovecs;
643 44 : msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
644 :
645 : ssize_t n;
646 : do
647 : {
648 44 : n = ::recvmsg(this->fd, &msg, msg_flags);
649 : }
650 44 : while (n < 0 && errno == EINTR);
651 :
652 44 : if (n >= 0)
653 : {
654 41 : source_addrlen = msg.msg_namelen;
655 41 : this->complete(0, static_cast<std::size_t>(n));
656 : }
657 : else
658 3 : this->complete(errno, 0);
659 44 : }
660 : };
661 :
662 : } // namespace boost::corosio::detail
663 :
664 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
|
