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