具有（非常）快速和可靠的切换的 C++ 生产者消费者队列答案

【问题标题】：C++ Producer consumer queue with (very) fast and reliable handover具有（非常）快速和可靠的切换的 C++ 生产者消费者队列
【发布时间】：2011-10-21 00:03:45
【问题描述】：

您好，我正在研究使用快速可靠的生产者消费者队列进行线程切换。我正在使用 VC++ 在 Windows 上工作。

我的设计基于Anthony Williams 队列，即基本上是带有 boost::condition_variable 的 boost::mutex。现在通常 notify_one() 和唤醒之间的时间在 10（罕见）和 100 微秒之间变化，大多数值在 50 微秒范围内。但是，每 1000 人中约有 1 人超过 1 毫秒，有些则超过 5 毫秒。

我只是想知道这些是否是典型值？有没有比旋转更快的信号方式？是从这里开始到管理线程优先级吗？我还没有开始玩优先级，但我只是想知道是否有机会将其置于大约 10 微秒的相当稳定的区域？

谢谢

编辑：使用 SetPriorityClass(GetCurrentProcess(),REALTIME_PRIORITY_CLASS)，平均唤醒时间仍约为 50 微，但是异常值要少得多，现在大多数约为 150-200 微。除了 7 毫秒的异常异常值。嗯……不好。

【问题讨论】：

你能使用额外的第三方库吗？我发现通常无锁实现的性能要好得多，英特尔的线程构建模块中有一个很好的实现。
题外话，但也许你想看看 Sutter 的 wait-free queue，它不使用锁。
@Chad：几乎不应该对锁有任何争用，我的理解是，在没有争用的情况下，Windows 上的 boost mutex 相当便宜并且保留在用户空间中，因此切换到无锁可能不会改善这种情况这么多......我认为这里的问题更多的是找到唤醒消费者线程的最快方法。
@Kerrek：感谢指点，好文章，但我想避免旋转的原因是生产者消费者队列的数量可能比核心数量大很多。我的理解是，这不是一个很好的场景？
@Cookie：确实不是，在这种情况下，自旋会阻碍真正工作的线程的进度，你也许可以使用微睡眠。

标签： c++ multithreading performance producer-consumer

【解决方案1】：

分摊锁定和线程唤醒开销的一种方法是添加第二个队列并实施双缓冲方法。这可以在消费者端进行批处理：

template<typename F>
std::size_t consume_all(F&& f)
{   
    // minimize the scope of the lock
    {
        std::lock_guard<std::mutex> lock(the_mutex);
        std::swap(the_queue, the_queue2);
    }

    // process all items from the_queue2 in batch
    for (auto& item : the_queue2)
    {
        f(item);
    }

    auto result = the_queue2.size();        
    the_queue2.clear(); // clears the queue and preserves the memory. perfect!
    return result;
}

Working sample code.

这并不能解决延迟问题，但可以提高吞吐量。如果发生打嗝，那么消费者将看到一个大批量，然后可以全速处理而没有任何锁定开销。这可以让消费者快速赶上生产者。

【讨论】：

【解决方案2】：

有很多事情可能会导致问题。

您可能需要尝试分析应用程序并查看可能出现减速的位置。

一些注意事项：

消费者和生产者是否在同一个进程中？如果是这样，Critical Section 比 Mutex 快得多。
尝试确保所有队列内存都在当前内存中。如果您必须交换页面，这会减慢速度。
将您的进程设置为实时优先级时要非常小心。那应该是针对系统进程的。如果进程做太多工作，它可以防止关键系统进程获取 cpu，这可能会非常糟糕地结束。除非您绝对需要实时，否则只需使用 HIGH_PRIORITY_CLASS

【讨论】：

确实——优先权属于用户，在较小程度上属于系统，而不是应用程序开发人员。我什至建议反对高优先级——这可能会导致使用任务管理器之类的东西出现问题，它充当用户的逃生舱口，默认情况下以高优先级运行。

【解决方案3】：

简短的回答是肯定的，从那里开始真正归结为操作系统管理和线程调度。 RTS（实时系统）可以将这 50 微秒提高到 15 微秒左右，更重要的是，它们可以去除异常值。否则，旋转是唯一的答案。如果队列多于核心，则想法可能是让 x 个线程旋转，立即做出反应，其余的阻塞。这将涉及某种“主”队列线程，不断旋转以检查所有队列，并且 - 要么自己处理项目 - 要么将它们交给工作线程，其中一些也可以旋转以节省这 50 微秒。不过，它变得复杂了。

最好的办法可能是只使用一个无锁的多生产者单消费者队列和一个旋转的消费者线程。然后进入队列的所有项目可能需要从一个通用的基本类型派生，并且需要包含一些关于如何处理这些项目的元信息。

复杂，但可能。如果我设置了它，我也可能会发布一些代码。

【讨论】：