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

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

77.8% Lines (70/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 253849x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::operator()() :126 253696x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::destroy() :134 153x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::invoke_deferred_io() :149 253696x 75.3% 65.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 253849x void add_ready_events(std::uint32_t ev) noexcept
121 {
122 253849x ready_events_.fetch_or(ev, std::memory_order_release);
123 253849x }
124
125 /// Invoke deferred I/O and dispatch completions.
126 253696x void operator()() override
127 {
128 253696x invoke_deferred_io();
129 253696x }
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 153x void destroy() override
135 {
136 153x impl_ref_.reset();
137 153x }
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 253696x reactor_descriptor_state::invoke_deferred_io()
150 {
151 253696x std::shared_ptr<void> prevent_impl_destruction;
152 253696x ready_queue local_ops;
153
154 {
155 253696x 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 253696x prevent_impl_destruction = std::move(impl_ref_);
163 253696x is_enqueued_.store(false, std::memory_order_release);
164
165 253696x std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
166 253696x if (ev == 0)
167 {
168 // Mutex unlocks here; compensate for work_cleanup's decrement
169 ✗ scheduler_->compensating_work_started();
170 ✗ return;
171 }
172
173 253696x int err = 0;
174 253696x if (ev & reactor_event_error)
175 {
176 6x socklen_t len = sizeof(err);
177 6x if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
178 ✗ err = errno;
179 6x if (err == 0)
180 ✗ err = EIO;
181 }
182
183 253696x if (ev & reactor_event_read)
184 {
185 228292x if (read_op)
186 {
187 7471x auto* rd = read_op;
188 7471x if (err)
189 ✗ rd->complete(err, 0);
190 else
191 7471x rd->perform_io();
192
193 7471x if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
194 {
195 350x rd->errn = 0;
196 }
197 else
198 {
199 7121x read_op = nullptr;
200 7121x local_ops.push(rd);
201 }
202 }
203 else
204 {
205 220821x read_ready = true;
206 }
207
208 // Complete any parked wait-for-read regardless of read_op presence.
209 228292x 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 253696x if (ev & reactor_event_write)
216 {
217 32386x bool had_write_op = (connect_op || write_op);
218 32386x if (connect_op)
219 {
220 6665x auto* cn = connect_op;
221 6665x if (err)
222 6x cn->complete(err, 0);
223 else
224 6659x cn->perform_io();
225 6665x connect_op = nullptr;
226 6665x local_ops.push(cn);
227 }
228 32386x 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 32386x if (!had_write_op)
247 25531x write_ready = true;
248
249 // Complete any parked wait-for-write regardless of write_op presence.
250 32386x 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 253696x if ((ev & reactor_event_error) || err)
258 {
259 6x 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 253696x if (err)
266 {
267 6x if (read_op)
268 {
269 ✗ read_op->complete(err, 0);
270 ✗ local_ops.push(std::exchange(read_op, nullptr));
271 }
272 6x if (write_op)
273 {
274 ✗ write_op->complete(err, 0);
275 ✗ local_ops.push(std::exchange(write_op, nullptr));
276 }
277 6x if (connect_op)
278 {
279 ✗ connect_op->complete(err, 0);
280 ✗ local_ops.push(std::exchange(connect_op, nullptr));
281 }
282 6x if (wait_read_op)
283 {
284 ✗ wait_read_op->complete(err, 0);
285 ✗ local_ops.push(std::exchange(wait_read_op, nullptr));
286 }
287 6x 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 253696x }
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 253696x scheduler_op* first = ready_as_op(local_ops.pop());
299 253696x if (first)
300 {
301 13991x scheduler_->post_deferred_completions(local_ops);
302 13991x (*first)();
303 }
304 else
305 {
306 239705x scheduler_->compensating_work_started();
307 }
308 253696x }
309
310 } // namespace boost::corosio::detail
311
312 #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
313