具有循環緩衝區的多生產者單一消費者

Question

需要幫助才能使以下工作成功。

我有一個多生產者線程（每個寫入說100個字節的數據）到環緩衝區。 （最終我想根據數據寫入文件）

通過這個實現，我得到了從環形緩衝區讀取的數據有時候錯了。見下面 由於環緩衝區的大小很小，它變滿了，部分數據丟失。這不是我目前的問題。

**問題：

1. 上的打印數據從ringbuffer讀多數民衆贊成，一些數據得到 互換!!我無法找到的bug。
2. 邏輯/方法是否正確？ （或）是否有更好的 辦法做到這一點

ringbuffer.h

#define RING_BUFFER_SIZE 500 
struct ringbuffer 
{ 
    char *buffer; 
    int wr_pointer; 
    int rd_pointer; 
    int size; 
    int fill_count; 
}; 

ringbuffer.c

#include <stdio.h> 
#include <stdlib.h> 
#include <string.h> 
#include "ringbuffer.h" 

int init_ringbuffer(char *rbuffer, struct ringbuffer *rb, size_t size) 
{ 
    rb->buffer = rbuffer; 
    rb->size = size; 
     rb->rd_pointer = 0; 
     rb->wr_pointer = 0; 
     rb->fill_count = 0; 
    return 0; 
} 

int rb_get_free_space (struct ringbuffer *rb) 
{ 
    return (rb->size - rb->fill_count); 
} 

int rb_write (struct ringbuffer *rb, unsigned char * buf, int len) 
{ 
    int availableSpace; 
    int i; 

    availableSpace = rb_get_free_space(rb); 
    printf("In Write AVAIL SPC=%d\n",availableSpace); 
    /* Check if Ring Buffer is FULL */ 
    if(len > availableSpace) 
    { 
     printf("NO SPACE TO WRITE - RETURN\n"); 
     return -1; 
    } 

    i = rb->wr_pointer; 
    if(i == rb->size) //At the end of Buffer 
    { 
     i = 0; 
    }  
    else if (i + len > rb->size) 
    { 
     memcpy(rb->buffer + i, buf, rb->size - i); 
     buf += rb->size - i; 
     len = len - (rb->size - i); 
     rb->fill_count += len; 
     i = 0; 
    } 
    memcpy(rb->buffer + i, buf, len); 
    rb->wr_pointer = i + len; 
    rb->fill_count += len; 

    printf("w...rb->write=%tx\n", rb->wr_pointer); 
    printf("w...rb->read=%tx\n", rb->rd_pointer); 
    printf("w...rb->fill_count=%d\n", rb->fill_count); 
    return 0; 
} 

int rb_read (struct ringbuffer *rb, unsigned char * buf, int max) 
{ 
    int i; 

    printf("In Read,Current DATA size in RB=%d\n",rb->fill_count); 
    /* Check if Ring Buffer is EMPTY */ 
    if(max > rb->fill_count) 
    { 
     printf("In Read, RB EMPTY - RETURN\n"); 
     return -1; 
    } 

    i = rb->rd_pointer; 
    if (i == rb->size) 
    { 
     i = 0; 
    } 
    else if(i + max > rb->size) 
    { 
     memcpy(buf, rb->buffer + i, rb->size - i); 
     buf += rb->size - i; 
     max = max - (rb->size - i); 
     rb->fill_count -= max; 
     i = 0; 
    } 
    memcpy(buf, rb->buffer + i, max); 
    rb->rd_pointer = i + max; 
    rb->fill_count -= max; 

    printf("r...rb->write=%tx\n", rb->wr_pointer); 
    printf("r...rb->read=%tx\n", rb->rd_pointer); 
    printf("DATA READ ---> %s\n",(char *)buf); 
    printf("r...rb->fill_count=%d\n", rb->fill_count); 
    return 0; 
} 

Answer 1

在生產者，您還需要等待has empty space條件的條件變量。這兩個條件變量都應無條件地發出信號，即當消費者從環形緩衝區中移除一個元素時，它應該向生產者發出信號;當生產者將某些東西放入緩衝區時，它應該向消費者發出信號。 另外，我會將這個等待/信令邏輯轉移到rb_read和rb_write實現中，因此您的環形緩衝區是您的程序其餘部分的「完全使用解決方案」。

Answer 2

至於你的問題 - 1.我找不到那個bug - 事實上，我試過你的代碼，並沒有看到這種行爲。 2.你問這是否邏輯/方法是正確的 - 好吧，就它而言，這確實實現了一種環形緩衝區。你的測試用例碰巧具有整數倍的大小，並且記錄大小是不變的，所以這不是最好的測試。

在試用你的代碼時，我發現有很多線程不足 - 運行的第一個生產者線程（最後創建的）非常困難，在第1次5次之後嘗試和失敗，緩衝區，而不是讓消費者線程有機會運行（甚至啓動）。然後，當消費者線程啓動時，在它釋放CPU之前它會保持啓動相當一段時間，並且下一個生產者線程最終啓動。這就是它在我的機器上的工作原理 - 我敢肯定，它在不同的機器上會有所不同。

這太糟糕了，您當前的代碼沒有辦法結束 - 創建10或100的MB文件...很難通過。

Answer 3

（可能有點晚了作者，但如果其他人搜索了「多個生產者單個消費者」）

我認爲在執行的根本問題是什麼rb_write修改全局狀態（RB-> fill_count和其他rb-> XX）不用在多個作者之間進行任何同步。

如需其他想法，請檢查：http://www.linuxjournal.com/content/lock-free-multi-producer-multi-consumer-queue-ring-buffer。