event/Loop: read eventfd with io_uring

This eliminates a roundtrip to epoll.
This commit is contained in:
Max Kellermann
2025-03-10 18:43:41 +01:00
parent 7797158ea6
commit 06bc373ace
2 changed files with 70 additions and 7 deletions

@ -17,6 +17,46 @@
#include "io/uring/Queue.hxx" #include "io/uring/Queue.hxx"
#endif #endif
#if defined(HAVE_THREADED_EVENT_LOOP) && defined(USE_EVENTFD) && defined(HAVE_URING)
#include <sys/eventfd.h>
/**
* Read from the eventfd using io_uring and invoke
* EventLoop::OnWake().
*/
class EventLoop::UringWake final : Uring::Operation {
EventLoop &event_loop;
eventfd_t value;
public:
explicit UringWake(EventLoop &_event_loop) noexcept
:event_loop(_event_loop) {}
void Start() {
assert(!IsUringPending());
assert(event_loop.GetUring());
auto &queue = *event_loop.GetUring();
auto &s = queue.RequireSubmitEntry();
io_uring_prep_read(&s, event_loop.wake_fd.GetSocket().Get(), &value, sizeof(value), 0);
queue.Push(s, *this);
}
private:
void OnUringCompletion(int res) noexcept override {
if (res <= 0)
return;
Start();
event_loop.OnWake();
}
};
#endif // USE_EVENTFD && HAVE_URING
EventLoop::EventLoop( EventLoop::EventLoop(
#ifdef HAVE_THREADED_EVENT_LOOP #ifdef HAVE_THREADED_EVENT_LOOP
ThreadId _thread ThreadId _thread
@ -40,6 +80,9 @@ EventLoop::~EventLoop() noexcept
/* if Run() was never called (maybe because startup failed and /* if Run() was never called (maybe because startup failed and
an exception is pending), we need to destruct the an exception is pending), we need to destruct the
Uring::Manager here or else the assertions below fail */ Uring::Manager here or else the assertions below fail */
#if defined(HAVE_THREADED_EVENT_LOOP) && defined(USE_EVENTFD)
uring_wake.reset();
#endif
uring_poll.reset(); uring_poll.reset();
uring.reset(); uring.reset();
#endif #endif
@ -403,7 +446,15 @@ EventLoop::Run() noexcept
assert(alive || quit_injected); assert(alive || quit_injected);
assert(busy); assert(busy);
wake_event.Schedule(SocketEvent::READ); #if defined(USE_EVENTFD) && defined(HAVE_URING)
if (uring) {
if (!uring_wake) {
uring_wake = std::make_unique<UringWake>(*this);
uring_wake->Start();
}
} else
#endif
wake_event.Schedule(SocketEvent::READ);
#endif #endif
#ifdef HAVE_THREADED_EVENT_LOOP #ifdef HAVE_THREADED_EVENT_LOOP
@ -537,13 +588,9 @@ EventLoop::HandleInject() noexcept
} }
} }
void inline void
EventLoop::OnSocketReady([[maybe_unused]] unsigned flags) noexcept EventLoop::OnWake() noexcept
{ {
assert(IsInside());
wake_fd.Read();
if (quit_injected) { if (quit_injected) {
Break(); Break();
return; return;
@ -553,4 +600,14 @@ EventLoop::OnSocketReady([[maybe_unused]] unsigned flags) noexcept
HandleInject(); HandleInject();
} }
void
EventLoop::OnSocketReady([[maybe_unused]] unsigned flags) noexcept
{
assert(IsInside());
wake_fd.Read();
OnWake();
}
#endif #endif

@ -107,6 +107,11 @@ class EventLoop final
#endif // HAVE_URING #endif // HAVE_URING
#ifdef HAVE_THREADED_EVENT_LOOP #ifdef HAVE_THREADED_EVENT_LOOP
#if defined(USE_EVENTFD) && defined(HAVE_URING)
class UringWake;
std::unique_ptr<UringWake> uring_wake;
#endif
/** /**
* A reference to the thread that is currently inside Run(). * A reference to the thread that is currently inside Run().
*/ */
@ -339,6 +344,7 @@ private:
void Wait(Event::Duration timeout) noexcept; void Wait(Event::Duration timeout) noexcept;
#ifdef HAVE_THREADED_EVENT_LOOP #ifdef HAVE_THREADED_EVENT_LOOP
void OnWake() noexcept;
void OnSocketReady(unsigned flags) noexcept; void OnSocketReady(unsigned flags) noexcept;
#endif #endif