尝试实现一个基于原子操作的多生产者-单消费者队列
来自 codereview。
我正在尝试使用原子操作实现一个多生产者单消费者（MPSC）队列。
目前我遇到了一些竞态条件，难以解决。
==================
WARNING: ThreadSanitizer: data race (pid=1)
Atomic write of size 8 at 0x720400000008 by thread T3:
#0 std::__atomic_base<mpsc_queue_node<int>*>::store(mpsc_queue_node<int>*, std::memory_order) /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/atomic_base.h:806 (output.s+0x4115f2) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#1 std::atomic<mpsc_queue_node<int>*>::store(mpsc_queue_node<int>*, std::memory_order) /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/atomic:578 (output.s+0x4115f2)
#2 mpsc_queue<int>::push(mpsc_queue_node<int>*) /app/example.cpp:23 (output.s+0x40f105) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#3 mpsc_queue<int>::push(int const&) /app/example.cpp:53 (output.s+0x40ae7a) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#4 operator() /app/example.cpp:100 (output.s+0x4046f6) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#5 __invoke_impl<void, main()::<lambda(int)>, int> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/invoke.h:63 (output.s+0x40524c) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#6 __invoke<main()::<lambda(int)>, int> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/invoke.h:98 (output.s+0x4051a9) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#7 _M_invoke<0, 1> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:303 (output.s+0x4050d0) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#8 operator() /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:310 (output.s+0x405060) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#9 _M_run /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:255 (output.s+0x40501a) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#10 <null> <null> (libstdc++.so.6+0xf2ca3) (BuildId: fac422a459bf619e821fd13769b77c0b934f64b1)
Previous write of size 8 at 0x720400000008 by thread T1:
#0 operator new(unsigned long) <null> (libtsan.so.2+0xa7421) (BuildId: d5d680a1035ee999a8de76940812285b233760e5)
#1 mpsc_queue<int>::push(int const&) /app/example.cpp:52 (output.s+0x40ae14) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#2 operator() /app/example.cpp:100 (output.s+0x4046f6) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#3 __invoke_impl<void, main()::<lambda(int)>, int> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/invoke.h:63 (output.s+0x40524c) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#4 __invoke<main()::<lambda(int)>, int> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/invoke.h:98 (output.s+0x4051a9) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#5 _M_invoke<0, 1> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:303 (output.s+0x4050d0) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#6 operator() /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:310 (output.s+0x405060) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#7 _M_run /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:255 (output.s+0x40501a) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#8 <null> <null> (libstdc++.so.6+0xf2ca3) (BuildId: fac422a459bf619e821fd13769b77c0b934f64b1)
Location is heap block of size 16 at 0x720400000000 allocated by thread T1:
#0 operator new(unsigned long) <null> (libtsan.so.2+0xa7421) (BuildId: d5d680a1035ee999a8de76940812285b233760e5)
#1 mpsc_queue<int>::push(int const&) /app/example.cpp:52 (output.s+0x40ae14) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#2 operator() /app/example.cpp:100 (output.s+0x4046f6) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#3 __invoke_impl<void, main()::<lambda(int)>, int> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/invoke.h:63 (output.s+0x40524c) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#4 __invoke<main()::<lambda(int)>, int> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/invoke.h:98 (output.s+0x4051a9) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#5 _M_invoke<0, 1> /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:303 (output.s+0x4050d0) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#6 operator() /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:310 (output.s+0x405060) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#7 _M_run /cefs/22/22e6cdc013c8541ce3d1548e_consolidated/compilers_c++_x86_gcc_15.2.0/include/c++/15.2.0/bits/std_thread.h:255 (output.s+0x40501a) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
#8 <null> <null> (libstdc++.so.6+0xf2ca3) (BuildId: fac422a459bf619e821fd13769b77c0b934f64b1)
Thread T3 (tid=5, running) created by main thread at:
#0 pthread_create <null> (libtsan.so.2+0x5f090) (BuildId: d5d680a1035ee999a8de76940812285b233760e5)
#1 std::thread::_M_start_thread(std::unique_ptr<std::thread::_State, std::default_delete<std::thread::_State> >, void (*)()) <null> (libstdc++.so.6+0xf2d78) (BuildId: fac422a459bf619e821fd13769b77c0b934f64b1)
#2 main /app/example.cpp:104 (output.s+0x40485b) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
Thread T1 (tid=3, finished) created by main thread at:
#0 pthread_create <null> (libtsan.so.2+0x5f090) (BuildId: d5d680a1035ee999a8de76940812285b233760e5)
#1 std::thread::_M_start_thread(std::unique_ptr<std::thread::_State, std::default_delete<std::thread::_State> >, void (*)()) <null> (libstdc++.so.6+0xf2d78) (BuildId: fac422a459bf619e821fd13769b77c0b934f64b1)
#2 main /app/example.cpp:104 (output.s+0x40485b) (BuildId: 45ca742e10c82fb540a7762fff55fea8a9eaba6a)
SUMMARY: ThreadSanitizer: data race /app/example.cpp:23 in mpsc_queue<int>::push(mpsc_queue_node<int>*)
==================
代码：
#include <atomic>
#include <latch>
#include <optional>
#include <print>
#include <thread>
#include <vector>
template <typename T>
struct mpsc_queue_node {
T value;
std::atomic<mpsc_queue_node<T>*> next;
};
template <typename T>
class mpsc_queue {
std::atomic<mpsc_queue_node<T>*> m_head{nullptr};
std::atomic<mpsc_queue_node<T>*> m_tail{nullptr};
void push(mpsc_queue_node<T>* node) {
mpsc_queue_node<T>* expected_empty_value = nullptr;
auto tail_ptr = m_tail.load(std::memory_order_relaxed);
if (tail_ptr != nullptr) {
tail_ptr->next.store(node, std::memory_order_release);
}
while (!m_tail.compare_exchange_weak(tail_ptr, node,
std::memory_order_release,
std::memory_order_relaxed)) {
; // try again
}
if (tail_ptr == nullptr) {
m_head.compare_exchange_strong(expected_empty_value, node,
std::memory_order_release,
std::memory_order_relaxed);
}
}
public:
~mpsc_queue() {
while (pop()) {
;
}
}
template <class... Args>
void emplace(Args&&... args) {
mpsc_queue_node<T>* node =
new mpsc_queue_node<T>{{std::forward<Args>(args)...}, nullptr};
push(node);
}
void push(const T& value) {
mpsc_queue_node<T>* node = new mpsc_queue_node<T>{value, nullptr};
push(node);
}
void push(T&& value) {
mpsc_queue_node<T>* node =
new mpsc_queue_node<T>{std::move(value), nullptr};
push(node);
}
std::optional<T> pop() {
auto head_ptr = m_head.load(std::memory_order_acquire);
if (head_ptr == nullptr) {
return std::nullopt;
}
auto head_next_ptr = head_ptr->next.load(std::memory_order_acquire);
while (!m_head.compare_exchange_weak(head_ptr, head_next_ptr,
std::memory_order_acq_rel,
std::memory_order_relaxed)) {
; // try again
}
auto tail_ptr = m_tail.load(std::memory_order_acquire);
// if this was the only node in the queue
if (tail_ptr == head_ptr && head_next_ptr == nullptr) {
while (!m_tail.compare_exchange_weak(tail_ptr, nullptr,
std::memory_order_release,
std::memory_order_relaxed))
{
; // try again
}
}
std::optional<T> popped_value = std::move(head_ptr->value);
delete head_ptr;
return popped_value;
}
};
int main() {
mpsc_queue<int> q;
std::vector<std::thread> threads(8);
std::latch pop_done{static_cast<std::ptrdiff_t>(threads.size())};
int i = 0;
auto push_fn = [&q](int i) {
std::println("Pushing {}", i);
q.push(i);
};
for (auto& t : threads) {
t = std::thread(push_fn, i);
i++;
}
while (!pop_done.try_wait()) {
auto value = q.pop();
if (value) {
std::println("Popped {}", value.value());
pop_done.count_down();
}
}
for (auto& t : threads) {
t.join();
}
}
请指教。
解决方案
请考虑下面这几行：
auto tail_ptr = m_tail.load(std::memory_order_relaxed);
if (tail_ptr != nullptr) {
tail_ptr->next.store(node, std::memory_order_release);
}
两条线程可以同时运行这段代码，因此两者都会覆盖 tail_ptr->next。你不能在同一时间原子地同时修改 tail_ptr->next 和 m_tail。常见的解决方案是改为将节点推送到队列的头部。这样，你可以准备一个新节点，其 next 指向当前的 m_head，然后用新的 node 与 m_head 进行比较并交换。如果成功，你就完成了。如果不成功，m_head 已被另一线程修改，你必须先把 next 更新为 m_head 的新值，然后再尝试更新后者。
当然，这样现在又给 pop() 带来一个问题，因为要成为一个FIFO队列，你需要弹出尾部，然后还需要更新倒数第二个元素的 next，这两件事同样不能原子地同时完成。
一个可能的解决方案是也从前端弹出，但这会把它变成一个后进先出（LIFO）队列。另一种方案是让 pop() 原子地将当前队列与一个空队列交换，并返回旧队列。这样，消费者就可以按自己的节奏处理旧队列，而生产者会把元素推送到新队列。
站内所有文章版权归属LeftHeroAI导航站，无授权禁止任何主体转载、抄袭、复制内容，亦不得私自架设镜像站点。一经侵权，本站将通过法律途径追责。
相关文章
汇集全球数千款 AI 工具，每日持续更新，配套 AI 开发学习资源，轻松用 AI 提升工作效率。快捷键 Ctrl+D/⌘+D 一键收藏本站。