From 92745e9c1ae18f282614ad76293ac87b6231cd91 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Sun, 2 Aug 2026 12:35:40 +0000 Subject: [PATCH 01/14] feat: wake io_uring workers through eventfd --- src/brpc/socket.cpp | 3 + src/bthread/ring_listener.cpp | 177 ++++++++++++++++++++++++++++++---- src/bthread/ring_listener.h | 33 ++++--- src/bthread/task_control.cpp | 13 +++ src/bthread/task_group.cpp | 50 +++++++++- 5 files changed, 243 insertions(+), 33 deletions(-) diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp index b2573b94..67f2eb96 100644 --- a/src/brpc/socket.cpp +++ b/src/brpc/socket.cpp @@ -64,6 +64,9 @@ DEFINE_bool(dispatch_lazily, false, "dispatcher lazily creates task"); #ifdef IO_URING_ENABLED DEFINE_bool(use_io_uring, false, "Use IO URING to do the polling."); +DEFINE_bool(brpc_use_event_fd_wakeup, false, + "Wake up idle brpc workers through eventfd polled by io_uring. " + "Requires use_io_uring and is fixed at startup."); #endif namespace bthread { diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index cd6799af..9f545b56 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -21,8 +21,11 @@ #include #include #include +#include #include #include +#include +#include #include @@ -43,19 +46,39 @@ DEFINE_int32(io_uring_registered_files, 1024, "inbound listener"); DEFINE_int32(io_uring_write_buffer_pool_size, 1024, "Number of buffers kept in the io_uring-based write buffer pool."); +DECLARE_bool(brpc_use_event_fd_wakeup); -RingListener::~RingListener() { - for (auto [fd, fd_idx]: reg_fds_) { - SocketUnRegisterData data; - data.fd_ = fd; - SubmitCancel(&data); - // Not wait here because the worker should have quit already. +void RingListener::Close() { + if (ring_init_) { + io_uring_queue_exit(&ring_); + ring_init_ = false; } - SubmitAll(); - poll_status_.store(PollStatus::Closed, std::memory_order_release); { - std::unique_lock lk(mux_); - cv_.notify_one(); + if (wakeup_event_fd_ >= 0) { + close(wakeup_event_fd_); + wakeup_event_fd_ = -1; + } + + if (in_buf_) { + free(in_buf_); + in_buf_ = nullptr; + } +} + +RingListener::~RingListener() { + if (!FLAGS_brpc_use_event_fd_wakeup) { + for (auto [fd, fd_idx]: reg_fds_) { + SocketUnRegisterData data; + data.fd_ = fd; + SubmitCancel(&data); + // Not wait here because the worker should have quit already. + } + SubmitAll(); + + poll_status_.store(PollStatus::Closed, std::memory_order_release); { + std::unique_lock lk(mux_); + cv_.notify_one(); + } } if (poll_thd_.joinable()) { @@ -112,7 +135,11 @@ int RingListener::Init() { const unsigned write_buf_slots = static_cast(flag_write_buffers); - int ret = io_uring_queue_init(queue_entries, &ring_, IORING_SETUP_SINGLE_ISSUER); + unsigned ring_flags = IORING_SETUP_SINGLE_ISSUER; + if (FLAGS_brpc_use_event_fd_wakeup) { + ring_flags |= IORING_SETUP_DEFER_TASKRUN | IORING_SETUP_TASKRUN_FLAG; + } + int ret = io_uring_queue_init(queue_entries, &ring_, ring_flags); if (ret < 0) { LOG(WARNING) << "Failed to initialize the IO uring of the inbound " @@ -206,17 +233,39 @@ int RingListener::Init() { std::make_unique(write_buf_slots, &ring_); if (write_buf_pool_->buf_pool_.empty()) { + Close(); return -1; } poll_status_.store(PollStatus::Sleep, std::memory_order_release); - poll_thd_ = std::thread([&]() { - std::string ring_listener = "ring_listener:"; - ring_listener.append(std::to_string(task_group_->group_id_)); - butil::PlatformThread::SetName(ring_listener.c_str()); + if (FLAGS_brpc_use_event_fd_wakeup) { + wakeup_event_fd_ = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); + if (wakeup_event_fd_ < 0) { + const int saved_errno = errno; + LOG(ERROR) << "Failed to create the brpc worker wakeup eventfd, errno: " + << saved_errno << " (" << strerror(saved_errno) << ")"; + Close(); + return -saved_errno; + } + ret = ArmEventFdPoll(); + if (ret != 0) { + Close(); + return ret; + } + ret = SubmitAll(); + if (ret < 0) { + Close(); + return ret; + } + } else { + poll_thd_ = std::thread([&]() { + std::string ring_listener = "ring_listener:"; + ring_listener.append(std::to_string(task_group_->group_id_)); + butil::PlatformThread::SetName(ring_listener.c_str()); - Run(); - }); + Run(); + }); + } return 0; } @@ -443,6 +492,83 @@ int RingListener::SubmitAll() { return ret; } +int RingListener::ArmEventFdPoll() { + io_uring_sqe *sqe = io_uring_get_sqe(&ring_); + if (sqe == nullptr) { + LOG(ERROR) << "Failed to get an SQE for the brpc worker wakeup eventfd"; + return -EAGAIN; + } + + io_uring_prep_poll_multishot(sqe, wakeup_event_fd_, POLLIN); + io_uring_sqe_set_data64(sqe, OpCodeToInt(OpCode::SchedulerWakeup)); + ++submit_cnt_; + return 0; +} + +void RingListener::DrainEventFd() { + uint64_t value = 0; + while (true) { + const ssize_t nread = read(wakeup_event_fd_, &value, sizeof(value)); + if (nread == static_cast(sizeof(value))) { + return; + } + if (nread < 0 && errno == EINTR) { + continue; + } + // EAGAIN means another notification consumer already drained the + // counter. There is only one consumer today, but treating it as + // drained keeps this helper safe if that implementation changes. + if (nread < 0 && errno == EAGAIN) { + return; + } + const int saved_errno = errno; + LOG(FATAL) << "Failed to drain the brpc worker wakeup eventfd, errno: " + << saved_errno << " (" << strerror(saved_errno) << ")"; + } +} + +void RingListener::NotifyEventFd() { + const uint64_t one = 1; + while (true) { + const ssize_t nwritten = write(wakeup_event_fd_, &one, sizeof(one)); + if (nwritten == static_cast(sizeof(one))) { + return; + } + if (nwritten < 0 && errno == EINTR) { + continue; + } + // A saturated eventfd is already readable, so the outstanding poll is + // sufficient to wake the worker. Notifications are also coalesced by + // TaskGroup::_notified, making this path exceptional. + if (nwritten < 0 && errno == EAGAIN) { + return; + } + const int saved_errno = errno; + LOG(FATAL) << "Failed to notify the brpc worker wakeup eventfd, errno: " + << saved_errno << " (" << strerror(saved_errno) << ")"; + } +} + +int RingListener::WaitForCqe() { + int ret; + do { + ret = io_uring_submit_and_wait(&ring_, 1); + } while (ret == -EAGAIN); + + if (ret >= 0) { + submit_cnt_ = submit_cnt_ >= ret ? submit_cnt_ - ret : 0; + cqe_ready_.store(true, std::memory_order_relaxed); + return 0; + } + // TaskControl interrupts worker pthreads during shutdown. Returning on + // EINTR lets the scheduler observe the stopped parking-lot state. + if (ret == -EINTR) { + return ret; + } + LOG(FATAL) << "Failed while waiting on the brpc worker io_uring, ret: " << ret; + return ret; +} + void RingListener::PollAndNotify() { io_uring_cqe *cqe = nullptr; while (true) { @@ -505,6 +631,9 @@ size_t RingListener::ExtPoll() { } void RingListener::ExtWakeup() { + if (FLAGS_brpc_use_event_fd_wakeup) { + return; + } has_external_.store(false, std::memory_order_relaxed); if (poll_status_.load(std::memory_order_relaxed) != PollStatus::Sleep) { return; @@ -711,6 +840,20 @@ void RingListener::HandleCqe(io_uring_cqe *cqe) { fsync_data->Notify(res); break; } + case OpCode::SchedulerWakeup: { + if (cqe->res < 0) { + LOG(FATAL) << "The brpc worker wakeup poll failed, ret: " + << cqe->res; + } + DrainEventFd(); + // A multishot poll stays armed after each readiness event. Losing + // IORING_CQE_F_MORE means the scheduler can no longer wake this + // worker, so fail instead of allowing a future permanent sleep. + if (!(cqe->flags & IORING_CQE_F_MORE)) { + LOG(FATAL) << "The brpc worker multishot wakeup poll terminated"; + } + break; + } default: break; } diff --git a/src/bthread/ring_listener.h b/src/bthread/ring_listener.h index 63b0c6af..837e0a79 100644 --- a/src/bthread/ring_listener.h +++ b/src/bthread/ring_listener.h @@ -128,17 +128,7 @@ class RingListener { int Init(); - void Close() { - if (ring_init_) { - io_uring_queue_exit(&ring_); - ring_init_ = false; - } - - if (in_buf_) { - free(in_buf_); - in_buf_ = nullptr; - } - } + void Close(); int Register(SocketRegisterData *data); @@ -173,6 +163,16 @@ class RingListener { void ExtWakeup(); + // Wakes a worker blocked in WaitForCqe through the poll request registered + // on wakeup_event_fd_. This is only used when + // FLAGS_brpc_use_event_fd_wakeup was enabled at startup. + void NotifyEventFd(); + + // Blocks the owning worker until this ring has at least one completion. + // DEFER_TASKRUN requires every io_uring_enter call to come from the ring's + // single issuer, so the legacy polling thread must never call this method. + int WaitForCqe(); + void Run(); void RecycleReadBuf(uint16_t bid, size_t bytes); @@ -209,6 +209,7 @@ class RingListener { NonFixedWriteFinish, WaitingNonFixedWrite, Fsync, + SchedulerWakeup, Noop = 255 }; @@ -232,6 +233,8 @@ class RingListener { return 7; case OpCode::Fsync: return 8; + case OpCode::SchedulerWakeup: + return 9; default: return UINT8_MAX; } @@ -257,6 +260,8 @@ class RingListener { return OpCode::WaitingNonFixedWrite; case 8: return OpCode::Fsync; + case 9: + return OpCode::SchedulerWakeup; default: return OpCode::Noop; } @@ -270,6 +275,11 @@ class RingListener { void RecycleReturnedWriteBufs(); + // Installs the persistent multishot poll used for scheduler wakeups. + int ArmEventFdPoll(); + + void DrainEventFd(); + enum struct PollStatus : uint8_t { Active = 0, Sleep, ExtPoll, Closed }; struct io_uring ring_; @@ -282,6 +292,7 @@ class RingListener { std::mutex mux_; std::condition_variable cv_; std::thread poll_thd_; + int wakeup_event_fd_{-1}; io_uring_buf_ring *in_buf_ring_{nullptr}; char *in_buf_{nullptr}; diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp index cede3e76..4db45e69 100644 --- a/src/bthread/task_control.cpp +++ b/src/bthread/task_control.cpp @@ -41,6 +41,9 @@ DEFINE_int32(task_group_runqueue_capacity, 4096, DEFINE_int32(task_group_yield_before_idle, 0, "TaskGroup yields so many times before idle"); DECLARE_bool(use_io_uring); +#ifdef IO_URING_ENABLED +DECLARE_bool(brpc_use_event_fd_wakeup); +#endif namespace bthread { @@ -280,6 +283,16 @@ void TaskControl::stop_and_join() { for (int i = 0; i < _parking_lot_num; ++i) { _pl[i].stop(); } +#ifdef IO_URING_ENABLED + if (FLAGS_brpc_use_event_fd_wakeup) { + // Workers in this mode wait in io_uring rather than on the parking + // lot. Reuse the scheduler's normal eventfd notification so shutdown + // does not depend on a signal interrupting io_uring_enter. + for (int i = 0; i < _parking_lot_num; ++i) { + _groups[i]->Notify(); + } + } +#endif // Interrupt blocking operations. for (size_t i = 0; i < _workers.size(); ++i) { interrupt_pthread(_workers[i]); diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index bfdf7695..8e43a921 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -52,6 +52,9 @@ std::atomic registered_module_version; DEFINE_int32(steal_task_rnd, 100, "Steal task frequency in wait_task"); DEFINE_bool(brpc_worker_as_ext_processor, false, "Work as external processor"); DECLARE_bool(use_io_uring); +#ifdef IO_URING_ENABLED +DECLARE_bool(brpc_use_event_fd_wakeup); +#endif namespace bthread { @@ -328,6 +331,10 @@ int TaskGroup::init(size_t runqueue_capacity) { _main_stack = stk; _last_run_ns = butil::cpuwide_time_ns(); #ifdef IO_URING_ENABLED + if (FLAGS_brpc_use_event_fd_wakeup && !FLAGS_use_io_uring) { + LOG(FATAL) << "brpc_use_event_fd_wakeup requires use_io_uring"; + return -1; + } if (FLAGS_use_io_uring) { ring_listener_ = std::make_unique(this); int ret = ring_listener_->Init(); @@ -1221,6 +1228,12 @@ void TaskGroup::Notify() { bool expect = false; // Only one caller gets the right to notify the worker. if (_notified.compare_exchange_strong(expect, true)) { +#ifdef IO_URING_ENABLED + if (FLAGS_brpc_use_event_fd_wakeup) { + ring_listener_->NotifyEventFd(); + return; + } +#endif std::unique_lock lk(_mux); _notified.store(true, std::memory_order_release); _cv.notify_one(); @@ -1233,6 +1246,12 @@ bool TaskGroup::NotifyIfWaiting() { bool expect = false; // Only one caller gets the right to notify the worker. if (_notified.compare_exchange_strong(expect, true)) { +#ifdef IO_URING_ENABLED + if (FLAGS_brpc_use_event_fd_wakeup) { + ring_listener_->NotifyEventFd(); + return true; + } +#endif std::unique_lock lk(_mux); _notified.store(true, std::memory_order_release); _cv.notify_one(); @@ -1246,10 +1265,7 @@ bool TaskGroup::Wait(){ _waiting.store(true, std::memory_order_release); _waiting_workers.fetch_add(1, std::memory_order_relaxed); - std::unique_lock lk(_mux); - // Before waiting and sleeping, reset the _notified status. - _notified.store(false, std::memory_order_release); - _cv.wait(lk, [this]()->bool { + const auto has_work = [this]()->bool { // Clear the _notified status every time the worker wakes up. _notified.store(false, std::memory_order_release); // No need to check _rq since _rq can only be pushed by itself. @@ -1267,7 +1283,31 @@ bool TaskGroup::Wait(){ // Check any module registered or deleted before checking modules' tasks. CheckAndUpdateModules(); return HasTasks(); - }); + }; + +#ifdef IO_URING_ENABLED + if (FLAGS_brpc_use_event_fd_wakeup) { + // has_work() clears _notified. If a producer races before or after that + // check, eventfd retains the wakeup until submit_and_wait observes it. + // A stop may happen immediately before this worker tries to sleep. A + // worker that is already blocked is woken through eventfd by + // TaskControl::stop_and_join(). +#ifndef BTHREAD_DONT_SAVE_PARKING_STATE + _last_pl_state = _pl->get_state(); + const bool stopped = _last_pl_state.stopped(); +#else + const bool stopped = false; +#endif + if (!stopped && !has_work()) { + ring_listener_->WaitForCqe(); + } + _notified.store(false, std::memory_order_release); + } else +#endif + { + std::unique_lock lk(_mux); + _cv.wait(lk, has_work); + } _waiting.store(false, std::memory_order_release); _waiting_workers.fetch_sub(1, std::memory_order_relaxed); return true; From fd2a6b66a85b25d3e6861f7d387da3c2e6538292 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 3 Aug 2026 03:54:38 +0000 Subject: [PATCH 02/14] fix: rely on destructor for init cleanup --- src/bthread/ring_listener.cpp | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index 9f545b56..77d9af17 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -233,7 +233,6 @@ int RingListener::Init() { std::make_unique(write_buf_slots, &ring_); if (write_buf_pool_->buf_pool_.empty()) { - Close(); return -1; } @@ -244,17 +243,14 @@ int RingListener::Init() { const int saved_errno = errno; LOG(ERROR) << "Failed to create the brpc worker wakeup eventfd, errno: " << saved_errno << " (" << strerror(saved_errno) << ")"; - Close(); return -saved_errno; } ret = ArmEventFdPoll(); if (ret != 0) { - Close(); return ret; } ret = SubmitAll(); if (ret < 0) { - Close(); return ret; } } else { From de458c5b5b8f08527f4ddc338f011cb6299e4177 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 3 Aug 2026 04:37:31 +0000 Subject: [PATCH 03/14] fix: preserve worker stop state in eventfd wait --- src/bthread/task_group.cpp | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 8e43a921..7fc1416e 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -1292,13 +1292,11 @@ bool TaskGroup::Wait(){ // A stop may happen immediately before this worker tries to sleep. A // worker that is already blocked is woken through eventfd by // TaskControl::stop_and_join(). + const ParkingLot::State pl_state = _pl->get_state(); #ifndef BTHREAD_DONT_SAVE_PARKING_STATE - _last_pl_state = _pl->get_state(); - const bool stopped = _last_pl_state.stopped(); -#else - const bool stopped = false; + _last_pl_state = pl_state; #endif - if (!stopped && !has_work()) { + if (!pl_state.stopped() && !has_work()) { ring_listener_->WaitForCqe(); } _notified.store(false, std::memory_order_release); From df50efb984ba2b852b7b0aeab9b0f6eb4f2c31b7 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 3 Aug 2026 05:39:27 +0000 Subject: [PATCH 04/14] test: enable eventfd worker wakeup by default --- src/brpc/socket.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp index 67f2eb96..b5dd3467 100644 --- a/src/brpc/socket.cpp +++ b/src/brpc/socket.cpp @@ -64,7 +64,7 @@ DEFINE_bool(dispatch_lazily, false, "dispatcher lazily creates task"); #ifdef IO_URING_ENABLED DEFINE_bool(use_io_uring, false, "Use IO URING to do the polling."); -DEFINE_bool(brpc_use_event_fd_wakeup, false, +DEFINE_bool(brpc_use_event_fd_wakeup, true, "Wake up idle brpc workers through eventfd polled by io_uring. " "Requires use_io_uring and is fixed at startup."); #endif From a192af98010f873f669652aa5d07c31ceee19521 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 3 Aug 2026 08:03:20 +0000 Subject: [PATCH 05/14] refactor: bind eventfd wakeup to io_uring --- src/brpc/socket.cpp | 3 --- src/bthread/ring_listener.cpp | 15 +++++++-------- src/bthread/ring_listener.h | 5 ++--- src/bthread/task_control.cpp | 5 +---- src/bthread/task_group.cpp | 13 +++---------- 5 files changed, 13 insertions(+), 28 deletions(-) diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp index b5dd3467..b2573b94 100644 --- a/src/brpc/socket.cpp +++ b/src/brpc/socket.cpp @@ -64,9 +64,6 @@ DEFINE_bool(dispatch_lazily, false, "dispatcher lazily creates task"); #ifdef IO_URING_ENABLED DEFINE_bool(use_io_uring, false, "Use IO URING to do the polling."); -DEFINE_bool(brpc_use_event_fd_wakeup, true, - "Wake up idle brpc workers through eventfd polled by io_uring. " - "Requires use_io_uring and is fixed at startup."); #endif namespace bthread { diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index 77d9af17..b2ffc6b6 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -46,7 +46,7 @@ DEFINE_int32(io_uring_registered_files, 1024, "inbound listener"); DEFINE_int32(io_uring_write_buffer_pool_size, 1024, "Number of buffers kept in the io_uring-based write buffer pool."); -DECLARE_bool(brpc_use_event_fd_wakeup); +DECLARE_bool(use_io_uring); void RingListener::Close() { if (ring_init_) { @@ -66,7 +66,7 @@ void RingListener::Close() { } RingListener::~RingListener() { - if (!FLAGS_brpc_use_event_fd_wakeup) { + if (!FLAGS_use_io_uring) { for (auto [fd, fd_idx]: reg_fds_) { SocketUnRegisterData data; data.fd_ = fd; @@ -135,10 +135,9 @@ int RingListener::Init() { const unsigned write_buf_slots = static_cast(flag_write_buffers); - unsigned ring_flags = IORING_SETUP_SINGLE_ISSUER; - if (FLAGS_brpc_use_event_fd_wakeup) { - ring_flags |= IORING_SETUP_DEFER_TASKRUN | IORING_SETUP_TASKRUN_FLAG; - } + unsigned ring_flags = IORING_SETUP_SINGLE_ISSUER | + IORING_SETUP_DEFER_TASKRUN | + IORING_SETUP_TASKRUN_FLAG; int ret = io_uring_queue_init(queue_entries, &ring_, ring_flags); if (ret < 0) { @@ -237,7 +236,7 @@ int RingListener::Init() { } poll_status_.store(PollStatus::Sleep, std::memory_order_release); - if (FLAGS_brpc_use_event_fd_wakeup) { + if (FLAGS_use_io_uring) { wakeup_event_fd_ = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); if (wakeup_event_fd_ < 0) { const int saved_errno = errno; @@ -627,7 +626,7 @@ size_t RingListener::ExtPoll() { } void RingListener::ExtWakeup() { - if (FLAGS_brpc_use_event_fd_wakeup) { + if (FLAGS_use_io_uring) { return; } has_external_.store(false, std::memory_order_relaxed); diff --git a/src/bthread/ring_listener.h b/src/bthread/ring_listener.h index 837e0a79..2138f03f 100644 --- a/src/bthread/ring_listener.h +++ b/src/bthread/ring_listener.h @@ -163,9 +163,8 @@ class RingListener { void ExtWakeup(); - // Wakes a worker blocked in WaitForCqe through the poll request registered - // on wakeup_event_fd_. This is only used when - // FLAGS_brpc_use_event_fd_wakeup was enabled at startup. + // Wakes an io_uring worker blocked in WaitForCqe through the poll request + // registered on wakeup_event_fd_. void NotifyEventFd(); // Blocks the owning worker until this ring has at least one completion. diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp index 4db45e69..c7a4eabf 100644 --- a/src/bthread/task_control.cpp +++ b/src/bthread/task_control.cpp @@ -41,9 +41,6 @@ DEFINE_int32(task_group_runqueue_capacity, 4096, DEFINE_int32(task_group_yield_before_idle, 0, "TaskGroup yields so many times before idle"); DECLARE_bool(use_io_uring); -#ifdef IO_URING_ENABLED -DECLARE_bool(brpc_use_event_fd_wakeup); -#endif namespace bthread { @@ -284,7 +281,7 @@ void TaskControl::stop_and_join() { _pl[i].stop(); } #ifdef IO_URING_ENABLED - if (FLAGS_brpc_use_event_fd_wakeup) { + if (FLAGS_use_io_uring) { // Workers in this mode wait in io_uring rather than on the parking // lot. Reuse the scheduler's normal eventfd notification so shutdown // does not depend on a signal interrupting io_uring_enter. diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 7fc1416e..19dd2c9f 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -52,9 +52,6 @@ std::atomic registered_module_version; DEFINE_int32(steal_task_rnd, 100, "Steal task frequency in wait_task"); DEFINE_bool(brpc_worker_as_ext_processor, false, "Work as external processor"); DECLARE_bool(use_io_uring); -#ifdef IO_URING_ENABLED -DECLARE_bool(brpc_use_event_fd_wakeup); -#endif namespace bthread { @@ -331,10 +328,6 @@ int TaskGroup::init(size_t runqueue_capacity) { _main_stack = stk; _last_run_ns = butil::cpuwide_time_ns(); #ifdef IO_URING_ENABLED - if (FLAGS_brpc_use_event_fd_wakeup && !FLAGS_use_io_uring) { - LOG(FATAL) << "brpc_use_event_fd_wakeup requires use_io_uring"; - return -1; - } if (FLAGS_use_io_uring) { ring_listener_ = std::make_unique(this); int ret = ring_listener_->Init(); @@ -1229,7 +1222,7 @@ void TaskGroup::Notify() { // Only one caller gets the right to notify the worker. if (_notified.compare_exchange_strong(expect, true)) { #ifdef IO_URING_ENABLED - if (FLAGS_brpc_use_event_fd_wakeup) { + if (FLAGS_use_io_uring) { ring_listener_->NotifyEventFd(); return; } @@ -1247,7 +1240,7 @@ bool TaskGroup::NotifyIfWaiting() { // Only one caller gets the right to notify the worker. if (_notified.compare_exchange_strong(expect, true)) { #ifdef IO_URING_ENABLED - if (FLAGS_brpc_use_event_fd_wakeup) { + if (FLAGS_use_io_uring) { ring_listener_->NotifyEventFd(); return true; } @@ -1286,7 +1279,7 @@ bool TaskGroup::Wait(){ }; #ifdef IO_URING_ENABLED - if (FLAGS_brpc_use_event_fd_wakeup) { + if (FLAGS_use_io_uring) { // has_work() clears _notified. If a producer races before or after that // check, eventfd retains the wakeup until submit_and_wait observes it. // A stop may happen immediately before this worker tries to sleep. A From 3cd056d5ed69de086e11f668bee0054e0d860fba Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 03:34:25 +0000 Subject: [PATCH 06/14] refactor: remove legacy ring polling thread --- src/bthread/ring_listener.cpp | 130 ++++------------------------------ src/bthread/ring_listener.h | 21 +----- src/bthread/ring_module.cpp | 8 +-- src/bthread/task_group.cpp | 5 -- 4 files changed, 18 insertions(+), 146 deletions(-) diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index b2ffc6b6..fbd934f0 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -34,7 +34,6 @@ #include "bthread/eloq_module.h" #include "bthread/inbound_ring_buf.h" #include "bthread/ring_write_buf_pool.h" -#include "butil/threading/platform_thread.h" #include "ring_listener.h" @@ -46,7 +45,6 @@ DEFINE_int32(io_uring_registered_files, 1024, "inbound listener"); DEFINE_int32(io_uring_write_buffer_pool_size, 1024, "Number of buffers kept in the io_uring-based write buffer pool."); -DECLARE_bool(use_io_uring); void RingListener::Close() { if (ring_init_) { @@ -66,24 +64,6 @@ void RingListener::Close() { } RingListener::~RingListener() { - if (!FLAGS_use_io_uring) { - for (auto [fd, fd_idx]: reg_fds_) { - SocketUnRegisterData data; - data.fd_ = fd; - SubmitCancel(&data); - // Not wait here because the worker should have quit already. - } - SubmitAll(); - - poll_status_.store(PollStatus::Closed, std::memory_order_release); { - std::unique_lock lk(mux_); - cv_.notify_one(); - } - } - - if (poll_thd_.joinable()) { - poll_thd_.join(); - } Close(); } @@ -235,31 +215,20 @@ int RingListener::Init() { return -1; } - poll_status_.store(PollStatus::Sleep, std::memory_order_release); - if (FLAGS_use_io_uring) { - wakeup_event_fd_ = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); - if (wakeup_event_fd_ < 0) { - const int saved_errno = errno; - LOG(ERROR) << "Failed to create the brpc worker wakeup eventfd, errno: " - << saved_errno << " (" << strerror(saved_errno) << ")"; - return -saved_errno; - } - ret = ArmEventFdPoll(); - if (ret != 0) { - return ret; - } - ret = SubmitAll(); - if (ret < 0) { - return ret; - } - } else { - poll_thd_ = std::thread([&]() { - std::string ring_listener = "ring_listener:"; - ring_listener.append(std::to_string(task_group_->group_id_)); - butil::PlatformThread::SetName(ring_listener.c_str()); - - Run(); - }); + wakeup_event_fd_ = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); + if (wakeup_event_fd_ < 0) { + const int saved_errno = errno; + LOG(ERROR) << "Failed to create the brpc worker wakeup eventfd, errno: " + << saved_errno << " (" << strerror(saved_errno) << ")"; + return -saved_errno; + } + ret = ArmEventFdPoll(); + if (ret != 0) { + return ret; + } + ret = SubmitAll(); + if (ret < 0) { + return ret; } return 0; @@ -564,46 +533,12 @@ int RingListener::WaitForCqe() { return ret; } -void RingListener::PollAndNotify() { - io_uring_cqe *cqe = nullptr; - while (true) { - int ret = io_uring_wait_cqe(&ring_, &cqe); - if (ret == -EINTR || ret == -EAGAIN) { - continue; - } - if (ret < 0) { - LOG(ERROR) << "Listener uring wait errno: " << ret; - poll_status_.store(PollStatus::Sleep, std::memory_order_relaxed); - return; - } - break; - } - - cqe_ready_.store(true, std::memory_order_relaxed); - poll_status_.store(PollStatus::Sleep, std::memory_order_relaxed); - RingModule::NotifyWorker(task_group_->group_id_); -} - - size_t RingListener::ExtPoll() { - if (!has_external_.load(std::memory_order_relaxed)) { - has_external_.store(true, std::memory_order_release); - } - - // has_external_ should be updated before poll_status_ is checked. - std::atomic_thread_fence(std::memory_order_release); - - PollStatus status = PollStatus::Sleep; - if (!poll_status_.compare_exchange_strong(status, PollStatus::ExtPoll)) { - return 0; - } - HandleBacklog(); io_uring_cqe *cqe = nullptr; int ret = io_uring_peek_cqe(&ring_, &cqe); if (ret != 0) { - poll_status_.store(PollStatus::Sleep, std::memory_order_relaxed); return 0; } @@ -614,51 +549,14 @@ size_t RingListener::ExtPoll() { ++processed; } - cqe_ready_.store(false, std::memory_order_relaxed); - if (processed > 0) { io_uring_cq_advance(&ring_, processed); } cqe_ready_.store(false, std::memory_order_relaxed); - poll_status_.store(PollStatus::Sleep, std::memory_order_relaxed); return processed; } -void RingListener::ExtWakeup() { - if (FLAGS_use_io_uring) { - return; - } - has_external_.store(false, std::memory_order_relaxed); - if (poll_status_.load(std::memory_order_relaxed) != PollStatus::Sleep) { - return; - } - std::unique_lock lk(mux_); - cv_.notify_one(); -} - -void RingListener::Run() { - while (poll_status_.load(std::memory_order_relaxed) != PollStatus::Closed) { - bool success = false; - if (!has_external_.load(std::memory_order_relaxed)) { - PollStatus status = PollStatus::Sleep; - success = poll_status_.compare_exchange_strong(status, PollStatus::Active, - std::memory_order_acq_rel); - if (success) { - PollAndNotify(); - } - } - std::unique_lock lk(mux_); - cv_.wait(lk, [this]() { - // wait for the worker to process the ready cqes and notify RingListener when it sleeps - return !has_external_.load(std::memory_order_relaxed) - && !cqe_ready_.load(std::memory_order_relaxed) || - poll_status_.load(std::memory_order_relaxed) == - PollStatus::Closed; - }); - } -} - void RingListener::RecycleReadBuf(uint16_t bid, size_t bytes) { // The socket has finished processing inbound messages. Returns the borrowed // buffers to the buffer ring. diff --git a/src/bthread/ring_listener.h b/src/bthread/ring_listener.h index 2138f03f..ee5428a9 100644 --- a/src/bthread/ring_listener.h +++ b/src/bthread/ring_listener.h @@ -21,12 +21,9 @@ #ifdef IO_URING_ENABLED -#include #include -#include #include #include -#include #include #include "brpc/socket.h" @@ -34,7 +31,6 @@ #undef BLOCK_SIZE #include "bthread/moodycamelqueue.h" #include "bthread/ring_write_buf_pool.h" -#include "butil/threading/platform_thread.h" #include "spsc_queue.h" namespace bthread { @@ -157,12 +153,8 @@ class RingListener { int SubmitAll(); - void PollAndNotify(); - size_t ExtPoll(); - void ExtWakeup(); - // Wakes an io_uring worker blocked in WaitForCqe through the poll request // registered on wakeup_event_fd_. void NotifyEventFd(); @@ -172,8 +164,6 @@ class RingListener { // single issuer, so the legacy polling thread must never call this method. int WaitForCqe(); - void Run(); - void RecycleReadBuf(uint16_t bid, size_t bytes); const char *GetReadBuf(uint16_t bid) const { @@ -279,18 +269,13 @@ class RingListener { void DrainEventFd(); - enum struct PollStatus : uint8_t { Active = 0, Sleep, ExtPoll, Closed }; - struct io_uring ring_; bool ring_init_{false}; - std::atomic poll_status_{PollStatus::Sleep}; - // cqe_ready_ is set by the ring listener and unset by the worker + // cqe_ready_ is set before a parked worker resumes and cleared after the + // owning worker drains the completion queue. std::atomic cqe_ready_{false}; uint16_t submit_cnt_{0}; std::unordered_map reg_fds_; - std::mutex mux_; - std::condition_variable cv_; - std::thread poll_thd_; int wakeup_event_fd_{-1}; io_uring_buf_ring *in_buf_ring_{nullptr}; @@ -306,8 +291,6 @@ class RingListener { buf_ring_size }; - std::atomic has_external_{true}; - std::vector free_reg_fd_idx_; std::unique_ptr write_buf_pool_; diff --git a/src/bthread/ring_module.cpp b/src/bthread/ring_module.cpp index 37d1a7f9..14f043a3 100644 --- a/src/bthread/ring_module.cpp +++ b/src/bthread/ring_module.cpp @@ -19,14 +19,10 @@ #include "ring_module.h" #include "ring_listener.h" -#include - #ifdef IO_URING_ENABLED -void RingModule::ExtThdStart(int thd_id) { - listeners_.at(thd_id)->has_external_.store(true, std::memory_order_relaxed); -} +void RingModule::ExtThdStart(int) {} -void RingModule::ExtThdEnd(int thd_id) { listeners_.at(thd_id)->ExtWakeup(); } +void RingModule::ExtThdEnd(int) {} void RingModule::Process(int thd_id) { RingListener *listener = listeners_.at(thd_id); diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 19dd2c9f..824009a8 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -193,11 +193,6 @@ bool TaskGroup::wait_task(bthread_t* tid) { if (FLAGS_worker_polling_time_us <= 0 || butil::cpuwide_time_us() - poll_start_us > FLAGS_worker_polling_time_us) { if (!HasTasks()) { -#ifdef IO_URING_ENABLED - if (FLAGS_use_io_uring && ring_listener_ != nullptr) { - ring_listener_->ExtWakeup(); - } -#endif NotifyRegisteredModules(WorkerStatus::Sleep); Wait(); From 3aa35062469dbdfd004d9e1c68f90f7f46f1b2f2 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 03:38:03 +0000 Subject: [PATCH 07/14] refactor: select worker wakeup by ring ownership --- src/bthread/task_group.cpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 824009a8..2c169320 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -1217,7 +1217,7 @@ void TaskGroup::Notify() { // Only one caller gets the right to notify the worker. if (_notified.compare_exchange_strong(expect, true)) { #ifdef IO_URING_ENABLED - if (FLAGS_use_io_uring) { + if (ring_listener_ != nullptr) { ring_listener_->NotifyEventFd(); return; } @@ -1235,7 +1235,7 @@ bool TaskGroup::NotifyIfWaiting() { // Only one caller gets the right to notify the worker. if (_notified.compare_exchange_strong(expect, true)) { #ifdef IO_URING_ENABLED - if (FLAGS_use_io_uring) { + if (ring_listener_ != nullptr) { ring_listener_->NotifyEventFd(); return true; } @@ -1274,7 +1274,7 @@ bool TaskGroup::Wait(){ }; #ifdef IO_URING_ENABLED - if (FLAGS_use_io_uring) { + if (ring_listener_ != nullptr) { // has_work() clears _notified. If a producer races before or after that // check, eventfd retains the wakeup until submit_and_wait observes it. // A stop may happen immediately before this worker tries to sleep. A From 535315cf71fb7911386ecbf9d87679dd4775ab10 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 03:43:39 +0000 Subject: [PATCH 08/14] fix: rearm scheduler wakeup poll --- src/bthread/ring_listener.cpp | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index fbd934f0..147a47b7 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -739,11 +739,16 @@ void RingListener::HandleCqe(io_uring_cqe *cqe) { << cqe->res; } DrainEventFd(); - // A multishot poll stays armed after each readiness event. Losing - // IORING_CQE_F_MORE means the scheduler can no longer wake this - // worker, so fail instead of allowing a future permanent sleep. + // A multishot poll stays armed only while the CQE carries MORE. + // Re-arm a terminated request so a later scheduler notification + // cannot leave this worker permanently asleep. if (!(cqe->flags & IORING_CQE_F_MORE)) { - LOG(FATAL) << "The brpc worker multishot wakeup poll terminated"; + const int ret = ArmEventFdPoll(); + if (ret != 0) { + LOG(ERROR) << "Failed to re-arm the brpc worker wakeup " + "poll, ret: " + << ret; + } } break; } From 05ef806c42c1c1300efe6376ffded94e7e2b5711 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 03:45:21 +0000 Subject: [PATCH 09/14] refactor: rename ring listener wait to park --- src/bthread/ring_listener.cpp | 2 +- src/bthread/ring_listener.h | 4 ++-- src/bthread/task_group.cpp | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index 147a47b7..b75d8722 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -513,7 +513,7 @@ void RingListener::NotifyEventFd() { } } -int RingListener::WaitForCqe() { +int RingListener::Park() { int ret; do { ret = io_uring_submit_and_wait(&ring_, 1); diff --git a/src/bthread/ring_listener.h b/src/bthread/ring_listener.h index ee5428a9..322ebf8b 100644 --- a/src/bthread/ring_listener.h +++ b/src/bthread/ring_listener.h @@ -155,14 +155,14 @@ class RingListener { size_t ExtPoll(); - // Wakes an io_uring worker blocked in WaitForCqe through the poll request + // Wakes an io_uring worker blocked in Park() through the poll request // registered on wakeup_event_fd_. void NotifyEventFd(); // Blocks the owning worker until this ring has at least one completion. // DEFER_TASKRUN requires every io_uring_enter call to come from the ring's // single issuer, so the legacy polling thread must never call this method. - int WaitForCqe(); + int Park(); void RecycleReadBuf(uint16_t bid, size_t bytes); diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 2c169320..472ba8f0 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -1285,7 +1285,7 @@ bool TaskGroup::Wait(){ _last_pl_state = pl_state; #endif if (!pl_state.stopped() && !has_work()) { - ring_listener_->WaitForCqe(); + ring_listener_->Park(); } _notified.store(false, std::memory_order_release); } else From 528894f417a81ca06acc0711298f1fdc42af1bf9 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 03:50:36 +0000 Subject: [PATCH 10/14] fix: handle busy ring park result --- src/bthread/ring_listener.cpp | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index b75d8722..a6c33bcc 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -525,8 +525,9 @@ int RingListener::Park() { return 0; } // TaskControl interrupts worker pthreads during shutdown. Returning on - // EINTR lets the scheduler observe the stopped parking-lot state. - if (ret == -EINTR) { + // EINTR lets the scheduler observe the stopped parking-lot state. EBUSY + // means the CQ overflow list must be reaped before entering the ring again. + if (ret == -EINTR || ret == -EBUSY) { return ret; } LOG(FATAL) << "Failed while waiting on the brpc worker io_uring, ret: " << ret; From 7179969fdae95b76d20f8e8156e6498c2d408112 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 03:51:24 +0000 Subject: [PATCH 11/14] fix: avoid terminating on ring park failure --- src/bthread/ring_listener.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index a6c33bcc..1dea4561 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -530,7 +530,7 @@ int RingListener::Park() { if (ret == -EINTR || ret == -EBUSY) { return ret; } - LOG(FATAL) << "Failed while waiting on the brpc worker io_uring, ret: " << ret; + LOG(ERROR) << "Failed while waiting on the brpc worker io_uring, ret: " << ret; return ret; } From 31ef946732d4a4c455cb5d3e92945f9e72d40c54 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 03:52:44 +0000 Subject: [PATCH 12/14] fix: recheck work while ring worker is parked --- src/bthread/task_group.cpp | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 472ba8f0..4b0cc663 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -1280,12 +1280,17 @@ bool TaskGroup::Wait(){ // A stop may happen immediately before this worker tries to sleep. A // worker that is already blocked is woken through eventfd by // TaskControl::stop_and_join(). - const ParkingLot::State pl_state = _pl->get_state(); + while (true) { + const ParkingLot::State pl_state = _pl->get_state(); #ifndef BTHREAD_DONT_SAVE_PARKING_STATE - _last_pl_state = pl_state; + _last_pl_state = pl_state; #endif - if (!pl_state.stopped() && !has_work()) { - ring_listener_->Park(); + if (pl_state.stopped() || has_work()) { + break; + } + if (ring_listener_->Park() < 0) { + break; + } } _notified.store(false, std::memory_order_release); } else From b4a9d86ea7deb3b91c922c2077a3b6f8f8f1c877 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 03:54:24 +0000 Subject: [PATCH 13/14] docs: simplify ring listener comments --- src/bthread/ring_listener.cpp | 6 +++--- src/bthread/ring_listener.h | 2 -- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index 1dea4561..835ec8bc 100644 --- a/src/bthread/ring_listener.cpp +++ b/src/bthread/ring_listener.cpp @@ -479,9 +479,9 @@ void RingListener::DrainEventFd() { if (nread < 0 && errno == EINTR) { continue; } - // EAGAIN means another notification consumer already drained the - // counter. There is only one consumer today, but treating it as - // drained keeps this helper safe if that implementation changes. + // Multiple multishot CQEs may already be queued for the same readable + // eventfd. An earlier CQE can drain the coalesced counter, leaving a + // later one with nothing to read. if (nread < 0 && errno == EAGAIN) { return; } diff --git a/src/bthread/ring_listener.h b/src/bthread/ring_listener.h index 322ebf8b..3ef4254b 100644 --- a/src/bthread/ring_listener.h +++ b/src/bthread/ring_listener.h @@ -160,8 +160,6 @@ class RingListener { void NotifyEventFd(); // Blocks the owning worker until this ring has at least one completion. - // DEFER_TASKRUN requires every io_uring_enter call to come from the ring's - // single issuer, so the legacy polling thread must never call this method. int Park(); void RecycleReadBuf(uint16_t bid, size_t bytes); From 81064566f3c29f27795c148cd14a03081597d06f Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Tue, 4 Aug 2026 06:45:12 +0000 Subject: [PATCH 14/14] refactor: remove redundant shutdown wakeup flag --- src/bthread/task_control.cpp | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp index c7a4eabf..97f1abc2 100644 --- a/src/bthread/task_control.cpp +++ b/src/bthread/task_control.cpp @@ -281,13 +281,10 @@ void TaskControl::stop_and_join() { _pl[i].stop(); } #ifdef IO_URING_ENABLED - if (FLAGS_use_io_uring) { - // Workers in this mode wait in io_uring rather than on the parking - // lot. Reuse the scheduler's normal eventfd notification so shutdown - // does not depend on a signal interrupting io_uring_enter. - for (int i = 0; i < _parking_lot_num; ++i) { - _groups[i]->Notify(); - } + // Notify() routes to eventfd when a ring listener exists and otherwise + // preserves the condition-variable wakeup path. + for (int i = 0; i < _parking_lot_num; ++i) { + _groups[i]->Notify(); } #endif // Interrupt blocking operations.