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

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

74.3% Lines (55/74) 100.0% List of functions (4/4)
[ ] prev/next file n p prev/next uncovered line
f(x) Functions (4)
Function Calls Lines Branches Blocks
boost::corosio::detail::reactor_descriptor_state::add_ready_events(unsigned int) :99 0 100.0% – – boost::corosio::detail::reactor_descriptor_state::operator()() :105 0 100.0% – – boost::corosio::detail::reactor_descriptor_state::destroy() :113 0 100.0% – – boost::corosio::detail::reactor_descriptor_state::invoke_deferred_io() :128 0 70.8% – –
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
16 #include <atomic>
17 #include <cstdint>
18 #include <memory>
19 #include <mutex>
20
21 #include <errno.h>
22 #include <sys/socket.h>
23
24 namespace boost::corosio::detail {
25
26 /// Shared reactor event constants.
27 /// These match epoll numeric values; kqueue maps its events to the same.
28 static constexpr std::uint32_t reactor_event_read = 0x001;
29 static constexpr std::uint32_t reactor_event_write = 0x004;
30 static constexpr std::uint32_t reactor_event_error = 0x008;
31
32 /** Per-descriptor state shared across reactor backends.
33
34 Tracks pending operations for a file descriptor. The fd is registered
35 once with the reactor and stays registered until closed. Uses deferred
36 I/O: the reactor sets ready_events atomically, then enqueues this state.
37 When popped by the scheduler, invoke_deferred_io() performs I/O under
38 the mutex and queues completed ops.
39
40 Non-template: uses reactor_op_base pointers so the scheduler and
41 descriptor_state code exist as a single copy in the binary regardless
42 of how many backends are compiled in.
43
44 @par Thread Safety
45 The mutex protects operation pointers and ready flags. ready_events_
46 and is_enqueued_ are atomic for lock-free reactor access.
47 */
48 struct reactor_descriptor_state : scheduler_op
49 {
50 /// Protects operation pointers and ready/cancel flags.
51 std::mutex mutex;
52
53 /// Pending read operation (guarded by `mutex`).
54 reactor_op_base* read_op = nullptr;
55
56 /// Pending write operation (guarded by `mutex`).
57 reactor_op_base* write_op = nullptr;
58
59 /// Pending connect operation (guarded by `mutex`).
60 reactor_op_base* connect_op = nullptr;
61
62 /// True if a read edge event arrived before an op was registered.
63 bool read_ready = false;
64
65 /// True if a write edge event arrived before an op was registered.
66 bool write_ready = false;
67
68 /// Deferred read cancellation (IOCP-style cancel semantics).
69 bool read_cancel_pending = false;
70
71 /// Deferred write cancellation (IOCP-style cancel semantics).
72 bool write_cancel_pending = false;
73
74 /// Deferred connect cancellation (IOCP-style cancel semantics).
75 bool connect_cancel_pending = false;
76
77 /// Event mask set during registration (no mutex needed).
78 std::uint32_t registered_events = 0;
79
80 /// File descriptor this state tracks.
81 int fd = -1;
82
83 /// Accumulated ready events (set by reactor, read by scheduler).
84 std::atomic<std::uint32_t> ready_events_{0};
85
86 /// True while this state is queued in the scheduler's completed_ops.
87 std::atomic<bool> is_enqueued_{false};
88
89 /// Owning scheduler for posting completions.
90 reactor_scheduler_base const* scheduler_ = nullptr;
91
92 /// Prevents impl destruction while queued in the scheduler.
93 std::shared_ptr<void> impl_ref_;
94
95 /// Add ready events atomically.
96 /// Release pairs with the consumer's acquire exchange on
97 /// ready_events_ so the consumer sees all flags. On x86 (TSO)
98 /// this compiles to the same LOCK OR as relaxed.
99 185812x void add_ready_events(std::uint32_t ev) noexcept
100 {
101 185812x ready_events_.fetch_or(ev, std::memory_order_release);
102 185812x }
103
104 /// Invoke deferred I/O and dispatch completions.
105 185760x void operator()() override
106 {
107 185760x invoke_deferred_io();
108 185760x }
109
110 /// Destroy without invoking.
111 /// Called during scheduler::shutdown() drain. Clear impl_ref_ to break
112 /// the self-referential cycle set by close_socket().
113 52x void destroy() override
114 {
115 52x impl_ref_.reset();
116 52x }
117
118 /** Perform deferred I/O and queue completions.
119
120 Performs I/O under the mutex and queues completed ops. EAGAIN
121 ops stay parked in their slot for re-delivery on the next
122 edge event.
123 */
124 void invoke_deferred_io();
125 };
126
127 inline void
128 185760x reactor_descriptor_state::invoke_deferred_io()
129 {
130 185760x std::shared_ptr<void> prevent_impl_destruction;
131 185760x op_queue local_ops;
132
133 {
134 185760x std::lock_guard lock(mutex);
135
136 // Must clear is_enqueued_ and move impl_ref_ under the same
137 // lock that processes I/O. close_socket() checks is_enqueued_
138 // under this mutex — without atomicity between the flag store
139 // and the ref move, close_socket() could see is_enqueued_==false,
140 // skip setting impl_ref_, and destroy the impl under us.
141 185760x prevent_impl_destruction = std::move(impl_ref_);
142 185760x is_enqueued_.store(false, std::memory_order_release);
143
144 std::uint32_t ev =
145 185760x ready_events_.exchange(0, std::memory_order_acquire);
146 185760x if (ev == 0)
147 {
148 // Mutex unlocks here; compensate for work_cleanup's decrement
149 ✗ scheduler_->compensating_work_started();
150 ✗ return;
151 }
152
153 185760x int err = 0;
154 185760x if (ev & reactor_event_error)
155 {
156 1x socklen_t len = sizeof(err);
157 1x if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
158 ✗ err = errno;
159 1x if (err == 0)
160 ✗ err = EIO;
161 }
162
163 185760x if (ev & reactor_event_read)
164 {
165 152789x if (read_op)
166 {
167 7621x auto* rd = read_op;
168 7621x if (err)
169 ✗ rd->complete(err, 0);
170 else
171 7621x rd->perform_io();
172
173 7621x if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
174 {
175 44x rd->errn = 0;
176 }
177 else
178 {
179 7577x read_op = nullptr;
180 7577x local_ops.push(rd);
181 }
182 }
183 else
184 {
185 145168x read_ready = true;
186 }
187 }
188 185760x if (ev & reactor_event_write)
189 {
190 41892x bool had_write_op = (connect_op || write_op);
191 41892x if (connect_op)
192 {
193 7485x auto* cn = connect_op;
194 7485x if (err)
195 1x cn->complete(err, 0);
196 else
197 7484x cn->perform_io();
198 7485x connect_op = nullptr;
199 7485x local_ops.push(cn);
200 }
201 41892x if (write_op)
202 {
203 ✗ auto* wr = write_op;
204 ✗ if (err)
205 ✗ wr->complete(err, 0);
206 else
207 ✗ wr->perform_io();
208
209 ✗ if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
210 {
211 ✗ wr->errn = 0;
212 }
213 else
214 {
215 ✗ write_op = nullptr;
216 ✗ local_ops.push(wr);
217 }
218 }
219 41892x if (!had_write_op)
220 34407x write_ready = true;
221 }
222 185760x if (err)
223 {
224 1x if (read_op)
225 {
226 ✗ read_op->complete(err, 0);
227 ✗ local_ops.push(std::exchange(read_op, nullptr));
228 }
229 1x if (write_op)
230 {
231 ✗ write_op->complete(err, 0);
232 ✗ local_ops.push(std::exchange(write_op, nullptr));
233 }
234 1x if (connect_op)
235 {
236 ✗ connect_op->complete(err, 0);
237 ✗ local_ops.push(std::exchange(connect_op, nullptr));
238 }
239 }
240 185760x }
241
242 // Execute first handler inline — the scheduler's work_cleanup
243 // accounts for this as the "consumed" work item
244 185760x scheduler_op* first = local_ops.pop();
245 185760x if (first)
246 {
247 15062x scheduler_->post_deferred_completions(local_ops);
248 15062x (*first)();
249 }
250 else
251 {
252 170698x scheduler_->compensating_work_started();
253 }
254 185760x }
255
256 } // namespace boost::corosio::detail
257
258 #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
259