将数据从一个线程传递到另一个线程的最快方法

Question

我正在使用 boost spsc_queue 将我的东西从一个线程移动到另一个线程。这是我软件中的关键位置之一，所以我想尽快完成。我写了这个测试程序：

#include <boost/lockfree/spsc_queue.hpp>
#include <stdint.h>

#include <condition_variable>
#include <thread>

const int N_TESTS = 1000;

int results[N_TESTS];

boost::lockfree::spsc_queue<int64_t, boost::lockfree::capacity<1024>> testQueue;

using std::chrono::nanoseconds;
using std::chrono::duration_cast;

int totalQueueNano(0);
int totalQueueCount(0);

void Consumer() {
    int i = 0;
    int64_t scheduledAt;
    while (i < N_TESTS - 1) {
        while (testQueue.pop(scheduledAt)) {
            int64_t dequeuedAt = (duration_cast<nanoseconds>(
                    std::chrono::high_resolution_clock::now().time_since_epoch())).count();
            auto diff = dequeuedAt - scheduledAt;
            totalQueueNano += diff;
            ++totalQueueCount;
            results[i] = diff;
            ++i;
        }
    }
    for (int i = 0; i < N_TESTS; i++) {
        printf("%d ", results[i]);
    }
    printf("\nspsc_queue latency average nano = %d\n", totalQueueNano / totalQueueCount);
}

int main() {
    std::thread t(Consumer);
    usleep(1000000);
    for (int i = 0; i < N_TESTS; i++) {
        usleep(1000);
        int64_t scheduledAt = (duration_cast<nanoseconds>(
                std::chrono::high_resolution_clock::now().time_since_epoch())).count();
        testQueue.push(scheduledAt);
    }
    usleep(1000000);
    return 0;
}

编译标志：

g++ -std=c++0x -O3 -Wall -c -fmessage-length=0 -march=native -mtune=native -pthread -MMD -MP -MF"src/TestProject.d" -MT"src/TestProject.d" -o "src/TestProject.o" "../src/TestProject.cpp"

g++ -pthread -o "TestProject"  ./src/TestProject.o   -lpthread

在我的机器上：RHEL 7.1、gcc 4.8.3、Xeon E5-2690 v3 我收到 290-300 纳秒。

• 我的测试应用程序有多好？我是否正确测量了“spsc_queue”延迟？
• 当前行业将数据从一个线程传递到另一个线程的最佳时间是什么时候？
• 使用 boost spsc_queue 将数据从一个线程移动到另一个线程是不是不错的选择？
• 你能推荐一些比 spsc_queue 更快的东西吗？
• 你能写一段代码来更快地完成相同的工作吗？

upd: 需要队列机制。如果第一个线程每 1000 纳秒产生一次数据，但第二个线程花费 10 000 纳秒来处理单个项目，我需要在短时间内“排队”几个项目。但我的“队列”永远不会“太大”。固定大小的短环形缓冲区就足够了。

upd2 所以简而言之，问题是——最快的单生产者单消费者队列是什么（最有可能基于固定大小的环形缓冲区）？我正在使用 boost spsc_queue，我实现了约 300 ns 的延迟，您能建议更快的方法吗？

upd3 在 java 世界中，有一个能达到 50 ns 延迟的破坏者 https://code.google.com/p/disruptor/wiki/PerformanceResults 我们在 c++ 中是否有具有相同 50 ns 延迟的东西？

Answer 1

由于您拥有ints，因此您（理想情况下）在上面测量的是从调用push() 到pop() 返回true 之间的总体延迟。

这没有意义：消费者线程忙于轮询队列，也就是说，它循环并忙于检查pop是否获取了一个值。

• 这很浪费，而且
• 如果您想最大限度地减少延迟，轮询肯定是不是要走的路

如果 (IFF) 你想最小化延迟（对于单个项目），我的 猜测 会使用信号同步机制，spsc_queue，据我所知，不会为此提供。 （您需要一个容器或自定义解决方案，在其中使用一种condition variable / Event，...）

但是，如果 (IFF)，您想要最大化吞吐量（每次的项目数），那么测量（单个）项目“唤醒”的延迟就更没有意义了。在这种情况下，您希望充分利用您拥有的并行性，如is mentioned in a comment：

通常，传递数据的最快方法是对每个数据块使用单个线程。也就是说，只使用数据中存在的并行性。

---

解决你的要点：

• 测试应用有多好：我认为它没有多大意义。

  • scheduledAt 在原子中是必需的，因为您从一个线程编写它并从另一个线程读取它。否则你有 UB。
  • 显然有任何测量差异。这纯粹是一个测量误差，并没有说明固有的延迟。 （您可以尝试将聚合 struct {int val; int64_t time; }; 放入队列中，从而避开原子栅栏。

• 当前行业最佳时间：没有线索。不确定有人关心这个。 （也许在一些内核的东西里面？）

• spsc_queue 的选择：我认为这不是一个好的选择，因为它需要轮询。

• 比 spsc_queue 更快？：见上文。使用非轮询通知。

• 写一个代码来做同样的工作明显更快？：不。或者更确切地说，我不会。 =>

引用"man"s answer：

1. 您定义问题并选择适当的同步机制

你的问题的问题在于没有问题定义。

就我目前而言，在常规操作系统上的用户级进程的上下文中，跨线程通知延迟似乎完全无关紧要。 您的用例是什么？

Answer 2

这取决于应用程序的语义以及涉及的线程数。到目前为止，您正在查看原始延迟。随着更多的线程，扩展也可能开始成为一个有趣的指标。

对于双线程情况，如果您对检索到的数据所做的操作允许，原子更新到单个位置（最好是在任何其他操作未触及的缓存行中）可能会更快。

Answer 3

首先，写这样的测试程序是完全没用的。您不对数据进行任何处理，因此结果会出现偏差。其次，您的测试在推送之间使用 usleep() - 在这个速率下，您可以使用任何类型的同步原语。 您的 Consumer() 似乎也永远不会退出......

你实现这样的事情的方式如下：

1. 您定义问题并选择合适的同步机制
2. 您实施该软件
3. 您配置软件以识别潜在热点
4. 您根据上一步的结果进行优化并重复。

您需要在第一步之前获得一些经验，或者您可以尝试实施不同的方法，看看哪种方法最有效。