C++ 无锁队列与多线程崩溃

Posted 2023-02-22

【中文标题】C++ 无锁队列与多线程崩溃

【英文标题】：C++ lockless queue crashes with multiple threads

【发布时间】：2017-10-27 13:10:43

【问题描述】：

我试图更好地理解在为多线程编码时控制内存顺序。我过去经常使用互斥锁来序列化变量访问，但我试图尽可能避免使用互斥锁以提高性能。

我有一个指针队列，它可能被许多线程填充并被许多线程消耗。它适用于单个线程，但当我使用多个线程运行时会崩溃。看起来消费者可能会得到指针的重复，这导致它们被释放两次。有点难以判断，因为当我输入任何打印语句时，它运行良好而不会崩溃。

首先，我使用预先分配的向量来保存指针。我保留 3 个原子索引变量来跟踪向量中需要处理的元素。值得注意的是，我尝试使用 _queue 类型，其中元素本身是原子的，但这似乎没有帮助。这是更简单的版本：

std::atomic<uint32_t> iread;
std::atomic<uint32_t> iwrite;
std::atomic<uint32_t> iend;
std::vector<JEvent*> _queue;

// Write to _queue (may be thread 1,2,3,...)
while(!_done)
    uint32_t idx = iwrite.load();
    uint32_t inext = (idx+1)%_queue.size();
    if( inext == iread.load() ) return kQUEUE_FULL;
    if( iwrite.compare_exchange_weak(idx, inext) )
        _queue[idx] = jevent; // jevent is JEvent* passed into this method
        while( !_done )
            if( iend.compare_exchange_weak(idx, inext) ) break;
        
        break;
    

和同一个班

// Read from _queue (may be thread 1,2,3,...)
while(!_done)
    uint32_t idx = iread.load();
    if(idx == iend.load()) return NULL;
    JEvent *Event = _queue[idx];
    uint32_t inext = (idx+1)%_queue.size();
    if( iread.compare_exchange_weak(idx, inext) )
        _nevents_processed++;
        return Event;
    

我应该强调，我真的很想了解为什么这不起作用。实现一些其他的预制包可以让我解决这个问题，但不会帮助我避免以后再次犯相同类型的错误。

更新 我将 Alexandr Konovalov 的答案标记为正确（请参阅我在下面的答案中的评论）。如果有人遇到此页面，“Write”部分的更正代码是：

std::atomic<uint32_t> iread;
std::atomic<uint32_t> iwrite;
std::atomic<uint32_t> iend;
std::vector<JEvent*> _queue;

// Write to _queue (may be thread 1,2,3,...)
while(!_done)
    uint32_t idx = iwrite.load();
    uint32_t inext = (idx+1)%_queue.size();
    if( inext == iread.load() ) return kQUEUE_FULL;
    if( iwrite.compare_exchange_weak(idx, inext) )
        _queue[idx] = jevent; // jevent is JEvent* passed into this method
        uint32_t save_idx = idx;
        while( !_done )
            if( iend.compare_exchange_weak(idx, inext) ) break;
            idx = save_idx;
        
        break;
    

【问题讨论】：

ABA 问题？ en.wikipedia.org/wiki/ABA_problem

你确定你有足够的争用，值得拥有一个无锁队列吗？

可能值得一读其他人如何解决 多生产者多消费者队列 (mpmc) 问题，因为这可能会让您深入了解您所缺少的内容。根据我对这个问题的回忆，它要求您的检查比您目前拥有的更复杂。

@UKMonkey 是的。我可以让它与锁一起工作，它会从 ~4MHz 减慢到 ~35kHz。

@Slava。这是可能的，因为队列用作振铃缓冲区，因此索引确实回滚到相同的值。无论我使用多少线程（只要它超过 1），崩溃似乎都很快发生。考虑到一些数字（队列大小与线程），似乎不太可能，但值得检查。我会做一些测试。

【参考方案1】：

对我来说，当有 2 位作者和 1 位读者时，可能会出现一个问题。假设第一个作者在之前停止

 _queue[0] = jevent;

和第二个写入器通过 iend 发出信号，表明它的 _queue[1] 已准备好被读取。然后，通过 iend 读取器看到 _queue[0] 已准备好被读取，因此我们有数据竞争。

我建议您尝试 Relacy Race Detector，它非常适用于此类分析。

【讨论】：

亚历山大：谢谢你的建议。 writer 部分中的内部 while 循环的目的是防止出现这种情况（尽管我可能会遗漏一些东西）。在您描述的场景中，第二个编写器线程应该卡在该内部 while 循环中，直到 iend 设置为“1”。这不会发生，直到第一个编写器线程通过相同的内部 while 循环并将 iend 从“0”更改为“1”。直到它写入 _queue 中的插槽时才会发生这种情况。

嗯，不。第 2 个写入线程中 iend.compare_exchange_weak() 的第 1 次运行肯定返回失败，但第 2 次运行成功（因此线程离开内部循环），因为第 1 次运行后 idx 已更新为当前 iend。

Alexandr：我不得不盯着你的评论和我的代码看了一会儿，最后才意识到我还没有完全理解 compare_exchange_weak。我没有意识到如果调用“失败”那么它会更新预期值（在我的例子中是“idx”）。这正是我来到SO的原因。对我来说这是一个令人尴尬的错误，但我很高兴现在能理解它。谢谢！