c++ - 使用 spsc_queue 避免在单一生产者单一消费者程序中忙于等待

Question

我正在尝试实现单个生产者(主线程)和单个消费者(从主线程生成的子线程)问题，因此根据我的搜索，我得到了 spsc_queue 作为最好的无锁数据结构由 boost 库提供。现在从他们的示例代码中获取，消费者函数看起来像这样:

void consumer(void)
{
    int value;
    while (!done) {
        while (queue.pop(value))
            ++consumer_count;
    }

    while (queue.pop(value))
        ++consumer_count;
}

现在，spsc_queue 可能会在一段时间内保持为空，因此为了避免忙等待，我在代码中引入了 sleep ，如下所示:

void consumer(void)
{
    int value;
    while (!done) {
        if (spsc_queue.empty())
        {
            cout << "Waiting for data....\n";
            this_thread::sleep_for (chrono::milliseconds(100));
        }
        else
        {
            while (spsc_queue.pop(value))
            {
                ++consumer_count;
            }
        }
    }

    while (spsc_queue.pop(value))
        ++consumer_count;
}

是正确的做法吗？或者，有没有更好的方法来做到这一点？我遇到了一些库，例如 libevent 或 libev 或 boost::asio::io_service - 有人能帮我找出最好的吗避免忙等待的方法？

我关心的是性能，代码必须是无锁和无等待的(如果可能的话)。任何帮助将不胜感激。

Answer 1

您的目标与您的要求不符。

免等待:等到元素可用时已经排除此属性。

无锁:您的目标是在元素可用之前不做任何工作，即您想要阻止。这再次与无等待和无锁相矛盾。

你真正想要的是类似的东西

if (spsc_queue.empty()) {
     doSomethingElse();
}

或者简单地说，继续忙循环。

也许最接近您想要的是:

if (spsc_queue.empty()) {
    std::this_thread::yield();
}

它重新安排线程并让其他线程完成它们的工作。但是，您正在放弃您的时间片并且可能不会在 25-100 毫秒之前重新安排。

代码必须无锁的任何具体原因？既然你的队列为空的概率看起来挺高的，为什么还需要无锁代码呢？在这种情况下，您没有任何好处。

另一方面，如果出现空队列的可能性很低，那么繁忙的循环会随着时间的推移而摊销。无论如何，您不会在繁忙的循环中花费太多时间，而是尽可能快地取出您的元素(以偶尔的繁忙等待为代价)。