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