C++无锁队列实现单生产者单消费者

如何解决C++无锁队列实现单生产者单消费者

我尝试了一个无锁的单生产者-单消费者实现link。该实现使用列表作为底层数据结构，依赖于只有生产者线程修改列表的事实。如果_head不等于_tail，消费者线程移动head变量。

int produced_count_,consumed_count_;
std::list<int> data_queue_;
std::list<int>::iterator head_,tail_;    

void ProducerConumer::produce() {
    static int count = 0;
    data_queue_.push_back(int(count++));
    ++produced_count_;
    tail_ = data_queue_.end();
    data_queue_.erase(data_queue_.begin(),head_);
}

bool ProducerConumer::consume() {
    auto it = head_;
    ++it;
    if(it != tail_) {
        head_ = it;
        ++consumed_count_;
        int t = *it;
        return true;
    } 
    
    return false;
}

在任何时候，头迭代器都指向一个已经被读取的值。

由于这里没有同步，我的印象是该实现不会工作，因为一个线程的写入可能对另一个线程不可见。但是当我测试我的代码生产者和消费者总是生产/消耗相同数量的单位。有人能解释一下这段代码如何在没有显式同步的情况下工作吗？（我没想到对其他线程可见的 tail_ 和 head_ 变量发生了变化）

控制生产者/消费者线程的代码如下

consumer_thread_ = std::thread([this]() {
set_cpu_affinity(0);
std::chrono::milliseconds start_time = current_time();
while((current_time() - start_time) < std::chrono::milliseconds(150)) {
        this->consume();
    }
    std::cout << "Data queue size from consumer is " << data_queue_.size() << " time " << current_time().count() << "\n";
});

producer_thread_ = std::thread([this]() {
    set_cpu_affinity(7);
    std::chrono::milliseconds start_time = current_time();
    while((current_time() - start_time) < std::chrono::milliseconds(100)) {
        this->produce();
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    std::cout << "Data queue size from producer is " << data_queue_.size() << " time " << current_time().count() << "\n";
});

我通过在生产者线程的末尾添加 sleep_for 来确保生产者线程比消费者线程存活时间更长。

顺便说一句，这里是 Herb Sutter 讨论它有什么问题的实现的剖析link。但是他从来没有谈到过tail_和head_的变化是否对其他线程可见。

解决方法

调试构建通常会“碰巧工作”，尤其是在 x86 上，因为代码生成块编译时重新排序的约束和 x86 硬件阻止了大多数运行时重新排序。

如果在调试模式下编译，内存访问将按程序顺序发生，the compiler won't keep values in registers across statements。（有点像 volatile，它可以用来滚动你自己的原子；但不要：When to use volatile with multi threading?）。尽管如此，缓存是一致的，简单地加载和存储 asm 就足以实现全局可见性（按某些顺序）。

它们将是原子的，因为它们是 int 大小和对齐的，并且编译器用一条指令完成它们，因为它不是 DeathStation 9000。自然对齐的 int 加载和存储 {{3 }}，在 C 中不保证。(are atomic in asm on normal machines like x86)

如果您只在 x86 上进行测试，硬件内存模型会为您提供程序顺序和存储缓冲区，因此您可以有效地获得与 std::atomic memory_order_acquire 和 {{1} 相同的 asm }. （因为调试版本不会在语句之间重新排序）。

C++ 未定义行为（包括这个数据竞争 UB）不意味着“保证失败或崩溃” - 这就是它如此讨厌的原因，也是测试不足以找到它的原因。

在启用优化的情况下编译，您可能会看到大问题，这取决于编译时重新排序和提升选择。例如如果编译器可以在循环期间将变量保存在寄存器中，它将永远不会从缓存/内存中重新读取，也永远不会看到其他线程存储的内容。除其他问题外。 https://lwn.net/Articles/793253/

如果代码恰好在“训练轮”模式下工作，那么它就不是很有用，因为您没有告诉编译器如何安全地优化它。（通过使用 release例如）。

我没有详细查看您的代码，但我认为您没有任何被两个线程修改的变量。在循环缓冲区队列中，您通常会在由生产者 RMW 处理但由消费者只读的变量上有 std::atomic 增量。对于读取位置，反之亦然。那些不需要是原子 RMW，只需原子 store 以便其他线程的原子负载可以看到未撕裂的值。这发生在 asm 中的“自然”中。

这里我认为你只是在存储一个新的头部，而另一个线程只是在读取它。

在链表中，释放可能是一个问题，尤其是对于多个消费者。在您确定没有线程具有指向它的指针之前，您无法释放或回收该节点。垃圾收集语言/运行时可以更轻松地将链表用于无锁队列，因为 GC 已经通常处理相同的对象检查。

因此，如果您自己动手，请确保您做对了；这可能很棘手。尽管只要在构造完节点后仅将其链接到链表中，并且只有一个使用者，您就永远无法看到半构造的节点。而且您永远不会有一个线程释放另一个线程可能会唤醒并继续读取的节点。

文章说：

另一个问题是使用标准的 std::list。虽然文章提到开发人员有责任检查读/写 std::list::iterator 是原子的，结果是太严格了。虽然 gcc/MSVC++2003 有 4 字节迭代器，但 MSVC++2005 在发布模式下有 8 字节迭代器和 12 字节迭代器在调试模式。

您有责任确保迭代器是原子的。 std::list 的情况并非如此。除非您明确指定数据为原子，否则无法保证来自不同线程的读/写操作。然而，即使“未定义行为”意味着“鼻恶魔”，如果这些恶魔被观察为一致的同步，也没有错。