Index/include / boost / corosio / native / detail / reactor / reactor_descriptor_state.hpp

include/boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp

82.3% Lines (107/130) 100.0% List of functions (6/6) 72.9% Branches (51/70)
[ ] prev/next file n p prev/next uncovered line
reactor_descriptor_state.hpp
f(x) Functions (6)
Function Calls Lines Branches Blocks
boost::corosio::detail::reactor_descriptor_state::reactor_descriptor_state() :49 39082x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::~reactor_descriptor_state() :49 39082x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::add_ready_events(unsigned int) :119 1255056x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::operator()() :125 1254794x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::destroy() :133 137x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::invoke_deferred_io() :148 1254794x 77.0% 72.9% 78.0%
Line Branch TLA Hits 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_DESCRIPTOR_STATE_HPP
11 #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
12
13 #include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
14 #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
15
16 #include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
17
18 #include <atomic>
19 #include <cstdint>
20 #include <memory>
21
22 #include <errno.h>
23 #include <sys/socket.h>
24
25 namespace boost::corosio::detail {
26
27 /// Shared reactor event constants.
28 /// These match epoll numeric values; kqueue maps its events to the same.
29 static constexpr std::uint32_t reactor_event_read = 0x001;
30 static constexpr std::uint32_t reactor_event_write = 0x004;
31 static constexpr std::uint32_t reactor_event_error = 0x008;
32
33 /** Per-descriptor state shared across reactor backends.
34
35 Tracks pending operations for a file descriptor. The fd is registered
36 once with the reactor and stays registered until closed. Uses deferred
37 I/O: the reactor sets ready_events atomically, then enqueues this state.
38 When popped by the scheduler, invoke_deferred_io() performs I/O under
39 the mutex and queues completed ops.
40
41 Non-template: uses reactor_op_base pointers so the scheduler and
42 descriptor_state code exist as a single copy in the binary regardless
43 of how many backends are compiled in.
44
45 @par Thread Safety
46 The mutex protects operation pointers and ready flags. ready_events_
47 and is_enqueued_ are atomic for lock-free reactor access.
48 */
49 19541x struct reactor_descriptor_state : scheduler_op
50 {
51 /// Protects operation pointers and ready/cancel flags.
52 /// Becomes a no-op in single-threaded mode.
53 19541x conditionally_enabled_mutex mutex{true};
54
55 /// Pending read operation (guarded by `mutex`).
56 19541x reactor_op_base* read_op = nullptr;
57
58 /// Pending write operation (guarded by `mutex`).
59 19541x reactor_op_base* write_op = nullptr;
60
61 /// Pending connect operation (guarded by `mutex`).
62 19541x reactor_op_base* connect_op = nullptr;
63
64 /// Pending wait-for-read operation (guarded by `mutex`).
65 19541x reactor_op_base* wait_read_op = nullptr;
66
67 /// Pending wait-for-write operation (guarded by `mutex`).
68 19541x reactor_op_base* wait_write_op = nullptr;
69
70 /// Pending wait-for-error operation (guarded by `mutex`).
71 19541x reactor_op_base* wait_error_op = nullptr;
72
73 /// True if a read edge event arrived before an op was registered.
74 19541x bool read_ready = false;
75
76 /// True if a write edge event arrived before an op was registered.
77 19541x bool write_ready = false;
78
79 /// Deferred read cancellation (IOCP-style cancel semantics).
80 19541x bool read_cancel_pending = false;
81
82 /// Deferred write cancellation (IOCP-style cancel semantics).
83 19541x bool write_cancel_pending = false;
84
85 /// Deferred connect cancellation (IOCP-style cancel semantics).
86 19541x bool connect_cancel_pending = false;
87
88 /// Deferred wait-read cancellation (IOCP-style cancel semantics).
89 19541x bool wait_read_cancel_pending = false;
90
91 /// Deferred wait-write cancellation (IOCP-style cancel semantics).
92 19541x bool wait_write_cancel_pending = false;
93
94 /// Deferred wait-error cancellation (IOCP-style cancel semantics).
95 19541x bool wait_error_cancel_pending = false;
96
97 /// Event mask set during registration (no mutex needed).
98 19541x std::uint32_t registered_events = 0;
99
100 /// File descriptor this state tracks.
101 19541x int fd = -1;
102
103 /// Accumulated ready events (set by reactor, read by scheduler).
104 19541x std::atomic<std::uint32_t> ready_events_{0};
105
106 /// True while this state is queued in the scheduler's completed_ops.
107 19541x std::atomic<bool> is_enqueued_{false};
108
109 /// Owning scheduler for posting completions.
110 19541x reactor_scheduler const* scheduler_ = nullptr;
111
112 /// Prevents impl destruction while queued in the scheduler.
113 std::shared_ptr<void> impl_ref_;
114
115 /// Add ready events atomically.
116 /// Release pairs with the consumer's acquire exchange on
117 /// ready_events_ so the consumer sees all flags. On x86 (TSO)
118 /// this compiles to the same LOCK OR as relaxed.
119 1255056x void add_ready_events(std::uint32_t ev) noexcept
120 {
121 1255056x ready_events_.fetch_or(ev, std::memory_order_release);
122 1255056x }
123
124 /// Invoke deferred I/O and dispatch completions.
125 1254794x void operator()() override
126 {
127 1254794x invoke_deferred_io();
128 1254794x }
129
130 /// Destroy without invoking.
131 /// Called during scheduler::shutdown() drain. Clear impl_ref_ to break
132 /// the self-referential cycle set by close_socket().
133 137x void destroy() override
134 {
135 137x impl_ref_.reset();
136 137x }
137
138 /** Perform deferred I/O and queue completions.
139
140 Performs I/O under the mutex and queues completed ops. EAGAIN
141 ops stay parked in their slot for re-delivery on the next
142 edge event.
143 */
144 void invoke_deferred_io();
145 };
146
147 inline void
148 1254794x reactor_descriptor_state::invoke_deferred_io()
149 {
150 1254794x std::shared_ptr<void> prevent_impl_destruction;
151 1254794x op_queue local_ops;
152
153 {
154
1/2
✓ Branch 0 taken 1254794 times.
✗ Branch 1 not taken.
 1254794x conditionally_enabled_mutex::scoped_lock lock(mutex);
155
156 // Must clear is_enqueued_ and move impl_ref_ under the same
157 // lock that processes I/O. close_socket() checks is_enqueued_
158 // under this mutex — without atomicity between the flag store
159 // and the ref move, close_socket() could see is_enqueued_==false,
160 // skip setting impl_ref_, and destroy the impl under us.
161 1254794x prevent_impl_destruction = std::move(impl_ref_);
162 1254794x is_enqueued_.store(false, std::memory_order_release);
163
164 1254794x std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
165
1/2
✓ Branch 0 taken 1254794 times.
✗ Branch 1 not taken.
 1254794x if (ev == 0)
166 {
167 // Mutex unlocks here; compensate for work_cleanup's decrement
168 ✗ scheduler_->compensating_work_started();
169 ✗ return;
170 }
171
172 1254794x int err = 0;
173
2/2
✓ Branch 0 taken 1254784 times.
✓ Branch 1 taken 10 times.
 1254794x if (ev & reactor_event_error)
174 {
175 10x socklen_t len = sizeof(err);
176
2/4
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 10 times.
 10x if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
177 ✗ err = errno;
178
2/2
✓ Branch 0 taken 1 time.
✓ Branch 1 taken 9 times.
 10x if (err == 0)
179 1x err = EIO;
180 10x }
181
182
2/2
✓ Branch 0 taken 58552 times.
✓ Branch 1 taken 1196242 times.
 1254794x if (ev & reactor_event_read)
183 {
184
2/2
✓ Branch 0 taken 79840 times.
✓ Branch 1 taken 1116402 times.
 1196242x if (read_op)
185 {
186 79840x auto* rd = read_op;
187
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 79840 times.
 79840x if (err)
188 ✗ rd->complete(err, 0);
189 else
190 79840x rd->perform_io();
191
192
3/4
✓ Branch 0 taken 78463 times.
✓ Branch 1 taken 1377 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 78463 times.
 79840x if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
193 {
194 1377x rd->errn = 0;
195 1377x }
196 else
197 {
198 78463x read_op = nullptr;
199 78463x local_ops.push(rd);
200 }
201 79840x }
202 else
203 {
204 1116402x read_ready = true;
205 }
206
207 // Complete any parked wait-for-read regardless of read_op presence.
208
2/2
✓ Branch 0 taken 1196226 times.
✓ Branch 1 taken 16 times.
 1196242x if (wait_read_op)
209 {
210 16x wait_read_op->complete(err, 0);
211 16x local_ops.push(std::exchange(wait_read_op, nullptr));
212 16x }
213 1196242x }
214
2/2
✓ Branch 0 taken 58677 times.
✓ Branch 1 taken 1196117 times.
 1254794x if (ev & reactor_event_write)
215 {
216
2/2
✓ Branch 0 taken 5939 times.
✓ Branch 1 taken 52738 times.
 58677x bool had_write_op = (connect_op || write_op);
217
2/2
✓ Branch 0 taken 52738 times.
✓ Branch 1 taken 5939 times.
 58677x if (connect_op)
218 {
219 5939x auto* cn = connect_op;
220
2/2
✓ Branch 0 taken 9 times.
✓ Branch 1 taken 5930 times.
 5939x if (err)
221 9x cn->complete(err, 0);
222 else
223 5930x cn->perform_io();
224 5939x connect_op = nullptr;
225 5939x local_ops.push(cn);
226 5939x }
227
2/2
✓ Branch 0 taken 58169 times.
✓ Branch 1 taken 508 times.
 58677x if (write_op)
228 {
229 508x auto* wr = write_op;
230
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 508 times.
 508x if (err)
231 ✗ wr->complete(err, 0);
232 else
233 508x wr->perform_io();
234
235
3/4
✓ Branch 0 taken 507 times.
✓ Branch 1 taken 1 time.
✗ Branch 2 not taken.
✓ Branch 3 taken 507 times.
 508x if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
236 {
237 1x wr->errn = 0;
238 1x }
239 else
240 {
241 507x write_op = nullptr;
242 507x local_ops.push(wr);
243 }
244 508x }
245
2/2
✓ Branch 0 taken 6447 times.
✓ Branch 1 taken 52230 times.
 58677x if (!had_write_op)
246 52230x write_ready = true;
247
248 // Complete any parked wait-for-write regardless of write_op presence.
249
1/2
✓ Branch 0 taken 58677 times.
✗ Branch 1 not taken.
 58677x if (wait_write_op)
250 {
251 ✗ wait_write_op->complete(err, 0);
252 ✗ local_ops.push(std::exchange(wait_write_op, nullptr));
253 ✗ }
254 58677x }
255 // Complete a parked wait-for-error on any error condition.
256
3/4
✓ Branch 0 taken 1254784 times.
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 1254784 times.
 1254794x if ((ev & reactor_event_error) || err)
257 {
258
2/2
✓ Branch 0 taken 9 times.
✓ Branch 1 taken 1 time.
 10x if (wait_error_op)
259 {
260 1x wait_error_op->complete(err, 0);
261 1x local_ops.push(std::exchange(wait_error_op, nullptr));
262 1x }
263 10x }
264
2/2
✓ Branch 0 taken 1254784 times.
✓ Branch 1 taken 10 times.
 1254794x if (err)
265 {
266
1/2
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
 10x if (read_op)
267 {
268 ✗ read_op->complete(err, 0);
269 ✗ local_ops.push(std::exchange(read_op, nullptr));
270 ✗ }
271
1/2
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
 10x if (write_op)
272 {
273 ✗ write_op->complete(err, 0);
274 ✗ local_ops.push(std::exchange(write_op, nullptr));
275 ✗ }
276
1/2
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
 10x if (connect_op)
277 {
278 ✗ connect_op->complete(err, 0);
279 ✗ local_ops.push(std::exchange(connect_op, nullptr));
280 ✗ }
281
1/2
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
 10x if (wait_read_op)
282 {
283 ✗ wait_read_op->complete(err, 0);
284 ✗ local_ops.push(std::exchange(wait_read_op, nullptr));
285 ✗ }
286
1/2
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
 10x if (wait_write_op)
287 {
288 ✗ wait_write_op->complete(err, 0);
289 ✗ local_ops.push(std::exchange(wait_write_op, nullptr));
290 ✗ }
291 10x }
292 1254794x }
293
294 // Execute first handler inline — the scheduler's work_cleanup
295 // accounts for this as the "consumed" work item
296 1254794x scheduler_op* first = local_ops.pop();
297
2/2
✓ Branch 0 taken 84925 times.
✓ Branch 1 taken 1169869 times.
 1254794x if (first)
298 {
299
1/2
✓ Branch 0 taken 84925 times.
✗ Branch 1 not taken.
 84925x scheduler_->post_deferred_completions(local_ops);
300
1/2
✓ Branch 0 taken 84925 times.
✗ Branch 1 not taken.
 84925x (*first)();
301 84925x }
302 else
303 {
304 1169869x scheduler_->compensating_work_started();
305 }
306 1254794x }
307
308 } // namespace boost::corosio::detail
309
310 #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
311