c - 在生产者/消费者多线程环境中优化共享缓冲区

Question

我有一些项目，其中有一个生产者线程将事件写入缓冲区，还有一个额外的单个消费者线程从缓冲区获取事件。我的目标是针对单个双核机器 优化这个东西 以实现最大吞吐量。

目前，我正在使用一些简单的无锁环形缓冲区(无锁是可能的，因为我只有一个消费者和一个生产者线程，因此指针仅由一个线程更新)。

#define BUF_SIZE 32768

struct buf_t { volatile int writepos; volatile void * buffer[BUF_SIZE]; 
    volatile int readpos;) };

void produce (buf_t *b, void * e) {
    int next = (b->writepos+1) % BUF_SIZE;
    while (b->readpos == next); // queue is full. wait
    b->buffer[b->writepos] = e; b->writepos = next;
}

void * consume (buf_t *b) {
    while (b->readpos == b->writepos); // nothing to consume. wait
    int next = (b->readpos+1) % BUF_SIZE;
    void * res = b->buffer[b->readpos]; b->readpos = next;
    return res;
}

buf_t *alloc () {
    buf_t *b = (buf_t *)malloc(sizeof(buf_t));
    b->writepos = 0; b->readpos = 0; return b;
}

但是，这种实现还不够快，应该进一步优化。我尝试了不同的 BUF_SIZE 值并获得了一些加速。另外，我在 writepos 之前移动了 buffer，在 readpos 之后移动了 buffer，以确保两个变量都在不同的缓存行上，这也导致了一些速度。

我需要的是大约 400% 的加速。你有什么想法我可以使用填充等来实现这一点吗？

Answer 1

这是我可以看到的一种优化:在 consume() 中，您不需要连续获取 b->readpos ，因为调用 consume() 的线程是唯一可以更新它的线程。因为它是 volatile ，编译器不能优化所有这些提取，所以你需要明确地做:

void * consume (buf_t *b) {
    int rp = b->readpos;
    while (rp == b->writepos); // nothing to consume. wait
    int next = (rp + 1) % BUF_SIZE;
    void * res = b->buffer[rp]; b->readpos = next;
    return res;
}

您还应该以每个至少一个缓存行的步幅单步通过缓冲区，否则您将在两个 CPU 之间得到缓存行乒乓(因为一个 CPU 希望缓存行读取 b->buffer[n] 并且 16 次中有 15 次，另一个使其无效写 b->buffer[n+1] )。例如:

#define STRIDE 16
#define STEPS 2048
#define BUF_SIZE (STRIDE * STEPS)

#define TO_INDEX(n) (STRIDE * (((n) + 1) % STEPS) + (((n) + 1) / STEPS))

void produce (buf_t *b, void * e) {
    unsigned wp = b->writepos;
    unsigned next = (wp + 1) % BUF_SIZE;
    while (b->readpos == next); // queue is full. wait
    b->buffer[TO_INDEX(wp)] = e; b->writepos = next;
}

void * consume (buf_t *b) {
    unsigned rp = b->readpos;
    while (rp == b->writepos); // nothing to consume. wait
    unsigned next = (rp + 1) % BUF_SIZE;
    void * res = b->buffer[TO_INDEX(rp)]; b->readpos = next;
    return res;
}

无论如何，必须值得一试。 (请注意，只要 STRIDE 和 STEPS 是 2 的幂，TO_INDEX() 中看起来很吓人的除法和模数就可以优化为移位和按位与，但前提是操作数是 unsigned - 因此我建议相应地更改这些类型)。