TLA Line data 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_EPOLL_EPOLL_SCHEDULER_HPP
12 : #define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
13 :
14 : #include <boost/corosio/detail/platform.hpp>
15 :
16 : #if BOOST_COROSIO_HAS_EPOLL
17 :
18 : #include <boost/corosio/detail/config.hpp>
19 : #include <boost/capy/ex/execution_context.hpp>
20 :
21 : #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
22 : #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp>
23 :
24 : #include <boost/corosio/native/detail/epoll/epoll_traits.hpp>
25 : #include <boost/corosio/detail/timer_service.hpp>
26 : #include <boost/corosio/native/detail/make_err.hpp>
27 :
28 : #include <boost/corosio/detail/except.hpp>
29 :
30 : #include <atomic>
31 : #include <chrono>
32 : #include <cstdint>
33 : #include <limits>
34 : #include <mutex>
35 : #include <vector>
36 :
37 : #include <errno.h>
38 : #include <sys/epoll.h>
39 : #include <sys/eventfd.h>
40 : #include <sys/timerfd.h>
41 : #include <unistd.h>
42 :
43 : namespace boost::corosio::detail {
44 :
45 : /** Linux scheduler using epoll for I/O multiplexing.
46 :
47 : This scheduler implements the scheduler interface using Linux epoll
48 : for efficient I/O event notification. It uses a single reactor model
49 : where one thread runs epoll_wait while other threads
50 : wait on a condition variable for handler work. This design provides:
51 :
52 : - Handler parallelism: N posted handlers can execute on N threads
53 : - No thundering herd: condition_variable wakes exactly one thread
54 : - IOCP parity: Behavior matches Windows I/O completion port semantics
55 :
56 : When threads call run(), they first try to execute queued handlers.
57 : If the queue is empty and no reactor is running, one thread becomes
58 : the reactor and runs epoll_wait. Other threads wait on a condition
59 : variable until handlers are available.
60 :
61 : @par Thread Safety
62 : All public member functions are thread-safe.
63 : */
64 : class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
65 : {
66 : public:
67 : /** Construct the scheduler.
68 :
69 : Creates an epoll instance and an eventfd for reactor
70 : interruption. The timerfd for timer expiry is created with the
71 : context's first timer.
72 :
73 : @param ctx Reference to the owning execution_context.
74 : @param concurrency_hint Hint for expected thread count (unused).
75 : */
76 : epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
77 :
78 : /// Destroy the scheduler.
79 : ~epoll_scheduler() override;
80 :
81 : epoll_scheduler(epoll_scheduler const&) = delete;
82 : epoll_scheduler& operator=(epoll_scheduler const&) = delete;
83 :
84 : /// Shut down the scheduler, draining pending operations.
85 : void shutdown() override;
86 :
87 : /// Apply runtime configuration, resizing the event buffer.
88 : void configure_reactor(
89 : unsigned max_events,
90 : unsigned budget_init,
91 : unsigned budget_max,
92 : unsigned unassisted) override;
93 :
94 : /** Return the epoll file descriptor.
95 :
96 : Used by socket services to register file descriptors
97 : for I/O event notification.
98 :
99 : @return The epoll file descriptor.
100 : */
101 : int epoll_fd() const noexcept
102 : {
103 : return epoll_fd_;
104 : }
105 :
106 : /** Register a descriptor for persistent monitoring.
107 :
108 : The fd is registered once and stays registered until explicitly
109 : deregistered. Events are dispatched via reactor_descriptor_state which
110 : tracks pending read/write/connect operations.
111 :
112 : @param fd The file descriptor to register.
113 : @param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
114 :
115 : @return The error if registration fails, otherwise a default
116 : constructed error code.
117 : */
118 : std::error_code
119 : register_descriptor(int fd, reactor_descriptor_state* desc) const;
120 :
121 : /// No-op: write readiness is watched from registration on.
122 : std::error_code
123 HIT 5401 : ensure_write_registered(int, reactor_descriptor_state*) const noexcept
124 : {
125 5401 : return {};
126 : }
127 :
128 : /** Deregister a persistently registered descriptor.
129 :
130 : @param fd The file descriptor to deregister.
131 : */
132 : void deregister_descriptor(int fd) const;
133 :
134 : /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
135 286 : [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
136 : {
137 286 : return register_descriptor(read_fd, signal_pipe_reader_.arm());
138 : }
139 :
140 : private:
141 : void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
142 : void interrupt_reactor() const override;
143 : void update_timerfd() const;
144 : static int timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept;
145 : void on_timer_service_attached(timer_service& svc) override;
146 :
147 : int epoll_fd_;
148 : int event_fd_;
149 : int timer_fd_;
150 :
151 : // Watches the global signal self-pipe's read end (armed lazily by
152 : // register_signal_reader on the first signal registration).
153 : reactor_signal_pipe_reader signal_pipe_reader_;
154 :
155 : // Edge-triggered eventfd state
156 : mutable std::atomic<bool> eventfd_armed_{false};
157 :
158 : // Set when the earliest timer changes; flushed before epoll_wait
159 : mutable std::atomic<bool> timerfd_stale_{false};
160 :
161 : // Event buffer sized from max_events_per_poll_ (set at construction,
162 : // resized by configure_reactor via io_context_options).
163 : std::vector<epoll_event> event_buffer_;
164 : };
165 :
166 2484 : inline epoll_scheduler::epoll_scheduler(capy::execution_context&, int)
167 2484 : : epoll_fd_(-1)
168 2484 : , event_fd_(-1)
169 2484 : , timer_fd_(-1)
170 4968 : , event_buffer_(max_events_per_poll_)
171 : {
172 2484 : epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
173 2484 : if (epoll_fd_ < 0)
174 3 : detail::throw_system_error(make_err(errno), "epoll_create1");
175 :
176 2481 : event_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
177 2481 : if (event_fd_ < 0)
178 : {
179 3 : int errn = errno;
180 3 : ::close(epoll_fd_);
181 3 : detail::throw_system_error(make_err(errn), "eventfd");
182 : }
183 :
184 2478 : epoll_event ev{};
185 2478 : ev.events = EPOLLIN | EPOLLET;
186 2478 : ev.data.ptr = nullptr;
187 2478 : if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
188 : {
189 3 : int errn = errno;
190 3 : ::close(event_fd_);
191 3 : ::close(epoll_fd_);
192 3 : detail::throw_system_error(make_err(errn), "epoll_ctl");
193 : }
194 :
195 2475 : completed_ops_.push(&task_op_);
196 2502 : }
197 :
198 : inline void
199 534 : epoll_scheduler::on_timer_service_attached(timer_service& svc)
200 : {
201 : // Without a timerfd the epoll_wait timeout drives the timers: the
202 : // first timer is often made where a throw can only terminate.
203 534 : int fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
204 534 : if (fd >= 0)
205 : {
206 : // Set before registration, which orders it for every thread
207 : // the timerfd's events can wake.
208 533 : timer_fd_ = fd;
209 533 : epoll_event timer_ev{};
210 533 : timer_ev.events = EPOLLIN | EPOLLERR;
211 533 : timer_ev.data.ptr = &timer_fd_;
212 533 : if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &timer_ev) < 0)
213 : {
214 1 : ::close(fd);
215 1 : timer_fd_ = -1;
216 : }
217 : }
218 534 : svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) {
219 4650 : auto* self = static_cast<epoll_scheduler*>(p);
220 4650 : self->timerfd_stale_.store(true, std::memory_order_release);
221 4650 : self->interrupt_reactor();
222 4650 : }));
223 534 : }
224 :
225 4950 : inline epoll_scheduler::~epoll_scheduler()
226 : {
227 2475 : if (timer_fd_ >= 0)
228 532 : ::close(timer_fd_);
229 2475 : if (event_fd_ >= 0)
230 2475 : ::close(event_fd_);
231 2475 : if (epoll_fd_ >= 0)
232 2475 : ::close(epoll_fd_);
233 4950 : }
234 :
235 : inline void
236 2475 : epoll_scheduler::shutdown()
237 : {
238 2475 : shutdown_drain();
239 :
240 2475 : if (event_fd_ >= 0)
241 2475 : interrupt_reactor();
242 2475 : }
243 :
244 : inline void
245 32 : epoll_scheduler::configure_reactor(
246 : unsigned max_events,
247 : unsigned budget_init,
248 : unsigned budget_max,
249 : unsigned unassisted)
250 : {
251 32 : reactor_scheduler::configure_reactor(
252 : max_events, budget_init, budget_max, unassisted);
253 30 : event_buffer_.resize(max_events_per_poll_);
254 30 : }
255 :
256 : inline std::error_code
257 13287 : epoll_scheduler::register_descriptor(
258 : int fd, reactor_descriptor_state* desc) const
259 : {
260 13287 : epoll_event ev{};
261 13287 : ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLERR | EPOLLHUP;
262 13287 : ev.data.ptr = desc;
263 :
264 : // Initialized before the add: another run thread's reactor can
265 : // dispatch an event for the descriptor as soon as it is added.
266 : // A completion queued by an earlier registration of the same state
267 : // may still be running, so the fields it reads change under its
268 : // mutex, and the mutex mode is only written when it differs.
269 13287 : if (desc->mutex.enabled() != reactor_io_locking_)
270 3 : desc->mutex.set_enabled(reactor_io_locking_);
271 : {
272 13287 : conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
273 13287 : desc->registered_events = ev.events;
274 13287 : desc->unpollable = false;
275 13287 : desc->fd = fd;
276 13287 : desc->scheduler_ = this;
277 13287 : desc->ready_events_.store(0, std::memory_order_release);
278 : // object_ref_ is not touched: a queued invocation from an
279 : // earlier registration still owns the reference in it.
280 13287 : desc->read_ready = false;
281 13287 : desc->write_ready = false;
282 13287 : }
283 :
284 13287 : if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
285 : {
286 12 : int const err = errno;
287 12 : conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
288 12 : desc->registered_events = 0;
289 : // EPERM: a file type epoll cannot watch. Its I/O does not
290 : // block, so adopt it unwatched, as asio does.
291 12 : if (err != EPERM)
292 7 : return make_err(err);
293 5 : desc->unpollable = true;
294 12 : }
295 13280 : return {};
296 : }
297 :
298 : inline void
299 12990 : epoll_scheduler::deregister_descriptor(int fd) const
300 : {
301 12990 : ::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
302 12990 : }
303 :
304 : inline void
305 12879 : epoll_scheduler::interrupt_reactor() const
306 : {
307 12879 : bool expected = false;
308 12879 : if (eventfd_armed_.compare_exchange_strong(
309 : expected, true, std::memory_order_release,
310 : std::memory_order_relaxed))
311 : {
312 10523 : std::uint64_t val = 1;
313 10523 : if (::write(event_fd_, &val, sizeof(val)) < 0)
314 : {
315 : // The flag is what coalesces later interrupts into a byte
316 : // already in the eventfd; a write that failed put no byte
317 : // there, so leaving it armed would swallow every interrupt
318 : // that follows. Disarming keeps the cost to the interrupts
319 : // already in flight -- the next one arms and writes again,
320 : // instead of every one after this coalescing into a byte
321 : // that does not exist.
322 2 : eventfd_armed_.store(false, std::memory_order_release);
323 : }
324 : }
325 12879 : }
326 :
327 : inline void
328 11568 : epoll_scheduler::update_timerfd() const
329 : {
330 11568 : auto* tsvc = timers();
331 11568 : if (!tsvc || timer_fd_ < 0)
332 6 : return;
333 11562 : auto nearest = tsvc->nearest_expiry();
334 :
335 11562 : itimerspec ts{};
336 11562 : int flags = 0;
337 :
338 11562 : if (nearest == timer_service::time_point::max())
339 : {
340 : // No timers — disarm by setting to 0 (relative)
341 : }
342 : else
343 : {
344 10281 : auto now = std::chrono::steady_clock::now();
345 10281 : if (nearest <= now)
346 : {
347 : // Use 1ns instead of 0 — zero disarms the timerfd
348 1109 : ts.it_value.tv_nsec = 1;
349 : }
350 : else
351 : {
352 9172 : auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
353 9172 : nearest - now)
354 9172 : .count();
355 9172 : ts.it_value.tv_sec = nsec / 1000000000;
356 9172 : ts.it_value.tv_nsec = nsec % 1000000000;
357 9172 : if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
358 MIS 0 : ts.it_value.tv_nsec = 1;
359 : }
360 : }
361 :
362 HIT 11562 : if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
363 1 : detail::throw_system_error(make_err(errno), "timerfd_settime");
364 : }
365 :
366 : // Rounded up, so the wait never wakes before the timer is due.
367 : inline int
368 4 : epoll_scheduler::timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept
369 : {
370 4 : auto nearest = svc.nearest_expiry();
371 4 : if (nearest == timer_service::time_point::max())
372 MIS 0 : return timeout_ms;
373 HIT 4 : auto now = std::chrono::steady_clock::now();
374 4 : if (nearest <= now)
375 MIS 0 : return 0;
376 : auto ms =
377 HIT 4 : std::chrono::ceil<std::chrono::milliseconds>(nearest - now).count();
378 8 : ms = (std::min)(ms,
379 4 : static_cast<decltype(ms)>(
380 4 : (std::numeric_limits<int>::max)()));
381 4 : if (timeout_ms >= 0 && timeout_ms < ms)
382 MIS 0 : return timeout_ms;
383 HIT 4 : return static_cast<int>(ms);
384 : }
385 :
386 : inline void
387 148711 : epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
388 : {
389 : int timeout_ms;
390 148711 : if (task_interrupted_)
391 128721 : timeout_ms = 0;
392 19990 : else if (timeout_us < 0)
393 19613 : timeout_ms = -1;
394 : else
395 377 : timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
396 :
397 148711 : if (lock.owns_lock())
398 19994 : lock.unlock();
399 :
400 148711 : task_cleanup on_exit{this, &lock, ctx};
401 :
402 : // Flush deferred timerfd programming before blocking
403 148711 : if (timerfd_stale_.exchange(false, std::memory_order_acquire))
404 3797 : update_timerfd();
405 :
406 : // timer_fd_ is read only once the service is seen, after which it
407 : // no longer changes.
408 148710 : auto* tsvc = timers();
409 148710 : bool const poll_timers = tsvc && timer_fd_ < 0;
410 148710 : if (poll_timers && timeout_ms != 0)
411 4 : timeout_ms = timer_timeout_ms(*tsvc, timeout_ms);
412 :
413 148710 : int nfds = ::epoll_wait(
414 148710 : epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
415 : timeout_ms);
416 :
417 148710 : if (nfds < 0 && errno != EINTR)
418 1 : detail::throw_system_error(make_err(errno), "epoll_wait");
419 :
420 148709 : bool check_timers = poll_timers;
421 148709 : ready_queue local_ops;
422 :
423 210020 : for (int i = 0; i < nfds; ++i)
424 : {
425 61311 : if (event_buffer_[i].data.ptr == nullptr)
426 : {
427 : std::uint64_t val;
428 : // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
429 8046 : [[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
430 8046 : eventfd_armed_.store(false, std::memory_order_relaxed);
431 8046 : continue;
432 8046 : }
433 :
434 53265 : if (event_buffer_[i].data.ptr == &timer_fd_)
435 : {
436 : std::uint64_t expirations;
437 : // NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
438 : [[maybe_unused]] auto r =
439 7767 : ::read(timer_fd_, &expirations, sizeof(expirations));
440 7767 : check_timers = true;
441 7767 : continue;
442 7767 : }
443 :
444 : auto* desc =
445 45498 : static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
446 :
447 : // EPOLLHUP maps to no reactor_event_* bit, so a HUP the widening
448 : // below did not translate would take no branch in
449 : // invoke_deferred_io() and, the registration being
450 : // edge-triggered, never get another chance.
451 : //
452 : // HUP without OUT means a non-socket: a pipe or tty whose peer
453 : // closed, with or without data still unread (HUP alone, or
454 : // IN|HUP). Treat it as readable, writable and faulted, the way
455 : // poll() and io_uring report it, so a parked error wait
456 : // completes too. Sockets carry at least OUT with HUP (a fresh
457 : // unconnected stream socket reports HUP|OUT), so they take the
458 : // plain widening, where forcing IN costs at most one spurious
459 : // EAGAIN.
460 45498 : std::uint32_t ev = event_buffer_[i].events;
461 45498 : if ((ev & (EPOLLHUP | EPOLLOUT)) == EPOLLHUP)
462 8 : ev |= EPOLLIN | EPOLLOUT | EPOLLERR;
463 45490 : else if (ev & EPOLLHUP)
464 2963 : ev |= EPOLLIN | EPOLLOUT;
465 45498 : desc->add_ready_events(ev);
466 :
467 45498 : bool expected = false;
468 45498 : if (desc->is_enqueued_.compare_exchange_strong(
469 : expected, true, std::memory_order_release,
470 : std::memory_order_relaxed))
471 : {
472 45476 : local_ops.push(desc);
473 : }
474 : }
475 :
476 148709 : if (check_timers)
477 : {
478 7771 : if (auto* svc = timers())
479 7771 : svc->process_expired();
480 7771 : update_timerfd();
481 : }
482 :
483 148709 : lock.lock();
484 :
485 148709 : completed_ops_.splice(local_ops);
486 148711 : }
487 :
488 : } // namespace boost::corosio::detail
489 :
490 : #endif // BOOST_COROSIO_HAS_EPOLL
491 :
492 : #endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
|
