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