include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp

97.7% Lines (173 / 177) 100.0% Functions (14 / 14)

[ ] prev/next file n p prev/next uncovered line

epoll_scheduler.hpp

f(x) Functions (14)

Function Calls Lines Blocks

Line	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_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	5688x	ensure_write_registered(int, reactor_descriptor_state*) const noexcept
124		{
125	5688x	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	287x	[[nodiscard]] std::error_code register_signal_reader(int read_fd) override
136		{
137	287x	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	2525x	inline epoll_scheduler::epoll_scheduler(capy::execution_context&, int)
167	2525x	: epoll_fd_(-1)
168	2525x	, event_fd_(-1)
169	2525x	, timer_fd_(-1)
170	5050x	, event_buffer_(max_events_per_poll_)
171		{
172	2525x	epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
173	2525x	if (epoll_fd_ < 0)
174	3x	detail::throw_system_error(make_err(errno), "epoll_create1");
175
176	2522x	event_fd_ = ::eventfd(0, EFD_NONBLOCK \| EFD_CLOEXEC);
177	2522x	if (event_fd_ < 0)
178		{
179	3x	int errn = errno;
180	3x	::close(epoll_fd_);
181	3x	detail::throw_system_error(make_err(errn), "eventfd");
182		}
183
184	2519x	epoll_event ev{};
185	2519x	ev.events = EPOLLIN \| EPOLLET;
186	2519x	ev.data.ptr = nullptr;
187	2519x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
188		{
189	3x	int errn = errno;
190	3x	::close(event_fd_);
191	3x	::close(epoll_fd_);
192	3x	detail::throw_system_error(make_err(errn), "epoll_ctl");
193		}
194
195	2516x	completed_ops_.push(&task_op_);
196	2543x	}
197
198		inline void
199	530x	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	530x	int fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK \| TFD_CLOEXEC);
204	530x	if (fd >= 0)
205		{
206		// Set before registration, which orders it for every thread
207		// the timerfd's events can wake.
208	529x	timer_fd_ = fd;
209	529x	epoll_event timer_ev{};
210	529x	timer_ev.events = EPOLLIN \| EPOLLERR;
211	529x	timer_ev.data.ptr = &timer_fd_;
212	529x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &timer_ev) < 0)
213		{
214	1x	::close(fd);
215	1x	timer_fd_ = -1;
216		}
217		}
218	530x	svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) {
219	4708x	auto* self = static_cast<epoll_scheduler*>(p);
220	4708x	self->timerfd_stale_.store(true, std::memory_order_release);
221	4708x	self->interrupt_reactor();
222	4708x	}));
223	530x	}
224
225	5032x	inline epoll_scheduler::~epoll_scheduler()
226		{
227	2516x	if (timer_fd_ >= 0)
228	528x	::close(timer_fd_);
229	2516x	if (event_fd_ >= 0)
230	2516x	::close(event_fd_);
231	2516x	if (epoll_fd_ >= 0)
232	2516x	::close(epoll_fd_);
233	5032x	}
234
235		inline void
236	2516x	epoll_scheduler::shutdown()
237		{
238	2516x	shutdown_drain();
239
240	2516x	if (event_fd_ >= 0)
241	2516x	interrupt_reactor();
242	2516x	}
243
244		inline void
245	30x	epoll_scheduler::configure_reactor(
246		unsigned max_events,
247		unsigned budget_init,
248		unsigned budget_max,
249		unsigned unassisted)
250		{
251	30x	reactor_scheduler::configure_reactor(
252		max_events, budget_init, budget_max, unassisted);
253	30x	event_buffer_.resize(max_events_per_poll_);
254	30x	}
255
256		inline std::error_code
257	13668x	epoll_scheduler::register_descriptor(
258		int fd, reactor_descriptor_state* desc) const
259		{
260	13668x	epoll_event ev{};
261	13668x	ev.events = EPOLLIN \| EPOLLOUT \| EPOLLET \| EPOLLERR \| EPOLLHUP;
262	13668x	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	13668x	if (desc->mutex.enabled() != reactor_io_locking_)
270	3x	desc->mutex.set_enabled(reactor_io_locking_);
271		{
272	13668x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
273	13668x	desc->registered_events = ev.events;
274	13668x	desc->unpollable = false;
275	13668x	desc->fd = fd;
276	13668x	desc->scheduler_ = this;
277	13668x	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	13668x	desc->read_ready = false;
281	13668x	desc->write_ready = false;
282	13668x	}
283
284	13668x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
285		{
286	12x	int const err = errno;
287	12x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
288	12x	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	12x	if (err != EPERM)
292	7x	return make_err(err);
293	5x	desc->unpollable = true;
294	12x	}
295	13661x	return {};
296		}
297
298		inline void
299	13370x	epoll_scheduler::deregister_descriptor(int fd) const
300		{
301	13370x	::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
302	13370x	}
303
304		inline void
305	13232x	epoll_scheduler::interrupt_reactor() const
306		{
307	13232x	bool expected = false;
308	13232x	if (eventfd_armed_.compare_exchange_strong(
309		expected, true, std::memory_order_release,
310		std::memory_order_relaxed))
311		{
312	10805x	std::uint64_t val = 1;
313	10805x	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	2x	eventfd_armed_.store(false, std::memory_order_release);
323		}
324		}
325	13232x	}
326
327		inline void
328	11401x	epoll_scheduler::update_timerfd() const
329		{
330	11401x	auto* tsvc = timers();
331	11401x	if (!tsvc \|\| timer_fd_ < 0)
332	6x	return;
333	11395x	auto nearest = tsvc->nearest_expiry();
334
335	11395x	itimerspec ts{};
336	11395x	int flags = 0;
337
338	11395x	if (nearest == timer_service::time_point::max())
339		{
340		// No timers — disarm by setting to 0 (relative)
341		}
342		else
343		{
344	10177x	auto now = std::chrono::steady_clock::now();
345	10177x	if (nearest <= now)
346		{
347		// Use 1ns instead of 0 — zero disarms the timerfd
348	1065x	ts.it_value.tv_nsec = 1;
349		}
350		else
351		{
352	9112x	auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
353	9112x	nearest - now)
354	9112x	.count();
355	9112x	ts.it_value.tv_sec = nsec / 1000000000;
356	9112x	ts.it_value.tv_nsec = nsec % 1000000000;
357	9112x	if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
358	✗	ts.it_value.tv_nsec = 1;
359		}
360		}
361
362	11395x	if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
363	1x	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	4x	epoll_scheduler::timer_timeout_ms(timer_service& svc, int timeout_ms) noexcept
369		{
370	4x	auto nearest = svc.nearest_expiry();
371	4x	if (nearest == timer_service::time_point::max())
372	✗	return timeout_ms;
373	4x	auto now = std::chrono::steady_clock::now();
374	4x	if (nearest <= now)
375	✗	return 0;
376		auto ms =
377	4x	std::chrono::ceil<std::chrono::milliseconds>(nearest - now).count();
378	8x	ms = (std::min)(ms,
379	4x	static_cast<decltype(ms)>(
380	4x	(std::numeric_limits<int>::max)()));
381	4x	if (timeout_ms >= 0 && timeout_ms < ms)
382	✗	return timeout_ms;
383	4x	return static_cast<int>(ms);
384		}
385
386		inline void
387	149360x	epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
388		{
389		int timeout_ms;
390	149360x	if (task_interrupted_)
391	129370x	timeout_ms = 0;
392	19990x	else if (timeout_us < 0)
393	19512x	timeout_ms = -1;
394		else
395	478x	timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
396
397	149360x	if (lock.owns_lock())
398	19994x	lock.unlock();
399
400	149360x	task_cleanup on_exit{this, &lock, ctx};
401
402		// Flush deferred timerfd programming before blocking
403	149360x	if (timerfd_stale_.exchange(false, std::memory_order_acquire))
404	3825x	update_timerfd();
405
406		// timer_fd_ is read only once the service is seen, after which it
407		// no longer changes.
408	149359x	auto* tsvc = timers();
409	149359x	bool const poll_timers = tsvc && timer_fd_ < 0;
410	149359x	if (poll_timers && timeout_ms != 0)
411	4x	timeout_ms = timer_timeout_ms(*tsvc, timeout_ms);
412
413	149359x	int nfds = ::epoll_wait(
414	149359x	epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
415		timeout_ms);
416
417	149359x	if (nfds < 0 && errno != EINTR)
418	1x	detail::throw_system_error(make_err(errno), "epoll_wait");
419
420	149358x	bool check_timers = poll_timers;
421	149358x	ready_queue local_ops;
422
423	211507x	for (int i = 0; i < nfds; ++i)
424		{
425	62149x	if (event_buffer_[i].data.ptr == nullptr)
426		{
427		std::uint64_t val;
428		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
429	8287x	[[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
430	8287x	eventfd_armed_.store(false, std::memory_order_relaxed);
431	8287x	continue;
432	8287x	}
433
434	53862x	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	7572x	::read(timer_fd_, &expirations, sizeof(expirations));
440	7572x	check_timers = true;
441	7572x	continue;
442	7572x	}
443
444		auto* desc =
445	46290x	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	46290x	std::uint32_t ev = event_buffer_[i].events;
461	46290x	if ((ev & (EPOLLHUP \| EPOLLOUT)) == EPOLLHUP)
462	8x	ev \|= EPOLLIN \| EPOLLOUT \| EPOLLERR;
463	46282x	else if (ev & EPOLLHUP)
464	3206x	ev \|= EPOLLIN \| EPOLLOUT;
465	46290x	desc->add_ready_events(ev);
466
467	46290x	bool expected = false;
468	46290x	if (desc->is_enqueued_.compare_exchange_strong(
469		expected, true, std::memory_order_release,
470		std::memory_order_relaxed))
471		{
472	46279x	local_ops.push(desc);
473		}
474		}
475
476	149358x	if (check_timers)
477		{
478	7576x	if (auto* svc = timers())
479	7576x	svc->process_expired();
480	7576x	update_timerfd();
481		}
482
483	149358x	lock.lock();
484
485	149358x	completed_ops_.splice(local_ops);
486	149360x	}
487
488		} // namespace boost::corosio::detail
489
490		#endif // BOOST_COROSIO_HAS_EPOLL
491
492		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
493