c++ - 使用无锁算法阻塞并发队列

Question

我正在构建一个阻塞并发队列(即，如果队列中没有任何内容，消费者就会进入休眠状态，直到被生产者唤醒)。我的用例受到限制，因为出列操作将总是阻塞在空队列上；没有 tryDequeue。

我目前的实现只是简单地使用一个 std::mutex 和一个 std::condition_variable。我想知道用无锁算法改进数据结构是否有意义。在队列为空的情况下，这可能没有意义，因为消费者无论如何都必须阻塞。但是在队列非空的情况下，我能做些什么吗？

Answer 1

我认为这个问题是在问你是否应该使用无锁。

在现代系统上，无锁只有在每个项目的工作时间基本上低于 1 毫秒或一些滑稽的大规模锁争用场景(不讨论，网络服务器就是一个例子)时才有意义。

您可以通过查看条件变量响应所需的时间来了解通知条件变量所需的时间:

#include <thread>
#include <array>
#include <mutex>
#include <iostream>
#include <condition_variable>

std::chrono::microseconds timestamp()
{
    const auto microseconds = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch());
    return microseconds;
}
struct measurement
{
    std::chrono::microseconds sent, acknowledged, received;
    measurement() :sent{ 0 }, acknowledged{ 0 }, received{ 0 } {}
};

int main(int argc, char** argv)
{
    std::array<measurement, 15> notifications;
    std::mutex notifications_m;
    std::condition_variable notification_cv;
    auto notifying_thread = std::thread([&] {
        for (auto i = 0; i < notifications.size(); ++i)
        {
            {
                std::unique_lock<std::mutex> lk(notifications_m);
                notifications[i].sent = timestamp();
            }
            notification_cv.notify_one();
            //Now we wait for the notification to be acknowledged
            std::unique_lock<std::mutex> lk(notifications_m);
            const auto check_that_acknowledged = [&] {return notifications[i].acknowledged != std::chrono::microseconds(0); };
            notification_cv.wait(lk, check_that_acknowledged);
            notifications[i].received = timestamp();

        }
    });
    for (auto i = 0; i < notifications.size(); ++i)
    {
        {
            std::unique_lock<std::mutex> lk(notifications_m);
            const auto check_that_sent = [&] {return notifications[i].sent != std::chrono::microseconds(0); };
            notification_cv.wait(lk, check_that_sent);
            notifications[i].acknowledged = timestamp();
        }
        notification_cv.notify_one();
    }
    notifying_thread.join();
    //
    for (const auto& notification : notifications)
    {
        const auto difference_us = notification.received - notification.sent;
        std::cout << difference_us.count() << " microseconds" << std::endl;
    }
    return 0;
}

即使在一些延迟最高的系统上，往返也需要大约 0.1 毫秒: https://coliru.stacked-crooked.com/a/9598c83dd05e09b5