. /include/boost/corosio /native /detail /coro_op.hpp
100.00% Lines (23/23) 100.00% Functions (6/6)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2026 Michael Vandeberg 2   // Copyright (c) 2026 Michael Vandeberg
3   // 3   //
4   // Distributed under the Boost Software License, Version 1.0. (See accompanying 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) 5   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6   // 6   //
7   // Official repository: https://github.com/cppalliance/corosio 7   // Official repository: https://github.com/cppalliance/corosio
8   // 8   //
9   9  
10   #ifndef BOOST_COROSIO_NATIVE_DETAIL_CORO_OP_HPP 10   #ifndef BOOST_COROSIO_NATIVE_DETAIL_CORO_OP_HPP
11   #define BOOST_COROSIO_NATIVE_DETAIL_CORO_OP_HPP 11   #define BOOST_COROSIO_NATIVE_DETAIL_CORO_OP_HPP
12   12  
13   #include <boost/corosio/detail/config.hpp> 13   #include <boost/corosio/detail/config.hpp>
14   #include <boost/capy/continuation.hpp> 14   #include <boost/capy/continuation.hpp>
15   #include <boost/corosio/detail/object_ref.hpp> 15   #include <boost/corosio/detail/object_ref.hpp>
16   #include <boost/corosio/detail/scheduler_op.hpp> 16   #include <boost/corosio/detail/scheduler_op.hpp>
17   #include <boost/capy/ex/abandon.hpp> 17   #include <boost/capy/ex/abandon.hpp>
18   #include <boost/capy/ex/executor_ref.hpp> 18   #include <boost/capy/ex/executor_ref.hpp>
19   19  
20   #include <atomic> 20   #include <atomic>
21   #include <coroutine> 21   #include <coroutine>
22   #include <cstddef> 22   #include <cstddef>
23   #include <memory> 23   #include <memory>
24   #include <optional> 24   #include <optional>
25   #include <stop_token> 25   #include <stop_token>
26   #include <system_error> 26   #include <system_error>
27   #include <utility> 27   #include <utility>
28   28  
29   /* 29   /*
30   Shared, non-template op envelope for every native backend — the readiness 30   Shared, non-template op envelope for every native backend — the readiness
31   reactors (epoll/kqueue/select), io_uring, and IOCP. It captures the part of 31   reactors (epoll/kqueue/select), io_uring, and IOCP. It captures the part of
32   an async operation that is identical regardless of how completion is 32   an async operation that is identical regardless of how completion is
33   reported: the coroutine to resume, the executor it dispatches on, the 33   reported: the coroutine to resume, the executor it dispatches on, the
34   output pointers, the stop_token wiring, the cancelled flag, and the 34   output pointers, the stop_token wiring, the cancelled flag, and the
35   keepalive that holds the owning impl alive while the op is in flight. 35   keepalive that holds the owning impl alive while the op is in flight.
36   36  
37   What is deliberately NOT here (it differs by backend and stays in the 37   What is deliberately NOT here (it differs by backend and stays in the
38   derived op layer): 38   derived op layer):
39   - the result model: the reactors re-run the syscall and record 39   - the result model: the reactors re-run the syscall and record
40   `errn`/`bytes_transferred` (reactor_op_base); io_uring stores the raw 40   `errn`/`bytes_transferred` (reactor_op_base); io_uring stores the raw
41   `res`/`cqe_flags`; IOCP stores `dwError`/`bytes_transferred`. Each 41   `res`/`cqe_flags`; IOCP stores `dwError`/`bytes_transferred`. Each
42   decodes its own result. 42   decodes its own result.
43   - the submission + the kernel cancel action. Cancellation is unified only 43   - the submission + the kernel cancel action. Cancellation is unified only
44   at the call site via the virtual `on_cancel()` hook: the stop_callback 44   at the call site via the virtual `on_cancel()` hook: the stop_callback
45   always targets `coro_op`, and each backend overrides `on_cancel()` — 45   always targets `coro_op`, and each backend overrides `on_cancel()` —
46   the reactors route to the owning impl's cancel(), io_uring submits an 46   the reactors route to the owning impl's cancel(), io_uring submits an
47   ASYNC_CANCEL SQE, IOCP calls the stored cancel_func_/CancelIoEx. 47   ASYNC_CANCEL SQE, IOCP calls the stored cancel_func_/CancelIoEx.
48   48  
49   See tasks/proactor-dedup-decisions.md and coro-op-unification-scope.md. 49   See tasks/proactor-dedup-decisions.md and coro-op-unification-scope.md.
50   */ 50   */
51   51  
52   namespace boost::corosio::detail { 52   namespace boost::corosio::detail {
53   53  
54   /** Non-template op envelope shared by every native backend's operations. 54   /** Non-template op envelope shared by every native backend's operations.
55   55  
56   `reactor_op_base`, `uring_op`, and `overlapped_op` all derive from this. 56   `reactor_op_base`, `uring_op`, and `overlapped_op` all derive from this.
57   Derives from scheduler_op so ops queue intrusively and dispatch through the 57   Derives from scheduler_op so ops queue intrusively and dispatch through the
58   function-pointer (io_uring/IOCP) or virtual (reactors) completion path — 58   function-pointer (io_uring/IOCP) or virtual (reactors) completion path —
59   hence both a default and a func_type constructor. 59   hence both a default and a func_type constructor.
60   60  
61   @note For IOCP, the concrete op multiply-inherits `OVERLAPPED` as its 61   @note For IOCP, the concrete op multiply-inherits `OVERLAPPED` as its
62   first base (so `static_cast<OVERLAPPED*>` round-trips); `coro_op` 62   first base (so `static_cast<OVERLAPPED*>` round-trips); `coro_op`
63   follows it. 63   follows it.
64   */ 64   */
65   struct coro_op : scheduler_op 65   struct coro_op : scheduler_op
66   { 66   {
67   /** Stop-callback handler: routes a stop_token firing to `on_cancel()`. 67   /** Stop-callback handler: routes a stop_token firing to `on_cancel()`.
68   68  
69   A single canceller type for both backends keeps `stop_cb` (and thus 69   A single canceller type for both backends keeps `stop_cb` (and thus
70   `start()`) in this shared base; the backend-specific action lives 70   `start()`) in this shared base; the backend-specific action lives
71   behind the `on_cancel()` virtual. 71   behind the `on_cancel()` virtual.
72   */ 72   */
73   struct canceller 73   struct canceller
74   { 74   {
75   coro_op* op; 75   coro_op* op;
HITCBC 76   801 void operator()() const noexcept 76   880 void operator()() const noexcept
77   { 77   {
HITCBC 78   801 op->on_cancel(); 78   880 op->on_cancel();
HITCBC 79   801 } 79   880 }
80   }; 80   };
81   81  
82   /// The awaiting coroutine's continuation. It lives in the awaitable, 82   /// The awaiting coroutine's continuation. It lives in the awaitable,
83   /// which stays in the awaiting frame until resumption, so a 83   /// which stays in the awaiting frame until resumption, so a
84   /// completion posted through an executor never shares this op's 84   /// completion posted through an executor never shares this op's
85   /// storage, which is reused by the next operation. 85   /// storage, which is reused by the next operation.
86   capy::continuation* cont = nullptr; 86   capy::continuation* cont = nullptr;
87   capy::executor_ref ex; 87   capy::executor_ref ex;
88   std::error_code* ec_out = nullptr; 88   std::error_code* ec_out = nullptr;
89   std::size_t* bytes_out = nullptr; 89   std::size_t* bytes_out = nullptr;
90   90  
91   /// True for receive/read ops (drives the zero-byte == EOF decision). 91   /// True for receive/read ops (drives the zero-byte == EOF decision).
92   bool is_read = false; 92   bool is_read = false;
93   /// True when the submitted buffer was zero-length (suppresses EOF). 93   /// True when the submitted buffer was zero-length (suppresses EOF).
94   bool empty_buffer = false; 94   bool empty_buffer = false;
95   95  
96   std::atomic<bool> cancelled{false}; 96   std::atomic<bool> cancelled{false};
97   std::optional<std::stop_callback<canceller>> stop_cb; 97   std::optional<std::stop_callback<canceller>> stop_cb;
98   98  
99   /// Keeps the owning impl alive while the op is in flight (the kernel 99   /// Keeps the owning impl alive while the op is in flight (the kernel
100   /// owns user buffers until completion). Released when the op resumes or 100   /// owns user buffers until completion). Released when the op resumes or
101   /// abandons its coroutine (see coro_op_complete.hpp). 101   /// abandons its coroutine (see coro_op_complete.hpp).
102   detail::object_ref object_ref_; 102   detail::object_ref object_ref_;
103   103  
104   /// Default-construct for virtual-dispatch backends (the reactors, which 104   /// Default-construct for virtual-dispatch backends (the reactors, which
105   /// override operator()/destroy() and leave func_ null). 105   /// override operator()/destroy() and leave func_ null).
HITCBC 106   49489 coro_op() noexcept = default; 106   49405 coro_op() noexcept = default;
107   107  
108   /// Construct with the completion function for func-pointer dispatch 108   /// Construct with the completion function for func-pointer dispatch
109   /// (io_uring / IOCP completion handlers). 109   /// (io_uring / IOCP completion handlers).
110   explicit coro_op(func_type func) noexcept : scheduler_op(func) {} 110   explicit coro_op(func_type func) noexcept : scheduler_op(func) {}
111   111  
112   /** Arm the stop-token callback. Call before the op is submitted. 112   /** Arm the stop-token callback. Call before the op is submitted.
113   113  
114   Resets the cancellation flag and (re)arms `stop_cb` against @a token. 114   Resets the cancellation flag and (re)arms `stop_cb` against @a token.
115   Derived ops that carry extra pre-submit state (e.g. io_uring's 115   Derived ops that carry extra pre-submit state (e.g. io_uring's
116   `sqe_set`) extend this. 116   `sqe_set`) extend this.
117   */ 117   */
HITCBC 118   109934 void start(std::stop_token const& token) 118   114494 void start(std::stop_token const& token)
119   { 119   {
HITCBC 120   109934 cancelled.store(false, std::memory_order_relaxed); 120   114494 cancelled.store(false, std::memory_order_relaxed);
HITCBC 121   109934 stop_cb.reset(); 121   114494 stop_cb.reset();
HITCBC 122   109934 if (token.stop_possible()) 122   114494 if (token.stop_possible())
HITCBC 123   3528 stop_cb.emplace(token, canceller{this}); 123   3780 stop_cb.emplace(token, canceller{this});
HITCBC 124   109934 } 124   114494 }
125   125  
126   /** Give up the awaiting coroutine without resuming it. 126   /** Give up the awaiting coroutine without resuming it.
127   127  
128   Used when the operation is drained at teardown: the operation 128   Used when the operation is drained at teardown: the operation
129   holds the suspended coroutine, so it abandons it, which unwinds 129   holds the suspended coroutine, so it abandons it, which unwinds
130   its chain from the chain's owner. The keepalive is held across 130   its chain from the chain's owner. The keepalive is held across
131   the call because the unwind may run at once and destroy the 131   the call because the unwind may run at once and destroy the
132   object whose impl stores this op; nothing here touches the op 132   object whose impl stores this op; nothing here touches the op
133   afterwards. 133   afterwards.
134   */ 134   */
HITCBC 135   1300 void destroy_awaiting() noexcept 135   1300 void destroy_awaiting() noexcept
136   { 136   {
HITCBC 137   1300 stop_cb.reset(); 137   1300 stop_cb.reset();
HITCBC 138   1300 auto keep = std::move(object_ref_); 138   1300 auto keep = std::move(object_ref_);
HITCBC 139   1300 if (auto* c = std::exchange(cont, nullptr)) 139   1300 if (auto* c = std::exchange(cont, nullptr))
140   { 140   {
HITCBC 141   1300 if (c->h) 141   1300 if (c->h)
HITCBC 142   1300 capy::abandon(*c); 142   1300 capy::abandon(*c);
143   } 143   }
HITCBC 144   1300 } 144   1300 }
145   145  
146   /// Mark this op cancellation-requested. Shared by every backend. 146   /// Mark this op cancellation-requested. Shared by every backend.
HITCBC 147   565410 void request_cancel() noexcept 147   559989 void request_cancel() noexcept
148   { 148   {
HITCBC 149   565410 cancelled.store(true, std::memory_order_release); 149   559989 cancelled.store(true, std::memory_order_release);
HITCBC 150   565410 } 150   559989 }
151   151  
152   /** Backend cancellation hook, invoked when the stop_token fires. 152   /** Backend cancellation hook, invoked when the stop_token fires.
153   153  
154   The default just records the request. Backends override to also 154   The default just records the request. Backends override to also
155   drive the kernel: io_uring submits an ASYNC_CANCEL SQE; IOCP calls 155   drive the kernel: io_uring submits an ASYNC_CANCEL SQE; IOCP calls
156   its stored cancel_func_ (CancelIoEx / wait-reactor deregister). 156   its stored cancel_func_ (CancelIoEx / wait-reactor deregister).
157   */ 157   */
HITCBC 158   204 virtual void on_cancel() noexcept 158   204 virtual void on_cancel() noexcept
159   { 159   {
HITCBC 160   204 request_cancel(); 160   204 request_cancel();
HITCBC 161   204 } 161   204 }
162   }; 162   };
163   163  
164   } // namespace boost::corosio::detail 164   } // namespace boost::corosio::detail
165   165  
166   #endif 166   #endif