單片機(jī)中的幾種環(huán)形緩沖區(qū)的分析和實(shí)現(xiàn)

一、簡(jiǎn)介

環(huán)形緩沖區(qū)（Ring Buffer）是一種高效的使用內(nèi)存的方法，它將一段固定長(zhǎng)度的內(nèi)存看成一個(gè)環(huán)形結(jié)構(gòu)，用于存儲(chǔ)數(shù)據(jù)，能夠避免使用動(dòng)態(tài)申請(qǐng)內(nèi)存導(dǎo)致的內(nèi)存碎片問(wèn)題，而且其能夠更高效的使用內(nèi)存。

在單片機(jī)中，由于內(nèi)存有限，而且需要盡可能避免使用動(dòng)態(tài)內(nèi)存，所以環(huán)形緩沖區(qū)在單片機(jī)中應(yīng)用非常廣泛。

二、原理

通常我們需要使用一個(gè)數(shù)組或者在程序開(kāi)頭申請(qǐng)一段不被釋放的內(nèi)存，將其作為緩沖區(qū)。然后使用兩個(gè)指令分別指向讀寫(xiě)位置。

使用讀指針管理讀位置，使用寫(xiě)指針管理寫(xiě)位置。當(dāng)讀指針追上寫(xiě)指針時(shí)，表示緩沖區(qū)為空，當(dāng)寫(xiě)指針追上讀指針時(shí)，表示緩沖區(qū)已滿。

環(huán)形緩沖區(qū)的讀寫(xiě)操作都是循環(huán)進(jìn)行的，當(dāng)讀指針或?qū)懼羔樀竭_(dá)緩沖區(qū)的末尾時(shí)，會(huì)自動(dòng)回到緩沖區(qū)的開(kāi)頭。

環(huán)形緩沖區(qū)的讀寫(xiě)操作都是原子操作，即一次只能進(jìn)行一個(gè)讀寫(xiě)操作，避免了多線程或多任務(wù)同時(shí)讀寫(xiě)緩沖區(qū)導(dǎo)致的數(shù)據(jù)混亂。

三、實(shí)現(xiàn)

緩沖區(qū)滿了以后可以選擇覆蓋寫(xiě)或者阻塞等待，需要注意的是，如果選擇覆蓋寫(xiě)，那么讀索引也應(yīng)該向前移動(dòng)，此時(shí)最前面的數(shù)據(jù)就會(huì)丟失；如果選擇阻塞等待的話，盡可能不要在中斷中使用，否則中斷嵌套會(huì)有無(wú)法預(yù)料的執(zhí)行流程。

注：我們可以使用數(shù)組索引代替讀寫(xiě)指針位置，畢竟對(duì)整數(shù)的加減還是比較容易理解的。下面我將讀指針用讀索引代替，寫(xiě)指針用寫(xiě)索引代替。

1、數(shù)據(jù)結(jié)構(gòu)設(shè)計(jì)

我們需要支持多個(gè)環(huán)形緩沖區(qū)，使用同一套代碼邏輯，那么就不能將緩沖區(qū)數(shù)組的大小進(jìn)行硬編碼，而是需要在初始化環(huán)形緩沖區(qū)的時(shí)候通過(guò)參數(shù)傳遞進(jìn)來(lái)，我們使用一個(gè)結(jié)構(gòu)體來(lái)表示緩沖區(qū)數(shù)組。

我們的緩沖區(qū)不能只是支持字節(jié)數(shù)組，而是支持任意類(lèi)型的數(shù)據(jù)，所以我們需要一個(gè)變量來(lái)保存緩沖區(qū)數(shù)組中每個(gè)元素的大小，這樣我們就可以根據(jù)這個(gè)大小然后結(jié)合讀寫(xiě)索引來(lái)獲取和寫(xiě)入數(shù)據(jù)。

typedef struct{ void *ptr; // 緩沖區(qū)數(shù)組指針 uint32_t elem_num; // 緩沖區(qū)數(shù)組元素個(gè)數(shù) uint32_t elem_size; // 緩沖區(qū)數(shù)組中每個(gè)元素的大小} Array;

我們通過(guò)一個(gè)環(huán)形緩沖區(qū)的結(jié)構(gòu)體來(lái)管理緩沖區(qū)，讀寫(xiě)索引，同時(shí)使用一個(gè)變量來(lái)記錄當(dāng)前緩沖區(qū)中有效的元素個(gè)數(shù)，以便判斷環(huán)形緩沖區(qū)是否為空或者是否已滿。

如果在RTOS中使用環(huán)形緩沖區(qū)，那么讀寫(xiě)索引需要使用原子操作，防止多線程或多任務(wù)同時(shí)讀寫(xiě)緩沖區(qū)導(dǎo)致的數(shù)據(jù)混亂。為保證可以實(shí)現(xiàn)原子操作，需要傳入RTOS提供的原子操作方法(進(jìn)入臨界區(qū)、互斥量等)

typedef struct{ Array *buffer; // 緩沖區(qū) uint32_t read_index; // 讀索引 uint32_t write_index; // 寫(xiě)索引 uint32_t count; // 環(huán)形緩沖區(qū)中元素個(gè)數(shù) void (*lock)(void); // 進(jìn)入原子操作 void (*unlock)(void); // 退出原子操作} RingBuffer;

為了及時(shí)了解我們的函數(shù)執(zhí)行結(jié)果，我們?cè)诤瘮?shù)執(zhí)行結(jié)束后需要返回一個(gè)錯(cuò)誤碼用于判斷執(zhí)行情況。

