RT-Thread 隐藏的宝藏之ringblk

一. ring blk 是什么

大家应该经常会听到 ringbuff , 那 ringblk(环形块缓冲区) 是什么呢？

ringblk 也是环形缓冲区，但是他的缓冲区的单位是一个 block , 每次对数据的操作都是以 block 为单位。

二. ringblk 怎么使用

ringblk 的使用先要创建环形块缓冲区，通过 alloc 申请到块之后，通过 put 写数据到缓冲区，通过 get 从缓冲区中获取数据。对于不需要在用的块，通过 free 释放。

连续的 block 可以组成 块队列。

1. 创建块缓冲区

rt_rbb_t rt_rbb_create(rt_size_t buf_size, rt_size_t blk_max_num)

buf_size: 缓冲区大小

blk_max_num: 支持最大的 block 的胡亮

2. 销毁块缓冲区

void rt_rbb_destroy(rt_rbb_t rbb)

rbb ：ringblk 的对象

3. 获取块缓冲区的大小

rt_size_t rt_rbb_get_buf_size(rt_rbb_t rbb)

rbb : 块缓冲区的对象

4. 从块缓冲区中分配一个块

rt_rbb_blk_t rt_rbb_blk_alloc(rt_rbb_t rbb, rt_size_t blk_size)

rbb : 块缓冲区的对象

blk_size : 块的大小

返回值：如果不为 NULL ，则表示返回分配的块对象

5. 将一个块放入到块缓冲区

void rt_rbb_blk_put(rt_rbb_blk_t block)

block ：从块缓冲区申请出来的块对象

6. 从块缓冲区获取一个块

rt_rbb_blk_t rt_rbb_blk_get(rt_rbb_t rbb)

rbb : 块缓冲区的对象

返回值：获取到块的对象

7. 释放一个块缓冲区上的块

void rt_rbb_blk_free(rt_rbb_t rbb, rt_rbb_blk_t block)

rbb : 块缓冲区的对象

block : 块对象

8 . 获取块队列

rt_size_t rt_rbb_blk_queue_get(rt_rbb_t rbb, rt_size_t queue_data_len, rt_rbb_blk_queue_t blk_queue)

rbb : 块缓冲区的对象

queue_data_len : 块队列中数据的最大长度，返回的长度必须小于它

blk_queue ：块队列

块队列中的块缓冲区地址连续.

             tail                         head+------------------+---------------+--------+----------+--------+|      block3      |  empty1       | block1 |  block2  |fragment|+------------------+------------------------+----------+--------+|<-- return_size -->|    ||<--- queue_data_len --->|tail                          head+------------------+---------------+--------+----------+--------+|      block3      |  empty1       | block1 |  block2  |fragment|+------------------+------------------------+----------+--------+|<-- return_size -->|    out of len(b1+b2+b3)    ||<-------------- queue_data_len ---------------->|

9. 获取块队列的数据长度

rt_size_t rt_rbb_blk_queue_len(rt_rbb_blk_queue_t blk_queue)

blk_queue: 块队列对象

10. 块队列的缓冲区

rt_uint8_t *rt_rbb_blk_queue_buf(rt_rbb_blk_queue_t blk_queue)

blk_queue: 块队列对象

11. 释放块队列

void rt_rbb_blk_queue_free(rt_rbb_t rbb, rt_rbb_blk_queue_t blk_queue)

rbb : 块缓冲区

blk_queue ：块队列对象

12. 下一个可以组成块队列的总长度

rt_size_t rt_rbb_next_blk_queue_len(rt_rbb_t rbb)

rbb : 块缓冲区的对象

三. 块缓冲区的实现

在分析原理之前先熟悉一下 ringblk 相关的结构体

struct rt_rbb_blk
{rt_rbb_status_t status :8; // 状态用 8 个bit 表示/* less then 2^24 */rt_size_t size :24; // 数据最长不能超过 2 的 24 次方rt_uint8_t *buf; // 块的缓冲区rt_slist_t list;// block 的链表
};
typedef struct rt_rbb_blk *rt_rbb_blk_t;

struct rt_rbb_blk_queue
{rt_rbb_blk_t blocks; // 连续的块rt_size_t blk_num;// block 的数量
};
typedef struct rt_rbb_blk_queue *rt_rbb_blk_queue_t;

struct rt_rbb
{rt_uint8_t *buf; // 块缓冲区的数据缓冲区rt_size_t buf_size; // buf 的大小/* all of blocks */rt_rbb_blk_t blk_set; // 块的集合rt_size_t blk_max_num; // 块的最大数量/* saved the initialized and put status blocks */rt_slist_t blk_list;// 链表
};
typedef struct rt_rbb *rt_rbb_t;

以下的源码都有删减：

1. 创建块缓冲区

