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

99.4% Lines (162 / 163) 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	5594x	ensure_write_registered(int, reactor_descriptor_state*) const noexcept
122		{
123	5594x	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	284x	[[nodiscard]] std::error_code register_signal_reader(int read_fd) override
134		{
135	284x	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	2473x	inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
163	2473x	: epoll_fd_(-1)
164	2473x	, event_fd_(-1)
165	2473x	, timer_fd_(-1)
166	4946x	, event_buffer_(max_events_per_poll_)
167		{
168	2473x	epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
169	2473x	if (epoll_fd_ < 0)
170	1x	detail::throw_system_error(make_err(errno), "epoll_create1");
171
172	2472x	event_fd_ = ::eventfd(0, EFD_NONBLOCK \| EFD_CLOEXEC);
173	2472x	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	2471x	timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK \| TFD_CLOEXEC);
181	2471x	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	2470x	epoll_event ev{};
190	2470x	ev.events = EPOLLIN \| EPOLLET;
191	2470x	ev.data.ptr = nullptr;
192	2470x	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	2469x	epoll_event timer_ev{};
202	2469x	timer_ev.events = EPOLLIN \| EPOLLERR;
203	2469x	timer_ev.data.ptr = &timer_fd_;
204	2469x	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	2468x	timer_svc_ = &get_timer_service(ctx, *this);
214	2468x	timer_svc_->set_on_earliest_changed(
215	7323x	timer_service::callback(this, [](void* p) {
216	4855x	auto* self = static_cast<epoll_scheduler*>(p);
217	4855x	self->timerfd_stale_.store(true, std::memory_order_release);
218	4855x	self->interrupt_reactor();
219	4855x	}));
220
221	2468x	completed_ops_.push(&task_op_);
222	2483x	}
223
224	4936x	inline epoll_scheduler::~epoll_scheduler()
225		{
226	2468x	if (timer_fd_ >= 0)
227	2468x	::close(timer_fd_);
228	2468x	if (event_fd_ >= 0)
229	2468x	::close(event_fd_);
230	2468x	if (epoll_fd_ >= 0)
231	2468x	::close(epoll_fd_);
232	4936x	}
233
234		inline void
235	2468x	epoll_scheduler::shutdown()
236		{
237	2468x	shutdown_drain();
238
239	2468x	if (event_fd_ >= 0)
240	2468x	interrupt_reactor();
241	2468x	}
242
243		inline void
244	32x	epoll_scheduler::configure_reactor(
245		unsigned max_events,
246		unsigned budget_init,
247		unsigned budget_max,
248		unsigned unassisted)
249		{
250	32x	reactor_scheduler::configure_reactor(
251		max_events, budget_init, budget_max, unassisted);
252	30x	event_buffer_.resize(max_events_per_poll_);
253	30x	}
254
255		inline std::error_code
256	13553x	epoll_scheduler::register_descriptor(
257		int fd, reactor_descriptor_state* desc) const
258		{
259	13553x	epoll_event ev{};
260	13553x	ev.events = EPOLLIN \| EPOLLOUT \| EPOLLET \| EPOLLERR \| EPOLLHUP;
261	13553x	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	13553x	if (desc->mutex.enabled() != reactor_io_locking_)
269	3x	desc->mutex.set_enabled(reactor_io_locking_);
270		{
271	13553x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
272	13553x	desc->registered_events = ev.events;
273	13553x	desc->unpollable = false;
274	13553x	desc->fd = fd;
275	13553x	desc->scheduler_ = this;
276	13553x	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	13553x	desc->read_ready = false;
280	13553x	desc->write_ready = false;
281	13553x	}
282
283	13553x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
284		{
285	12x	int const err = errno;
286	12x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
287	12x	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	12x	if (err != EPERM)
291	7x	return make_err(err);
292	5x	desc->unpollable = true;
293	12x	}
294	13546x	return {};
295		}
296
297		inline void
298	13258x	epoll_scheduler::deregister_descriptor(int fd) const
299		{
300	13258x	::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
301	13258x	}
302
303		inline void
304	13186x	epoll_scheduler::interrupt_reactor() const
305		{
306	13186x	bool expected = false;
307	13186x	if (eventfd_armed_.compare_exchange_strong(
308		expected, true, std::memory_order_release,
309		std::memory_order_relaxed))
310		{
311	10919x	std::uint64_t val = 1;
312	10919x	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	2x	eventfd_armed_.store(false, std::memory_order_release);
322		}
323		}
324	13186x	}
325
326		inline void
327	12259x	epoll_scheduler::update_timerfd() const
328		{
329	12259x	auto nearest = timer_svc_->nearest_expiry();
330
331	12259x	itimerspec ts{};
332	12259x	int flags = 0;
333
334	12259x	if (nearest == timer_service::time_point::max())
335		{
336		// No timers — disarm by setting to 0 (relative)
337		}
338		else
339		{
340	10952x	auto now = std::chrono::steady_clock::now();
341	10952x	if (nearest <= now)
342		{
343		// Use 1ns instead of 0 — zero disarms the timerfd
344	733x	ts.it_value.tv_nsec = 1;
345		}
346		else
347		{
348	10219x	auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
349	10219x	nearest - now)
350	10219x	.count();
351	10219x	ts.it_value.tv_sec = nsec / 1000000000;
352	10219x	ts.it_value.tv_nsec = nsec % 1000000000;
353	10219x	if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
354	✗	ts.it_value.tv_nsec = 1;
355		}
356		}
357
358	12259x	if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
359	1x	detail::throw_system_error(make_err(errno), "timerfd_settime");
360	12258x	}
361
362		inline void
363	154989x	epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
364		{
365		int timeout_ms;
366	154989x	if (task_interrupted_)
367	136446x	timeout_ms = 0;
368	18543x	else if (timeout_us < 0)
369	17900x	timeout_ms = -1;
370		else
371	643x	timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
372
373	154989x	if (lock.owns_lock())
374	18547x	lock.unlock();
375
376	154989x	task_cleanup on_exit{this, &lock, ctx};
377
378		// Flush deferred timerfd programming before blocking
379	154989x	if (timerfd_stale_.exchange(false, std::memory_order_acquire))
380	4098x	update_timerfd();
381
382	154988x	int nfds = ::epoll_wait(
383	154988x	epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
384		timeout_ms);
385
386	154988x	if (nfds < 0 && errno != EINTR)
387	1x	detail::throw_system_error(make_err(errno), "epoll_wait");
388
389	154987x	bool check_timers = false;
390	154987x	ready_queue local_ops;
391
392	219689x	for (int i = 0; i < nfds; ++i)
393		{
394	64702x	if (event_buffer_[i].data.ptr == nullptr)
395		{
396		std::uint64_t val;
397		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
398	8449x	[[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
399	8449x	eventfd_armed_.store(false, std::memory_order_relaxed);
400	8449x	continue;
401	8449x	}
402
403	56253x	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	8161x	::read(timer_fd_, &expirations, sizeof(expirations));
409	8161x	check_timers = true;
410	8161x	continue;
411	8161x	}
412
413		auto* desc =
414	48092x	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	48092x	std::uint32_t ev = event_buffer_[i].events;
430	48092x	if ((ev & (EPOLLHUP \| EPOLLOUT)) == EPOLLHUP)
431	8x	ev \|= EPOLLIN \| EPOLLOUT \| EPOLLERR;
432	48084x	else if (ev & EPOLLHUP)
433	3094x	ev \|= EPOLLIN \| EPOLLOUT;
434	48092x	desc->add_ready_events(ev);
435
436	48092x	bool expected = false;
437	48092x	if (desc->is_enqueued_.compare_exchange_strong(
438		expected, true, std::memory_order_release,
439		std::memory_order_relaxed))
440		{
441	48086x	local_ops.push(desc);
442		}
443		}
444
445	154987x	if (check_timers)
446		{
447	8161x	timer_svc_->process_expired();
448	8161x	update_timerfd();
449		}
450
451	154987x	lock.lock();
452
453	154987x	completed_ops_.splice(local_ops);
454	154989x	}
455
456		} // namespace boost::corosio::detail
457
458		#endif // BOOST_COROSIO_HAS_EPOLL
459
460		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
461