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

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

80.0% Lines (72/90) 100.0% List of functions (4/4)
[ ] prev/next file n p prev/next uncovered line
reactor_descriptor_state.hpp
f(x) Functions (4)
Function Calls Lines Blocks
boost::corosio::detail::reactor_descriptor_state::add_ready_events(unsigned int) :120 462037x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::operator()() :126 461849x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::destroy() :134 188x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::invoke_deferred_io() :149 461849x 77.8% 67.0%
Line 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 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 conditionally_enabled_mutex mutex{true};
55
56 /// Pending read operation (guarded by `mutex`).
57 reactor_op_base* read_op = nullptr;
58
59 /// Pending write operation (guarded by `mutex`).
60 reactor_op_base* write_op = nullptr;
61
62 /// Pending connect operation (guarded by `mutex`).
63 reactor_op_base* connect_op = nullptr;
64
65 /// Pending wait-for-read operation (guarded by `mutex`).
66 reactor_op_base* wait_read_op = nullptr;
67
68 /// Pending wait-for-write operation (guarded by `mutex`).
69 reactor_op_base* wait_write_op = nullptr;
70
71 /// Pending wait-for-error operation (guarded by `mutex`).
72 reactor_op_base* wait_error_op = nullptr;
73
74 /// True if a read edge event arrived before an op was registered.
75 bool read_ready = false;
76
77 /// True if a write edge event arrived before an op was registered.
78 bool write_ready = false;
79
80 /// Deferred read cancellation (IOCP-style cancel semantics).
81 bool read_cancel_pending = false;
82
83 /// Deferred write cancellation (IOCP-style cancel semantics).
84 bool write_cancel_pending = false;
85
86 /// Deferred connect cancellation (IOCP-style cancel semantics).
87 bool connect_cancel_pending = false;
88
89 /// Deferred wait-read cancellation (IOCP-style cancel semantics).
90 bool wait_read_cancel_pending = false;
91
92 /// Deferred wait-write cancellation (IOCP-style cancel semantics).
93 bool wait_write_cancel_pending = false;
94
95 /// Deferred wait-error cancellation (IOCP-style cancel semantics).
96 bool wait_error_cancel_pending = false;
97
98 /// Event mask set during registration (no mutex needed).
99 std::uint32_t registered_events = 0;
100
101 /// File descriptor this state tracks.
102 int fd = -1;
103
104 /// Accumulated ready events (set by reactor, read by scheduler).
105 std::atomic<std::uint32_t> ready_events_{0};
106
107 /// True while this state is queued in the scheduler's completed_ops.
108 std::atomic<bool> is_enqueued_{false};
109
110 /// Owning scheduler for posting completions.
111 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 462037x void add_ready_events(std::uint32_t ev) noexcept
121 {
122 462037x ready_events_.fetch_or(ev, std::memory_order_release);
123 462037x }
124
125 /// Invoke deferred I/O and dispatch completions.
126 461849x void operator()() override
127 {
128 461849x invoke_deferred_io();
129 461849x }
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 188x void destroy() override
135 {
136 188x impl_ref_.reset();
137 188x }
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 461849x reactor_descriptor_state::invoke_deferred_io()
150 {
151 461849x std::shared_ptr<void> prevent_impl_destruction;
152 461849x ready_queue local_ops;
153
154 {
155 461849x 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 461849x prevent_impl_destruction = std::move(impl_ref_);
163 461849x is_enqueued_.store(false, std::memory_order_release);
164
165 461849x std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
166 461849x if (ev == 0)
167 {
168 // Mutex unlocks here; compensate for work_cleanup's decrement
169 ✗ scheduler_->compensating_work_started();
170 ✗ return;
171 }
172
173 461849x int err = 0;
174 461849x if (ev & reactor_event_error)
175 {
176 11x socklen_t len = sizeof(err);
177 11x if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
178 ✗ err = errno;
179 11x if (err == 0)
180 1x err = EIO;
181 }
182
183 461849x if (ev & reactor_event_read)
184 {
185 428979x if (read_op)
186 {
187 10016x auto* rd = read_op;
188 10016x if (err)
189 3x rd->complete(err, 0);
190 else
191 10013x rd->perform_io();
192
193 10016x if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
194 {
195 352x rd->errn = 0;
196 }
197 else
198 {
199 9664x read_op = nullptr;
200 9664x local_ops.push(rd);
201 }
202 }
203 else
204 {
205 418963x read_ready = true;
206 }
207
208 // Complete any parked wait-for-read regardless of read_op presence.
209 428979x 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 }
214 }
215 461849x if (ev & reactor_event_write)
216 {
217 46354x bool had_write_op = (connect_op || write_op);
218 46354x if (connect_op)
219 {
220 7586x auto* cn = connect_op;
221 7586x if (err)
222 6x cn->complete(err, 0);
223 else
224 7580x cn->perform_io();
225 7586x connect_op = nullptr;
226 7586x local_ops.push(cn);
227 }
228 46354x if (write_op)
229 {
230 190x auto* wr = write_op;
231 190x if (err)
232 ✗ wr->complete(err, 0);
233 else
234 190x wr->perform_io();
235
236 190x if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
237 {
238 1x wr->errn = 0;
239 }
240 else
241 {
242 189x write_op = nullptr;
243 189x local_ops.push(wr);
244 }
245 }
246 46354x if (!had_write_op)
247 38578x write_ready = true;
248
249 // Complete any parked wait-for-write regardless of write_op presence.
250 46354x if (wait_write_op)
251 {
252 ✗ wait_write_op->complete(err, 0);
253 ✗ local_ops.push(std::exchange(wait_write_op, nullptr));
254 }
255 }
256 // Complete a parked wait-for-error on any error condition.
257 461849x if ((ev & reactor_event_error) || err)
258 {
259 11x if (wait_error_op)
260 {
261 ✗ wait_error_op->complete(err, 0);
262 ✗ local_ops.push(std::exchange(wait_error_op, nullptr));
263 }
264 }
265 461849x if (err)
266 {
267 11x if (read_op)
268 {
269 ✗ read_op->complete(err, 0);
270 ✗ local_ops.push(std::exchange(read_op, nullptr));
271 }
272 11x if (write_op)
273 {
274 ✗ write_op->complete(err, 0);
275 ✗ local_ops.push(std::exchange(write_op, nullptr));
276 }
277 11x if (connect_op)
278 {
279 ✗ connect_op->complete(err, 0);
280 ✗ local_ops.push(std::exchange(connect_op, nullptr));
281 }
282 11x if (wait_read_op)
283 {
284 ✗ wait_read_op->complete(err, 0);
285 ✗ local_ops.push(std::exchange(wait_read_op, nullptr));
286 }
287 11x if (wait_write_op)
288 {
289 ✗ wait_write_op->complete(err, 0);
290 ✗ local_ops.push(std::exchange(wait_write_op, nullptr));
291 }
292 }
293 461849x }
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 461849x scheduler_op* first = ready_as_op(local_ops.pop());
299 461849x if (first)
300 {
301 17455x scheduler_->post_deferred_completions(local_ops);
302 17455x (*first)();
303 }
304 else
305 {
306 444394x scheduler_->compensating_work_started();
307 }
308 461849x }
309
310 } // namespace boost::corosio::detail
311
312 #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
313