You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2026/04/30 21:19:53 UTC
(kudu) branch master updated: [rpc] use min priority queue for service queue container
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new aa7cc0aa9 [rpc] use min priority queue for service queue container
aa7cc0aa9 is described below
commit aa7cc0aa9c559b87f264235eaadd03ec492277e4
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Thu Apr 23 18:03:26 2026 -0700
[rpc] use min priority queue for service queue container
A lot of contention in libtcmalloc might result in longer times spent
by reactor and RPC worker threads waiting on the lock in RPC service
queue when invoking LifoServiceQueue::{Put,BlockingGet}(). Per analysis
of captured stack traces, the culprit seemed to be a thread that would
acquire LifoServiceQueue::lock_ while executing LifoServiceQueue::Put()
to add an element into the queue. The latter lead to memory
(de)allocations via tcmalloc that went through the central free list
perhaps due to exhaustion of the thread cache's free list. Meanwhile,
the central free list's lock had already been acquired by some other
thread performing memory (de)allocation.
It's prudent to switch to a container which doesn't do any dynamic
memory allocations while adding and removing elements to/from the
service queue. Using a pre-allocated std::vector instance to maintain
a min priority queue turns to be a good option. Instead of using direct
API of the std::priority_queue container adaptor, std::push_stack() and
std::pop_stack() are used on top of the underlying array/vector to
handle an edge case to evict an element while replacing it with the
newly arrived one.
Various scenarios of the dedicated perf benchmark run on a 8 CPU-core
machine reported improvement even without artificially induced
contention in libtcmalloc. The service queue length was set to 200.
With larger service queue the performance boost was even higher for
scenarios where the queue often stays full.
KUDU_ALLOW_SLOW_TESTS=1 ./bin/service_queue-test --num_producers=P --num_consumers=C --max_queue_size=200
Balanced P=4/C=4 scenario reported ~1.56x improvement:
before:
Reqs/sec: 1288901
User CPU per req: 2.89116us
Sys CPU per req: 2.71363us
Avg rpc queue length: 102.518
Avg idle workers: 0
after:
Reqs/sec: 2022126
User CPU per req: 1.9856us
Sys CPU per req: 1.78929us
Avg rpc queue length: 50.964
Avg idle workers: 0.03
Balanced P=3/C=3 scenario reported ~1.30x improvement:
before:
Reqs/sec: 1470446
User CPU per req: 2.3612us
Sys CPU per req: 1.37508us
Avg rpc queue length: 76.184
Avg idle workers: 0
after:
Reqs/sec: 1917650
User CPU per req: 1.97289us
Sys CPU per req: 1.0409us
Avg rpc queue length: 145.424
Avg idle workers: 0
Skewed P=4/C=2 scenario reported ~1.76x improvement:
before:
Reqs/sec: 1053219
User CPU per req: 3.59901us
Sys CPU per req: 1.5802us
Avg rpc queue length: 194.678
Avg idle workers: 0
after:
Reqs/sec: 1862593
User CPU per req: 2.32343us
Sys CPU per req: 0.823388us
Avg rpc queue length: 194.37
Avg idle workers: 0
Skewed P=4/C=3 scenario reported ~1.55x improvement:
before:
Reqs/sec: 1185783
User CPU per req: 3.08805us
Sys CPU per req: 2.16397us
Avg rpc queue length: 191.486
Avg idle workers: 0
after:
Reqs/sec: 1834414
User CPU per req: 2.35371us
Sys CPU per req: 1.32012us
Avg rpc queue length: 192.042
Avg idle workers: 0
Skewed P=3/C=5 scenario reported ~1.29x improvement:
before:
Reqs/sec: 1622451
User CPU per req: 2.081us
Sys CPU per req: 1.86138us
Avg rpc queue length: 5.072
Avg idle workers: 0.104
after:
Reqs/sec: 2095612
User CPU per req: 1.65551us
Sys CPU per req: 1.39116us
Avg rpc queue length: 6.296
Avg idle workers: 0.176
Skewed P=2/C=5 scenario reported ~1.23x improvement:
before:
Reqs/sec: 1168078
User CPU per req: 1.83038us
Sys CPU per req: 1.94243us
Avg rpc queue length: 0.85
Avg idle workers: 0.858
after:
Reqs/sec: 1436187
User CPU per req: 1.57106us
Sys CPU per req: 1.56673us
Avg rpc queue length: 1.078
Avg idle workers: 0.878
Change-Id: I0d95828177ae6458b427ce43dd292877d03065cf
Reviewed-on: http://gerrit.cloudera.org:8080/24230
Tested-by: Kudu Jenkins
Reviewed-by: Zoltan Martonka <zm...@cloudera.com>
Reviewed-by: Abhishek Chennaka <ac...@cloudera.com>
---
src/kudu/rpc/service_pool.cc | 7 ++-
src/kudu/rpc/service_queue-test.cc | 58 +++++++++++++----------
src/kudu/rpc/service_queue.cc | 94 +++++++++++++++++++++++++-------------
src/kudu/rpc/service_queue.h | 50 ++++++++------------
4 files changed, 119 insertions(+), 90 deletions(-)
diff --git a/src/kudu/rpc/service_pool.cc b/src/kudu/rpc/service_pool.cc
index c95db25e7..223ab3512 100644
--- a/src/kudu/rpc/service_pool.cc
+++ b/src/kudu/rpc/service_pool.cc
@@ -19,7 +19,6 @@
#include <cstdint>
#include <memory>
-#include <optional>
#include <ostream>
#include <string>
#include <utility>
@@ -168,15 +167,15 @@ Status ServicePool::QueueInboundCall(unique_ptr<InboundCall> call) {
TRACE_TO(c->trace(), "Inserting onto call queue");
// Queue message on service queue
- std::optional<InboundCall*> evicted;
+ InboundCall* evicted = nullptr;
const auto queue_status = service_queue_.Put(c, &evicted);
if (queue_status == QUEUE_FULL) {
RejectTooBusy(c);
return Status::OK();
}
- if (PREDICT_FALSE(evicted.has_value())) {
- RejectTooBusy(*evicted);
+ if (PREDICT_FALSE(evicted != nullptr)) {
+ RejectTooBusy(evicted);
}
if (PREDICT_TRUE(queue_status == QUEUE_SUCCESS)) {
diff --git a/src/kudu/rpc/service_queue-test.cc b/src/kudu/rpc/service_queue-test.cc
index a970c3474..c32022dea 100644
--- a/src/kudu/rpc/service_queue-test.cc
+++ b/src/kudu/rpc/service_queue-test.cc
@@ -18,7 +18,6 @@
#include <atomic>
#include <cstdint>
#include <memory>
-#include <optional>
#include <ostream>
#include <string>
#include <thread>
@@ -36,8 +35,9 @@
#include "kudu/util/stopwatch.h"
#include "kudu/util/test_util.h"
-using std::shared_ptr;
+using std::atomic;
using std::string;
+using std::thread;
using std::unique_ptr;
using std::vector;
@@ -50,34 +50,42 @@ DEFINE_int32(num_consumers, 20,
DEFINE_int32(max_queue_size, 50,
"Max queue length");
+DEFINE_int32(consumer_delay_cycles, 0,
+ "Number of CPU yield cycles for a consumer thread to run after "
+ "each BlockingGet() call: this is to give the producer threads "
+ "a chance to fill up the queue");
+
namespace kudu {
namespace rpc {
-static std::atomic<uint32_t> inprogress;
+static atomic<uint32_t> inprogress;
-static std::atomic<uint32_t> total;
+static atomic<uint32_t> total;
template <typename Queue>
void ProducerThread(Queue* queue) {
- int max_inprogress = FLAGS_max_queue_size - FLAGS_num_producers;
+ const int max_inprogress = FLAGS_max_queue_size - FLAGS_num_producers;
while (true) {
- while (inprogress > max_inprogress) {
+ // For atomics, std::memory_relaxed is enough since producer threads
+ // back off when hitting QUEUE_FULL anyway. As for the re-ordering concern,
+ // it's addressed by the presence of a lock in the Put() call below.
+ while (inprogress.load(std::memory_order_relaxed) > max_inprogress) {
base::subtle::PauseCPU();
}
- inprogress++;
+ inprogress.fetch_add(1, std::memory_order_relaxed);
InboundCall* call = new InboundCall(nullptr);
- std::optional<InboundCall*> evicted;
+ InboundCall* evicted = nullptr;
auto status = queue->Put(call, &evicted);
if (status == QUEUE_FULL) {
- LOG(INFO) << "queue full: producer exiting";
delete call;
- break;
+ base::subtle::PauseCPU();
+ continue;
}
- if (PREDICT_FALSE(evicted.has_value())) {
- LOG(INFO) << "call evicted: producer exiting";
- delete *evicted;
- break;
+ if (PREDICT_FALSE(evicted != nullptr)) {
+ delete evicted;
+ base::subtle::PauseCPU();
+ continue;
}
if (PREDICT_TRUE(status == QUEUE_SHUTDOWN)) {
@@ -89,18 +97,22 @@ void ProducerThread(Queue* queue) {
template <typename Queue>
void ConsumerThread(Queue* queue) {
+ const int32_t delay_cycles = FLAGS_consumer_delay_cycles;
unique_ptr<InboundCall> call;
while (queue->BlockingGet(&call)) {
- inprogress--;
- total++;
+ inprogress.fetch_sub(1, std::memory_order_relaxed);
+ total.fetch_add(1, std::memory_order_relaxed);
call.reset();
+ for (int32_t cnt = delay_cycles; cnt > 0; --cnt) {
+ base::subtle::PauseCPU();
+ }
}
}
TEST(TestServiceQueue, LifoServiceQueuePerf) {
LifoServiceQueue queue(FLAGS_max_queue_size);
- vector<std::thread> producers;
- vector<std::thread> consumers;
+ vector<thread> producers;
+ vector<thread> consumers;
for (int i = 0; i < FLAGS_num_producers; i++) {
producers.emplace_back(&ProducerThread<LifoServiceQueue>, &queue);
@@ -120,7 +132,7 @@ TEST(TestServiceQueue, LifoServiceQueuePerf) {
for (int i = 0; i < seconds * 50; i++) {
SleepFor(MonoDelta::FromMilliseconds(20));
- total_sample++;
+ ++total_sample;
total_queue_len += queue.estimated_queue_length();
total_idle_workers += queue.estimated_idle_worker_count();
}
@@ -129,11 +141,11 @@ TEST(TestServiceQueue, LifoServiceQueuePerf) {
int32_t delta = total - before;
queue.Shutdown();
- for (int i = 0; i < FLAGS_num_producers; i++) {
- producers[i].join();
+ for (auto& p : producers) {
+ p.join();
}
- for (int i = 0; i < FLAGS_num_consumers; i++) {
- consumers[i].join();
+ for (auto& c : consumers) {
+ c.join();
}
float reqs_per_second = static_cast<float>(delta / sw.elapsed().wall_seconds());
diff --git a/src/kudu/rpc/service_queue.cc b/src/kudu/rpc/service_queue.cc
index a3259a621..7c18b9e82 100644
--- a/src/kudu/rpc/service_queue.cc
+++ b/src/kudu/rpc/service_queue.cc
@@ -17,21 +17,31 @@
#include "kudu/rpc/service_queue.h"
+#include <algorithm>
#include <mutex>
-#include <optional>
#include <ostream>
#include "kudu/gutil/port.h"
+using std::lock_guard;
+using std::pop_heap;
+using std::push_heap;
+using std::string;
+using std::unique_lock;
+using std::unique_ptr;
+
namespace kudu {
namespace rpc {
thread_local LifoServiceQueue::ConsumerState* LifoServiceQueue::tl_consumer_ = nullptr;
LifoServiceQueue::LifoServiceQueue(size_t max_size)
- : max_queue_size_(max_size),
- shutdown_(false) {
+ : max_queue_size_(max_size),
+ shutdown_(false) {
DCHECK_GT(max_queue_size_, 0);
+ // Reserving all the required memory upfront, no re-allocations are necessary
+ // during the lifecycle of the instance of this class.
+ queue_.reserve(max_queue_size_);
}
LifoServiceQueue::~LifoServiceQueue() {
@@ -39,21 +49,21 @@ LifoServiceQueue::~LifoServiceQueue() {
<< "ServiceQueue holds bare pointers at destruction time";
}
-bool LifoServiceQueue::BlockingGet(std::unique_ptr<InboundCall>* out) {
+bool LifoServiceQueue::BlockingGet(unique_ptr<InboundCall>* out) {
auto* consumer = tl_consumer_;
if (PREDICT_FALSE(!consumer)) {
consumer = tl_consumer_ = new ConsumerState(this);
- std::lock_guard l(lock_);
+ lock_guard l(lock_);
consumers_.emplace_back(consumer);
}
while (true) {
{
- std::lock_guard l(lock_);
+ lock_guard l(lock_);
if (!queue_.empty()) {
- auto it = queue_.begin();
- out->reset(*it);
- queue_.erase(it);
+ out->reset(queue_.front());
+ pop_heap(queue_.begin(), queue_.end(), kMinHeapCompare);
+ queue_.pop_back();
return true;
}
if (PREDICT_FALSE(shutdown_)) {
@@ -74,9 +84,8 @@ bool LifoServiceQueue::BlockingGet(std::unique_ptr<InboundCall>* out) {
}
}
-QueueStatus LifoServiceQueue::Put(InboundCall* call,
- std::optional<InboundCall*>* evicted) {
- std::unique_lock l(lock_);
+QueueStatus LifoServiceQueue::Put(InboundCall* call, InboundCall** evicted) {
+ unique_lock l(lock_);
if (PREDICT_FALSE(shutdown_)) {
return QUEUE_SHUTDOWN;
}
@@ -87,32 +96,45 @@ QueueStatus LifoServiceQueue::Put(InboundCall* call,
if (queue_.empty() && !waiting_consumers_.empty()) {
auto* consumer = waiting_consumers_.back();
waiting_consumers_.pop_back();
- // Notify condition var(and wake up consumer thread) takes time,
+ // Notifying the condition and waking up the consumer thread takes time,
// so put it out of spinlock scope.
l.unlock();
consumer->Post(call);
return QUEUE_SUCCESS;
}
- if (PREDICT_FALSE(queue_.size() >= max_queue_size_)) {
- // eviction
- DCHECK_EQ(queue_.size(), max_queue_size_);
- auto it = queue_.end();
- --it;
- if (DeadlineLess(*it, call)) {
- return QUEUE_FULL;
- }
-
- *evicted = *it;
- queue_.erase(it);
+ if (queue_.size() < max_queue_size_) {
+ queue_.push_back(call);
+ push_heap(queue_.begin(), queue_.end(), kMinHeapCompare);
+ return QUEUE_SUCCESS;
+ }
+ DCHECK_EQ(queue_.size(), max_queue_size_);
+
+ // If the deadline of the new call is not earlier than the deadline of the
+ // call in the very end of the array backing the min heap, reject it with
+ // QUEUE_FULL status.
+ auto* back = queue_.back();
+ if (!DeadlineGreater(back, call)) {
+ return QUEUE_FULL;
}
- queue_.insert(call);
+ // Otherwise, replace the call in the very end of the container with the new
+ // one. The replaced call isn't guaranteed to be have the latest deadline
+ // among all the elements in the queue because it's a binary heap,
+ // not a sorted sequence. However, usually it's the latest one because most
+ // of the times the elements are added into the queue timestamped by their
+ // arrival time, and it increases monotonically. Removing the last element
+ // of the array representing a binary heap leaves the rest of the array still
+ // a binary heap, of one element smaller size, but with all the invariants
+ // of a binary heap preserved.
+ *evicted = back;
+ queue_.back() = call;
+ push_heap(queue_.begin(), queue_.end(), kMinHeapCompare);
return QUEUE_SUCCESS;
}
void LifoServiceQueue::Shutdown() {
- std::lock_guard l(lock_);
+ lock_guard l(lock_);
shutdown_ = true;
// Post a nullptr to wake up any consumers which are waiting.
@@ -122,14 +144,20 @@ void LifoServiceQueue::Shutdown() {
waiting_consumers_.clear();
}
-std::string LifoServiceQueue::ToString() const {
- std::string ret;
-
- std::lock_guard l(lock_);
- for (const auto* t : queue_) {
- ret.append(t->ToString());
- ret.append("\n");
+string LifoServiceQueue::ToString() const {
+ string ret;
+
+ // It's necessary to take a lock while accessing InboundCall pointers
+ // in the queue. There isn't a guarantee the underlying memory stays valid
+ // and not freed otherwise.
+ lock_guard l(lock_);
+ decltype(queue_) tmp(queue_);
+ while (!tmp.empty()) {
+ ret += tmp.front()->ToString() + "\n";
+ pop_heap(tmp.begin(), tmp.end(), kMinHeapCompare);
+ tmp.pop_back();
}
+
return ret;
}
diff --git a/src/kudu/rpc/service_queue.h b/src/kudu/rpc/service_queue.h
index 55d9e5b63..6caca76dd 100644
--- a/src/kudu/rpc/service_queue.h
+++ b/src/kudu/rpc/service_queue.h
@@ -18,8 +18,6 @@
#include <cstddef>
#include <memory>
-#include <optional>
-#include <set>
#include <string>
#include <vector>
@@ -46,13 +44,8 @@ enum QueueStatus {
// Blocking queue used for passing inbound RPC calls to the service handler pool.
// Calls are dequeued in 'earliest-deadline first' order. The queue also maintains a
-// bounded number of calls. If the queue overflows, then calls with deadlines farthest
-// in the future are evicted.
-//
-// When calls do not provide deadlines, the RPC layer considers their deadline to
-// be infinitely in the future. This means that any call that does have a deadline
-// can evict any call that does not have a deadline. This incentivizes clients to
-// provide accurate deadlines for their calls.
+// bounded number of calls. If the queue is about to overflow, then one call
+// with the deadline later than the deadline of the call being added is evicted.
//
// In order to improve concurrent throughput, this class uses a LIFO design:
// Each consumer thread has its own lock and condition variable. If a
@@ -81,21 +74,20 @@ class LifoServiceQueue final {
return max_queue_size_;
}
- // Get an element from the queue. Returns false if we were shut down prior to
- // getting the element.
+ // Get an element from the queue. Returns false if the queue is shut down.
bool BlockingGet(std::unique_ptr<InboundCall>* out);
// Add a new call to the queue.
// Returns:
// - QUEUE_SHUTDOWN if Shutdown() has already been called.
- // - QUEUE_FULL if the queue is full and 'call' has a later deadline than any
- // RPC already in the queue.
+ // - QUEUE_FULL if the queue is full and 'call' has a later deadline than most
+ // of the RPCs already in the queue.
// - QUEUE_SUCCESS if 'call' was enqueued.
//
// In the case of a 'QUEUE_SUCCESS' response, the new element may have bumped
// another call out of the queue. In that case, *evicted will be set to the
// call that was bumped.
- QueueStatus Put(InboundCall* call, std::optional<InboundCall*>* evicted);
+ QueueStatus Put(InboundCall* call, InboundCall** evicted);
// Shut down the queue.
// When a blocking queue is shut down, no more elements can be added to it,
@@ -110,8 +102,8 @@ class LifoServiceQueue final {
FRIEND_TEST(TestServiceQueue, LifoServiceQueuePerf);
// Comparison function which orders calls by their deadlines.
- static bool DeadlineLess(const InboundCall* a,
- const InboundCall* b) {
+ static bool DeadlineGreater(const InboundCall* a,
+ const InboundCall* b) {
auto time_a = a->GetClientDeadline();
auto time_b = b->GetClientDeadline();
if (time_a == time_b) {
@@ -120,21 +112,21 @@ class LifoServiceQueue final {
time_a = a->GetTimeReceived();
time_b = b->GetTimeReceived();
}
- return time_a < time_b;
+ return time_a > time_b;
}
- // Struct functor wrapper for DeadlineLess.
- struct DeadlineLessStruct {
- bool operator()(const InboundCall* a, const InboundCall* b) const {
- return DeadlineLess(a, b);
+ // Struct functor wrapper for DeadlineGreater.
+ static const struct DeadlineGreaterStruct {
+ bool operator()(const InboundCall* a, const InboundCall* b) const noexcept {
+ return DeadlineGreater(a, b);
}
- };
+ } kMinHeapCompare;
// The thread-local record corresponding to a single consumer thread.
// Threads push this record onto the waiting_consumers_ stack when
// they are awaiting work. Producers pop the top waiting consumer and
// post work using Post().
- class ConsumerState {
+ class ConsumerState final {
public:
explicit ConsumerState(LifoServiceQueue* queue) :
cond_(&lock_),
@@ -180,10 +172,7 @@ class LifoServiceQueue final {
// Return an estimate of the current queue length.
size_t estimated_queue_length() const {
ANNOTATE_IGNORE_READS_BEGIN();
- // The C++ standard says that std::multiset::size must be constant time,
- // so this method won't try to traverse any actual nodes of the underlying
- // RB tree. Investigation of the libstdcxx implementation confirms that
- // size() is a simple field access of the _Rb_tree structure.
+ // Size of a vector is a simple field access so this is safe.
auto ret = queue_.size();
ANNOTATE_IGNORE_READS_END();
return ret;
@@ -207,9 +196,10 @@ class LifoServiceQueue final {
// Stack of consumer threads which are currently waiting for work.
std::vector<ConsumerState*> waiting_consumers_;
- // The actual queue. Work is only added to the queue when there were no
- // consumers available for a "direct hand-off".
- std::multiset<InboundCall*, DeadlineLessStruct> queue_;
+ // The container backing the queue. This is operated as a min priority queue.
+ // Items are only added into the queue when there aren't any consumers
+ // available for a "direct hand-off".
+ std::vector<InboundCall*> queue_;
// The total set of consumers who have ever accessed this queue.
// This container is necessary to maintain proper lifecycle and ownership