用无锁算法阻塞并发队列

Question

我正在构建一个阻塞并发队列(即，在被生产者唤醒之前，如果队列中没有任何内容，消费者就会睡觉)。我的用例受到限制，因为去队列操作总是在空队列上阻塞；没有tryDequeue。

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

Answer 1

我想这个问题是问你应该不要锁。

在现代系统中，只有当每个项的工作时间基本上小于1ms，或者是一些搞笑的大规模锁争用场景(不讨论，web服务器就是一个例子)时，才有意义。

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

#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.1ms：https://coliru.stacked-crooked.com/a/9598c83dd05e09b5