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 Blocks
boost::corosio::detail::reactor_descriptor_state::add_ready_events(unsigned int) :99 160360x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::operator()() :105 160290x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::destroy() :113 70x 100.0% 100.0% boost::corosio::detail::reactor_descriptor_state::invoke_deferred_io() :128 160290x 70.8% 65.0%
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 160360x void add_ready_events(std::uint32_t ev) noexcept
100 {
101 160360x ready_events_.fetch_or(ev, std::memory_order_release);
102 160360x }
103
104 /// Invoke deferred I/O and dispatch completions.
105 160290x void operator()() override
106 {
107 160290x invoke_deferred_io();
108 160290x }
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 70x void destroy() override
114 {
115 70x impl_ref_.reset();
116 70x }
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 160290x reactor_descriptor_state::invoke_deferred_io()
129 {
130 160290x std::shared_ptr<void> prevent_impl_destruction;
131 160290x op_queue local_ops;
132
133 {
134 160290x 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 160290x prevent_impl_destruction = std::move(impl_ref_);
142 160290x is_enqueued_.store(false, std::memory_order_release);
143
144 160290x std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
145 160290x if (ev == 0)
146 {
147 // Mutex unlocks here; compensate for work_cleanup's decrement
148 ✗ scheduler_->compensating_work_started();
149 ✗ return;
150 }
151
152 160290x int err = 0;
153 160290x if (ev & reactor_event_error)
154 {
155 2x socklen_t len = sizeof(err);
156 2x if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
157 ✗ err = errno;
158 2x if (err == 0)
159 ✗ err = EIO;
160 }
161
162 160290x if (ev & reactor_event_read)
163 {
164 133345x if (read_op)
165 {
166 4622x auto* rd = read_op;
167 4622x if (err)
168 ✗ rd->complete(err, 0);
169 else
170 4622x rd->perform_io();
171
172 4622x if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK)
173 {
174 43x rd->errn = 0;
175 }
176 else
177 {
178 4579x read_op = nullptr;
179 4579x local_ops.push(rd);
180 }
181 }
182 else
183 {
184 128723x read_ready = true;
185 }
186 }
187 160290x if (ev & reactor_event_write)
188 {
189 35861x bool had_write_op = (connect_op || write_op);
190 35861x if (connect_op)
191 {
192 4490x auto* cn = connect_op;
193 4490x if (err)
194 2x cn->complete(err, 0);
195 else
196 4488x cn->perform_io();
197 4490x connect_op = nullptr;
198 4490x local_ops.push(cn);
199 }
200 35861x if (write_op)
201 {
202 ✗ auto* wr = write_op;
203 ✗ if (err)
204 ✗ wr->complete(err, 0);
205 else
206 ✗ wr->perform_io();
207
208 ✗ if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK)
209 {
210 ✗ wr->errn = 0;
211 }
212 else
213 {
214 ✗ write_op = nullptr;
215 ✗ local_ops.push(wr);
216 }
217 }
218 35861x if (!had_write_op)
219 31371x write_ready = true;
220 }
221 160290x if (err)
222 {
223 2x if (read_op)
224 {
225 ✗ read_op->complete(err, 0);
226 ✗ local_ops.push(std::exchange(read_op, nullptr));
227 }
228 2x if (write_op)
229 {
230 ✗ write_op->complete(err, 0);
231 ✗ local_ops.push(std::exchange(write_op, nullptr));
232 }
233 2x if (connect_op)
234 {
235 ✗ connect_op->complete(err, 0);
236 ✗ local_ops.push(std::exchange(connect_op, nullptr));
237 }
238 }
239 160290x }
240
241 // Execute first handler inline — the scheduler's work_cleanup
242 // accounts for this as the "consumed" work item
243 160290x scheduler_op* first = local_ops.pop();
244 160290x if (first)
245 {
246 9069x scheduler_->post_deferred_completions(local_ops);
247 9069x (*first)();
248 }
249 else
250 {
251 151221x scheduler_->compensating_work_started();
252 }
253 160290x }
254
255 } // namespace boost::corosio::detail
256
257 #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
258