實現一個通用的生產者消費者隊列（c語言版本）

背景：筆者之前一直從事嵌入式音視頻相關的開發工作，對于音視頻的數據的處理，生產者消費者隊列必不可少，而如何實現一個高效穩定的生產者消費者隊列則十分重要，不過按照筆者從業的經驗，所看到的現象，不容樂觀，很多知名大廠在這種基礎組件的開發能力上十分堪憂。

音視頻數據處理的特點：

音視頻數據量大：音視頻數據特別是視頻數據，占據了計算機數據的很大一塊，不信就看看每個人的硬盤里，去除電影，照片，mp3是不是很空蕩蕩的。
實時性要求高：音視頻的延時如果大于200ms，使用體驗會十分糟糕。
處理流程復雜：一幀數據從sensor捕獲到最終網傳輸出或者lcd顯示，都需要經過一系列的模塊進行處理。特別是網絡傳輸，一般要經過原始數據捕獲，視頻數據格式轉換，數據編碼壓縮，數據封裝打包，網絡傳輸。

生產者消費者隊列在視頻數據處理的必要性：

視頻數據的處理為什么需要生產者消費者隊列？其實上面提到的音視頻數據處理的特點就是答案。

一般對視頻數據處理的模塊都是運行在多線程中，每一個處理模塊運行在一個線程中（也是軟件工程分模塊的思想），相互之間通過生產者消費者隊列進行數據的交互，之所以用線程模型，而沒有用我一直推崇的進程模型是因為線程間的共享內存比較方便，而進程則要相對復雜的多。
利用多線程/多進程的并行處理能力，如果采用單線程單進程的單線處理模式，一幀數據從采集到輸出線性經過幾個模塊，時效性無法保證。
緩存數據，保持平滑，通過隊列緩存視頻數據，可以有效的去除一些數據抖動，幫助音視頻數據的平滑播放，同時這個數據的緩存又不易過多，否則加大了延時，損傷實時性，所以隊列大小的設置是一個平衡的藝術。

常見的生產者消費者隊列實現存在的問題：

不注重效率性能：

對于buffer狀態的檢測采用loop輪詢方式。
loop輪詢是任何有追求的程序員都要避免的處理方式，而筆者以自己經歷經常看到以下類似的代碼：

pthread_mutex_lock();
state = check_some_state();
if (state == xxx) {do_some_process();
} else {usleep(x)；
}
pthread_mutex_unlock();

以上代碼loop一個狀態，如果狀態成立做有效的處理，不成立則睡眠一定時間，之后再次調用該段代碼進行下一次的狀態檢測，而這個睡眠時間是一個隨機經驗值，很有可能下次仍然是無效的檢測，接著睡眠再loop，多余的loop是一種資源的浪費。

數據的傳遞采用copy方式：

數據的傳遞，采用copy的方式，一幀數據在一個完整的處理流程中經過n次copy（筆者見過一個系統中一幀數據copy了8次之多）

生產者消費者隊列和業務代碼混雜在一起，沒有分離：

對于開發者，都希望用最簡單的接口完成某個功能，而不關心內部實現，在這里就是，只需要生產者生產數據，消費者消費數據，而內部的處理（同步，數據的處理等）完全不關心，這樣開發者就不需要去弄很多鎖，降低了開發難度，也可以使代碼組件化，模塊化。

有經驗的開發者應該感覺以上都是基礎點，不會有人犯這樣的錯誤，不過筆者以自己的經歷肯定的說，以上兩種問題在某世界級大廠的視頻設備上隨處可見。

讓我悲哀的是，當我指出這些問題時，某些開發者完全無動于衷。而我更無力的是，現在的cpu，memory性能實在是高，在某些不太高端的嵌入式芯片上，優化過的數據并沒有十分明顯，在一個實際項目上，經過優化后cpu大概降低1%（原總系統cpu占用7%），有經驗的開發者又會說原7%的cpu占用率說明這個芯片做這個系統浪費了，不過也沒辦法其實已經用了比較低端的芯片了。。。

以上的吐槽主要是想說明：由于cpu，memory性能的提升，讓很多開發者感覺軟件的優化意義不大了，而我是一個理想主義者，對于某些設計ugly的代碼真的是零容忍啊。

如何優化：

以上說了這么多，那如何操作呢？