rt_rbb_t rt_rbb_create(rt_size_t buf_size, rt_size_t blk_max_num)
{rbb = (rt_rbb_t)rt_malloc(sizeof(struct rt_rbb));// 申请块缓冲区的内存buf = (rt_uint8_t *)rt_malloc(buf_size);// 申请数据缓冲区的内存blk_set = (rt_rbb_blk_t)rt_malloc(sizeof(struct rt_rbb_blk) * blk_max_num);//申请最大 blk 数量的内存rt_rbb_init(rbb, buf, buf_size, blk_set, blk_max_num); // 初始化块缓冲区return rbb;
}

2. 销毁块缓冲区

void rt_rbb_destroy(rt_rbb_t rbb)
{// 释放申请的内存rt_free(rbb->buf);rt_free(rbb->blk_set);rt_free(rbb);}

3. 获取块缓冲区的大小

rt_size_t rt_rbb_get_buf_size(rt_rbb_t rbb)
{return rbb->buf_size; // 返回 块缓冲区的大小
}

4. 从块缓冲区分配一个块

rt_rbb_blk_t rt_rbb_blk_alloc(rt_rbb_t rbb, rt_size_t blk_size)
{rt_base_t level;rt_size_t empty1 = 0, empty2 = 0;rt_rbb_blk_t head, tail, new_rbb = NULL;level = rt_hw_interrupt_disable();// 关中断new_rbb = find_empty_blk_in_set(rbb); // 找到一个空闲块// 判断 申请出来的块是不是在 最大范围之内if (rt_slist_len(&rbb->blk_list) < rbb->blk_max_num && new_rbb){if (rt_slist_len(&rbb->blk_list) > 0) // 检查是不是第一次申请blk{   // 获取头节点的结构体起始地址head = rt_slist_first_entry(&rbb->blk_list, struct rt_rbb_blk, list);// 获取尾节点的结构体起始地址tail = rt_slist_tail_entry(&rbb->blk_list, struct rt_rbb_blk, list);if (head->buf <= tail->buf) // 头节点数据缓冲区的地址小于尾节点的数据缓存区的地址{/***                      head                     tail* +--------------------------------------+-----------------+------------------+* |      empty2     | block1 |   block2  |      block3     |       empty1     |* +--------------------------------------+-----------------+------------------+*                            rbb->buf*/// 求出空 block 的大小empty1 = (rbb->buf + rbb->buf_size) - (tail->buf + tail->size);empty2 = head->buf - rbb->buf;// 判断新的 block 可以存放的区域if (empty1 >= blk_size){ // 给 block 结构体赋值rt_slist_append(&rbb->blk_list, &new_rbb->list);new_rbb->status = RT_RBB_BLK_INITED;new_rbb->buf = tail->buf + tail->size;new_rbb->size = blk_size;}else if (empty2 >= blk_size){// 给 block 结构体赋值rt_slist_append(&rbb->blk_list, &new_rbb->list);new_rbb->status = RT_RBB_BLK_INITED;new_rbb->buf = rbb->buf;new_rbb->size = blk_size;}else{/* no space */new_rbb = NULL;}}else{/***        tail                                              head* +----------------+-------------------------------------+--------+-----------+* |     block3     |                empty1               | block1 |  block2   |* +----------------+-------------------------------------+--------+-----------+*                            rbb->buf*/// 获取空闲的空间empty1 = head->buf - (tail->buf + tail->size);// 判断剩余空间是否够本次的分配if (empty1 >= blk_size){// 给 block 结构体赋值rt_slist_append(&rbb->blk_list, &new_rbb->list);new_rbb->status = RT_RBB_BLK_INITED;new_rbb->buf = tail->buf + tail->size;new_rbb->size = blk_size;}else{   /* no space */new_rbb = NULL;}}}else{/* the list is empty */rt_slist_append(&rbb->blk_list, &new_rbb->list); // 把bew_rbb 链表插入到 rbbnew_rbb->status = RT_RBB_BLK_INITED; // 修改状态为 已经初始化new_rbb->buf = rbb->buf; // 设置缓冲区new_rbb->size = blk_size;// 设置块大小}}else{new_rbb = NULL;}rt_hw_interrupt_enable(level);return new_rbb;
}

5. 将一个块放入到块缓冲区

void rt_rbb_blk_put(rt_rbb_blk_t block)
{block->status = RT_RBB_BLK_PUT; // 设置状态为 PUT
}

6. 从块缓冲区获取一个块

rt_rbb_blk_t rt_rbb_blk_get(rt_rbb_t rbb)
{rt_base_t level;rt_rbb_blk_t block = NULL;rt_slist_t *node;if (rt_slist_isempty(&rbb->blk_list)) // 链表为空，就是没有数据，就无法获取到数据了return 0;level = rt_hw_interrupt_disable();// 关中断// 遍历链表for (node = rt_slist_first(&rbb->blk_list); node; node = rt_slist_next(node)){   // 获取当前 list 的结构体地址block = rt_slist_entry(node, struct rt_rbb_blk, list);if (block->status == RT_RBB_BLK_PUT) // 判断当前块的状态为 PUT{block->status = RT_RBB_BLK_GET; // 设置状态为 GETgoto __exit;}}/* not found */block = NULL;__exit:rt_hw_interrupt_enable(level); // 开中断return block; //返回 block
}

7. 释放块缓冲区上的块

void rt_rbb_blk_free(rt_rbb_t rbb, rt_rbb_blk_t block)
{/* remove it on rbb block list */rt_slist_remove(&rbb->blk_list, &block->list);// 从链表上脱离block->status = RT_RBB_BLK_UNUSED; // 设置状态为 未使用
}

8. 获取块队列

rt_size_t rt_rbb_blk_queue_get(rt_rbb_t rbb, rt_size_t queue_data_len, rt_rbb_blk_queue_t blk_queue)
{rt_base_t level;rt_size_t data_total_size = 0;rt_slist_t *node;rt_rbb_blk_t last_block = NULL, block;if (rt_slist_isempty(&rbb->blk_list))// 如果链表为空，就不会有块队列了return 0;level = rt_hw_interrupt_disable(); // 关中断// 遍历链表for (node = rt_slist_first(&rbb->blk_list); node; node = rt_slist_next(node)){if (!last_block) // 如果下一个 block 为空{   // 获取 list 节点上的结构体的地址last_block = rt_slist_entry(node, struct rt_rbb_blk, list);if (last_block->status == RT_RBB_BLK_PUT) // 如果状态是 PUT{/* save the first put status block to queue */blk_queue->blocks = last_block; // 保存第一个 blockblk_queue->blk_num = 0;}else{/* the first block must be put status */last_block = NULL; // 第一个 block 必须是 PUT 状态continue;}}else // 执行了 else 说明不是第一个 block 了{  //  获取的结构体的首地址block = rt_slist_entry(node, struct rt_rbb_blk, list);/** these following conditions will break the loop:* 1. the current block is not put status* 2. the last block and current block is not continuous* 3. the data_total_size will out of range*/if (block->status != RT_RBB_BLK_PUT ||last_block->buf > block->buf ||data_total_size + block->size > queue_data_len){break;}/* backup last block */last_block = block; // 下一个 block}/* remove current block */rt_slist_remove(&rbb->blk_list, &last_block->list); // 移除链表data_total_size += last_block->size; // 设置数据的总长度last_block->status = RT_RBB_BLK_GET; // 设置状态为 GETblk_queue->blk_num++;// 数量 +1}rt_hw_interrupt_enable(level); // 开中断return data_total_size; // 数据的总长度
}

9. 获取块队列的数据长度

rt_size_t rt_rbb_blk_queue_len(rt_rbb_blk_queue_t blk_queue)
{rt_size_t i, data_total_size = 0;for (i = 0; i < blk_queue->blk_num; i++){data_total_size += blk_queue->blocks[i].size; // 循环累加数据块的大小}return data_total_size;
}

10. 块队列的缓冲区

rt_uint8_t *rt_rbb_blk_queue_buf(rt_rbb_blk_queue_t blk_queue)
{return blk_queue->blocks[0].buf; // 返回块队列的数据的首地址
}

11. 释放块队列

void rt_rbb_blk_queue_free(rt_rbb_t rbb, rt_rbb_blk_queue_t blk_queue)
{rt_size_t i;for (i = 0; i < blk_queue->blk_num; i++){rt_rbb_blk_free(rbb, &blk_queue->blocks[i]); // 循环释放已申请的空间}
}

12. 下一个可以组成块队列的总长度

rt_size_t rt_rbb_next_blk_queue_len(rt_rbb_t rbb)
{rt_base_t level;rt_size_t data_len = 0;rt_slist_t *node;rt_rbb_blk_t last_block = NULL, block;if (rt_slist_isempty(&rbb->blk_list)) // 检查链表是否为空return 0;level = rt_hw_interrupt_disable(); // 关中断// 遍历链表for (node = rt_slist_first(&rbb->blk_list); node; node = rt_slist_next(node)){if (!last_block) // 第一次为 NULL 才可以进入这里的判断{   // 获取 last_lock 的结构体的首地址last_block = rt_slist_entry(node, struct rt_rbb_blk, list);if (last_block->status != RT_RBB_BLK_PUT){   // 检查块的状态/* the first block must be put status */last_block = NULL;continue;}}else{   // 获取 block 结构体块的首地址block = rt_slist_entry(node, struct rt_rbb_blk, list);/** these following conditions will break the loop:* 1. the current block is not put status* 2. the last block and current block is not continuous*/if (block->status != RT_RBB_BLK_PUT || last_block->buf > block->buf){break;}/* backup last block */last_block = block; // last_block 指向  block}data_len += last_block->size;// 累加 数据的长度}rt_hw_interrupt_enable(level); // 开中断return data_len; // 返回数据的长度
}

四. 总结

ringblk: 是由 多个不同长度 的 block 组成的，ringbuff : 是由单字节的数据组成的。ringblk 每一个 block 有多少个字节可以由用户自己设定。
ringblk 支持零字节拷贝(不需要额外的 memcpy 操作)。所以 rbb 非常适合用于生产者顺序 put 数据块，消费者顺序 get 数据块的场景，例如：DMA 传输，通信帧的接收与发送等等。