TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Steve Gerbino
3 : // Copyright (c) 2026 Michael Vandeberg
4 : //
5 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 : //
8 : // Official repository: https://github.com/cppalliance/corosio
9 : //
10 :
11 : #ifndef BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
12 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
13 :
14 : #include <boost/corosio/native/detail/reactor/reactor_events.hpp>
15 : #include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
16 : #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
17 : #include <boost/corosio/detail/ready_queue.hpp>
18 :
19 : #include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
20 :
21 : #include <atomic>
22 : #include <cstdint>
23 : #include <memory>
24 :
25 : #include <errno.h>
26 : #include <sys/socket.h>
27 :
28 : namespace boost::corosio::detail {
29 :
30 : /** Per-descriptor state shared across reactor backends.
31 :
32 : Tracks pending operations for a file descriptor. The fd is registered
33 : once with the reactor and stays registered until closed. Uses deferred
34 : I/O: the reactor sets ready_events atomically, then enqueues this state.
35 : When popped by the scheduler, invoke_deferred_io() performs I/O under
36 : the mutex and queues completed ops.
37 :
38 : Non-template: uses reactor_op_base pointers so the scheduler and
39 : descriptor_state code exist as a single copy in the binary regardless
40 : of how many backends are compiled in.
41 :
42 : @par Thread Safety
43 : The mutex protects operation pointers and ready flags. ready_events_
44 : and is_enqueued_ are atomic for lock-free reactor access.
45 : */
46 : struct reactor_descriptor_state
47 : : scheduler_op
48 : , reactor_retirable
49 : {
50 : /// Protects operation pointers and ready/cancel flags.
51 : /// Becomes a no-op in single-threaded mode.
52 : conditionally_enabled_mutex mutex{true};
53 :
54 : /// Pending read operation (guarded by `mutex`).
55 : reactor_op_base* read_op = nullptr;
56 :
57 : /// Pending write operation (guarded by `mutex`).
58 : reactor_op_base* write_op = nullptr;
59 :
60 : /// Pending connect operation (guarded by `mutex`).
61 : reactor_op_base* connect_op = nullptr;
62 :
63 : /// Pending wait-for-read operation (guarded by `mutex`).
64 : reactor_op_base* wait_read_op = nullptr;
65 :
66 : /// Pending wait-for-write operation (guarded by `mutex`).
67 : reactor_op_base* wait_write_op = nullptr;
68 :
69 : /// Pending wait-for-error operation (guarded by `mutex`).
70 : reactor_op_base* wait_error_op = nullptr;
71 :
72 : /// True if a read edge event arrived before an op was registered.
73 : bool read_ready = false;
74 :
75 : /// True if a write edge event arrived before an op was registered.
76 : bool write_ready = false;
77 :
78 : /// Event mask set during registration (no mutex needed).
79 : std::uint32_t registered_events = 0;
80 :
81 : /// The reactor refused to watch this fd (e.g. /dev/null on epoll);
82 : /// its I/O never blocks, and an op that would park must not.
83 : bool unpollable = false;
84 :
85 : /// File descriptor this state tracks.
86 : int fd = -1;
87 :
88 : /// Accumulated ready events (set by reactor, read by scheduler).
89 : std::atomic<std::uint32_t> ready_events_{0};
90 :
91 : /// True while this state is queued in the scheduler's completed_ops.
92 : std::atomic<bool> is_enqueued_{false};
93 :
94 : /// Owning scheduler for posting completions.
95 : reactor_scheduler const* scheduler_ = nullptr;
96 :
97 : /// Prevents impl destruction while queued in the scheduler.
98 : detail::object_ref object_ref_;
99 :
100 : /// Add ready events atomically.
101 : /// Release pairs with the consumer's acquire exchange on
102 : /// ready_events_ so the consumer sees all flags. Acquire pairs with
103 : /// registration's release store, so a reactor pass on another
104 : /// thread sees the state registration initialized before handing
105 : /// it to the kernel. On x86 (TSO) this compiles to the same LOCK OR
106 : /// as relaxed.
107 HIT 124497 : void add_ready_events(std::uint32_t ev) noexcept
108 : {
109 124497 : ready_events_.fetch_or(ev, std::memory_order_acq_rel);
110 124497 : }
111 :
112 : /// Invoke deferred I/O and dispatch completions.
113 54214 : void operator()() override
114 : {
115 54214 : invoke_deferred_io();
116 54214 : }
117 :
118 : /// Move @p keep into this state if it is queued without one.
119 22992 : void hand_off(object_ref& keep) noexcept override
120 : {
121 : // Locked even to read the flag: a popped invocation clears it at
122 : // the start of its critical section and still uses this state
123 : // until it unlocks, so an unlocked false does not mean idle.
124 22992 : conditionally_enabled_mutex::scoped_lock lock(mutex);
125 22992 : if (is_enqueued_.load(std::memory_order_acquire) && !object_ref_)
126 886 : object_ref_ = std::move(keep);
127 22992 : }
128 :
129 : /// Destroy without invoking.
130 : /// Called during scheduler::shutdown() drain. Clear object_ref_ to break
131 : /// the self-referential cycle set by close_socket().
132 228 : void destroy() override
133 : {
134 228 : object_ref_.reset();
135 228 : }
136 :
137 : /** Perform deferred I/O and queue completions.
138 :
139 : Performs I/O under the mutex and queues completed ops. EAGAIN
140 : ops stay parked in their slot for re-delivery on the next
141 : edge event.
142 : */
143 : void invoke_deferred_io();
144 : };
145 :
146 : inline void
147 54216 : reactor_descriptor_state::invoke_deferred_io()
148 : {
149 54216 : detail::object_ref prevent_impl_destruction;
150 54216 : ready_queue local_ops;
151 : // Read under the lock: once it drops, a retired owner may be
152 : // recycled, so nothing below touches this state.
153 54216 : reactor_scheduler const* sched = nullptr;
154 :
155 : {
156 54216 : conditionally_enabled_mutex::scoped_lock lock(mutex);
157 54216 : sched = scheduler_;
158 :
159 : // Must clear is_enqueued_ and move object_ref_ under the same
160 : // lock that processes I/O. close_socket() checks is_enqueued_
161 : // under this mutex — without atomicity between the flag store
162 : // and the ref move, close_socket() could see is_enqueued_==false,
163 : // skip setting object_ref_, and destroy the impl under us.
164 54216 : prevent_impl_destruction = std::move(object_ref_);
165 54216 : is_enqueued_.store(false, std::memory_order_release);
166 :
167 54216 : std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
168 54216 : if (ev == 0)
169 : {
170 : // Mutex unlocks here; compensate for work_cleanup's decrement
171 MIS 0 : scheduler_->compensating_work_started();
172 0 : return;
173 : }
174 :
175 HIT 54216 : int err = 0;
176 54216 : if (ev & reactor_event_error)
177 : {
178 : // Force the read/write dispatch below to run: an
179 : // edge-triggered EPOLLERR can arrive alone, and without this
180 : // a parked op never calls perform_io() and, the edge being
181 : // one-shot, never gets another chance -- a permanent hang.
182 : // Every parked op then re-runs its own syscall or probe.
183 : //
184 : // Assumes at least one parked op's own syscall makes
185 : // non-EAGAIN progress; if every op re-parks with EAGAIN this
186 : // sticky error is never redelivered and they hang. No such
187 : // case is known -- a future descriptor type that hits one
188 : // should be handled here.
189 365 : ev |= reactor_event_read | reactor_event_write;
190 :
191 : // SO_ERROR clears on read, so take it only for a parked op
192 : // that reports it. A readiness wait reports readiness and
193 : // leaves the error for the next read or write to name, as
194 : // asio does; reading it here with nothing to report it to
195 : // would turn a reset into a clean EOF.
196 365 : bool const reports_error =
197 365 : read_op || write_op || connect_op || wait_error_op;
198 365 : socklen_t len = sizeof(err);
199 395 : if (reports_error &&
200 30 : ::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
201 : {
202 : // Non-socket fd (pipe, chardev, ...): no SO_ERROR, so
203 : // let the op's own syscall name the real failure.
204 11 : err = (errno == ENOTSOCK) ? 0 : errno;
205 : }
206 : // select raises its exceptional set for out-of-band/urgent
207 : // data as well as for genuine faults; on a healthy socket the
208 : // probe then reads SO_ERROR == 0. Faulting a pending read or
209 : // write on that is wrong, so an I/O operation completes only
210 : // on a real (non-zero) error. wait(error) still names a code
211 : // below.
212 : }
213 :
214 54216 : if (ev & reactor_event_read)
215 : {
216 22332 : if (read_op)
217 : {
218 7702 : auto* rd = read_op;
219 7702 : if (err)
220 6 : rd->complete(err, 0);
221 : else
222 7696 : rd->perform_io();
223 :
224 7702 : if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
225 : {
226 5 : rd->errn = 0;
227 : }
228 : else
229 : {
230 7697 : read_op = nullptr;
231 7697 : local_ops.push(rd);
232 : }
233 : }
234 : else
235 : {
236 14630 : read_ready = true;
237 : }
238 :
239 : // The event does not prove the socket is still readable: a
240 : // parked read op above may have drained it, or a speculative
241 : // read consumed the data before this dispatch ran. The wait
242 : // op's perform_io() re-probes and reports EAGAIN to stay
243 : // parked.
244 22332 : if (wait_read_op)
245 : {
246 38 : auto* wo = wait_read_op;
247 38 : wo->perform_io();
248 :
249 38 : if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK)
250 : {
251 2 : wo->errn = 0;
252 : }
253 : else
254 : {
255 36 : wait_read_op = nullptr;
256 36 : local_ops.push(wo);
257 : }
258 : }
259 : }
260 54216 : if (ev & reactor_event_write)
261 : {
262 46690 : bool had_write_op = (connect_op || write_op);
263 : // A writable event on a socket still in SYN_SENT (e.g. the
264 : // spurious pre-connect readiness of a fresh socket) must
265 : // not complete the connect; perform_io() reports EAGAIN
266 : // until a peer is actually established.
267 46690 : if (connect_op)
268 : {
269 8110 : auto* cn = connect_op;
270 8110 : if (err)
271 12 : cn->complete(err, 0);
272 : else
273 8098 : cn->perform_io();
274 :
275 8110 : if (cn->errn == EAGAIN || cn->errn == EWOULDBLOCK)
276 : {
277 MIS 0 : cn->errn = 0;
278 : }
279 : else
280 : {
281 HIT 8110 : connect_op = nullptr;
282 8110 : local_ops.push(cn);
283 : }
284 : }
285 46690 : if (write_op)
286 : {
287 200 : auto* wr = write_op;
288 200 : if (err)
289 2 : wr->complete(err, 0);
290 : else
291 198 : wr->perform_io();
292 :
293 200 : if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
294 : {
295 1 : wr->errn = 0;
296 : }
297 : else
298 : {
299 199 : write_op = nullptr;
300 199 : local_ops.push(wr);
301 : }
302 : }
303 46690 : if (!had_write_op)
304 38380 : write_ready = true;
305 :
306 : // Same re-probe discipline as the wait-for-read dispatch.
307 46690 : if (wait_write_op)
308 : {
309 11 : auto* wo = wait_write_op;
310 11 : wo->perform_io();
311 :
312 11 : if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK)
313 : {
314 MIS 0 : wo->errn = 0;
315 : }
316 : else
317 : {
318 HIT 11 : wait_write_op = nullptr;
319 11 : local_ops.push(wo);
320 : }
321 : }
322 : }
323 : // Complete a parked wait-for-error on any error condition.
324 54216 : if (ev & reactor_event_error)
325 : {
326 365 : if (wait_error_op)
327 : {
328 : // wait(error) fired on the exceptional condition; name a
329 : // code even when the kernel exposed none (e.g. urgent
330 : // data leaves SO_ERROR == 0).
331 6 : int const werr = err ? err : EIO;
332 6 : wait_error_op->complete(werr, 0);
333 6 : local_ops.push(std::exchange(wait_error_op, nullptr));
334 : }
335 : }
336 54216 : if (err)
337 : {
338 23 : if (read_op)
339 : {
340 MIS 0 : read_op->complete(err, 0);
341 0 : local_ops.push(std::exchange(read_op, nullptr));
342 : }
343 HIT 23 : if (write_op)
344 : {
345 MIS 0 : write_op->complete(err, 0);
346 0 : local_ops.push(std::exchange(write_op, nullptr));
347 : }
348 HIT 23 : if (connect_op)
349 : {
350 MIS 0 : connect_op->complete(err, 0);
351 0 : local_ops.push(std::exchange(connect_op, nullptr));
352 : }
353 : }
354 HIT 54216 : }
355 :
356 : // Execute first handler inline — the scheduler's work_cleanup
357 : // accounts for this as the "consumed" work item. local_ops holds
358 : // only ops, so the popped entry decodes directly.
359 54216 : scheduler_op* first = ready_as_op(local_ops.pop());
360 54216 : if (first)
361 : {
362 16057 : sched->post_deferred_completions(local_ops);
363 16057 : (*first)();
364 : }
365 : else
366 : {
367 38159 : sched->compensating_work_started();
368 : }
369 54216 : }
370 :
371 : } // namespace boost::corosio::detail
372 :
373 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
|
