在两个线程之间传递数据时的时间不一致

Question

在将数据从生产者（线程 1）传递到消费者（线程 2）时，我有一段代码用于测试各种容器（例如双端队列和循环缓冲区）。数据由具有一对时间戳的结构表示。第一个时间戳是在生产者推送之前获取的，第二个时间戳是在消费者弹出数据时获取的。 容器受到 pthread 自旋锁的保护。

这台机器运行 redhat 5.5 和 2.6.18 内核（旧！），它是一个禁用超线程的 4 核系统。所有测试都使用带有 -std=c++11 标志的 gcc 4.7。

生产者获取锁，为数据加上时间戳并将其推入队列，解锁并在繁忙的循环中休眠 2 微秒（我发现在该系统上准确休眠 2 微秒的唯一可靠方法）。

消费者锁定、弹出数据、为其添加时间戳并生成一些统计信息（运行平均延迟和标准偏差）。统计信息每 5 秒打印一次（M 是平均值，M2 是标准差）并重置。我使用 gettimeofday() 来获取时间戳，这意味着平均延迟数可以认为是延迟超过 1 微秒的百分比。

大多数时候输出是这样的：

    CNT=2500000 M=0.00935 M2=0.910238
    CNT=2500000 M=0.0204112 M2=1.57601
    CNT=2500000 M=0.0045016 M2=0.372065

但有时（可能是 20 次试验中的 1 次）是这样的：

    CNT=2500000 M=0.523413 M2=4.83898
    CNT=2500000 M=0.558525 M2=4.98872
    CNT=2500000 M=0.581157 M2=5.05889

（请注意，平均数比第一种情况要差得多，并且随着程序运行它永远不会恢复）。

我会感谢您对为什么会发生这种情况的想法。谢谢。

#include <iostream>
#include <string.h>
#include <stdexcept>
#include <sys/time.h>
#include <deque>
#include <thread>
#include <cstdint>
#include <cmath>
#include <unistd.h>
#include <xmmintrin.h> // _mm_pause()

int64_t timestamp() {
    struct timeval tv;
    gettimeofday(&tv, 0);
    return 1000000L * tv.tv_sec + tv.tv_usec;
}

//running mean and a second moment
struct StatsM2 {
    StatsM2() {}
    double m = 0;
    double m2 = 0;
    long count = 0;
    inline void update(long x, long c) {
        count = c;
        double delta = x - m;
        m += delta / count;
        m2 += delta * (x - m);
    }
    inline void reset() {
        m = m2 = 0;
        count = 0;
    }
    inline double getM2() { // running second moment
        return (count > 1) ? m2 / (count - 1) : 0.;
    }
    inline double getDeviation() {
        return std::sqrt(getM2() );
    }
    inline double getM() { // running mean
        return m;
    }
};

// pause for usec microseconds using busy loop
int64_t busyloop_microsec_sleep(unsigned long usec) {
    int64_t t, tend;
    tend = t = timestamp();
    tend += usec;
    while (t < tend) {
        t = timestamp();
    }
    return t;
}

struct Data {
    Data() : time_produced(timestamp() ) {}
    int64_t time_produced;
    int64_t time_consumed;
};

int64_t sleep_interval = 2;
StatsM2 statsm2;
std::deque<Data> queue;
bool producer_running = true;
bool consumer_running = true;
pthread_spinlock_t spin;

void producer() {
    producer_running = true;
    while(producer_running) {
        pthread_spin_lock(&spin);
        queue.push_back(Data() );
        pthread_spin_unlock(&spin);
        busyloop_microsec_sleep(sleep_interval);
    }
}

void consumer() {
    int64_t count = 0;
    int64_t print_at = 1000000/sleep_interval * 5;
    Data data;
    consumer_running = true;
    while (consumer_running) {
        pthread_spin_lock(&spin);
        if (queue.empty() ) {
            pthread_spin_unlock(&spin);
            // _mm_pause();
            continue;
        }
        data = queue.front();
        queue.pop_front();
        pthread_spin_unlock(&spin);
        ++count;
        data.time_consumed = timestamp();
        statsm2.update(data.time_consumed - data.time_produced, count);
        if (count >= print_at) {
            std::cerr << "CNT=" << count << " M=" << statsm2.getM() << " M2=" << statsm2.getDeviation() << "\n";
            statsm2.reset();
            count = 0;
        }
    }
}

int main(void) {
    if (pthread_spin_init(&spin, PTHREAD_PROCESS_PRIVATE) < 0)
        exit(2);
    std::thread consumer_thread(consumer);
    std::thread producer_thread(producer);
    sleep(40);
    consumer_running = false;
    producer_running = false;
    consumer_thread.join();
    producer_thread.join();
    return 0;
}

Answer 1

编辑：
我相信下面的 5 是唯一可以解释 1/2 秒延迟的东西。当在同一个核心上时，每个都会运行很长时间，然后才切换到另一个。
列表中的其余内容太小，不会导致 1/2 秒的延迟。
您可以使用 pthread_setaffinity_np 将线程固定到特定内核。您可以尝试不同的组合，看看效果如何变化。

编辑#2：
您应该注意的更多事项：（谁说测试很简单...）
1.确保生产者开始生产时消费者已经在运行。在您的情况下不太重要，因为生产者并没有真正在紧密的循环中生产。
2. 这很重要：你每次都除以计数，这对你的统计数据来说是不正确的。这意味着每个统计窗口中的第一次测量的权重都比最后一次大得多。要测量中位数，您必须收集所有值。在不收集所有数字的情况下测量平均值和最小值/最大值，应该可以很好地了解延迟情况。

---

这并不奇怪，真的。
1.时间是在Data()中占用的，但是随后容器花时间调用malloc。
2. 你运行的是 64 位还是 32 位？在 32 位中，gettimeofday 是一个系统调用，而在 64 位中，它是一个不进入内核的 VDSO……您可能想要对 gettimeofday 本身进行计时并记录差异。或者使用 rdtsc 注册您自己的。
最好的办法是使用循环而不是微，因为微对于这种情况来说真的太大了......只有四舍五入到微会让你在处理这么小的事情时变得非常歪斜
3.你保证不会在生产者和消费者之间被抢占吗？我猜不是。但这不应该在专门用于测试的盒子上经常发生...
4. 单插槽4核还是2核？如果它是 2 个插槽的盒子，您希望在同一个插槽上拥有 2 个线程，或者您为数据传输支付（至少）双倍的费用。
5. 确保线程不在同一个内核上运行。
6. 如果您传输的数据和附加数据（容器节点）与其他数据+节点共享缓存线（很可能），则生产者在写入消费时间戳时会被消费者延迟。这称为虚假共享。您可以通过填充/对齐到 64 字节并使用侵入式容器来消除这种情况。

Answer 2

gettimeofday 不是分析计算开销的好方法。它是挂钟，您的计算机是多处理的。即使您认为您没有运行其他任何东西，操作系统调度程序也总是有一些其他活动来保持系统运行。要分析您的流程开销，您至少必须提高您正在分析的流程的优先级。还可以使用高分辨率计时器或 cpu 滴答来进行计时测量。