Zephys OS nano 核心篇：環形緩衝 Ring Buffer
阿新 • • 發佈：2019-02-16
Zephyr OS 所有的學習筆記已託管到 Github，CSDN 部落格裡的內容只是 Github 裡內容的拷貝，因此連結會有錯誤，請諒解。
與棧有點類似，環形緩衝的大小也是在初始化時就固定了。
環形緩衝的型別定義
struct ring_buf {
uint32_t head; /**< Index in buf for the head element */
uint32_t tail; /**< Index in buf for the tail element */
uint32_t dropped_put_count; /**< Running tally of the number of failed put attempts */ 不考慮用於除錯的 __next，環形buffer一個有6個成員：
- head：“指向”環形buffer中資料的頭部。注意，head不是指標，而是一個下標索引。
- tail：“指向”環形buffer中資料的尾部的下一個地址。注意，tail也不是指標，是一個下標索引。
- dropped_put_count：記錄向buffer中新增資料時由於buffer剩餘空間不夠而被丟棄的資料的次數。
- size：環形buffer的總容量。
- buf：指向環形buffer的起始地址。
- mask：當buffer的長度是2的整數次冪時，使用該成員進行“取模”運算，將buffer的下標索引表示在(0, size-1)範圍內。
環形緩衝的初始化
nanokernel 提供了三種初始化環形buffer的方法，它們分別有各自的使用場景
SYS_RING_BUF_DECLARE_POW2
#define SYS_RING_BUF_DECLARE_POW2(name, pow) \
static uint32_t _ring_buffer_data_##name[1 << (pow)]; \
struct ring_buf name = { \
.size = (1 << (pow)), \
.mask = (1 << (pow)) - 1, \
.buf = _ring_buffer_data_##name \
};當buffer的長度是2的整數次冪時，推薦使用該巨集進行初始化。當使用該巨集進行初始化後，再向該buffer中新增資料和取出資料時的效率比使用第二種方法的效率高。該巨集做了兩件事兒：
- 先定義了一個buffer，即環形緩衝實際用來存放資料的buffer。
- 定義了一個描述環形緩衝的結構體，並對其相關成員進行初始化。
- size：初始化為
2pow 。 - mask：初始化為二進位制的0b111…111，即後面一共有pow個1。
- buf：指向所分配的記憶體空間。
- buffer中其它成員被初始化為 0。
- size：初始化為
使用該巨集進行初始化時，所開闢的buffer的記憶體空間的長度是
SYS_RING_BUF_DECLARE_SIZE
#define SYS_RING_BUF_DECLARE_SIZE(name, size32) \
static uint32_t _ring_buffer_data_##name[size32]; \
struct ring_buf name = { \
.size = size32, \
.buf = _ring_buffer_data_##name \
};當buffer的長度不是2的整數次冪時，可以使用該巨集進行初始化。使用該巨集初始化後，再向該buffer中新增資料和取出資料時的效率比使用第一種方法的效率低。該巨集做了兩件事兒：
- 先定義了一個buffer，即環形緩衝實際用來存放資料的buffer。
- 定義了一個描述環形緩衝的結構體，並對其相關成員進行初始化。
- size：初始化為size32。
- buf：指向所分配的記憶體空間。
- buffer中其它成員被初始化為 0。
使用該巨集進行初始化時，所開闢的buffer的記憶體空間的長度是
sys_ring_buf_init
static inline void sys_ring_buf_init(struct ring_buf *buf, uint32_t size,
uint32_t *data)
{
buf->head = 0;
buf->tail = 0;
buf->dropped_put_count = 0;
buf->size = size;
buf->buf = data;
// is_power_of_two()用於判斷一個數是不是2的整數次冪，其實現非常
// 巧妙，有興趣的可以研究研究
if (is_power_of_two(size)) {
buf->mask = size - 1;
} else {
buf->mask = 0;
}
SYS_TRACING_OBJ_INIT(sys_ring_buf, buf);
}該函式能判斷buffer的大小是不是2的整數次冪，然後進行相應的初始化。使用該函式進行初始化的一般步驟為：
#define MY_RING_BUF_SIZE 64
struct my_struct {
struct ring_buffer rb;
uint32_t buffer[MY_RING_BUF_SIZE];
...
};
struct my_struct ms;
void init_my_struct {
sys_ring_buf_init(&ms.rb, sizeof(ms.buffer), ms.buffer);
...
}假設我們初始化了一個長度為16的buffer，那麼其記憶體空間分佈情況如圖 1 所示。
圖 1. 環形緩衝的記憶體空間儲存情況
新增資料
int sys_ring_buf_put(struct ring_buf *buf, uint16_t type, uint8_t value,
uint32_t *data, uint8_t size32)
{
uint32_t i, space, index, rc;
// 先獲取buffer中剩餘空間的長度
space = sys_ring_buf_space_get(buf);
if (space >= (size32 + 1)) {
// 如果剩餘空間足夠存放資料，才將其存放到buffer中
// 這裡判斷剩餘空間的長度是否大於 size32 + 1，是因為除了真實的資料外，
// 還有一個數據頭(佔一個整型的長度)，用於記錄資料的 metadata。
struct ring_element *header =
(struct ring_element *)&buf->buf[buf->tail];
header->type = type;
header->length = size32;
header->value = value;
// 通過mask的值是否為0，判斷buffer的長度是否是2的整數次冪，然後分開處理
if (likely(buf->mask)) {
// 如果是2的整數次冪
for (i = 0; i < size32; ++i) {
// 一個位元組一個位元組地將資料複製到buffer中
// 先通過與 mask 按位與取得index
index = (i + buf->tail + 1) & buf->mask;
// 再複製資料
buf->buf[index] = data[i];
}
// 將tail“指向”資料buffer中資料末尾的下一個地址處
buf->tail = (buf->tail + size32 + 1) & buf->mask;
} else {
// 如果不是2的整數次冪
for (i = 0; i < size32; ++i) {
// 一個位元組一個位元組地將資料複製到buffer中
// 先通過對 size 進行取模運算取得 index
index = (i + buf->tail + 1) % buf->size;
// 再複製資料
buf->buf[index] = data[i];
}
// 將tail“指向”資料buffer中資料末尾的下一個地址處
buf->tail = (buf->tail + size32 + 1) % buf->size;
}
rc = 0;
} else {
// 如果buffer的剩餘空間不夠存放資料，dropped_put_count 遞增
buf->dropped_put_count++;
// 並返回錯誤碼
rc = -EMSGSIZE;
}
return rc;
} sys_ring_buf_space_get()用於獲取buffer的剩餘空間，其實現如下：
static inline int sys_ring_buf_space_get(struct ring_buf *buf)
{
if (sys_ring_buf_is_empty(buf)) {
// 當buffer為空，參考圖1
return buf->size - 1;
}
// 當tail < head，參考圖 2
if (buf->tail < buf->head) {
return buf->head - buf->tail - 1;
}
// 當 tail > head，參考圖3
return (buf->size - buf->tail) + buf->head - 1;
}
圖 2. tail < head 的示意圖
圖 3. tail < head 的示意圖
取出資料
int sys_ring_buf_get(struct ring_buf *buf, uint16_t *type, uint8_t *value,
uint32_t *data, uint8_t *size32)
{
struct ring_element *header;
uint32_t i, index;
if (sys_ring_buf_is_empty(buf)) {
// 如果 buffer 是空的，直接返回錯誤碼
return -EAGAIN;
}
header = (struct ring_element *) &buf->buf[buf->head];
if (header->length > *size32) {
// 資料長度不匹配，將實際長度返回給呼叫者
*size32 = header->length;
// 然後返回錯誤碼
return -EMSGSIZE;
}
*size32 = header->length;
*type = header->type;
*value = header->value;
if (likely(buf->mask)) {
// 如果 buffer 的長度是 2 的整數次冪
for (i = 0; i < header->length; ++i) {
// 通過與 mask 的按位與運算，取得 index
index = (i + buf->head + 1) & buf->mask;
// 然後複製資料到 data 中
data[i] = buf->buf[index];
}
// 重新調整頭部 index
buf->head = (buf->head + header->length + 1) & buf->mask;
} else {
// 如果 buffer 的長度不是 2 的整數次冪
for (i = 0; i < header->length; ++i) {
// 通過與 buffer 的 size 進行取模運算，獲取 index
index = (i + buf->head + 1) % buf->size;
// 然後複製資料到 data 中
data[i] = buf->buf[index];
}
// 重新調整頭部 index
buf->head = (buf->head + header->length + 1) % buf->size;
}
return 0;
}