include/boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp
92.3% Lines (96 / 104)
100.0% Functions (5 / 5)
Functions (5)
Function
Calls
Lines
Blocks
boost::corosio::detail::reactor_descriptor_state::add_ready_events(unsigned int)
:107
50987x
100.0%
100.0%
boost::corosio::detail::reactor_descriptor_state::operator()()
:113
49547x
100.0%
100.0%
boost::corosio::detail::reactor_descriptor_state::hand_off(boost::corosio::detail::object_ref&)
:119
23011x
100.0%
100.0%
boost::corosio::detail::reactor_descriptor_state::destroy()
:132
225x
100.0%
100.0%
boost::corosio::detail::reactor_descriptor_state::invoke_deferred_io()
:147
49549x
91.1%
86.0%
| Line | TLA | Hits | 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 | 50987x | void add_ready_events(std::uint32_t ev) noexcept | |
| 108 | { | ||
| 109 | 50987x | ready_events_.fetch_or(ev, std::memory_order_acq_rel); | |
| 110 | 50987x | } | |
| 111 | |||
| 112 | /// Invoke deferred I/O and dispatch completions. | ||
| 113 | 49547x | void operator()() override | |
| 114 | { | ||
| 115 | 49547x | invoke_deferred_io(); | |
| 116 | 49547x | } | |
| 117 | |||
| 118 | /// Move @p keep into this state if it is queued without one. | ||
| 119 | 23011x | 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 | 23011x | conditionally_enabled_mutex::scoped_lock lock(mutex); | |
| 125 | 23011x | if (is_enqueued_.load(std::memory_order_acquire) && !object_ref_) | |
| 126 | 866x | object_ref_ = std::move(keep); | |
| 127 | 23011x | } | |
| 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 | 225x | void destroy() override | |
| 133 | { | ||
| 134 | 225x | object_ref_.reset(); | |
| 135 | 225x | } | |
| 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 | 49549x | reactor_descriptor_state::invoke_deferred_io() | |
| 148 | { | ||
| 149 | 49549x | detail::object_ref prevent_impl_destruction; | |
| 150 | 49549x | 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 | 49549x | reactor_scheduler const* sched = nullptr; | |
| 154 | |||
| 155 | { | ||
| 156 | 49549x | conditionally_enabled_mutex::scoped_lock lock(mutex); | |
| 157 | 49549x | 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 | 49549x | prevent_impl_destruction = std::move(object_ref_); | |
| 165 | 49549x | is_enqueued_.store(false, std::memory_order_release); | |
| 166 | |||
| 167 | 49549x | std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire); | |
| 168 | 49549x | if (ev == 0) | |
| 169 | { | ||
| 170 | // Mutex unlocks here; compensate for work_cleanup's decrement | ||
| 171 | 1x | scheduler_->compensating_work_started(); | |
| 172 | 1x | return; | |
| 173 | } | ||
| 174 | |||
| 175 | 49548x | int err = 0; | |
| 176 | 49548x | 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 | 281x | 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 | 281x | bool const reports_error = | |
| 197 | 281x | read_op || write_op || connect_op || wait_error_op; | |
| 198 | 281x | socklen_t len = sizeof(err); | |
| 199 | 313x | if (reports_error && | |
| 200 | 32x | ::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 | 11x | 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 | 49548x | if (ev & reactor_event_read) | |
| 215 | { | ||
| 216 | 22421x | if (read_op) | |
| 217 | { | ||
| 218 | 7668x | auto* rd = read_op; | |
| 219 | 7668x | if (err) | |
| 220 | 8x | rd->complete(err, 0); | |
| 221 | else | ||
| 222 | 7660x | rd->perform_io(); | |
| 223 | |||
| 224 | 7668x | if (rd->errn == EAGAIN || rd->errn == EWOULDBLOCK) | |
| 225 | { | ||
| 226 | 5x | rd->errn = 0; | |
| 227 | } | ||
| 228 | else | ||
| 229 | { | ||
| 230 | 7663x | read_op = nullptr; | |
| 231 | 7663x | local_ops.push(rd); | |
| 232 | } | ||
| 233 | } | ||
| 234 | else | ||
| 235 | { | ||
| 236 | 14753x | 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 | 22421x | if (wait_read_op) | |
| 245 | { | ||
| 246 | 38x | auto* wo = wait_read_op; | |
| 247 | 38x | wo->perform_io(); | |
| 248 | |||
| 249 | 38x | if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK) | |
| 250 | { | ||
| 251 | 2x | wo->errn = 0; | |
| 252 | } | ||
| 253 | else | ||
| 254 | { | ||
| 255 | 36x | wait_read_op = nullptr; | |
| 256 | 36x | local_ops.push(wo); | |
| 257 | } | ||
| 258 | } | ||
| 259 | } | ||
| 260 | 49548x | if (ev & reactor_event_write) | |
| 261 | { | ||
| 262 | 42181x | 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 | 42181x | if (connect_op) | |
| 268 | { | ||
| 269 | 8230x | auto* cn = connect_op; | |
| 270 | 8230x | if (err) | |
| 271 | 12x | cn->complete(err, 0); | |
| 272 | else | ||
| 273 | 8218x | cn->perform_io(); | |
| 274 | |||
| 275 | 8230x | if (cn->errn == EAGAIN || cn->errn == EWOULDBLOCK) | |
| 276 | { | ||
| 277 | ✗ | cn->errn = 0; | |
| 278 | } | ||
| 279 | else | ||
| 280 | { | ||
| 281 | 8230x | connect_op = nullptr; | |
| 282 | 8230x | local_ops.push(cn); | |
| 283 | } | ||
| 284 | } | ||
| 285 | 42181x | if (write_op) | |
| 286 | { | ||
| 287 | 200x | auto* wr = write_op; | |
| 288 | 200x | if (err) | |
| 289 | 2x | wr->complete(err, 0); | |
| 290 | else | ||
| 291 | 198x | wr->perform_io(); | |
| 292 | |||
| 293 | 200x | if (wr->errn == EAGAIN || wr->errn == EWOULDBLOCK) | |
| 294 | { | ||
| 295 | 1x | wr->errn = 0; | |
| 296 | } | ||
| 297 | else | ||
| 298 | { | ||
| 299 | 199x | write_op = nullptr; | |
| 300 | 199x | local_ops.push(wr); | |
| 301 | } | ||
| 302 | } | ||
| 303 | 42181x | if (!had_write_op) | |
| 304 | 33751x | write_ready = true; | |
| 305 | |||
| 306 | // Same re-probe discipline as the wait-for-read dispatch. | ||
| 307 | 42181x | if (wait_write_op) | |
| 308 | { | ||
| 309 | 11x | auto* wo = wait_write_op; | |
| 310 | 11x | wo->perform_io(); | |
| 311 | |||
| 312 | 11x | if (wo->errn == EAGAIN || wo->errn == EWOULDBLOCK) | |
| 313 | { | ||
| 314 | ✗ | wo->errn = 0; | |
| 315 | } | ||
| 316 | else | ||
| 317 | { | ||
| 318 | 11x | wait_write_op = nullptr; | |
| 319 | 11x | local_ops.push(wo); | |
| 320 | } | ||
| 321 | } | ||
| 322 | } | ||
| 323 | // Complete a parked wait-for-error on any error condition. | ||
| 324 | 49548x | if (ev & reactor_event_error) | |
| 325 | { | ||
| 326 | 281x | 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 | 6x | int const werr = err ? err : EIO; | |
| 332 | 6x | wait_error_op->complete(werr, 0); | |
| 333 | 6x | local_ops.push(std::exchange(wait_error_op, nullptr)); | |
| 334 | } | ||
| 335 | } | ||
| 336 | 49548x | if (err) | |
| 337 | { | ||
| 338 | 25x | if (read_op) | |
| 339 | { | ||
| 340 | ✗ | read_op->complete(err, 0); | |
| 341 | ✗ | local_ops.push(std::exchange(read_op, nullptr)); | |
| 342 | } | ||
| 343 | 25x | if (write_op) | |
| 344 | { | ||
| 345 | ✗ | write_op->complete(err, 0); | |
| 346 | ✗ | local_ops.push(std::exchange(write_op, nullptr)); | |
| 347 | } | ||
| 348 | 25x | if (connect_op) | |
| 349 | { | ||
| 350 | ✗ | connect_op->complete(err, 0); | |
| 351 | ✗ | local_ops.push(std::exchange(connect_op, nullptr)); | |
| 352 | } | ||
| 353 | } | ||
| 354 | 49549x | } | |
| 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 | 49548x | scheduler_op* first = ready_as_op(local_ops.pop()); | |
| 360 | 49548x | if (first) | |
| 361 | { | ||
| 362 | 16143x | sched->post_deferred_completions(local_ops); | |
| 363 | 16143x | (*first)(); | |
| 364 | } | ||
| 365 | else | ||
| 366 | { | ||
| 367 | 33405x | sched->compensating_work_started(); | |
| 368 | } | ||
| 369 | 49549x | } | |
| 370 | |||
| 371 | } // namespace boost::corosio::detail | ||
| 372 | |||
| 373 | #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP | ||
| 374 |