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

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

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