【发布时间】：2020-07-26 16:04:49
【问题描述】：
我设计了这个函数，用来实现Lock-Free队列，但是在实际执行过程中(dequeue)存在死锁问题。我检查了很多次，我认为这很好。 我在 x86 平台上运行，有 12 个线程可以读写。
现在我想弄清楚是什么导致了这种情况，我想知道这是否是线程安全的设计，还是需要继续优化以获得更高的性能。
12 个线程出队和 12 个线程入队。
开发工具：Visual Studio 2019
我非常期待您的回复。谢谢你。
#include <iostream>
#include <functional>
#include<atomic>
#include<cassert>
#include<thread>
#include<vector>
template<typename T>
class mpmc_queue_t
{
public:
mpmc_queue_t(size_t size) :
_size(size),
_mask(_size - 1),
_buffer((node_t*)(new aligned_node_t[_size]))
{
assert((_size != 0) && ((_size & (~_size + 1)) == _size));
_read.store(0, std::memory_order_relaxed);
_write.store(0, std::memory_order_relaxed);
for (size_t i = 0; i < _size; ++i)
{
_buffer[i].status.store(false, std::memory_order_relaxed);
}
}
~mpmc_queue_t()
{
delete[] _buffer;
}
bool enqueue(const T& data)
{
auto write = _write.fetch_add(1, std::memory_order_relaxed);
node_t* node = &_buffer[write & _mask];
while (true)
{
if (!node->status.load(std::memory_order_acquire))
{
node->data = data;
node->status.store(true, std::memory_order_release);
return true;
}
std::this_thread::yield();
}
}
bool dequeue(T& data)
{
auto read = _read.fetch_add(1, std::memory_order_relaxed);
node_t* node = &_buffer[read & _mask];
while (true)
{
if (node->status.load(std::memory_order_acquire))
{
data = node->data;
node->status.store(false, std::memory_order_release);
return true;
}
std::this_thread::yield();
}
}
private:
struct node_t
{
T data;
std::atomic_bool status;
};
typedef typename std::aligned_storage<sizeof(node_t), std::alignment_of<node_t>::value>::type aligned_node_t;
typedef char cache_line_pad_t[64];
cache_line_pad_t _pad0;
size_t _size;
size_t _mask;
node_t* const _buffer;
cache_line_pad_t _pad1;
std::atomic_size_t _read;
cache_line_pad_t _pad2;
std::atomic_size_t _write;
cache_line_pad_t _pad3;
};
#define COUNT 100000000
#define THREAD 12
typedef mpmc_queue_t<size_t> queue_t;
template<typename T>
void consumer_func(T* queue)
{
size_t count = COUNT;
size_t value = 0;
while (count > 0) {
if (queue->dequeue(value)) {
--count;
}
}
std::cout << "consumer_func ID: " << std::this_thread::get_id() << " ok" << std::endl;
}
template<typename T>
void producer_func(T* queue)
{
size_t count = COUNT;
while (count > 0) {
if (queue->enqueue(count)) {
--count;
}
}
std::cout << "producer_func ID: " << std::this_thread::get_id() << " ok" << std::endl;
}
template<typename T>
long double
run_test(
T producer_func,
T consumer_func)
{
typedef std::chrono::high_resolution_clock clock_t;
typedef std::chrono::time_point<clock_t> time_t;
time_t start;
time_t end;
start = clock_t::now();
std::thread producer0(producer_func);
std::thread producer1(producer_func);
std::thread producer2(producer_func);
std::thread producer3(producer_func);
std::thread producer4(producer_func);
std::thread producer5(producer_func);
std::thread producer6(producer_func);
std::thread producer7(producer_func);
std::thread producer8(producer_func);
std::thread producer9(producer_func);
std::thread producer10(producer_func);
std::thread producer11(producer_func);
std::thread consumer0(consumer_func);
std::thread consumer1(consumer_func);
std::thread consumer2(consumer_func);
std::thread consumer3(consumer_func);
std::thread consumer4(consumer_func);
std::thread consumer5(consumer_func);
std::thread consumer6(consumer_func);
std::thread consumer7(consumer_func);
std::thread consumer8(consumer_func);
std::thread consumer9(consumer_func);
std::thread consumer10(consumer_func);
std::thread consumer11(consumer_func);
producer0.join();
producer1.join();
producer2.join();
producer3.join();
producer4.join();
producer5.join();
producer6.join();
producer7.join();
producer8.join();
producer9.join();
producer10.join();
producer11.join();
consumer0.join();
consumer1.join();
consumer2.join();
consumer3.join();
consumer4.join();
consumer5.join();
consumer6.join();
consumer7.join();
consumer8.join();
consumer9.join();
consumer10.join();
consumer11.join();
end = clock_t::now();
return
(end - start).count()
* ((double)std::chrono::high_resolution_clock::period::num
/ std::chrono::high_resolution_clock::period::den);
}
int main()
{
{
queue_t queue(65536);
long double seconds = run_test(std::bind(&producer_func<queue_t>, &queue),
std::bind(&consumer_func<queue_t>, &queue));
std::cout << "The control group completed "
<< COUNT * THREAD
<< " iterations in "
<< seconds
<< " seconds. "
<< ((long double)COUNT * THREAD / seconds) / 1000000
<< " million enqueue/dequeue pairs per second."
<< std::endl;
}
return 0;
}
【问题讨论】：
-
除非您的缓冲区非常大，否则我看不出有任何方法可以确保在您等待时读取或写入指针不会一直环绕。其他线程可以在一次调用
yield期间执行很多 入队和出队 -
我会先修复一些通用的 C++ 问题，例如可复制性和可分配性以及带有警告的编译。此外，该课程以使用
reinterpret_cast的谎言开始其生命。想想这在这里做了什么以及这两种指针类型实际上有何不同！然后，提取minimal reproducible example 以在此处发布。此外，使用 cmets 描述您的策略会有所帮助，既有助于审查，也有助于理解和修正。 -
@UlrichEckhardt 谢谢你，我改了代码，现在可以直接运行了。
-
@MattTimmermans 谢谢你的回复，如果缓冲区满了就不能继续写了，因为每个节点都有一个节点状态，真：可读，假：可写。
标签： c++ multithreading atomic