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

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

87.9% Lines (123/140) 100.0% List of functions (6/6) 76.7% Branches (66/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 49202x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::~reactor_descriptor_state() :45 49202x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::add_ready_events(unsigned int) :97 1062396x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::operator()() :103 1061822x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::destroy() :111 399x 100.0% – 100.0% boost::corosio::detail::reactor_descriptor_state::invoke_deferred_io() :126 1061822x 85.3% 76.7% 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 24601x 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 24601x conditionally_enabled_mutex mutex{true};
50
51 /// Pending read operation (guarded by `mutex`).
52 24601x reactor_op_base* read_op = nullptr;
53
54 /// Pending write operation (guarded by `mutex`).
55 24601x reactor_op_base* write_op = nullptr;
56
57 /// Pending connect operation (guarded by `mutex`).
58 24601x reactor_op_base* connect_op = nullptr;
59
60 /// Pending wait-for-read operation (guarded by `mutex`).
61 24601x reactor_op_base* wait_read_op = nullptr;
62
63 /// Pending wait-for-write operation (guarded by `mutex`).
64 24601x reactor_op_base* wait_write_op = nullptr;
65
66 /// Pending wait-for-error operation (guarded by `mutex`).
67 24601x reactor_op_base* wait_error_op = nullptr;
68
69 /// True if a read edge event arrived before an op was registered.
70 24601x bool read_ready = false;
71
72 /// True if a write edge event arrived before an op was registered.
73 24601x bool write_ready = false;
74
75 /// Event mask set during registration (no mutex needed).
76 24601x std::uint32_t registered_events = 0;
77
78 /// File descriptor this state tracks.
79 24601x int fd = -1;
80
81 /// Accumulated ready events (set by reactor, read by scheduler).
82 24601x std::atomic<std::uint32_t> ready_events_{0};
83
84 /// True while this state is queued in the scheduler's completed_ops.
85 24601x std::atomic<bool> is_enqueued_{false};
86
87 /// Owning scheduler for posting completions.
88 24601x reactor_scheduler const* scheduler_ = nullptr;
89
90 /// Prevents impl destruction while queued in the scheduler.
91 std::shared_ptr<void> impl_ref_;
92
93 /// Add ready events atomically.
94 /// Release pairs with the consumer's acquire exchange on
95 /// ready_events_ so the consumer sees all flags. On x86 (TSO)
96 /// this compiles to the same LOCK OR as relaxed.
97 1062396x void add_ready_events(std::uint32_t ev) noexcept
98 {
99 1062396x ready_events_.fetch_or(ev, std::memory_order_release);
100 1062396x }
101
102 /// Invoke deferred I/O and dispatch completions.
103 1061822x void operator()() override
104 {
105 1061822x invoke_deferred_io();
106 1061822x }
107
108 /// Destroy without invoking.
109 /// Called during scheduler::shutdown() drain. Clear impl_ref_ to break
110 /// the self-referential cycle set by close_socket().
111 399x void destroy() override
112 {
113 399x impl_ref_.reset();
114 399x }
115
116 /** Perform deferred I/O and queue completions.
117
118 Performs I/O under the mutex and queues completed ops. EAGAIN
119 ops stay parked in their slot for re-delivery on the next
120 edge event.
121 */
122 void invoke_deferred_io();
123 };
124
125 inline void
126 1061822x reactor_descriptor_state::invoke_deferred_io()
127 {
128 1061822x std::shared_ptr<void> prevent_impl_destruction;
129 1061822x ready_queue local_ops;
130
131 {
132
1/2
✓ Branch 0 taken 1061822 times.
✗ Branch 1 not taken.
 1061822x conditionally_enabled_mutex::scoped_lock lock(mutex);
133
134 // Must clear is_enqueued_ and move impl_ref_ under the same
135 // lock that processes I/O. close_socket() checks is_enqueued_
136 // under this mutex — without atomicity between the flag store
137 // and the ref move, close_socket() could see is_enqueued_==false,
138 // skip setting impl_ref_, and destroy the impl under us.
139 1061822x prevent_impl_destruction = std::move(impl_ref_);
140 1061822x is_enqueued_.store(false, std::memory_order_release);
141
142 1061822x std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
143
2/2
✓ Branch 0 taken 1061818 times.
✓ Branch 1 taken 4 times.
 1061822x if (ev == 0)
144 {
145 // Mutex unlocks here; compensate for work_cleanup's decrement
146 4x scheduler_->compensating_work_started();
147 4x return;
148 }
149
150 1061818x int err = 0;
151
2/2
✓ Branch 0 taken 1061774 times.
✓ Branch 1 taken 44 times.
 1061818x if (ev & reactor_event_error)
152 {
153 44x socklen_t len = sizeof(err);
154
3/4
✓ Branch 0 taken 44 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 6 times.
✓ Branch 3 taken 38 times.
 44x if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
155
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6x err = errno;
156 // select raises its exceptional set for out-of-band/urgent
157 // data as well as for genuine faults; on a healthy socket the
158 // probe then reads SO_ERROR == 0. Faulting a pending read or
159 // write on that is wrong, so an I/O operation completes only
160 // on a real (non-zero) error. wait(error) still names a code
161 // below.
162 44x }
163
164
2/2
✓ Branch 0 taken 45318 times.
✓ Branch 1 taken 1016500 times.
 1061818x if (ev & reactor_event_read)
165 {
166
2/2
✓ Branch 0 taken 64334 times.
✓ Branch 1 taken 952166 times.
 1016500x if (read_op)
167 {
168 64334x auto* rd = read_op;
169
2/2
✓ Branch 0 taken 9 times.
✓ Branch 1 taken 64325 times.
 64334x if (err)
170 9x rd->complete(err, 0);
171 else
172 64325x rd->perform_io();
173
174
3/4
✓ Branch 0 taken 63617 times.
✓ Branch 1 taken 717 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 63617 times.
 64334x if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
175 {
176 717x rd->errn = 0;
177 717x }
178 else
179 {
180 63617x read_op = nullptr;
181 63617x local_ops.push(rd);
182 }
183 64334x }
184 else
185 {
186 952166x read_ready = true;
187 }
188
189 // The event does not prove the socket is still readable: a
190 // parked read op above may have drained it, or a speculative
191 // read consumed the data before this dispatch ran. The wait
192 // op's perform_io() re-probes and reports EAGAIN to stay
193 // parked.
194
2/2
✓ Branch 0 taken 1016476 times.
✓ Branch 1 taken 24 times.
 1016500x if (wait_read_op)
195 {
196 24x auto* wo = wait_read_op;
197
2/2
✓ Branch 0 taken 1 time.
✓ Branch 1 taken 23 times.
 24x if (err)
198 1x wo->complete(err, 0);
199 else
200 23x wo->perform_io();
201
202
3/4
✓ Branch 0 taken 23 times.
✓ Branch 1 taken 1 time.
✗ Branch 2 not taken.
✓ Branch 3 taken 23 times.
 24x if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK)
203 {
204 1x wo->errn = 0;
205 1x }
206 else
207 {
208 23x wait_read_op = nullptr;
209 23x local_ops.push(wo);
210 }
211 24x }
212 1016500x }
213
2/2
✓ Branch 0 taken 45481 times.
✓ Branch 1 taken 1016337 times.
 1061818x if (ev & reactor_event_write)
214 {
215
2/2
✓ Branch 0 taken 6383 times.
✓ Branch 1 taken 39098 times.
 45481x bool had_write_op = (connect_op || write_op);
216 // A writable event on a socket still in SYN_SENT (e.g. the
217 // spurious pre-connect readiness of a fresh socket) must
218 // not complete the connect; perform_io() reports EAGAIN
219 // until a peer is actually established.
220
2/2
✓ Branch 0 taken 39098 times.
✓ Branch 1 taken 6383 times.
 45481x if (connect_op)
221 {
222 6383x auto* cn = connect_op;
223
2/2
✓ Branch 0 taken 17 times.
✓ Branch 1 taken 6366 times.
 6383x if (err)
224 17x cn->complete(err, 0);
225 else
226 6366x cn->perform_io();
227
228
2/4
✓ Branch 0 taken 6383 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6383 times.
 6383x if (cn->errn == EAGAIN || cn->errn == EWOULDBLOCK)
229 {
230 ✗ cn->errn = 0;
231 ✗ }
232 else
233 {
234 6383x connect_op = nullptr;
235 6383x local_ops.push(cn);
236 }
237 6383x }
238
2/2
✓ Branch 0 taken 44463 times.
✓ Branch 1 taken 1018 times.
 45481x if (write_op)
239 {
240 1018x auto* wr = write_op;
241
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1018 times.
 1018x if (err)
242 ✗ wr->complete(err, 0);
243 else
244 1018x wr->perform_io();
245
246
2/4
✓ Branch 0 taken 1018 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 1018 times.
 1018x if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
247 {
248 ✗ wr->errn = 0;
249 ✗ }
250 else
251 {
252 1018x write_op = nullptr;
253 1018x local_ops.push(wr);
254 }
255 1018x }
256
2/2
✓ Branch 0 taken 7401 times.
✓ Branch 1 taken 38080 times.
 45481x if (!had_write_op)
257 38080x write_ready = true;
258
259 // Same re-probe discipline as the wait-for-read dispatch.
260
2/2
✓ Branch 0 taken 45477 times.
✓ Branch 1 taken 4 times.
 45481x if (wait_write_op)
261 {
262 4x auto* wo = wait_write_op;
263
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 4 times.
 4x if (err)
264 ✗ wo->complete(err, 0);
265 else
266 4x wo->perform_io();
267
268
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)
269 {
270 ✗ wo->errn = 0;
271 ✗ }
272 else
273 {
274 4x wait_write_op = nullptr;
275 4x local_ops.push(wo);
276 }
277 4x }
278 45481x }
279 // Complete a parked wait-for-error on any error condition.
280
3/4
✓ Branch 0 taken 1061774 times.
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 1061774 times.
 1061818x if ((ev & reactor_event_error) || err)
