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

99.4% Lines (159 / 160) 100.0% Functions (12 / 12)

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

epoll_scheduler.hpp

f(x) Functions (12)

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 <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	2626x	ensure_write_registered(int, reactor_descriptor_state*) const noexcept
122		{
123	2626x	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	76x	[[nodiscard]] std::error_code register_signal_reader(int read_fd) override
134		{
135	76x	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	1577x	inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
163	1577x	: epoll_fd_(-1)
164	1577x	, event_fd_(-1)
165	1577x	, timer_fd_(-1)
166	3154x	, event_buffer_(max_events_per_poll_)
167		{
168	1577x	epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
169	1577x	if (epoll_fd_ < 0)
170	1x	detail::throw_system_error(make_err(errno), "epoll_create1");
171
172	1576x	event_fd_ = ::eventfd(0, EFD_NONBLOCK \| EFD_CLOEXEC);
173	1576x	if (event_fd_ < 0)
174		{
175	1x	int errn = errno;
176	1x	::close(epoll_fd_);
177	1x	detail::throw_system_error(make_err(errn), "eventfd");
178		}
179
180	1575x	timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK \| TFD_CLOEXEC);
181	1575x	if (timer_fd_ < 0)
182		{
183	1x	int errn = errno;
184	1x	::close(event_fd_);
185	1x	::close(epoll_fd_);
186	1x	detail::throw_system_error(make_err(errn), "timerfd_create");
187		}
188
189	1574x	epoll_event ev{};
190	1574x	ev.events = EPOLLIN \| EPOLLET;
191	1574x	ev.data.ptr = nullptr;
192	1574x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
193		{
194	1x	int errn = errno;
195	1x	::close(timer_fd_);
196	1x	::close(event_fd_);
197	1x	::close(epoll_fd_);
198	1x	detail::throw_system_error(make_err(errn), "epoll_ctl");
199		}
200
201	1573x	epoll_event timer_ev{};
202	1573x	timer_ev.events = EPOLLIN \| EPOLLERR;
203	1573x	timer_ev.data.ptr = &timer_fd_;
204	1573x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
205		{
206	1x	int errn = errno;
207	1x	::close(timer_fd_);
208	1x	::close(event_fd_);
209	1x	::close(epoll_fd_);
210	1x	detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
211		}
212
213	1572x	timer_svc_ = &get_timer_service(ctx, *this);
214	1572x	timer_svc_->set_on_earliest_changed(
215	5976x	timer_service::callback(this, [](void* p) {
216	4404x	auto* self = static_cast<epoll_scheduler*>(p);
217	4404x	self->timerfd_stale_.store(true, std::memory_order_release);
218	4404x	self->interrupt_reactor();
219	4404x	}));
220
221	1572x	completed_ops_.push(&task_op_);
222	1587x	}
223
224	3144x	inline epoll_scheduler::~epoll_scheduler()
225		{
226	1572x	if (timer_fd_ >= 0)
227	1572x	::close(timer_fd_);
228	1572x	if (event_fd_ >= 0)
229	1572x	::close(event_fd_);
230	1572x	if (epoll_fd_ >= 0)
231	1572x	::close(epoll_fd_);
232	3144x	}
233
234		inline void
235	1572x	epoll_scheduler::shutdown()
236		{
237	1572x	shutdown_drain();
238
239	1572x	if (event_fd_ >= 0)
240	1572x	interrupt_reactor();
241	1572x	}
242
243		inline void
244	28x	epoll_scheduler::configure_reactor(
245		unsigned max_events,
246		unsigned budget_init,
247		unsigned budget_max,
248		unsigned unassisted)
249		{
250	28x	reactor_scheduler::configure_reactor(
251		max_events, budget_init, budget_max, unassisted);
252	26x	event_buffer_.resize(max_events_per_poll_);
253	26x	}
254
255		inline std::error_code
256	5939x	epoll_scheduler::register_descriptor(
257		int fd, reactor_descriptor_state* desc) const
258		{
259	5939x	epoll_event ev{};
260	5939x	ev.events = EPOLLIN \| EPOLLOUT \| EPOLLET \| EPOLLERR \| EPOLLHUP;
261	5939x	ev.data.ptr = desc;
262
263	5939x	bool unpollable = false;
264	5939x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
265		{
266		// EPERM: a file type epoll cannot watch. Its I/O does not
267		// block, so adopt it unwatched, as asio does.
268	11x	if (errno != EPERM)
269	7x	return make_err(errno);
270	4x	unpollable = true;
271		}
272
273	5932x	desc->registered_events = unpollable ? 0 : ev.events;
274	5932x	desc->unpollable = unpollable;
275	5932x	desc->fd = fd;
276	5932x	desc->scheduler_ = this;
277	5932x	desc->mutex.set_enabled(reactor_io_locking_);
278	5932x	desc->ready_events_.store(0, std::memory_order_relaxed);
279
280	5932x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
281	5932x	desc->impl_ref_.reset();
282	5932x	desc->read_ready = false;
283	5932x	desc->write_ready = false;
284	5932x	return {};
285	5932x	}
286
287		inline void
288	5853x	epoll_scheduler::deregister_descriptor(int fd) const
289		{
290	5853x	::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
291	5853x	}
292
293		inline void
294	8636x	epoll_scheduler::interrupt_reactor() const
295		{
296	8636x	bool expected = false;
297	8636x	if (eventfd_armed_.compare_exchange_strong(
298		expected, true, std::memory_order_release,
299		std::memory_order_relaxed))
300		{
301	6887x	std::uint64_t val = 1;
302	6887x	if (::write(event_fd_, &val, sizeof(val)) < 0)
303		{
304		// The flag is what coalesces later interrupts into a byte
305		// already in the eventfd; a write that failed put no byte
306		// there, so leaving it armed would swallow every interrupt
307		// that follows. Disarming keeps the cost to the interrupts
308		// already in flight -- the next one arms and writes again,
309		// instead of every one after this coalescing into a byte
310		// that does not exist.
311	2x	eventfd_armed_.store(false, std::memory_order_release);
312		}
313		}
314	8636x	}
315
316		inline void
317	10892x	epoll_scheduler::update_timerfd() const
318		{
319	10892x	auto nearest = timer_svc_->nearest_expiry();
320
321	10892x	itimerspec ts{};
322	10892x	int flags = 0;
323
324	10892x	if (nearest == timer_service::time_point::max())
325		{
326		// No timers — disarm by setting to 0 (relative)
327		}
328		else
329		{
330	9631x	auto now = std::chrono::steady_clock::now();
331	9631x	if (nearest <= now)
332		{
333		// Use 1ns instead of 0 — zero disarms the timerfd
334	971x	ts.it_value.tv_nsec = 1;
335		}
336		else
337		{
338	8660x	auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
339	8660x	nearest - now)
340	8660x	.count();
341	8660x	ts.it_value.tv_sec = nsec / 1000000000;
342	8660x	ts.it_value.tv_nsec = nsec % 1000000000;
343	8660x	if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
344	✗	ts.it_value.tv_nsec = 1;
345		}
346		}
347
348	10892x	if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
349	1x	detail::throw_system_error(make_err(errno), "timerfd_settime");
350	10891x	}
351
352		inline void
353	41337x	epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
354		{
355		int timeout_ms;
356	41337x	if (task_interrupted_)
357	29469x	timeout_ms = 0;
358	11868x	else if (timeout_us < 0)
359	11501x	timeout_ms = -1;
360		else
361	367x	timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
362
363	41337x	if (lock.owns_lock())
364	11870x	lock.unlock();
365
366	41337x	task_cleanup on_exit{this, &lock, ctx};
367
368		// Flush deferred timerfd programming before blocking
369	41337x	if (timerfd_stale_.exchange(false, std::memory_order_acquire))
370	3799x	update_timerfd();
371
372	41336x	int nfds = ::epoll_wait(
373	41336x	epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
374		timeout_ms);
375
376	41336x	if (nfds < 0 && errno != EINTR)
377	1x	detail::throw_system_error(make_err(errno), "epoll_wait");
378
379	41335x	bool check_timers = false;
380	41335x	ready_queue local_ops;
381
382	89874x	for (int i = 0; i < nfds; ++i)
383		{
384	48539x	if (event_buffer_[i].data.ptr == nullptr)
385		{
386		std::uint64_t val;
387		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
388	5313x	[[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
389	5313x	eventfd_armed_.store(false, std::memory_order_relaxed);
390	5313x	continue;
391	5313x	}
392
393	43226x	if (event_buffer_[i].data.ptr == &timer_fd_)
394		{
395		std::uint64_t expirations;
396		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
397		[[maybe_unused]] auto r =
398	7093x	::read(timer_fd_, &expirations, sizeof(expirations));
399	7093x	check_timers = true;
400	7093x	continue;
401	7093x	}
402
403		auto* desc =
404	36133x	static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
405
406		// EPOLLHUP maps to no reactor_event_* bit, so a HUP the widening
407		// below did not translate would take no branch in
408		// invoke_deferred_io() and, the registration being
409		// edge-triggered, never get another chance.
410		//
411		// HUP without OUT means a non-socket: a pipe or tty whose peer
412		// closed, with or without data still unread (HUP alone, or
413		// IN\|HUP). Treat it as readable, writable and faulted, the way
414		// poll() and io_uring report it, so a parked error wait
415		// completes too. Sockets carry at least OUT with HUP (a fresh
416		// unconnected stream socket reports HUP\|OUT), so they take the
417		// plain widening, where forcing IN costs at most one spurious
418		// EAGAIN.
419	36133x	std::uint32_t ev = event_buffer_[i].events;
420	36133x	if ((ev & (EPOLLHUP \| EPOLLOUT)) == EPOLLHUP)
421	8x	ev \|= EPOLLIN \| EPOLLOUT \| EPOLLERR;
422	36125x	else if (ev & EPOLLHUP)
423	234x	ev \|= EPOLLIN \| EPOLLOUT;
424	36133x	desc->add_ready_events(ev);
425
426	36133x	bool expected = false;
427	36133x	if (desc->is_enqueued_.compare_exchange_strong(
428		expected, true, std::memory_order_release,
429		std::memory_order_relaxed))
430		{
431	36133x	local_ops.push(desc);
432		}
433		}
434
435	41335x	if (check_timers)
436		{
437	7093x	timer_svc_->process_expired();
438	7093x	update_timerfd();
439		}
440
441	41335x	lock.lock();
442
443	41335x	completed_ops_.splice(local_ops);
444	41337x	}
445
446		} // namespace boost::corosio::detail
447
448		#endif // BOOST_COROSIO_HAS_EPOLL
449
450		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
451