typedef enum{ ARRAY_OK, // 成功 ARRAY_PARAMS_NULL, // 參數(shù)為空 ARRAY_INDEX_OUT_OF_RANGE, // 索引越界} ArrayError;typedef enum{ RB_OK, // 操作成功 RB_READ_NOT_ENOUGH, // 緩沖區(qū)中元素個(gè)數(shù)不足，這只是一個(gè)警告，程序可以進(jìn)行進(jìn)行 RB_PARAM_ERROR, // 參數(shù)錯(cuò)誤 RB_FULL, // 緩沖區(qū)已滿 RB_EMPTY, // 緩沖區(qū)為空} RingBufferStatus;

2、方法實(shí)現(xiàn)

我們的緩沖區(qū)其實(shí)就是一個(gè)數(shù)組，但是我們這里為了支持存儲(chǔ)不同類(lèi)型的元素，使用了Array來(lái)對(duì)這個(gè)數(shù)組進(jìn)行管理（類(lèi)似C++中的Vector）。

在讀寫(xiě)某個(gè)元素的時(shí)候，需要判斷輸入的索引是否在范圍內(nèi)。

// 初始化數(shù)組// pthis: 數(shù)組結(jié)構(gòu)體指針// ptr: 數(shù)組指針// size: 數(shù)組元素個(gè)數(shù)// elem_size: 數(shù)組中每個(gè)元素的大小ArrayError array_init(Array *pthis, void *ptr, const uint32_t elem_num, const uint32_t elem_size){ if (pthis == NULL || ptr == NULL) { return ARRAY_PARAMS_NULL; } pthis->ptr = ptr; pthis->elem_num = elem_num; pthis->elem_size = elem_size; return ARRAY_OK;}// 獲取數(shù)組元素個(gè)數(shù)// pthis: 數(shù)組結(jié)構(gòu)體指針uint32_t array_get_elem_num(Array *pthis){ return pthis->elem_num;}// 獲取數(shù)組中每個(gè)元素的大小// pthis: 數(shù)組結(jié)構(gòu)體指針uint32_t array_get_elem_size(Array *pthis){ return pthis->elem_size;}// 向數(shù)組寫(xiě)入一個(gè)元素// pthis: 數(shù)組結(jié)構(gòu)體指針// index: 要寫(xiě)入的元素索引// elem: 要寫(xiě)入的元素ArrayError array_write_elem(Array *pthis, const uint32_t index, const void *elem){ if (pthis == NULL || elem == NULL) { return ARRAY_PARAMS_NULL; } if (index >= pthis->elem_num) { return ARRAY_INDEX_OUT_OF_RANGE; } memcpy((char *)pthis->ptr + index * pthis->elem_size, elem, pthis->elem_size); return ARRAY_OK;}// 從數(shù)組讀取一個(gè)元素// pthis: 數(shù)組結(jié)構(gòu)體指針// index: 要讀取的元素索引// elem: 讀取的元素ArrayError array_read_elem(Array *pthis, const uint32_t index, void *elem){ if (pthis == NULL || elem == NULL) { return ARRAY_PARAMS_NULL; } if (index >= pthis->elem_num) { return ARRAY_INDEX_OUT_OF_RANGE; } memcpy(elem, (char *)pthis->ptr + index * pthis->elem_size, pthis->elem_size); return ARRAY_OK;}

為了方便判斷環(huán)形緩沖區(qū)是否滿或者空了，我使用一個(gè)變量來(lái)記錄有效的元素?cái)?shù)量，當(dāng)有效元素?cái)?shù)量為0時(shí)表示環(huán)形緩沖區(qū)空了，當(dāng)有效元素?cái)?shù)量為緩存數(shù)組的長(zhǎng)度時(shí)表示環(huán)形緩沖區(qū)滿了。

讀寫(xiě)多個(gè)元素時(shí)，在內(nèi)部都是一個(gè)一個(gè)進(jìn)行的讀寫(xiě)，只有在讀寫(xiě)某個(gè)元素前才后判斷環(huán)形緩沖區(qū)是否滿或者空。那么讀寫(xiě)多個(gè)元素的操作就不一定都能成功，在這里如果全部讀寫(xiě)成功
或者只是因?yàn)榄h(huán)形緩沖區(qū)滿、空導(dǎo)致失敗，只需要返回成功讀寫(xiě)的數(shù)據(jù)，如果是其他原因?qū)е碌淖x寫(xiě)失敗，那么就需要根據(jù)RingBufferStatus的枚舉類(lèi)型返回相應(yīng)的負(fù)數(shù)。

