TLA Line data 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 : 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 : conditionally_enabled_mutex mutex{true};
50 :
51 : /// Pending read operation (guarded by `mutex`).
52 : reactor_op_base* read_op = nullptr;
53 :
54 : /// Pending write operation (guarded by `mutex`).
55 : reactor_op_base* write_op = nullptr;
56 :
57 : /// Pending connect operation (guarded by `mutex`).
58 : reactor_op_base* connect_op = nullptr;
59 :
60 : /// Pending wait-for-read operation (guarded by `mutex`).
61 : reactor_op_base* wait_read_op = nullptr;
62 :
63 : /// Pending wait-for-write operation (guarded by `mutex`).
64 : reactor_op_base* wait_write_op = nullptr;
65 :
66 : /// Pending wait-for-error operation (guarded by `mutex`).
67 : reactor_op_base* wait_error_op = nullptr;
68 :
69 : /// True if a read edge event arrived before an op was registered.
70 : bool read_ready = false;
71 :
72 : /// True if a write edge event arrived before an op was registered.
73 : bool write_ready = false;
74 :
75 : /// Event mask set during registration (no mutex needed).
76 : std::uint32_t registered_events = 0;
77 :
78 : /// File descriptor this state tracks.
79 : int fd = -1;
80 :
81 : /// Accumulated ready events (set by reactor, read by scheduler).
82 : std::atomic<std::uint32_t> ready_events_{0};
83 :
84 : /// True while this state is queued in the scheduler's completed_ops.
85 : std::atomic<bool> is_enqueued_{false};
86 :
87 : /// Owning scheduler for posting completions.
88 : 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 HIT 379348 : void add_ready_events(std::uint32_t ev) noexcept
98 : {
99 379348 : ready_events_.fetch_or(ev, std::memory_order_release);
100 379348 : }
101 :
102 : /// Invoke deferred I/O and dispatch completions.
103 379058 : void operator()() override
104 : {
105 379058 : invoke_deferred_io();
106 379058 : }
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 290 : void destroy() override
112 : {
113 290 : impl_ref_.reset();
114 290 : }
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 379058 : reactor_descriptor_state::invoke_deferred_io()
127 : {
128 379058 : std::shared_ptr<void> prevent_impl_destruction;
129 379058 : ready_queue local_ops;
130 :
131 : {
132 379058 : 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 379058 : prevent_impl_destruction = std::move(impl_ref_);
140 379058 : is_enqueued_.store(false, std::memory_order_release);
141 :
142 379058 : std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
143 379058 : if (ev == 0)
144 : {
145 : // Mutex unlocks here; compensate for work_cleanup's decrement
146 4 : scheduler_->compensating_work_started();
147 4 : return;
148 : }
149 :
150 379054 : int err = 0;
151 379054 : if (ev & reactor_event_error)
152 : {
153 32 : socklen_t len = sizeof(err);
154 32 : if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
155 10 : 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 : }
163 :
164 379054 : if (ev & reactor_event_read)
165 : {
166 352319 : if (read_op)
167 : {
168 7393 : auto* rd = read_op;
169 7393 : if (err)
170 3 : rd->complete(err, 0);
171 : else
172 7390 : rd->perform_io();
173 :
174 7393 : if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
175 : {
176 360 : rd->errn = 0;
177 : }
178 : else
179 : {
180 7033 : read_op = nullptr;
181 7033 : local_ops.push(rd);
182 : }
183 : }
184 : else
185 : {
186 344926 : 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 352319 : if (wait_read_op)
195 : {
196 27 : auto* wo = wait_read_op;
197 27 : if (err)
198 1 : wo->complete(err, 0);
199 : else
200 26 : wo->perform_io();
201 :
202 27 : if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK)
203 : {
204 4 : wo->errn = 0;
205 : }
206 : else
207 : {
208 23 : wait_read_op = nullptr;
209 23 : local_ops.push(wo);
210 : }
211 : }
212 : }
213 379054 : if (ev & reactor_event_write)
214 : {
215 35883 : 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 35883 : if (connect_op)
221 : {
222 6562 : auto* cn = connect_op;
223 6562 : if (err)
224 8 : cn->complete(err, 0);
225 : else
226 6554 : cn->perform_io();
227 :
228 6562 : if (cn->errn == EAGAIN || cn->errn == EWOULDBLOCK)
229 : {
230 MIS 0 : cn->errn = 0;
231 : }
232 : else
233 : {
234 HIT 6562 : connect_op = nullptr;
235 6562 : local_ops.push(cn);
236 : }
237 : }
238 35883 : if (write_op)
239 : {
240 196 : auto* wr = write_op;
241 196 : if (err)
242 2 : wr->complete(err, 0);
243 : else
244 194 : wr->perform_io();
245 :
246 196 : if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
247 : {
248 1 : wr->errn = 0;
249 : }
250 : else
251 : {
252 195 : write_op = nullptr;
253 195 : local_ops.push(wr);
254 : }
255 : }
256 35883 : if (!had_write_op)
257 29125 : write_ready = true;
258 :
259 : // Same re-probe discipline as the wait-for-read dispatch.
260 35883 : if (wait_write_op)
261 : {
262 7 : auto* wo = wait_write_op;
263 7 : if (err)
264 2 : wo->complete(err, 0);
265 : else
266 5 : wo->perform_io();
267 :
268 7 : if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK)
269 : {
270 MIS 0 : wo->errn = 0;
271 : }
272 : else
273 : {
274 HIT 7 : wait_write_op = nullptr;
275 7 : local_ops.push(wo);
276 : }
277 : }
278 : }
279 : // Complete a parked wait-for-error on any error condition.
280 379054 : if ((ev & reactor_event_error) || err)
281 : {
282 32 : 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 3 : int const werr = err ? err : EIO;
288 3 : wait_error_op->complete(werr, 0);
289 3 : local_ops.push(std::exchange(wait_error_op, nullptr));
290 : }
291 : }
292 379054 : if (err)
293 : {
294 27 : if (read_op)
295 : {
296 1 : read_op->complete(err, 0);
297 1 : local_ops.push(std::exchange(read_op, nullptr));
298 : }
299 27 : if (write_op)
300 : {
301 MIS 0 : write_op->complete(err, 0);
302 0 : local_ops.push(std::exchange(write_op, nullptr));
303 : }
304 HIT 27 : if (connect_op)
305 : {
306 MIS 0 : connect_op->complete(err, 0);
307 0 : local_ops.push(std::exchange(connect_op, nullptr));
308 : }
309 HIT 27 : if (wait_read_op)
310 : {
311 1 : wait_read_op->complete(err, 0);
312 1 : local_ops.push(std::exchange(wait_read_op, nullptr));
313 : }
314 27 : if (wait_write_op)
315 : {
316 MIS 0 : wait_write_op->complete(err, 0);
317 0 : local_ops.push(std::exchange(wait_write_op, nullptr));
318 : }
319 : }
320 HIT 379058 : }
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 379054 : scheduler_op* first = ready_as_op(local_ops.pop());
326 379054 : if (first)
327 : {
328 13823 : scheduler_->post_deferred_completions(local_ops);
329 13823 : (*first)();
330 : }
331 : else
332 : {
333 365231 : scheduler_->compensating_work_started();
334 : }
335 379058 : }
336 :
337 : } // namespace boost::corosio::detail
338 :
339 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
|
