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_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 HIT 261745 : void add_ready_events(std::uint32_t ev) noexcept
121 : {
122 261745 : ready_events_.fetch_or(ev, std::memory_order_release);
123 261745 : }
124 :
125 : /// Invoke deferred I/O and dispatch completions.
126 261592 : void operator()() override
127 : {
128 261592 : invoke_deferred_io();
129 261592 : }
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 153 : void destroy() override
135 : {
136 153 : impl_ref_.reset();
137 153 : }
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 261592 : reactor_descriptor_state::invoke_deferred_io()
150 : {
151 261592 : std::shared_ptr<void> prevent_impl_destruction;
152 261592 : ready_queue local_ops;
153 :
154 : {
155 261592 : 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 261592 : prevent_impl_destruction = std::move(impl_ref_);
163 261592 : is_enqueued_.store(false, std::memory_order_release);
164 :
165 261592 : std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
166 261592 : if (ev == 0)
167 : {
168 : // Mutex unlocks here; compensate for work_cleanup's decrement
169 MIS 0 : scheduler_->compensating_work_started();
170 0 : return;
171 : }
172 :
173 HIT 261592 : int err = 0;
174 261592 : if (ev & reactor_event_error)
175 : {
176 9 : socklen_t len = sizeof(err);
177 9 : if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
178 MIS 0 : err = errno;
179 HIT 9 : if (err == 0)
180 MIS 0 : err = EIO;
181 : }
182 :
183 HIT 261592 : if (ev & reactor_event_read)
184 : {
185 232711 : if (read_op)
186 : {
187 6574 : auto* rd = read_op;
188 6574 : if (err)
189 1 : rd->complete(err, 0);
190 : else
191 6573 : rd->perform_io();
192 :
193 6574 : if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
194 : {
195 350 : rd->errn = 0;
196 : }
197 : else
198 : {
199 6224 : read_op = nullptr;
200 6224 : local_ops.push(rd);
201 : }
202 : }
203 : else
204 : {
205 226137 : read_ready = true;
206 : }
207 :
208 : // Complete any parked wait-for-read regardless of read_op presence.
209 232711 : if (wait_read_op)
210 : {
211 16 : wait_read_op->complete(err, 0);
212 16 : local_ops.push(std::exchange(wait_read_op, nullptr));
213 : }
214 : }
215 261592 : if (ev & reactor_event_write)
216 : {
217 37457 : bool had_write_op = (connect_op || write_op);
218 37457 : if (connect_op)
219 : {
220 5767 : auto* cn = connect_op;
221 5767 : if (err)
222 7 : cn->complete(err, 0);
223 : else
224 5760 : cn->perform_io();
225 5767 : connect_op = nullptr;
226 5767 : local_ops.push(cn);
227 : }
228 37457 : if (write_op)
229 : {
230 190 : auto* wr = write_op;
231 190 : if (err)
232 MIS 0 : wr->complete(err, 0);
233 : else
234 HIT 190 : wr->perform_io();
235 :
236 190 : if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
237 : {
238 1 : wr->errn = 0;
239 : }
240 : else
241 : {
242 189 : write_op = nullptr;
243 189 : local_ops.push(wr);
244 : }
245 : }
246 37457 : if (!had_write_op)
247 31500 : write_ready = true;
248 :
249 : // Complete any parked wait-for-write regardless of write_op presence.
250 37457 : if (wait_write_op)
251 : {
252 MIS 0 : wait_write_op->complete(err, 0);
253 0 : local_ops.push(std::exchange(wait_write_op, nullptr));
254 : }
255 : }
256 : // Complete a parked wait-for-error on any error condition.
257 HIT 261592 : if ((ev & reactor_event_error) || err)
258 : {
259 9 : if (wait_error_op)
260 : {
261 MIS 0 : wait_error_op->complete(err, 0);
262 0 : local_ops.push(std::exchange(wait_error_op, nullptr));
263 : }
264 : }
265 HIT 261592 : if (err)
266 : {
267 9 : if (read_op)
268 : {
269 MIS 0 : read_op->complete(err, 0);
270 0 : local_ops.push(std::exchange(read_op, nullptr));
271 : }
272 HIT 9 : if (write_op)
273 : {
274 MIS 0 : write_op->complete(err, 0);
275 0 : local_ops.push(std::exchange(write_op, nullptr));
276 : }
277 HIT 9 : if (connect_op)
278 : {
279 MIS 0 : connect_op->complete(err, 0);
280 0 : local_ops.push(std::exchange(connect_op, nullptr));
281 : }
282 HIT 9 : if (wait_read_op)
283 : {
284 MIS 0 : wait_read_op->complete(err, 0);
285 0 : local_ops.push(std::exchange(wait_read_op, nullptr));
286 : }
287 HIT 9 : if (wait_write_op)
288 : {
289 MIS 0 : wait_write_op->complete(err, 0);
290 0 : local_ops.push(std::exchange(wait_write_op, nullptr));
291 : }
292 : }
293 HIT 261592 : }
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 261592 : scheduler_op* first = ready_as_op(local_ops.pop());
299 261592 : if (first)
300 : {
301 12196 : scheduler_->post_deferred_completions(local_ops);
302 12196 : (*first)();
303 : }
304 : else
305 : {
306 249396 : scheduler_->compensating_work_started();
307 : }
308 261592 : }
309 :
310 : } // namespace boost::corosio::detail
311 :
312 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
|