281 {
282
2/2
✓ Branch 0 taken 41 times.
✓ Branch 1 taken 3 times.
 44x if (wait_error_op)
283 {
284 // wait(error) fired on the exceptional condition; name a
285 // code even when the kernel exposed none (e.g. urgent
286 // data leaves SO_ERROR == 0).
287
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
 3x int const werr = err ? err : EIO;
288 3x wait_error_op->complete(werr, 0);
289 3x local_ops.push(std::exchange(wait_error_op, nullptr));
290 3x }
291 44x }
292
2/2
✓ Branch 0 taken 1061779 times.
✓ Branch 1 taken 39 times.
 1061818x if (err)
293 {
294
2/2
✓ Branch 0 taken 38 times.
✓ Branch 1 taken 1 time.
 39x if (read_op)
295 {
296 1x read_op->complete(err, 0);
297 1x local_ops.push(std::exchange(read_op, nullptr));
298 1x }
299
1/2
✓ Branch 0 taken 39 times.
✗ Branch 1 not taken.
 39x if (write_op)
300 {
301 ✗ write_op->complete(err, 0);
302 ✗ local_ops.push(std::exchange(write_op, nullptr));
303 ✗ }
304
1/2
✓ Branch 0 taken 39 times.
✗ Branch 1 not taken.
 39x if (connect_op)
305 {
306 ✗ connect_op->complete(err, 0);
307 ✗ local_ops.push(std::exchange(connect_op, nullptr));
308 ✗ }
309
2/2
✓ Branch 0 taken 38 times.
✓ Branch 1 taken 1 time.
 39x if (wait_read_op)
310 {
311 1x wait_read_op->complete(err, 0);
312 1x local_ops.push(std::exchange(wait_read_op, nullptr));
313 1x }
314
1/2
✓ Branch 0 taken 39 times.
✗ Branch 1 not taken.
 39x if (wait_write_op)
315 {
316 ✗ wait_write_op->complete(err, 0);
317 ✗ local_ops.push(std::exchange(wait_write_op, nullptr));
318 ✗ }
319 39x }
320 1061822x }
321
322 // Execute first handler inline — the scheduler's work_cleanup
323 // accounts for this as the "consumed" work item. local_ops holds
324 // only ops, so the popped entry decodes directly.
325 1061818x scheduler_op* first = ready_as_op(local_ops.pop());
326
2/2
✓ Branch 0 taken 71048 times.
✓ Branch 1 taken 990770 times.
 1061818x if (first)
327 {
328
1/2
✓ Branch 0 taken 71048 times.
✗ Branch 1 not taken.
 71048x scheduler_->post_deferred_completions(local_ops);
329
1/2
✓ Branch 0 taken 71048 times.
✗ Branch 1 not taken.
 71048x (*first)();
330 71048x }
331 else
332 {
333 990770x scheduler_->compensating_work_started();
334 }
335 1061822x }
336
337 } // namespace boost::corosio::detail
338
339 #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
340