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	76x	[[nodiscard]] std::error_code register_signal_reader(int read_fd) override
131		{
132	76x	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	1303x	inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
160	1303x	: epoll_fd_(-1)
161	1303x	, event_fd_(-1)
162	1303x	, timer_fd_(-1)
163	2606x	, event_buffer_(max_events_per_poll_)
164		{
165	1303x	epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
166	1303x	if (epoll_fd_ < 0)
167	1x	detail::throw_system_error(make_err(errno), "epoll_create1");
168
169	1302x	event_fd_ = ::eventfd(0, EFD_NONBLOCK \| EFD_CLOEXEC);
170	1302x	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	1301x	timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK \| TFD_CLOEXEC);
178	1301x	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	1300x	epoll_event ev{};
187	1300x	ev.events = EPOLLIN \| EPOLLET;
188	1300x	ev.data.ptr = nullptr;
189	1300x	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	1299x	epoll_event timer_ev{};
199	1299x	timer_ev.events = EPOLLIN \| EPOLLERR;
200	1299x	timer_ev.data.ptr = &timer_fd_;
201	1299x	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	1298x	timer_svc_ = &get_timer_service(ctx, *this);
211	1298x	timer_svc_->set_on_earliest_changed(
212	5490x	timer_service::callback(this, [](void* p) {
213	4192x	auto* self = static_cast<epoll_scheduler*>(p);
214	4192x	self->timerfd_stale_.store(true, std::memory_order_release);
215	4192x	self->interrupt_reactor();
216	4192x	}));
217
218	1298x	get_resolver_service(ctx, *this);
219	1298x	get_signal_service(ctx, *this);
220	1298x	get_stream_file_service(ctx, *this);
221	1298x	get_random_access_file_service(ctx, *this);
222
223	1298x	completed_ops_.push(&task_op_);
224	1313x	}
225
226	2596x	inline epoll_scheduler::~epoll_scheduler()
227		{
228	1298x	if (timer_fd_ >= 0)
229	1298x	::close(timer_fd_);
230	1298x	if (event_fd_ >= 0)
231	1298x	::close(event_fd_);
232	1298x	if (epoll_fd_ >= 0)
233	1298x	::close(epoll_fd_);
234	2596x	}
235
236		inline void
237	1298x	epoll_scheduler::shutdown()
238		{
239	1298x	shutdown_drain();
240
241	1298x	if (event_fd_ >= 0)
242	1298x	interrupt_reactor();
243	1298x	}
244
245		inline void
246	27x	epoll_scheduler::configure_reactor(
247		unsigned max_events,
248		unsigned budget_init,
249		unsigned budget_max,
250		unsigned unassisted)
251		{
252	27x	reactor_scheduler::configure_reactor(
253		max_events, budget_init, budget_max, unassisted);
254	25x	event_buffer_.resize(max_events_per_poll_);
255	25x	}
256
257		inline std::error_code
258	5709x	epoll_scheduler::register_descriptor(
259		int fd, reactor_descriptor_state* desc) const
260		{
261	5709x	epoll_event ev{};
262	5709x	ev.events = EPOLLIN \| EPOLLOUT \| EPOLLET \| EPOLLERR \| EPOLLHUP;
263	5709x	ev.data.ptr = desc;
264
265	5709x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
266	7x	return make_err(errno);
267
268	5702x	desc->registered_events = ev.events;
269	5702x	desc->fd = fd;
270	5702x	desc->scheduler_ = this;
271	5702x	desc->mutex.set_enabled(reactor_io_locking_);
272	5702x	desc->ready_events_.store(0, std::memory_order_relaxed);
273
274	5702x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
275	5702x	desc->impl_ref_.reset();
276	5702x	desc->read_ready = false;
277	5702x	desc->write_ready = false;
278	5702x	return {};
279	5702x	}
280
281		inline void
282	5627x	epoll_scheduler::deregister_descriptor(int fd) const
283		{
284	5627x	::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
285	5627x	}
286
287		inline void
288	7559x	epoll_scheduler::interrupt_reactor() const
289		{
290	7559x	bool expected = false;
291	7559x	if (eventfd_armed_.compare_exchange_strong(
292		expected, true, std::memory_order_release,
293		std::memory_order_relaxed))
294		{
295	6028x	std::uint64_t val = 1;
296	6028x	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	7559x	}
309
310		inline void
311	11013x	epoll_scheduler::update_timerfd() const
312		{
313	11013x	auto nearest = timer_svc_->nearest_expiry();
314
315	11013x	itimerspec ts{};
316	11013x	int flags = 0;
317
318	11013x	if (nearest == timer_service::time_point::max())
319		{
320		// No timers — disarm by setting to 0 (relative)
321		}
322		else
323		{
324	9867x	auto now = std::chrono::steady_clock::now();
325	9867x	if (nearest <= now)
326		{
327		// Use 1ns instead of 0 — zero disarms the timerfd
328	1265x	ts.it_value.tv_nsec = 1;
329		}
330		else
331		{
332	8602x	auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
333	8602x	nearest - now)
334	8602x	.count();
335	8602x	ts.it_value.tv_sec = nsec / 1000000000;
336	8602x	ts.it_value.tv_nsec = nsec % 1000000000;
337	8602x	if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
338	✗	ts.it_value.tv_nsec = 1;
339		}
340		}
341
342	11013x	if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
343	1x	detail::throw_system_error(make_err(errno), "timerfd_settime");
344	11012x	}
345
346		inline void
347	41849x	epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
348		{
349		int timeout_ms;
350	41849x	if (task_interrupted_)
351	30430x	timeout_ms = 0;
352	11419x	else if (timeout_us < 0)
353	11051x	timeout_ms = -1;
354		else
355	368x	timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
356
357	41849x	if (lock.owns_lock())
358	11421x	lock.unlock();
359
360	41849x	task_cleanup on_exit{this, &lock, ctx};
361
362		// Flush deferred timerfd programming before blocking
363	41849x	if (timerfd_stale_.exchange(false, std::memory_order_acquire))
364	3552x	update_timerfd();
365
366	41848x	int nfds = ::epoll_wait(
367	41848x	epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
368		timeout_ms);
369
370	41848x	if (nfds < 0 && errno != EINTR)
371	1x	detail::throw_system_error(make_err(errno), "epoll_wait");
372
373	41847x	bool check_timers = false;
374	41847x	ready_queue local_ops;
375
376	92178x	for (int i = 0; i < nfds; ++i)
377		{
378	50331x	if (event_buffer_[i].data.ptr == nullptr)
379		{
380		std::uint64_t val;
381		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
382	4728x	[[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
383	4728x	eventfd_armed_.store(false, std::memory_order_relaxed);
384	4728x	continue;
385	4728x	}
386
387	45603x	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	7461x	::read(timer_fd_, &expirations, sizeof(expirations));
393	7461x	check_timers = true;
394	7461x	continue;
395	7461x	}
396
397		auto* desc =
398	38142x	static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
399	38142x	desc->add_ready_events(event_buffer_[i].events);
400
401	38142x	bool expected = false;
402	38142x	if (desc->is_enqueued_.compare_exchange_strong(
403		expected, true, std::memory_order_release,
404		std::memory_order_relaxed))
405		{
406	38142x	local_ops.push(desc);
407		}
408		}
409
410	41847x	if (check_timers)
411		{
412	7461x	timer_svc_->process_expired();
413	7461x	update_timerfd();
414		}
415
416	41847x	lock.lock();
417
418	41847x	completed_ops_.splice(local_ops);
419	41849x	}
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