This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new aff4d82e8 Fix MPSC queue publication races in butil, bthread, and
Socket (#3671)
aff4d82e8 is described below
commit aff4d82e82f47a0a35be532d9add6a06828ea84a
Author: nkysg <[email protected]>
AuthorDate: Sun Nov 8 12:29:39 2026 +0800
Fix MPSC queue publication races in butil, bthread, and Socket (#3671)
* Fix MPSCQueue publication races
* Fix execution and socket queue publication races
---
src/brpc/socket.cpp | 46 ++++++++++++++++-------------
src/bthread/execution_queue.cpp | 28 +++++++++---------
src/bthread/execution_queue_inl.h | 15 +++++-----
src/butil/containers/mpsc_queue.h | 33 +++++++++++----------
test/mpsc_queue_unittest.cc | 61 +++++++++++++++++++++++++++++++++++++++
5 files changed, 127 insertions(+), 56 deletions(-)
diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp
index c4411c960..2985a35b6 100644
--- a/src/brpc/socket.cpp
+++ b/src/brpc/socket.cpp
@@ -311,7 +311,7 @@ struct BAIDU_CACHELINE_ALIGNMENT Socket::WriteRequest {
static WriteRequest* const UNCONNECTED;
butil::IOBuf data;
- WriteRequest* next;
+ butil::atomic<WriteRequest*> next;
bthread_id_t id_wait;
void clear_and_set_control_bits(bool notify_on_success,
@@ -533,8 +533,9 @@ void Socket::ReturnFailedWriteRequest(Socket::WriteRequest*
p, int error_code,
Socket::WriteRequest* Socket::ReleaseWriteRequestsExceptLast(
Socket::WriteRequest* req, int error_code, const std::string& error_text) {
WriteRequest* p = req;
- while (p->next != nullptr) {
- WriteRequest* const saved_next = p->next;
+ while (p->next.load(butil::memory_order_relaxed) != nullptr) {
+ WriteRequest* const saved_next =
+ p->next.load(butil::memory_order_relaxed);
ReturnFailedWriteRequest(p, error_code, error_text);
p = saved_next;
}
@@ -1195,7 +1196,7 @@ int Socket::Status(SocketId id, int32_t* nref) {
bool Socket::IsWriteComplete(Socket::WriteRequest* old_head,
bool singular_node,
Socket::WriteRequest** new_tail) {
- CHECK(nullptr == old_head->next);
+ CHECK(nullptr == old_head->next.load(butil::memory_order_relaxed));
// Try to set _write_head to nullptr to mark that the write is done.
WriteRequest* new_head = old_head;
WriteRequest* desired = nullptr;
@@ -1222,23 +1223,25 @@ bool Socket::IsWriteComplete(Socket::WriteRequest*
old_head,
WriteRequest* tail = nullptr;
WriteRequest* p = new_head;
do {
- while (p->next == WriteRequest::UNCONNECTED) {
+ WriteRequest* saved_next;
+ while ((saved_next = p->next.load(butil::memory_order_acquire)) ==
+ WriteRequest::UNCONNECTED) {
// TODO(gejun): elaborate this
sched_yield();
}
- WriteRequest* const saved_next = p->next;
- p->next = tail;
+ p->next.store(tail, butil::memory_order_relaxed);
tail = p;
p = saved_next;
CHECK(p != nullptr);
} while (p != old_head);
// Link old list with new list.
- old_head->next = tail;
+ old_head->next.store(tail, butil::memory_order_relaxed);
// Call Setup() from oldest to newest, notice that the calling sequence
// matters for protocols using pipelined_count, this is why we don't
// call Setup in above loop which is from newest to oldest.
- for (WriteRequest* q = tail; q; q = q->next) {
+ for (WriteRequest* q = tail; q;
+ q = q->next.load(butil::memory_order_relaxed)) {
q->Setup(this);
}
if (new_tail) {
@@ -1669,7 +1672,7 @@ int Socket::Write(butil::IOBuf* data, const WriteOptions*
options_in) {
req->data.swap(*data);
// Set `req->next' to UNCONNECTED so that the KeepWrite thread will
// wait until it points to a valid WriteRequest or nullptr.
- req->next = WriteRequest::UNCONNECTED;
+ req->next.store(WriteRequest::UNCONNECTED, butil::memory_order_relaxed);
req->id_wait = opt.id_wait;
req->clear_and_set_control_bits(opt.notify_on_success, opt.shutdown_write);
req->set_pipelined_count_and_user_message(
@@ -1706,7 +1709,7 @@ int Socket::Write(SocketMessagePtr<>& msg, const
WriteOptions* options_in) {
// Set `req->next' to UNCONNECTED so that the KeepWrite thread will
// wait until it points to a valid WriteRequest or nullptr.
- req->next = WriteRequest::UNCONNECTED;
+ req->next.store(WriteRequest::UNCONNECTED, butil::memory_order_relaxed);
req->id_wait = opt.id_wait;
req->clear_and_set_control_bits(opt.notify_on_success, opt.shutdown_write);
req->set_pipelined_count_and_user_message(
@@ -1724,7 +1727,7 @@ int Socket::StartWrite(WriteRequest* req, const
WriteOptions& opt) {
// lock-free, but the duration is so short(1~2 instructions,
// depending on compiler) that the spin rarely occurs in practice
// (I've not seen any spin in highly contended tests).
- req->next = prev_head;
+ req->next.store(prev_head, butil::memory_order_release);
return 0;
}
@@ -1737,7 +1740,7 @@ int Socket::StartWrite(WriteRequest* req, const
WriteOptions& opt) {
int ret = 0;
// We've got the right to write.
- req->next = nullptr;
+ req->next.store(nullptr, butil::memory_order_relaxed);
// Fast fail when write has been shutdown.
if (_is_write_shutdown) {
@@ -1827,10 +1830,11 @@ void* Socket::KeepWrite(void* void_arg) {
do {
// req was written, skip it.
bool need_shutdown = false;
- if (req->next != nullptr && req->data.empty()) {
+ if (req->next.load(butil::memory_order_relaxed) != nullptr &&
+ req->data.empty()) {
WriteRequest* const saved_req = req;
need_shutdown = req->need_shutdown_write();
- req = req->next;
+ req = req->next.load(butil::memory_order_relaxed);
s->ReturnSuccessfulWriteRequest(saved_req);
}
if (need_shutdown) {
@@ -1851,10 +1855,11 @@ void* Socket::KeepWrite(void* void_arg) {
s->AddOutputBytes(nw);
}
// Release WriteRequest until non-empty data, last request or shutdown
write.
- while (req->next != nullptr && req->data.empty()) {
+ while (req->next.load(butil::memory_order_relaxed) != nullptr &&
+ req->data.empty()) {
WriteRequest* const saved_req = req;
need_shutdown = req->need_shutdown_write();
- req = req->next;
+ req = req->next.load(butil::memory_order_relaxed);
s->ReturnSuccessfulWriteRequest(saved_req);
if (need_shutdown) {
break;
@@ -1884,8 +1889,9 @@ void* Socket::KeepWrite(void* void_arg) {
}
}
if (nullptr == cur_tail) {
- for (cur_tail = req; cur_tail->next != nullptr;
- cur_tail = cur_tail->next);
+ for (cur_tail = req;
+ cur_tail->next.load(butil::memory_order_relaxed) != nullptr;
+ cur_tail = cur_tail->next.load(butil::memory_order_relaxed));
}
// Return when there's no more WriteRequests and req is completely
// written.
@@ -1906,7 +1912,7 @@ ssize_t Socket::DoWrite(WriteRequest* req) {
butil::IOBuf* data_list[DATA_LIST_MAX];
size_t ndata = 0;
for (WriteRequest* p = req; p != nullptr && ndata < DATA_LIST_MAX;
- p = p->next) {
+ p = p->next.load(butil::memory_order_relaxed)) {
data_list[ndata++] = &p->data;
if (p->need_shutdown_write()) {
// Write WriteRequest until shutdown write.
diff --git a/src/bthread/execution_queue.cpp b/src/bthread/execution_queue.cpp
index de4069b12..2d48deb50 100644
--- a/src/bthread/execution_queue.cpp
+++ b/src/bthread/execution_queue.cpp
@@ -66,7 +66,7 @@ inline ExecutionQueueVars* get_execq_vars() {
}
void ExecutionQueueBase::start_execute(TaskNode* node) {
- node->next = TaskNode::UNCONNECTED;
+ node->next.store(TaskNode::UNCONNECTED, butil::memory_order_relaxed);
node->status = TaskNode::UNEXECUTED;
node->iterated = false;
if (node->high_priority) {
@@ -79,12 +79,12 @@ void ExecutionQueueBase::start_execute(TaskNode* node) {
}
TaskNode* const prev_head = _head.exchange(node,
butil::memory_order_release);
if (prev_head != nullptr) {
- node->next = prev_head;
+ node->next.store(prev_head, butil::memory_order_release);
return;
}
// Get the right to execute the task, start a bthread to avoid deadlock
// or stack overflow
- node->next = nullptr;
+ node->next.store(nullptr, butil::memory_order_relaxed);
node->q = this;
ExecutionQueueVars* const vars = get_execq_vars();
@@ -149,9 +149,9 @@ void* ExecutionQueueBase::_execute_tasks(void* arg) {
bool destroy_queue = false;
for (;;) {
if (head->iterated) {
- CHECK(head->next != nullptr);
+ CHECK(head->next.load(butil::memory_order_relaxed) != nullptr);
TaskNode* saved_head = head;
- head = head->next;
+ head = head->next.load(butil::memory_order_relaxed);
m->return_task_node(saved_head);
}
int rc = 0;
@@ -172,14 +172,16 @@ void* ExecutionQueueBase::_execute_tasks(void* arg) {
destroy_queue = true;
}
// Release TaskNode until uniterated task or last task
- while (head->next != nullptr && head->iterated) {
+ while (head->next.load(butil::memory_order_relaxed) != nullptr &&
+ head->iterated) {
TaskNode* saved_head = head;
- head = head->next;
+ head = head->next.load(butil::memory_order_relaxed);
m->return_task_node(saved_head);
}
if (cur_tail == nullptr) {
- for (cur_tail = head; cur_tail->next != nullptr;
- cur_tail = cur_tail->next) {}
+ for (cur_tail = head;
+ cur_tail->next.load(butil::memory_order_relaxed) != nullptr;
+ cur_tail = cur_tail->next.load(butil::memory_order_relaxed))
{}
}
// break when no more tasks and head has been executed
if (!m->_more_tasks(cur_tail, &cur_tail, !head->iterated)) {
@@ -301,7 +303,7 @@ int ExecutionQueueBase::stop() {
int ExecutionQueueBase::_execute(TaskNode* head, bool high_priority, int*
niterated) {
if (head != nullptr && head->stop_task) {
- CHECK(head->next == nullptr);
+ CHECK(head->next.load(butil::memory_order_relaxed) == nullptr);
head->iterated = true;
head->status = TaskNode::EXECUTED;
TaskIteratorBase iter(nullptr, this, true, false);
@@ -431,7 +433,7 @@ void TaskIteratorBase::operator++() {
return;
}
if (_cur_node->iterated) {
- _cur_node = _cur_node->next;
+ _cur_node = _cur_node->next.load(butil::memory_order_relaxed);
}
if (should_break_for_high_priority_tasks()) {
return;
@@ -447,7 +449,7 @@ void TaskIteratorBase::operator++() {
_num_iterated += !_cur_node->iterated;
_cur_node->iterated = true;
}
- _cur_node = _cur_node->next;
+ _cur_node = _cur_node->next.load(butil::memory_order_relaxed);
}
return;
}
@@ -463,7 +465,7 @@ TaskIteratorBase::~TaskIteratorBase() {
if (_head->iterated && _head->high_priority == _high_priority) {
_head->set_executed();
}
- _head = _head->next;
+ _head = _head->next.load(butil::memory_order_relaxed);
}
if (_should_break && _cur_node != nullptr
&& _cur_node->high_priority == _high_priority &&
_cur_node->iterated) {
diff --git a/src/bthread/execution_queue_inl.h
b/src/bthread/execution_queue_inl.h
index 2a85d27be..23c753311 100644
--- a/src/bthread/execution_queue_inl.h
+++ b/src/bthread/execution_queue_inl.h
@@ -57,7 +57,7 @@ struct BAIDU_CACHELINE_ALIGNMENT TaskNode {
, iterated(false)
, high_priority(false)
, in_place(false)
- , next(UNCONNECTED)
+ , next(nullptr)
, q(nullptr)
{}
~TaskNode() {}
@@ -91,7 +91,7 @@ struct BAIDU_CACHELINE_ALIGNMENT TaskNode {
bool iterated;
bool high_priority;
bool in_place;
- TaskNode* next;
+ butil::atomic<TaskNode*> next;
ExecutionQueueBase* q;
union {
char static_task_mem[56]; // Make sizeof TaskNode exactly 128 bytes
@@ -486,7 +486,7 @@ inline bool ExecutionQueueBase::_more_tasks(
TaskNode* old_head, TaskNode** new_tail,
bool has_uniterated) {
- CHECK(old_head->next == nullptr);
+ CHECK(old_head->next.load(butil::memory_order_relaxed) == nullptr);
// Try to set _head to nullptr to mark that the execute is done.
TaskNode* new_head = old_head;
TaskNode* desired = nullptr;
@@ -512,19 +512,20 @@ inline bool ExecutionQueueBase::_more_tasks(
}
TaskNode* p = new_head;
do {
- while (p->next == TaskNode::UNCONNECTED) {
+ TaskNode* saved_next;
+ while ((saved_next = p->next.load(butil::memory_order_acquire)) ==
+ TaskNode::UNCONNECTED) {
// TODO(gejun): elaborate this
sched_yield();
}
- TaskNode* const saved_next = p->next;
- p->next = tail;
+ p->next.store(tail, butil::memory_order_relaxed);
tail = p;
p = saved_next;
CHECK(p != nullptr);
} while (p != old_head);
// Link old list with new list.
- old_head->next = tail;
+ old_head->next.store(tail, butil::memory_order_relaxed);
return true;
}
diff --git a/src/butil/containers/mpsc_queue.h
b/src/butil/containers/mpsc_queue.h
index ff5aee0b2..87fd7ac22 100644
--- a/src/butil/containers/mpsc_queue.h
+++ b/src/butil/containers/mpsc_queue.h
@@ -22,6 +22,7 @@
#ifndef BUTIL_MPSC_QUEUE_H
#define BUTIL_MPSC_QUEUE_H
+#include <new>
#include "butil/object_pool.h"
#include "butil/type_traits.h"
#include "butil/memory/manual_constructor.h"
@@ -32,7 +33,7 @@ template <typename T>
struct BAIDU_CACHELINE_ALIGNMENT MPSCQueueNode {
static MPSCQueueNode* const UNCONNECTED;
- MPSCQueueNode* next{nullptr};
+ atomic<MPSCQueueNode*> next{nullptr};
ManualConstructor<T> data_mem;
};
@@ -107,8 +108,8 @@ void MPSCQueue<T, Alloc>::Enqueue(T&& data) {
template <typename T, typename Alloc>
template <typename U>
void MPSCQueue<T, Alloc>::EnqueueInternal(U&& data) {
- auto node = (MPSCQueueNode<T>*)_alloc.Alloc();
- node->next = MPSCQueueNode<T>::UNCONNECTED;
+ auto node = new (_alloc.Alloc()) MPSCQueueNode<T>;
+ node->next.store(MPSCQueueNode<T>::UNCONNECTED, memory_order_relaxed);
node->data_mem.Init(std::forward<U>(data));
EnqueueImpl(node);
}
@@ -117,11 +118,11 @@ template <typename T, typename Alloc>
void MPSCQueue<T, Alloc>::EnqueueImpl(MPSCQueueNode<T>* node) {
MPSCQueueNode<T>* prev = _head.exchange(node, memory_order_release);
if (prev) {
- node->next = prev;
+ node->next.store(prev, memory_order_release);
return;
}
- node->next = nullptr;
- _cur_enqueue_node.store(node, memory_order_relaxed);
+ node->next.store(nullptr, memory_order_relaxed);
+ _cur_enqueue_node.store(node, memory_order_release);
}
template <typename T, typename Alloc>
@@ -135,7 +136,7 @@ bool MPSCQueue<T, Alloc>::DequeueImpl(T* data) {
if (_cur_dequeue_node) {
node = _cur_dequeue_node;
} else {
- node = _cur_enqueue_node.exchange(nullptr, memory_order_relaxed);
+ node = _cur_enqueue_node.exchange(nullptr, memory_order_acquire);
}
if (!node) {
return false;
@@ -146,10 +147,10 @@ bool MPSCQueue<T, Alloc>::DequeueImpl(T* data) {
*data = std::move(*mem);
}
MPSCQueueNode<T>* old_node = node;
- if (!node->next) {
+ if (!node->next.load(memory_order_relaxed)) {
ReverseList(node);
}
- _cur_dequeue_node = node->next;
+ _cur_dequeue_node = node->next.load(memory_order_relaxed);
node->data_mem.Destroy();
_alloc.Free(old_node);
@@ -167,27 +168,27 @@ void MPSCQueue<T, Alloc>::ReverseList(MPSCQueueNode<T>*
old_head) {
return;
}
CHECK_NE(new_head, old_head);
- // Above acquire fence pairs release fence of exchange in Enqueue() to make
- // sure that we see all fields of requests set.
+ // Acquire the new head's initialization, including its UNCONNECTED link.
// Someone added new requests.
// Reverse the list until old_head.
MPSCQueueNode<T>* tail = nullptr;
MPSCQueueNode<T>* p = new_head;
do {
- while (p->next == MPSCQueueNode<T>::UNCONNECTED) {
- // TODO(gejun): elaborate this
+ MPSCQueueNode<T>* saved_next;
+ // The producer connects the link after publishing the node in _head.
+ while ((saved_next = p->next.load(memory_order_acquire)) ==
+ MPSCQueueNode<T>::UNCONNECTED) {
sched_yield();
}
- MPSCQueueNode<T>* const saved_next = p->next;
- p->next = tail;
+ p->next.store(tail, memory_order_relaxed);
tail = p;
p = saved_next;
CHECK(p);
} while (p != old_head);
// Link old list with new list.
- old_head->next = tail;
+ old_head->next.store(tail, memory_order_relaxed);
}
}
diff --git a/test/mpsc_queue_unittest.cc b/test/mpsc_queue_unittest.cc
index 97fb9ee45..0a741863a 100644
--- a/test/mpsc_queue_unittest.cc
+++ b/test/mpsc_queue_unittest.cc
@@ -1,5 +1,9 @@
#include <gtest/gtest.h>
#include <pthread.h>
+#include <chrono>
+#include <thread>
+#include <utility>
+#include <vector>
#include "butil/containers/mpsc_queue.h"
namespace {
@@ -120,6 +124,63 @@ TEST(MPSCQueueTest, mpsc_multi_thread) {
}
+template <typename Alloc>
+void CheckConcurrentPublication() {
+ const unsigned producer_count = 4;
+ const unsigned items_per_producer = 10000;
+ const unsigned total = producer_count * items_per_producer;
+ using Item = std::pair<unsigned, unsigned>;
+ butil::MPSCQueue<Item, Alloc> queue;
+ butil::atomic<bool> start(false);
+ std::vector<std::thread> producers;
+ for (unsigned p = 0; p < producer_count; ++p) {
+ producers.emplace_back([&, p] {
+ while (!start.load(butil::memory_order_acquire)) {
+ std::this_thread::yield();
+ }
+ for (unsigned i = 0; i < items_per_producer; ++i) {
+ const unsigned value = p * items_per_producer + i;
+ queue.Enqueue(Item(value, ~value));
+ }
+ });
+ }
+ std::vector<bool> seen(total, false);
+ unsigned consumed = 0;
+ const auto deadline = std::chrono::steady_clock::now() +
+ std::chrono::seconds(30);
+ start.store(true, butil::memory_order_release);
+ while (consumed < total && std::chrono::steady_clock::now() < deadline) {
+ Item item;
+ if (!queue.Dequeue(item)) {
+ std::this_thread::yield();
+ continue;
+ }
+ EXPECT_EQ(~item.first, item.second);
+ EXPECT_LT(item.first, total);
+ if (item.first < total) {
+ EXPECT_FALSE(seen[item.first]);
+ seen[item.first] = true;
+ }
+ ++consumed;
+ }
+ for (auto& producer : producers) {
+ producer.join();
+ }
+ EXPECT_EQ(total, consumed);
+ Item item;
+ EXPECT_FALSE(queue.Dequeue(item));
+}
+
+TEST(MPSCQueueTest, concurrent_publication) {
+ CheckConcurrentPublication<
+ butil::DefaultAllocator<std::pair<unsigned, unsigned>>>();
+}
+
+TEST(MPSCQueueTest, concurrent_publication_with_object_pool) {
+ CheckConcurrentPublication<
+ butil::ObjectPoolAllocator<std::pair<unsigned, unsigned>>>();
+}
+
struct MyObject {};
struct DestructionTracked {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