#define RB_LOCK() \ if (rb->lock) \ rb->lock()#define RB_UNLOCK() \ if (rb->unlock) \ rb->unlock()// 初始化環(huán)形緩沖區(qū)// @rb: 環(huán)形緩沖區(qū)// @array: 環(huán)形緩沖區(qū)使用的數(shù)組// @lock: 鎖函數(shù)，進(jìn)入原子操作時(shí)調(diào)用// @unlock: 解鎖函數(shù)，退出原子操作時(shí)調(diào)用// 返回值: RB_OK, RB_PARAM_ERRORRingBufferStatus ring_buffer_init(RingBuffer *rb, Array *array, void (*lock)(void), void (*unlock)(void)){ if (rb == NULL || array == NULL) { return RB_PARAM_ERROR; } rb->buffer = array; rb->lock = lock; rb->unlock = unlock; rb->write_index = 0; rb->read_index = 0; rb->count = 0; return RB_OK;}// 向環(huán)形緩沖區(qū)寫(xiě)入一個(gè)元素// @rb: 環(huán)形緩沖區(qū)// @data: 要寫(xiě)入的數(shù)據(jù)// 返回值: RB_OK, RB_FULL, RB_PARAM_ERRORstatic RingBufferStatus ring_buffer_write_one(RingBuffer *rb, const void *data){ ArrayError err = ARRAY_OK; if (rb == NULL || data == NULL) { return RB_PARAM_ERROR; } if (rb->count == array_get_elem_num(rb->buffer)) { return RB_FULL; } err = array_write_elem(rb->buffer, rb->write_index, data); if (err != ARRAY_OK) { return RB_PARAM_ERROR; } rb->write_index = (rb->write_index + 1) % array_get_elem_num(rb->buffer); rb->count++; return RB_OK;}// 從環(huán)形緩沖區(qū)讀取一個(gè)元素// @rb: 環(huán)形緩沖區(qū)// @data: 讀取的數(shù)據(jù)// 返回值: RB_OK, RB_EMPTY, RB_PARAM_ERRORstatic RingBufferStatus ring_buffer_read_one(RingBuffer *rb, void *data){ ArrayError err = ARRAY_OK; if (rb == NULL || data == NULL) { return RB_PARAM_ERROR; } if (rb->count == 0) { return RB_EMPTY; } err = array_read_elem(rb->buffer, rb->read_index, data); if (err != ARRAY_OK) { return RB_PARAM_ERROR; } rb->read_index = (rb->read_index + 1) % array_get_elem_num(rb->buffer); rb->count--; return RB_OK;}// 向環(huán)形緩沖區(qū)寫(xiě)入數(shù)據(jù)// @rb: 環(huán)形緩沖區(qū)// @data: 要寫(xiě)入的數(shù)據(jù)// @elem_num: 要寫(xiě)入的元素個(gè)數(shù)// 返回值: 當(dāng)至少能寫(xiě)入一個(gè)元素時(shí)，返回實(shí)際寫(xiě)入的元素個(gè)數(shù)；其他時(shí)候返回 -RB_PARAM_ERROR, -RB_FULLint32_t ring_buffer_write(RingBuffer *rb, const void *data, const uint32_t elem_num){ RingBufferStatus ret = RB_OK; int32_t write_num = 0; if (rb == NULL || data == NULL || elem_num == 0) { return -RB_PARAM_ERROR; } RB_LOCK(); for (int32_t i = 0; i < elem_num; i++) { ret = ring_buffer_write_one(rb, (char *)data + i * array_get_elem_size(rb->buffer)); write_num++; if (ret != RB_OK) { break; } } RB_UNLOCK(); if (ret == RB_OK || ret == RB_FULL) { return write_num; // 返回實(shí)際寫(xiě)入的元素個(gè)數(shù) } else { return -ret; // 返回負(fù)數(shù)表示寫(xiě)入失敗 }}// 從環(huán)形緩沖區(qū)讀取數(shù)據(jù)// @rb: 環(huán)形緩沖區(qū)// @data: 讀取的數(shù)據(jù)// @elem_num: 要讀取的元素個(gè)數(shù)// 返回值: 當(dāng)至少能讀取一個(gè)元素時(shí)，返回實(shí)際讀取的元素個(gè)數(shù)；其他時(shí)候返回 -RB_PARAM_ERROR, -RB_EMPTYint32_t ring_buffer_read(RingBuffer *rb, void *data, const uint32_t elem_num){ RingBufferStatus ret = RB_OK; int32_t read_num = 0; if (rb == NULL || data == NULL || elem_num == 0) { return -RB_PARAM_ERROR; } RB_LOCK(); for (int32_t i = 0; i < elem_num; i++) { ret = ring_buffer_read_one(rb, (char *)data + i * array_get_elem_size(rb->buffer)); read_num++; if (ret != RB_OK) { break; } } RB_UNLOCK(); if (ret == RB_OK || ret == RB_READ_NOT_ENOUGH) { return read_num; // 返回實(shí)際讀取的元素個(gè)數(shù) } else { return -ret; // 返回負(fù)數(shù)表示讀取失敗 }}

總結(jié)

本章主要介紹了一種在單片機(jī)中常用的環(huán)形緩沖區(qū)，分析了設(shè)計(jì)思路和代碼實(shí)現(xiàn)。