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

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值，并获得了一些加速。另外，我已经在buffer和readpos之前移动了writepos，在buffer之后确保两个变量都在不同的缓存行上，这也导致了一些速度。

我需要的是加速大约400％。你有什么想法如何使用填充等东西来实现这一点？

Answer 1

我已经在cafs的第一个代码块中实现了优化。他们实际上给了一些加速（谢谢），但它还不够。导致缓存被列填充而不是按行填充的第二次优化会导致更差的性能。

消费者落后于生产者的想法并没有带来任何加速。

现在，我在300％。我已经

另外一个变化是关于挥发性writepos和readpos变量的一些黑客：

void produce (void * e) { 
    unsigned int oldpos = buffer.writepos; 
    unsigned int next = (oldpos+1) % BUF_SIZE; 
    while (next == buffer.rpos) { // rpos is not volatile 
     buffer.rpos = buffer.readpos; 
     usleep(1); 
    } 
    buffer.buffer[oldpos] = e; buffer.writepos = next; 
} 

，为消费类似的（）。

对结构的其他更改会导致以下新缓冲区结构体（在全局范围内，因为它在一个回答中建议，而不是在堆上）。

#define STRIDE 16 
#define STEPS 524288 

struct buf_t { 
    volatile unsigned int writepos; 
    int offset [STRIDE - 1]; 
    unsigned int wpos; 
    int offset2 [STRIDE - 1]; 
    volatile void * buffer[BUF_SIZE]; 
    int offset4 [STRIDE]; 
    volatile unsigned int readpos; 
    int offset3 [STRIDE - 1]; 
    unsigned int rpos; 
} 

其中300％加速失踪，并将其推到了我必须达到的性能极限以下。

如果你有可能被用来进一步提高性能的一些额外的黑客，不要犹豫，后他们也:-)

感谢您的帮助。

Answer 2

这里有一个优化的，我可以看到：在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之间获得cachelines乒乓（因为一个CPU想让cacheline读取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 - 因此我建议相应地改变的那些类型）。

Answer 3

我假设您使用的机器具有多个处理器或核心。如果没有，那么你的忙碌的等待将会伤害事物。如果你在一个操作系统下运行，那么它们可能会受到伤害，因为它决定你没有足够的睡眠，并且会降低动态优先级，并且还有其他程序正在运行。

您需要收集有关缓冲区充满程度的数据。在某个时候，太大的开始也会伤害你的缓存。

如果使用全局数组而不是从堆中分配它，那么指向缓冲区的指针将变成指针字面值，并且两个线程都不必从缓存中的同一位置读取该指针值，因为它只是推入代码。

如果吞吐量对您来说很重要（以延迟为代价），并且缓存真的在大幅增加，那么您可以考虑让消费者滞后于生产者，以免他们试图读取和写入缓冲区中的相同位置。

您可能希望将接口更改为您的消费者函数，以便它可以消耗缓存大小（或多个）大小的块（这对于消费者滞后于上面制定的生产者建议而言效果很好），除个别或部分缓存行块。尝试保持消费高速缓存对齐。如果您将可用数据视为蛇，那么头和尾可能未对齐。当没有其他数据要消耗时，您应该只消耗未对齐的尾部。如果您可以在通话中使用任何其他数据来消费，那么您应该留下尾部以便进行下一次通话。

除此之外，caf已经提到了什么，我不得不怀疑在这段代码之外发生的任何事情都必须发挥更大的作用。