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

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

81.6% Lines (120/147) 100.0% List of functions (6/6) 70.9% Branches (61/86)
[ ] 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() :45 55700x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::~reactor_descriptor_state() :45 55700x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::add_ready_events(unsigned int) :115 1710644x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::operator()() :121 1710090x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::destroy() :129 335x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::invoke_deferred_io() :144 1710090x 76.9% 70.9% 79.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_events.hpp>
14 #include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
15 #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
16 #include <boost/corosio/detail/ready_queue.hpp>
17
18 #include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
19
20 #include <atomic>
21 #include <cstdint>
22 #include <memory>
23
24 #include <errno.h>
25 #include <sys/socket.h>
26
27 namespace boost::corosio::detail {
28
29 /** Per-descriptor state shared across reactor backends.
30
31 Tracks pending operations for a file descriptor. The fd is registered
32 once with the reactor and stays registered until closed. Uses deferred
33 I/O: the reactor sets ready_events atomically, then enqueues this state.
34 When popped by the scheduler, invoke_deferred_io() performs I/O under
35 the mutex and queues completed ops.
36
37 Non-template: uses reactor_op_base pointers so the scheduler and
38 descriptor_state code exist as a single copy in the binary regardless
39 of how many backends are compiled in.
40
41 @par Thread Safety
42 The mutex protects operation pointers and ready flags. ready_events_
43 and is_enqueued_ are atomic for lock-free reactor access.
44 */
45 27850x struct reactor_descriptor_state : scheduler_op
46 {
47 /// Protects operation pointers and ready/cancel flags.
48 /// Becomes a no-op in single-threaded mode.
49 27850x conditionally_enabled_mutex mutex{true};
50
51 /// Pending read operation (guarded by `mutex`).
52 27850x reactor_op_base* read_op = nullptr;
53
54 /// Pending write operation (guarded by `mutex`).
55 27850x reactor_op_base* write_op = nullptr;
56
57 /// Pending connect operation (guarded by `mutex`).
58 27850x reactor_op_base* connect_op = nullptr;
59
60 /// Pending wait-for-read operation (guarded by `mutex`).
61 27850x reactor_op_base* wait_read_op = nullptr;
62
63 /// Pending wait-for-write operation (guarded by `mutex`).
64 27850x reactor_op_base* wait_write_op = nullptr;
65
66 /// Pending wait-for-error operation (guarded by `mutex`).
67 27850x reactor_op_base* wait_error_op = nullptr;
68
69 /// True if a read edge event arrived before an op was registered.
70 27850x bool read_ready = false;
71
72 /// True if a write edge event arrived before an op was registered.
73 27850x bool write_ready = false;
74
75 /// Deferred read cancellation (IOCP-style cancel semantics).
76 27850x bool read_cancel_pending = false;
77
78 /// Deferred write cancellation (IOCP-style cancel semantics).
79 27850x bool write_cancel_pending = false;
80
81 /// Deferred connect cancellation (IOCP-style cancel semantics).
82 27850x bool connect_cancel_pending = false;
83
84 /// Deferred wait-read cancellation (IOCP-style cancel semantics).
85 27850x bool wait_read_cancel_pending = false;
86
87 /// Deferred wait-write cancellation (IOCP-style cancel semantics).
88 27850x bool wait_write_cancel_pending = false;
89
90 /// Deferred wait-error cancellation (IOCP-style cancel semantics).
91 27850x bool wait_error_cancel_pending = false;
92
93 /// Event mask set during registration (no mutex needed).
94 27850x std::uint32_t registered_events = 0;
95
96 /// File descriptor this state tracks.
97 27850x int fd = -1;
98
99 /// Accumulated ready events (set by reactor, read by scheduler).
100 27850x std::atomic<std::uint32_t> ready_events_{0};
101
102 /// True while this state is queued in the scheduler's completed_ops.
103 27850x std::atomic<bool> is_enqueued_{false};
104
105 /// Owning scheduler for posting completions.
106 27850x reactor_scheduler const* scheduler_ = nullptr;
107
108 /// Prevents impl destruction while queued in the scheduler.
109 std::shared_ptr<void> impl_ref_;
110
111 /// Add ready events atomically.
112 /// Release pairs with the consumer's acquire exchange on
113 /// ready_events_ so the consumer sees all flags. On x86 (TSO)
114 /// this compiles to the same LOCK OR as relaxed.
115 1710644x void add_ready_events(std::uint32_t ev) noexcept
116 {
117 1710644x ready_events_.fetch_or(ev, std::memory_order_release);
118 1710644x }
119
120 /// Invoke deferred I/O and dispatch completions.
121 1710090x void operator()() override
122 {
123 1710090x invoke_deferred_io();
124 1710090x }
125
126 /// Destroy without invoking.
127 /// Called during scheduler::shutdown() drain. Clear impl_ref_ to break
128 /// the self-referential cycle set by close_socket().
129 335x void destroy() override
130 {
131 335x impl_ref_.reset();
132 335x }
133
134 /** Perform deferred I/O and queue completions.
135
136 Performs I/O under the mutex and queues completed ops. EAGAIN
137 ops stay parked in their slot for re-delivery on the next
138 edge event.
139 */
140 void invoke_deferred_io();
141 };
142
143 inline void
144 1710090x reactor_descriptor_state::invoke_deferred_io()
145 {
146 1710090x std::shared_ptr<void> prevent_impl_destruction;
147 1710090x ready_queue local_ops;
148
149 {
150
1/2
✓ Branch 0 taken 1710090 times.
✗ Branch 1 not taken.
 1710090x conditionally_enabled_mutex::scoped_lock lock(mutex);
151
152 // Must clear is_enqueued_ and move impl_ref_ under the same
153 // lock that processes I/O. close_socket() checks is_enqueued_
154 // under this mutex — without atomicity between the flag store
155 // and the ref move, close_socket() could see is_enqueued_==false,
156 // skip setting impl_ref_, and destroy the impl under us.
157 1710090x prevent_impl_destruction = std::move(impl_ref_);
158 1710090x is_enqueued_.store(false, std::memory_order_release);
159
160 1710090x std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
161
2/2
✓ Branch 0 taken 1710086 times.
✓ Branch 1 taken 4 times.
 1710090x if (ev == 0)
162 {
163 // Mutex unlocks here; compensate for work_cleanup's decrement
164 4x scheduler_->compensating_work_started();
165 4x return;
166 }
167
168 1710086x int err = 0;
169
2/2
✓ Branch 0 taken 1710059 times.
✓ Branch 1 taken 27 times.
 1710086x if (ev & reactor_event_error)
170 {
171 27x socklen_t len = sizeof(err);
172
2/4
✓ Branch 0 taken 27 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 27 times.
 27x if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
173 ✗ err = errno;
174
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 27 times.
 27x if (err == 0)
175 ✗ err = EIO;
176 27x }
177
178
2/2
✓ Branch 0 taken 67961 times.
✓ Branch 1 taken 1642125 times.
 1710086x if (ev & reactor_event_read)
179 {
180
2/2
✓ Branch 0 taken 95419 times.
✓ Branch 1 taken 1546706 times.
 1642125x if (read_op)
181 {
182 95419x auto* rd = read_op;
183
2/2
✓ Branch 0 taken 5 times.
✓ Branch 1 taken 95414 times.
 95419x if (err)
184 5x rd->complete(err, 0);
185 else
186 95414x rd->perform_io();
187
188
3/4
✓ Branch 0 taken 94400 times.
✓ Branch 1 taken 1019 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 94400 times.
 95419x if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
189 {
190 1019x rd->errn = 0;
191 1019x }
192 else
193 {
194 94400x read_op = nullptr;
195 94400x local_ops.push(rd);
196 }
197 95419x }
198 else
199 {
200 1546706x read_ready = true;
201 }
202
203 // The event does not prove the socket is still readable: a
204 // parked read op above may have drained it, or a speculative
205 // read consumed the data before this dispatch ran. The wait
206 // op's perform_io() re-probes and reports EAGAIN to stay
207 // parked.
208
2/2
✓ Branch 0 taken 1642105 times.
✓ Branch 1 taken 20 times.
 1642125x if (wait_read_op)
209 {
210 20x auto* wo = wait_read_op;
211
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 20 times.
 20x if (err)
212 ✗ wo->complete(err, 0);
213 else
214 20x wo->perform_io();
215
216
3/4
✓ Branch 0 taken 18 times.
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 18 times.
 20x if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK)
217 {
218 2x wo->errn = 0;
219 2x }
220 else
221 {
222 18x wait_read_op = nullptr;
223 18x local_ops.push(wo);
224 }
225 20x }
226 1642125x }
227
2/2
✓ Branch 0 taken 68164 times.
✓ Branch 1 taken 1641922 times.
 1710086x if (ev & reactor_event_write)
228 {
229
2/2
✓ Branch 0 taken 7540 times.
✓ Branch 1 taken 60624 times.
 68164x bool had_write_op = (connect_op || write_op);
230 // A writable event on a socket still in SYN_SENT (e.g. the
231 // spurious pre-connect readiness of a fresh socket) must
232 // not complete the connect; perform_io() reports EAGAIN
233 // until a peer is actually established.
234
2/2
✓ Branch 0 taken 60624 times.
✓ Branch 1 taken 7540 times.
 68164x if (connect_op)
235 {
236 7540x auto* cn = connect_op;
237
2/2
✓ Branch 0 taken 17 times.
✓ Branch 1 taken 7523 times.
 7540x if (err)
238 17x cn->complete(err, 0);
239 else
240 7523x cn->perform_io();
241
242
2/4
✓ Branch 0 taken 7540 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 7540 times.
 7540x if (cn->errn == EAGAIN || cn->errn == EWOULDBLOCK)
243 {
244 ✗ cn->errn = 0;
245 ✗ }
246 else
247 {
248 7540x connect_op = nullptr;
249 7540x local_ops.push(cn);
250 }
251 7540x }
252
2/2
✓ Branch 0 taken 67147 times.
✓ Branch 1 taken 1017 times.
 68164x if (write_op)
253 {
254 1017x auto* wr = write_op;
255
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1017 times.
 1017x if (err)
256 ✗ wr->complete(err, 0);
257 else
258 1017x wr->perform_io();
259
260
3/4
✓ Branch 0 taken 1016 times.
✓ Branch 1 taken 1 time.
✗ Branch 2 not taken.
✓ Branch 3 taken 1016 times.
 1017x if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
261 {
262 1x wr->errn = 0;
263 1x }
264 else
265 {
266 1016x write_op = nullptr;
267 1016x local_ops.push(wr);
268 }
269 1017x }
270
2/2
✓ Branch 0 taken 8557 times.
✓ Branch 1 taken 59607 times.
 68164x if (!had_write_op)
271 59607x write_ready = true;
272
273 // Same re-probe discipline as the wait-for-read dispatch.
274
2/2
✓ Branch 0 taken 68160 times.
✓ Branch 1 taken 4 times.
 68164x if (wait_write_op)
275 {
276 4x auto* wo = wait_write_op;
277
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 4 times.
 4x if (err)
278 ✗ wo->complete(err, 0);
279 else
280 4x wo->perform_io();
281
282
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 4 times.
 4x if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK)
283 {
284 ✗ wo->errn = 0;
285 ✗ }
286 else
287 {
288 4x wait_write_op = nullptr;
289 4x local_ops.push(wo);
290 }
291 4x }
292 68164x }
293 // Complete a parked wait-for-error on any error condition.
294
3/4
✓ Branch 0 taken 1710059 times.
✓ Branch 1 taken 27 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 1710059 times.
 1710086x if ((ev & reactor_event_error) || err)
295 {
296
1/2
✓ Branch 0 taken 27 times.
✗ Branch 1 not taken.
 27x if (wait_error_op)
297 {
298 ✗ wait_error_op->complete(err, 0);
299 ✗ local_ops.push(std::exchange(wait_error_op, nullptr));
300 ✗ }
301 27x }
302
2/2
✓ Branch 0 taken 1710059 times.
✓ Branch 1 taken 27 times.
 1710086x if (err)
303 {
304
1/2
✓ Branch 0 taken 27 times.
✗ Branch 1 not taken.
 27x if (read_op)
305 {
306 ✗ read_op->complete(err, 0);
307 ✗ local_ops.push(std::exchange(read_op, nullptr));
308 ✗ }
309
1/2
✓ Branch 0 taken 27 times.
✗ Branch 1 not taken.
 27x if (write_op)
310 {
311 ✗ write_op->complete(err, 0);
312 ✗ local_ops.push(std::exchange(write_op, nullptr));
313 ✗ }
314
1/2
✓ Branch 0 taken 27 times.
✗ Branch 1 not taken.
 27x if (connect_op)
315 {
316 ✗ connect_op->complete(err, 0);
317 ✗ local_ops.push(std::exchange(connect_op, nullptr));
318 ✗ }
319
1/2
✓ Branch 0 taken 27 times.
✗ Branch 1 not taken.
 27x if (wait_read_op)
320 {
321 ✗ wait_read_op->complete(err, 0);
322 ✗ local_ops.push(std::exchange(wait_read_op, nullptr));
323 ✗ }
324
1/2
✓ Branch 0 taken 27 times.
✗ Branch 1 not taken.
 27x if (wait_write_op)
325 {
326 ✗ wait_write_op->complete(err, 0);
327 ✗ local_ops.push(std::exchange(wait_write_op, nullptr));
328 ✗ }
329 27x }
330 1710090x }
331
332 // Execute first handler inline — the scheduler's work_cleanup
333 // accounts for this as the "consumed" work item. local_ops holds
334 // only ops, so the popped entry decodes directly.
335 1710086x scheduler_op* first = ready_as_op(local_ops.pop());
336
2/2
✓ Branch 0 taken 102978 times.
✓ Branch 1 taken 1607108 times.
 1710086x if (first)
337 {
338
1/2
✓ Branch 0 taken 102978 times.
✗ Branch 1 not taken.
 102978x scheduler_->post_deferred_completions(local_ops);
339
1/2
✓ Branch 0 taken 102978 times.
✗ Branch 1 not taken.
 102978x (*first)();
340 102978x }
341 else
342 {
343 1607108x scheduler_->compensating_work_started();
344 }
345 1710090x }
346
347 } // namespace boost::corosio::detail
348
349 #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
350