diff --git a/src/bthread/ring_listener.cpp b/src/bthread/ring_listener.cpp index cd6799af..835ec8bc 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 @@ -31,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" @@ -44,23 +46,24 @@ DEFINE_int32(io_uring_registered_files, 1024, DEFINE_int32(io_uring_write_buffer_pool_size, 1024, "Number of buffers kept in the io_uring-based write buffer pool."); -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 (poll_thd_.joinable()) { - poll_thd_.join(); + if (in_buf_) { + free(in_buf_); + in_buf_ = nullptr; } +} + +RingListener::~RingListener() { Close(); } @@ -112,7 +115,10 @@ 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 | + 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 " @@ -209,14 +215,21 @@ int RingListener::Init() { 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()); - - 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; } @@ -443,46 +456,90 @@ int RingListener::SubmitAll() { return ret; } -void RingListener::PollAndNotify() { - io_uring_cqe *cqe = nullptr; +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) { - int ret = io_uring_wait_cqe(&ring_, &cqe); - if (ret == -EINTR || ret == -EAGAIN) { + const ssize_t nread = read(wakeup_event_fd_, &value, sizeof(value)); + if (nread == static_cast(sizeof(value))) { + return; + } + if (nread < 0 && errno == EINTR) { continue; } - if (ret < 0) { - LOG(ERROR) << "Listener uring wait errno: " << ret; - poll_status_.store(PollStatus::Sleep, std::memory_order_relaxed); + // 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; } - break; + const int saved_errno = errno; + LOG(FATAL) << "Failed to drain the brpc worker wakeup eventfd, errno: " + << saved_errno << " (" << strerror(saved_errno) << ")"; } - - 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); +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) << ")"; } +} - // has_external_ should be updated before poll_status_ is checked. - std::atomic_thread_fence(std::memory_order_release); +int RingListener::Park() { + int ret; + do { + ret = io_uring_submit_and_wait(&ring_, 1); + } while (ret == -EAGAIN); - PollStatus status = PollStatus::Sleep; - if (!poll_status_.compare_exchange_strong(status, PollStatus::ExtPoll)) { + 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. EBUSY + // means the CQ overflow list must be reaped before entering the ring again. + if (ret == -EINTR || ret == -EBUSY) { + return ret; + } + LOG(ERROR) << "Failed while waiting on the brpc worker io_uring, ret: " << ret; + return ret; +} +size_t RingListener::ExtPoll() { 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; } @@ -493,48 +550,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() { - 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. @@ -711,6 +734,25 @@ 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 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)) { + const int ret = ArmEventFdPoll(); + if (ret != 0) { + LOG(ERROR) << "Failed to re-arm the brpc worker wakeup " + "poll, ret: " + << ret; + } + } + break; + } default: break; } diff --git a/src/bthread/ring_listener.h b/src/bthread/ring_listener.h index 63b0c6af..3ef4254b 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 { @@ -128,17 +124,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); @@ -167,13 +153,14 @@ class RingListener { int SubmitAll(); - void PollAndNotify(); - size_t ExtPoll(); - void ExtWakeup(); + // Wakes an io_uring worker blocked in Park() through the poll request + // registered on wakeup_event_fd_. + void NotifyEventFd(); - void Run(); + // Blocks the owning worker until this ring has at least one completion. + int Park(); void RecycleReadBuf(uint16_t bid, size_t bytes); @@ -209,6 +196,7 @@ class RingListener { NonFixedWriteFinish, WaitingNonFixedWrite, Fsync, + SchedulerWakeup, Noop = 255 }; @@ -232,6 +220,8 @@ class RingListener { return 7; case OpCode::Fsync: return 8; + case OpCode::SchedulerWakeup: + return 9; default: return UINT8_MAX; } @@ -257,6 +247,8 @@ class RingListener { return OpCode::WaitingNonFixedWrite; case 8: return OpCode::Fsync; + case 9: + return OpCode::SchedulerWakeup; default: return OpCode::Noop; } @@ -270,18 +262,19 @@ class RingListener { void RecycleReturnedWriteBufs(); - enum struct PollStatus : uint8_t { Active = 0, Sleep, ExtPoll, Closed }; + // Installs the persistent multishot poll used for scheduler wakeups. + int ArmEventFdPoll(); + + void DrainEventFd(); 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}; char *in_buf_{nullptr}; @@ -296,8 +289,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_control.cpp b/src/bthread/task_control.cpp index cede3e76..97f1abc2 100644 --- a/src/bthread/task_control.cpp +++ b/src/bthread/task_control.cpp @@ -280,6 +280,13 @@ void TaskControl::stop_and_join() { for (int i = 0; i < _parking_lot_num; ++i) { _pl[i].stop(); } +#ifdef IO_URING_ENABLED + // 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. 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..4b0cc663 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(); @@ -1221,6 +1216,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 (ring_listener_ != nullptr) { + ring_listener_->NotifyEventFd(); + return; + } +#endif std::unique_lock lk(_mux); _notified.store(true, std::memory_order_release); _cv.notify_one(); @@ -1233,6 +1234,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 (ring_listener_ != nullptr) { + ring_listener_->NotifyEventFd(); + return true; + } +#endif std::unique_lock lk(_mux); _notified.store(true, std::memory_order_release); _cv.notify_one(); @@ -1246,10 +1253,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 +1271,34 @@ bool TaskGroup::Wait(){ // Check any module registered or deleted before checking modules' tasks. CheckAndUpdateModules(); return HasTasks(); - }); + }; + +#ifdef IO_URING_ENABLED + 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 + // worker that is already blocked is woken through eventfd by + // TaskControl::stop_and_join(). + while (true) { + const ParkingLot::State pl_state = _pl->get_state(); +#ifndef BTHREAD_DONT_SAVE_PARKING_STATE + _last_pl_state = pl_state; +#endif + if (pl_state.stopped() || has_work()) { + break; + } + if (ring_listener_->Park() < 0) { + break; + } + } + _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;