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

99.3% Lines (152 / 153) 100.0% Functions (11 / 11)

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

epoll_scheduler.hpp

f(x) Functions (11)

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		#include <boost/corosio/native/detail/posix/posix_resolver_service.hpp>
28		#include <boost/corosio/native/detail/posix/posix_signal_service.hpp>
29		#include <boost/corosio/native/detail/posix/posix_stream_file_service.hpp>
30		#include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp>
31
32		#include <boost/corosio/detail/except.hpp>
33
34		#include <atomic>
35		#include <chrono>
36		#include <cstdint>
37		#include <mutex>
38		#include <vector>
39
40		#include <errno.h>
41		#include <sys/epoll.h>
42		#include <sys/eventfd.h>
43		#include <sys/timerfd.h>
44		#include <unistd.h>
45
46		namespace boost::corosio::detail {
47
48		/** Linux scheduler using epoll for I/O multiplexing.
49
50		This scheduler implements the scheduler interface using Linux epoll
51		for efficient I/O event notification. It uses a single reactor model
52		where one thread runs epoll_wait while other threads
53		wait on a condition variable for handler work. This design provides:
54
55		- Handler parallelism: N posted handlers can execute on N threads
56		- No thundering herd: condition_variable wakes exactly one thread
57		- IOCP parity: Behavior matches Windows I/O completion port semantics
58
59		When threads call run(), they first try to execute queued handlers.
60		If the queue is empty and no reactor is running, one thread becomes
61		the reactor and runs epoll_wait. Other threads wait on a condition
62		variable until handlers are available.
63
64		@par Thread Safety
65		All public member functions are thread-safe.
66		*/
67		class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
68		{
69		public:
70		/** Construct the scheduler.
71
72		Creates an epoll instance, eventfd for reactor interruption,
73		and timerfd for kernel-managed timer expiry.
74
75		@param ctx Reference to the owning execution_context.
76		@param concurrency_hint Hint for expected thread count (unused).
77		*/
78		epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
79
80		/// Destroy the scheduler.
81		~epoll_scheduler() override;
82
83		epoll_scheduler(epoll_scheduler const&) = delete;
84		epoll_scheduler& operator=(epoll_scheduler const&) = delete;
85
86		/// Shut down the scheduler, draining pending operations.
87		void shutdown() override;
88
89		/// Apply runtime configuration, resizing the event buffer.
90		void configure_reactor(
91		unsigned max_events,
92		unsigned budget_init,
93		unsigned budget_max,
94		unsigned unassisted) override;
95
96		/** Return the epoll file descriptor.
97
98		Used by socket services to register file descriptors
99		for I/O event notification.
100
101		@return The epoll file descriptor.
102		*/
103		int epoll_fd() const noexcept
104		{
105		return epoll_fd_;
106		}
107
108		/** Register a descriptor for persistent monitoring.
109
110		The fd is registered once and stays registered until explicitly
111		deregistered. Events are dispatched via reactor_descriptor_state which
112		tracks pending read/write/connect operations.
113
114		@param fd The file descriptor to register.
115		@param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
116
117		@return The error if registration fails, otherwise a default
118		constructed error code.
119		*/
120		std::error_code
121		register_descriptor(int fd, reactor_descriptor_state* desc) const;
122
123		/** Deregister a persistently registered descriptor.
124
125		@param fd The file descriptor to deregister.
126		*/
127		void deregister_descriptor(int fd) const;
128
129		/// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
130	69x	[[nodiscard]] std::error_code register_signal_reader(int read_fd) override
131		{
132	69x	return register_descriptor(read_fd, signal_pipe_reader_.arm());
133		}
134
135		private:
136		void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
137		void interrupt_reactor() const override;
138		void update_timerfd() const;
139
140		int epoll_fd_;
141		int event_fd_;
142		int timer_fd_;
143
144		// Watches the global signal self-pipe's read end (armed lazily by
145		// register_signal_reader on the first signal registration).
146		reactor_signal_pipe_reader signal_pipe_reader_;
147
148		// Edge-triggered eventfd state
149		mutable std::atomic<bool> eventfd_armed_{false};
150
151		// Set when the earliest timer changes; flushed before epoll_wait
152		mutable std::atomic<bool> timerfd_stale_{false};
153
154		// Event buffer sized from max_events_per_poll_ (set at construction,
155		// resized by configure_reactor via io_context_options).
156		std::vector<epoll_event> event_buffer_;
157		};
158
159	1232x	inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
160	1232x	: epoll_fd_(-1)
161	1232x	, event_fd_(-1)
162	1232x	, timer_fd_(-1)
163	2464x	, event_buffer_(max_events_per_poll_)
164		{
165	1232x	epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
166	1232x	if (epoll_fd_ < 0)
167	1x	detail::throw_system_error(make_err(errno), "epoll_create1");
168
169	1231x	event_fd_ = ::eventfd(0, EFD_NONBLOCK \| EFD_CLOEXEC);
170	1231x	if (event_fd_ < 0)
171		{
172	1x	int errn = errno;
173	1x	::close(epoll_fd_);
174	1x	detail::throw_system_error(make_err(errn), "eventfd");
175		}
176
177	1230x	timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK \| TFD_CLOEXEC);
178	1230x	if (timer_fd_ < 0)
179		{
180	1x	int errn = errno;
181	1x	::close(event_fd_);
182	1x	::close(epoll_fd_);
183	1x	detail::throw_system_error(make_err(errn), "timerfd_create");
184		}
185
186	1229x	epoll_event ev{};
187	1229x	ev.events = EPOLLIN \| EPOLLET;
188	1229x	ev.data.ptr = nullptr;
189	1229x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
190		{
191	1x	int errn = errno;
192	1x	::close(timer_fd_);
193	1x	::close(event_fd_);
194	1x	::close(epoll_fd_);
195	1x	detail::throw_system_error(make_err(errn), "epoll_ctl");
196		}
197
198	1228x	epoll_event timer_ev{};
199	1228x	timer_ev.events = EPOLLIN \| EPOLLERR;
200	1228x	timer_ev.data.ptr = &timer_fd_;
201	1228x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
202		{
203	1x	int errn = errno;
204	1x	::close(timer_fd_);
205	1x	::close(event_fd_);
206	1x	::close(epoll_fd_);
207	1x	detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
208		}
209
210	1227x	timer_svc_ = &get_timer_service(ctx, *this);
211	1227x	timer_svc_->set_on_earliest_changed(
212	5476x	timer_service::callback(this, [](void* p) {
213	4249x	auto* self = static_cast<epoll_scheduler*>(p);
214	4249x	self->timerfd_stale_.store(true, std::memory_order_release);
215	4249x	self->interrupt_reactor();
216	4249x	}));
217
218	1227x	get_resolver_service(ctx, *this);
219	1227x	get_signal_service(ctx, *this);
220	1227x	get_stream_file_service(ctx, *this);
221	1227x	get_random_access_file_service(ctx, *this);
222
223	1227x	completed_ops_.push(&task_op_);
224	1242x	}
225
226	2454x	inline epoll_scheduler::~epoll_scheduler()
227		{
228	1227x	if (timer_fd_ >= 0)
229	1227x	::close(timer_fd_);
230	1227x	if (event_fd_ >= 0)
231	1227x	::close(event_fd_);
232	1227x	if (epoll_fd_ >= 0)
233	1227x	::close(epoll_fd_);
234	2454x	}
235
236		inline void
237	1227x	epoll_scheduler::shutdown()
238		{
239	1227x	shutdown_drain();
240
241	1227x	if (event_fd_ >= 0)
242	1227x	interrupt_reactor();
243	1227x	}
244
245		inline void
246	23x	epoll_scheduler::configure_reactor(
247		unsigned max_events,
248		unsigned budget_init,
249		unsigned budget_max,
250		unsigned unassisted)
251		{
252	23x	reactor_scheduler::configure_reactor(
253		max_events, budget_init, budget_max, unassisted);
254	21x	event_buffer_.resize(max_events_per_poll_);
255	21x	}
256
257		inline std::error_code
258	5667x	epoll_scheduler::register_descriptor(
259		int fd, reactor_descriptor_state* desc) const
260		{
261	5667x	epoll_event ev{};
262	5667x	ev.events = EPOLLIN \| EPOLLOUT \| EPOLLET \| EPOLLERR \| EPOLLHUP;
263	5667x	ev.data.ptr = desc;
264
265	5667x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
266	7x	return make_err(errno);
267
268	5660x	desc->registered_events = ev.events;
269	5660x	desc->fd = fd;
270	5660x	desc->scheduler_ = this;
271	5660x	desc->mutex.set_enabled(reactor_io_locking_);
272	5660x	desc->ready_events_.store(0, std::memory_order_relaxed);
273
274	5660x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
275	5660x	desc->impl_ref_.reset();
276	5660x	desc->read_ready = false;
277	5660x	desc->write_ready = false;
278	5660x	return {};
279	5660x	}
280
281		inline void
282	5592x	epoll_scheduler::deregister_descriptor(int fd) const
283		{
284	5592x	::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
285	5592x	}
286
287		inline void
288	6644x	epoll_scheduler::interrupt_reactor() const
289		{
290	6644x	bool expected = false;
291	6644x	if (eventfd_armed_.compare_exchange_strong(
292		expected, true, std::memory_order_release,
293		std::memory_order_relaxed))
294		{
295	5175x	std::uint64_t val = 1;
296	5175x	if (::write(event_fd_, &val, sizeof(val)) < 0)
297		{
298		// The flag is what coalesces later interrupts into a byte
299		// already in the eventfd; a write that failed put no byte
300		// there, so leaving it armed would swallow every interrupt
301		// that follows. Disarming keeps the cost to the interrupts
302		// already in flight -- the next one arms and writes again,
303		// instead of every one after this coalescing into a byte
304		// that does not exist.
305	2x	eventfd_armed_.store(false, std::memory_order_release);
306		}
307		}
308	6644x	}
309
310		inline void
311	10847x	epoll_scheduler::update_timerfd() const
312		{
313	10847x	auto nearest = timer_svc_->nearest_expiry();
314
315	10847x	itimerspec ts{};
316	10847x	int flags = 0;
317
318	10847x	if (nearest == timer_service::time_point::max())
319		{
320		// No timers — disarm by setting to 0 (relative)
321		}
322		else
323		{
324	9707x	auto now = std::chrono::steady_clock::now();
325	9707x	if (nearest <= now)
326		{
327		// Use 1ns instead of 0 — zero disarms the timerfd
328	1117x	ts.it_value.tv_nsec = 1;
329		}
330		else
331		{
332	8590x	auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
333	8590x	nearest - now)
334	8590x	.count();
335	8590x	ts.it_value.tv_sec = nsec / 1000000000;
336	8590x	ts.it_value.tv_nsec = nsec % 1000000000;
337	8590x	if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
338	✗	ts.it_value.tv_nsec = 1;
339		}
340		}
341
342	10847x	if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
343	1x	detail::throw_system_error(make_err(errno), "timerfd_settime");
344	10846x	}
345
346		inline void
347	41216x	epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
348		{
349		int timeout_ms;
350	41216x	if (task_interrupted_)
351	30377x	timeout_ms = 0;
352	10839x	else if (timeout_us < 0)
353	10824x	timeout_ms = -1;
354		else
355	15x	timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
356
357	41216x	if (lock.owns_lock())
358	10841x	lock.unlock();
359
360	41216x	task_cleanup on_exit{this, &lock, ctx};
361
362		// Flush deferred timerfd programming before blocking
363	41216x	if (timerfd_stale_.exchange(false, std::memory_order_acquire))
364	3599x	update_timerfd();
365
366	41215x	int nfds = ::epoll_wait(
367	41215x	epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
368		timeout_ms);
369
370	41215x	if (nfds < 0 && errno != EINTR)
371	1x	detail::throw_system_error(make_err(errno), "epoll_wait");
372
373	41214x	bool check_timers = false;
374	41214x	ready_queue local_ops;
375
376	91237x	for (int i = 0; i < nfds; ++i)
377		{
378	50023x	if (event_buffer_[i].data.ptr == nullptr)
379		{
380		std::uint64_t val;
381		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
382	3946x	[[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
383	3946x	eventfd_armed_.store(false, std::memory_order_relaxed);
384	3946x	continue;
385	3946x	}
386
387	46077x	if (event_buffer_[i].data.ptr == &timer_fd_)
388		{
389		std::uint64_t expirations;
390		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
391		[[maybe_unused]] auto r =
392	7248x	::read(timer_fd_, &expirations, sizeof(expirations));
393	7248x	check_timers = true;
394	7248x	continue;
395	7248x	}
396
397		auto* desc =
398	38829x	static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
399	38829x	desc->add_ready_events(event_buffer_[i].events);
400
401	38829x	bool expected = false;
402	38829x	if (desc->is_enqueued_.compare_exchange_strong(
403		expected, true, std::memory_order_release,
404		std::memory_order_relaxed))
405		{
406	38829x	local_ops.push(desc);
407		}
408		}
409
410	41214x	if (check_timers)
411		{
412	7248x	timer_svc_->process_expired();
413	7248x	update_timerfd();
414		}
415
416	41214x	lock.lock();
417
418	41214x	completed_ops_.splice(local_ops);
419	41216x	}
420
421		} // namespace boost::corosio::detail
422
423		#endif // BOOST_COROSIO_HAS_EPOLL
424
425		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
426