提高效率，去除多余的loop輪詢：
采用線程的同步機制，當狀態條件符合要求時，通過通知機制觸發后續處理，這樣去除了無效的loop輪詢檢測，linux系統下可以采用條件變量，信號量來實現，我更傾向于使用條件變量，因為它是linux系統原生支持的接口。
提到條件變量，不得不提一個概念：同步互斥，這么一個基本的操作系統概念，我最喜歡作為面試第一題，不過能用一句話切中要害的說出之間區別和各自特點的人不是很多，答不出這題的，基本上就pass了。
減少多余copy：
采用預分配的方式，將buffer分為free和active兩大類，每一類buffer又切成幾個小buffer，然后通過指針將兩類buffer下的小buffer鏈接成兩個鏈表，使用者獲取buffer通過free鏈表獲取buffer，再將buffer put到active鏈表上，以上都是指針的操作，沒有數據的copy，極大的減少了copy操作。（再次強調指針是個好東西）
模塊化組件化：
將生產者消費者隊列的處理部分完全剝離成一個獨立的模塊組件，對外只提供幾個基本的接口，內部完成同步通知的處理。

一個簡單的實現：

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <sys/ioctl.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <fcntl.h>
#include <pthread.h>
#include <signal.h>
#include <time.h>#include "sfifo.h"//#define CONFIG_COND_FREE 1
#define CONFIG_COND_ACTIVE 1#define MAX_SFIFO_NUM   32struct sfifo_des_s sfifo_des[MAX_SFIFO_NUM];
struct sfifo_des_s *my_sfifo_des;struct sfifo_s* sfifo_get_free_buf(struct sfifo_des_s *sfifo_des_p)
{static long empty_count = 0;struct sfifo_s *sfifo = NULL;pthread_mutex_lock(&(sfifo_des_p->free_list.lock_mutex));
#ifdef CONFIG_COND_FREEwhile (sfifo_des_p->free_list.head == NULL) {pthread_cond_wait(&(sfifo_des_p->free_list.cond), &(sfifo_des_p->free_list.lock_mutex));}
#elseif (sfifo_des_p->free_list.head == NULL) {if (empty_count++ % 120 == 0) {printf("free list empty\n");}goto EXIT;}
#endifsfifo = sfifo_des_p->free_list.head;sfifo_des_p->free_list.head = sfifo->next;EXIT:pthread_mutex_unlock(&(sfifo_des_p->free_list.lock_mutex));return sfifo;
}int sfifo_put_free_buf(struct sfifo_s *sfifo, struct sfifo_des_s *sfifo_des_p)
{int send_cond = 0;pthread_mutex_lock(&(sfifo_des_p->free_list.lock_mutex));if (sfifo_des_p->free_list.head == NULL) {sfifo_des_p->free_list.head = sfifo;sfifo_des_p->free_list.tail = sfifo;sfifo_des_p->free_list.tail->next = NULL;send_cond = 1;} else {sfifo_des_p->free_list.tail->next = sfifo;sfifo_des_p->free_list.tail = sfifo;sfifo_des_p->free_list.tail->next = NULL;}pthread_mutex_unlock(&(sfifo_des_p->free_list.lock_mutex));
#ifdef CONFIG_COND_FREEif (send_cond) {pthread_cond_signal(&(sfifo_des_p->free_list.cond));}
#endifreturn 0;
}struct sfifo_s* sfifo_get_active_buf(struct sfifo_des_s *sfifo_des_p)
{struct sfifo_s *sfifo = NULL;pthread_mutex_lock(&(sfifo_des_p->active_list.lock_mutex));
#ifdef CONFIG_COND_ACTIVEwhile (sfifo_des_p->active_list.head == NULL) {//pthread_cond_timedwait(&(sfifo_des_p->active_list.cond), &(sfifo_des_p->active_list.lock_mutex), &outtime);pthread_cond_wait(&(sfifo_des_p->active_list.cond), &(sfifo_des_p->active_list.lock_mutex));}
#elseif (sfifo_des_p->active_list.head == NULL) {printf("active list empty\n");goto EXIT;}
#endifsfifo = sfifo_des_p->active_list.head;sfifo_des_p->active_list.head = sfifo->next;EXIT:pthread_mutex_unlock(&(sfifo_des_p->active_list.lock_mutex));return sfifo;
}int sfifo_put_active_buf(struct sfifo_s *sfifo, struct sfifo_des_s *sfifo_des_p)
{int send_cond = 0;pthread_mutex_lock(&(sfifo_des_p->active_list.lock_mutex));if (sfifo_des_p->active_list.head == NULL) {sfifo_des_p->active_list.head = sfifo;sfifo_des_p->active_list.tail = sfifo;sfifo_des_p->active_list.tail->next = NULL;send_cond = 1;} else {sfifo_des_p->active_list.tail->next = sfifo;sfifo_des_p->active_list.tail = sfifo;sfifo_des_p->active_list.tail->next = NULL;}pthread_mutex_unlock(&(sfifo_des_p->active_list.lock_mutex));
#ifdef CONFIG_COND_ACTIVEif (send_cond) {pthread_cond_signal(&(sfifo_des_p->active_list.cond));}
#endifreturn 0;
}int dump_sfifo_list(struct sfifo_list_des_s *list)
{struct sfifo_s *sfifo = NULL;sfifo = list->head;do {printf("dump : %x\n", sfifo->buffer[0]);usleep(500 * 1000);} while (sfifo->next != NULL && (sfifo = sfifo->next));return 0;
}struct sfifo_des_s *sfifo_init(int sfifo_num, int sfifo_buffer_size, int sfifo_active_max_num)
{int i = 0;struct sfifo_s *sfifo;struct sfifo_des_s *sfifo_des_p;sfifo_des_p = (struct sfifo_des_s *)malloc(sizeof(struct sfifo_des_s));sfifo_des_p->sfifos_num = sfifo_num;sfifo_des_p->sfifos_active_max_num = sfifo_active_max_num;sfifo_des_p->free_list.sfifo_num = 0;sfifo_des_p->free_list.head = NULL;sfifo_des_p->free_list.tail = NULL;pthread_mutex_init(&sfifo_des_p->free_list.lock_mutex, NULL);pthread_cond_init(&sfifo_des_p->free_list.cond, NULL);sfifo_des_p->active_list.sfifo_num = 0;sfifo_des_p->active_list.head = NULL;sfifo_des_p->active_list.tail = NULL;pthread_mutex_init(&sfifo_des_p->active_list.lock_mutex, NULL);pthread_cond_init(&sfifo_des_p->active_list.cond, NULL);for (i = 0; i < sfifo_num; i++) {sfifo = (struct sfifo_s *)malloc(sizeof(struct sfifo_s));sfifo->buffer = (unsigned char *)malloc(sfifo_buffer_size);printf("sfifo_init: %x\n", sfifo->buffer);memset(sfifo->buffer, i, sfifo_buffer_size);sfifo->size = sfifo_buffer_size;sfifo->next = NULL;sfifo_put_free_buf(sfifo, sfifo_des_p);}return sfifo_des_p;
}void *productor_thread_func(void *arg)
{struct sfifo_s *sfifo;while (1) {sfifo = sfifo_get_free_buf(my_sfifo_des);if (sfifo != NULL) {printf("+++++++++++++++++ put : %x\n", sfifo->buffer[0]);sfifo_put_active_buf(sfifo, my_sfifo_des);}//usleep(20*1000);}
}void *comsumer_thread_func(void *arg)
{struct sfifo_s *sfifo;int count = 0;while (1) {sfifo = sfifo_get_active_buf(my_sfifo_des);if (sfifo != NULL) {printf("---------------- get %x\n", sfifo->buffer[0]);sfifo_put_free_buf(sfifo, my_sfifo_des);}//usleep(10 * 1000);// if (count++ > 10000) {//  exit(-1);// }}
}int main()
{int ret;static pthread_t productor_thread;static pthread_t consumer_thread;struct sfifo_s *r_sfifo;my_sfifo_des = sfifo_init(10, 4096, 5);ret = pthread_create(&productor_thread, NULL, productor_thread_func, NULL);ret = pthread_create(&consumer_thread, NULL, comsumer_thread_func, NULL);while (1) {sleep(1);}return 0;
}

以上是一個簡單的生產者消費者隊列的c語言的實現，對應的頭文件在本文底部（貼代碼太長看起來很崩潰）。

仔細的同學可能會發現，以上代碼sfifo_get_free_buf（）中默認是loop輪詢檢測free buffer鏈表的，你前面不是說了一大堆不能loop嗎？怎么還用loop呢？
這里其實有兩個原因：

生產者的loop是可以接受的，當發生多余loop，無法命中時，說明生產者太快，消費者太慢，而其實對于一個生產者消費者模型出現以上問題時，說明整個業務流程要重新考慮，因為正常的情況是消費者總是要快于生產者，這個業務模型才能正常的運行下去。
對于有些業務模型，生產者業務模塊部分是不能阻塞的，也就是說，如果free list沒有數據，我們采用pthread_cond_wait（）阻塞后，會導致生產者出現問題，這樣最好的處理方式就是生產者模塊接口返回出錯，生產者業務方丟棄數據（此時就是丟幀了，這種情況如果頻繁發生是不能接受了，不過也說明了消費者要有足夠的能力處理生產者生產出的數據，否則整個業務都是有問題）

這個實現有那些優勢：

走讀和運行以上代碼的同學應該可以發現這里做了一個簡單可運行的demo模擬了生產者和消費者雙方：

void *productor_thread_func(void *arg)
{struct sfifo_s *sfifo;while (1) {sfifo = sfifo_get_free_buf(my_sfifo_des);if (sfifo != NULL) {printf("+++++++++++++++++ put : %x\n", sfifo->buffer[0]);sfifo_put_active_buf(sfifo, my_sfifo_des);}//usleep(20*1000);}
}void *comsumer_thread_func(void *arg)
{struct sfifo_s *sfifo;int count = 0;while (1) {sfifo = sfifo_get_active_buf(my_sfifo_des);if (sfifo != NULL) {printf("---------------- get %x\n", sfifo->buffer[0]);sfifo_put_free_buf(sfifo, my_sfifo_des);}//usleep(10 * 1000);// if (count++ > 10000) {//  exit(-1);// }}
}

這里面對于使用者的優點有：

接口簡單：只需要get free，put active；get active，put free。
沒有了數據copy，只需要操作鏈表上的buffer就可以了，而這些buffer的參數控制通過init接口設置。
不用再控制sleep的時間值：前面提到，在loop模型下，如果狀態不成立需要sleep一段時間，再次檢查，這樣來控制同步狀態，而這個時間值很難確定，如果時間值過長，則會導致狀態檢測不及時，延誤數據處理，如果時間值太短，則會增加狀態檢測miss cache的次數，耗費更多cpu資源。而采用本模塊的實現則完全不需要考慮這些問題，只需要銜接業務處理，sleep，同步，yield cpu的操作都由這個模塊實現吧，完全不需要關心。
模塊化，完全和業務處理無關，可以毫無壓力的運用在不同的業務處理邏輯中，沒有剝離代碼的工作。

以上描述了一個生產者消費者隊列c語言的實現，為什么是c語言版本的？因為其他高級語言，有很多成熟的庫提供了該功能，完全不用自己寫，而c就沒這么完善了，不過這也說明了c的簡單靈活。但悲哀的是很多人因此進行了很ugly的實現。
多吐槽幾句，嵌入式行業由于各種技術原因，導致開發語言還是采用c，這樣對開發人員有了不小的要求，而如何才能寫一些優雅的代碼，對人的素質有了要求，但現狀是優秀的開發者都被互聯網行業搶走了，導致嵌入式行業開發人員的水平參差不齊，本來應該是一個對編碼能力要求很高的行業被一些水平低下的開發者占據。so，我離開了這個行業了。。。

附：模塊頭文件，類linux用戶可通過gcc xxx.c命令build該demo，然后運行測試。

#ifndef SFIFO_H_
#define SFIFO_H_struct sfifo_list_des_s {int sfifo_num;struct sfifo_s *head;struct sfifo_s *tail;pthread_mutex_t lock_mutex;pthread_cond_t cond;
};struct sfifo_des_s {int sfifo_init;unsigned int sfifos_num;unsigned int sfifos_active_max_num;struct sfifo_list_des_s free_list;struct sfifo_list_des_s active_list;
};struct sfifo_s {unsigned char *buffer;unsigned int size;struct sfifo_s *next;
};extern struct sfifo_des_s *sfifo_init(int sfifo_num, int sfifo_buffer_size, int sfifo_active_max_num);/* productor */
extern struct sfifo_s* sfifo_get_free_buf(struct sfifo_des_s *sfifo_des_p);
extern int sfifo_put_free_buf(struct sfifo_s *sfifo, struct sfifo_des_s *sfifo_des_p);/* consumer */
extern struct sfifo_s* sfifo_get_active_buf(struct sfifo_des_s *sfifo_des_p);
extern int sfifo_put_active_buf(struct sfifo_s *sfifo, struct sfifo_des_s *sfifo_des_p);#endif