Skip to content

Commit

Permalink
event/Loop: read eventfd with io_uring
Browse files Browse the repository at this point in the history
This eliminates a roundtrip to epoll.
  • Loading branch information
MaxKellermann committed Mar 10, 2025
1 parent 7797158 commit 06bc373
Show file tree
Hide file tree
Showing 2 changed files with 70 additions and 7 deletions.
71 changes: 64 additions & 7 deletions src/event/Loop.cxx
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,46 @@
#include "io/uring/Queue.hxx"
#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(
#ifdef HAVE_THREADED_EVENT_LOOP
ThreadId _thread
Expand All @@ -40,6 +80,9 @@ EventLoop::~EventLoop() noexcept
/* if Run() was never called (maybe because startup failed and
an exception is pending), we need to destruct the
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.reset();
#endif
Expand Down Expand Up @@ -403,7 +446,15 @@ EventLoop::Run() noexcept
assert(alive || quit_injected);
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

#ifdef HAVE_THREADED_EVENT_LOOP
Expand Down Expand Up @@ -537,13 +588,9 @@ EventLoop::HandleInject() noexcept
}
}

void
EventLoop::OnSocketReady([[maybe_unused]] unsigned flags) noexcept
inline void
EventLoop::OnWake() noexcept
{
assert(IsInside());

wake_fd.Read();

if (quit_injected) {
Break();
return;
Expand All @@ -553,4 +600,14 @@ EventLoop::OnSocketReady([[maybe_unused]] unsigned flags) noexcept
HandleInject();
}

void
EventLoop::OnSocketReady([[maybe_unused]] unsigned flags) noexcept
{
assert(IsInside());

wake_fd.Read();

OnWake();
}

#endif
6 changes: 6 additions & 0 deletions src/event/Loop.hxx
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,11 @@ class EventLoop final
#endif // HAVE_URING

#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().
*/
Expand Down Expand Up @@ -339,6 +344,7 @@ private:
void Wait(Event::Duration timeout) noexcept;

#ifdef HAVE_THREADED_EVENT_LOOP
void OnWake() noexcept;
void OnSocketReady(unsigned flags) noexcept;
#endif

Expand Down

0 comments on commit 06bc373

Please sign in to comment.