在生產者/消費者多線程環境中優化共享緩衝區

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已經提到了什麼，我不得不懷疑在這段代碼之外發生的任何事情都必須發揮更大的作用。