环形缓冲区(Ring Buffer)

什么是环形缓冲区?

嵌入式C中,环形缓冲区是一个基于数组的简单可靠的环回线性数据存取结构。核心是通过一个写指针(我习惯叫head),一个读指针(tail)管理数据的写入和读取,并用循环取模或位与等操作实现回绕的特性。环形缓冲区解决了传统线性缓冲区频繁分配与释放导致的内存碎片、效率低下等问题。

接下来给一个简单流程示意图例:

  • 初始时,读写指针均位于缓冲区起始,此时缓冲区为空状态,如图
  • 数据由Head写入,写入后Head索引移动到下一待写入位置。
  • 缓冲区中数据由Tail读出,读出后Tail索引移动到下一待读取位置。
  • Head写满后回绕到缓冲区起始,Tail同理。

以上便是环形缓冲区正常工况下数据存取的一个简单模型。内部一些具体细节将在下文展开,上述流程只是为了简单了解下环形缓冲区的工作原理。从上述可看出:

  1. 环形缓冲区本质是一个FIFO队列,先进先出。
  2. 读写数据时间复杂度均为O(1),无需移动任何数据。
  3. 读写索引到缓冲区边界后自动发生回绕。

环形缓冲区是嵌入式系统中十分重要且常用的一个数据结构,在串口收发场景;DMA数据缓冲场景;传感器数据缓存场景等均有很广泛的运用。


环形缓冲区的数据结构

要设计一个环形缓冲区,有3个要素是必须的:

  • 缓冲区数组:这就是我们的缓冲区物理空间,后续我们所有的操作均基于该缓冲区上进行。
  • 读/写索引:环形缓冲区灵魂所在,利用两个索引的平移和回绕在线性数组上天然实现环形特性。注意我们这里一直说的是“索引”,所以本质其实应该是一个u16/u32变量,不要定义为指针类型复杂化问题!

这里提前说明一下:还有一个要素是 有效数据长度 ,但是该要素是否必须取决于你的缓冲区实现策略,这里我使用的是牺牲槽判空法的策略,因此不需要该元素。

通过上面的叙述,我们可以定义出一个结构如下:

1
2
3
4
5
6
#define RING_BUF_SIZE	(64u)
typedef struct {
uint8_t buffer[RING_BUF_SIZE]; /* 缓冲区 */
uint16_t head; /* 写索引 */
uint16_t tail; /* 读索引 */
} rBuf_t;

每向缓冲区写一次数据,head索引步进一格;每从缓冲区读一次数据,tail索引步进一格。这个流程是明确且易懂的,但是这其中涉及到几个问题需要考虑清楚:怎么判断缓冲区是空状态?怎么判断缓冲区是满状态?回绕具体是怎么个回绕法?接下来我们就来详细讲讲。


环形缓冲区判空判满策略

关于状态判断,你可能会这么想:“既然是环形缓冲区,那么当head追上tail时不就满了吗?那么等量关系很简单直接就是 head == tail 。” 这么想其实没错,但是忽略了一个前提:我们在初始化状态(也就是初始空状态)时已经满足等量关系 head == tail 。若 head == tail 也用来判满,那么会出现一个等量关系对应两种不同的状态,这肯定是不能成立的。

因此我们需要一个办法来区分空和满状态,这里有三种常见方法。这里详细讲讲我使用最多的方法:

  • 牺牲一个空槽位进行判满。

顾名思义,该方法的核心思想是:永久牺牲掉一个槽位用于判满操作,例如我们缓冲区大小为64个槽,那么在该方法下,我们最多可用的空间只有63个槽位。

原理是这样的,既然 head == tail 无法使用,那么我们使用 head + 1 == tail 不就可以了吗?在写入槽位之前提前判断下一个槽位是不是读索引,如果是那么说明除开当前这个空槽之外,缓冲区已经满了。此时我们牺牲掉这个空位置,并抛出缓冲区满事件供上层处理。

上图示例中,缓冲区总大小 Size == 8。但是在该方法下可用的大小永远只有7个槽。

Ps:这里要强调下,判满条件不能直接写 head + 1 == tail ,我们要将回绕考虑进去,因此正确的逻辑应该是 (head + 1) % Size == tail。上述的写法是我未将回绕考虑进去的简写,注意区分。

牺牲一个空槽位的方法在嵌入式实践中比较常见,因为其不需要维护额外的变量,简单可靠。其代价仅仅只是浪费一格存储单元,这个代价我们基本可以忽略不计。

  • 维护一个计数变量进行判满。

如前文所述,我们在缓冲区结构中可以多维护一个有效数据长度的要素。如下

1
2
3
4
5
6
7
#define RING_BUF_SIZE	(64u)
typedef struct {
uint8_t buffer[RING_BUF_SIZE]; /* 缓冲区 */
uint16_t head; /* 写索引 */
uint16_t tail; /* 读索引 */
uint16_t count; /* 缓冲区中有效数据条目计数 */
} rBuf_t;

每当往缓冲区写入一个数据时,count++;从缓冲区读出一个数据时,count--。可用 RING_BUF_SIZE == count ? 来判断缓冲区是否已满。 用 count == 0 来判断缓冲区是否已空。

这个方法就是多维护一个变量,多一次非原子的读写操作(生产者和消费者均需要访问该变量),但是只要保证在临界区操作或串行化访问的话,其实并不会出任何问题。这种方式其实并没有什么太大的弊端,也比较直观,所以可以采用该方法实现,不过我个人还是习惯上一种方式。

  • 镜像指示,标志翻转判满。

这个和第二个有些类似,需要在缓冲区结构中维护一个标志位flag。

1
2
3
4
5
6
7
#define RING_BUF_SIZE	(64u)
typedef struct {
uint8_t buffer[RING_BUF_SIZE]; /* 缓冲区 */
uint16_t head; /* 写索引 */
uint16_t tail; /* 读索引 */
uint16_t flag; /* 标志位(应初始化为0) */
} rBuf_t;

当初始为空状态时,我们用 (head == tail) && (flag == 0) 可判断当前缓冲区为空;当我们往缓冲区写入数据时,该次写入完毕后若 head == tail 则表示写索引追上了读索引,此时我们应将 flag = 1 表示缓冲区已满并向上层抛出一个事件;同样的,当本次读取完毕后若 head == tail 则表示读索引追上了写索引,应将 flag = 0 表示缓冲区为空。

镜像指示位法的优势在于单生产者单消费者场景下天然无锁安全:flag 只在“写追上读”时被写方置 1,只在“读追上写”时被读方置 0,不存在竞态。缺点是无法直接知道当前有多少数据。

这个方式了解一下即可,我个人并不建议使用这种方式,因为该方式类似于是上一个的下位替代,唯一的线程安全优势其实并不是那么绝对——在可能存在竞争的场景下我们均会采取临界区等手段来规避风险,所以在该种场景下采用count其实也是安全的,还能够精确进行计数。


环形缓冲区绕回特性

缓冲区在物理上是线性的,因此要实现环形缓冲,自动回绕就必须在软件逻辑上下功夫。我们通常使用对索引取余的方式来完成回绕,其计算公式如下:

1
2
3
关于写索引: (head + 1) % Size
关于读索引: (tail + 1) % Size
其中,Size为缓冲区大小.

举个例子:我们有一个32槽位的缓冲区,已经装了29个数据,其布局如下所示。目前head的值为31(注意不是30,因为写索引指向的是下一个待写的槽位)。

当再次发生写入操作时,该包数据被放入head指向的最后一个槽位(Buf[31]),随后执行 (head + 1) % Size更新索引。head + 1 = 32;32 % 32 = 0 看见了吗,索引自动的正确回绕到了Buf[0]。后续的写入操作又继续重复上述的缓冲区写入流程。tail读取索引也是同样的回绕方式。

这种取余的方式进行回环是通用且常见的方式,接下来介绍另一种效率更高的方式。

  • 通过按位与(&)操作进行逻辑绕回。

这里要说明一下:该方法并不通用,只有在缓冲区大小Size为2的幂等数情况下才可以使用。这是因为只有当Size是2的幂时,size - 1的二进制才是全1,按位与才能等价于取模。

其索引更新的计算公式为

1
(head + 1) & (Size - 1);

举个例子:

1
2
3
4
5
6
Size = 8  --> 0b1000
head = 7
(head + 1) = 0b1000;
(Size - 1) = 0b0111;
(head + 1) & (Size - 1) = 0b1000 & 0b0111;
很明显,最后按位与出来的结果是0. 索引正确回绕

这种方式最明显的优势就是速度极快,比通用的取余方式能快出不少,所以在设计缓冲区时其缓冲区大小尽量设置为2的幂等数,这样就能使用该方式进一步提升程序效率。


环形缓冲区本质上它就是一段线性数组加上两个索引,靠取模或位与实现逻辑上的首尾相连;判空判满的重点在于“如何区分空和满”,牺牲槽位法、计数变量法、镜像指示位法各有取舍,选哪个取决于你的场景。回绕是这套机制的灵魂——物理上是线性的,逻辑上却是环形的。

C黑魔法 X-Macro 宏

最近在写一个简易的内存管理逻辑用于 CCM_RAM 以及其余通用 RAM 管理。过程中涉及到了内存池的配置:如果需要添加一个新类型的内存池,常规做法是在相关 .c/.h 中再定义一个数组以及维护一套描述头结构。但是这涉及到多个文件的修改,如果漏改一个地方那么将导致程序编译不过。而且如果后续逻辑写大之后,维护将及其不便。此时便想起了可以使用 X-Macro 来解决这个问题。

什么是 X-Macro ?

X-Macro 是一种用宏来定义数据列表,然后反复重定义 X 来生成不同代码的预处理器技巧。是一种可靠维护代码或数据的并行列表的技术,你只需要维护一份数据列表,编译器自动帮你生成枚举、数组、函数声明、结构体、初始化代码等所有重复性内容。

X-Macro 并不是C语言的新特性,它本质其实就是普通的宏定义与展开,在这个基础之上才衍生出了 X-Macro,因此 X-Macro 准确来说应该是一个宏的使用技巧。

使用方法

单看概念会觉得十分抽象,这里我们给出一个简单的示例:

1
2
3
4
#define MY_DATA_LIST \
X(ITEM_A, 100, "hello") \
X(ITEM_B, 200, "world") \
X(ITEM_C, 300, "foo")

上面我们定义了一个数据表,现在我们想要利用这个数据表,一次性定义3个变量,并使用表中的数值参数对其进行初始化,该怎么做? 这里可以使用 X-Macro 来完成:

1
2
3
#define X(name, val, str) int name##_value = val;
MY_DATA_LIST
#undef X

上面逻辑中,我们定义了 X(参数) 为我们想做的操作(变量定义):int 参数1##_value = 参数2;

在预处理阶段,MY_DATA_LIST 展开后得到下列程序:

1
2
3
int ITEM_A_value = 100;
int ITEM_B_value = 200;
int ITEM_C_value = 300;

发现没有?我们仅仅通过一次展开 MY_DATA_LIST 宏就得到了所有的变量定义,如果此刻我们想要加入第四个变量该怎么做?很简单,只需要在 MY_DATA_LIST 中多引入一个表项就行。

1
2
3
4
5
6
7
8
9
10
11
#define MY_DATA_LIST \
X(ITEM_A, 100, "hello") \
X(ITEM_B, 200, "world") \
X(ITEM_C, 300, "foo") \
X(ITEM_D, 400, "happy")

// 展开后:
int ITEM_A_value = 100;
int ITEM_B_value = 200;
int ITEM_C_value = 300;
int ITEM_D_value = 400;

我们通过一个简单的变量定义场景就阐述出了 X-Macro 宏的用法。相信你此刻已经见识到了这玩意儿的强大之处:没错,我们只需在 conf.h 中维护一个全局表项,后续的 数组声明/枚举定义 等等均采用 X-Macro 进行展开。如果后期需要对表项做出修改,只需要改 conf.h 文件就行,其余模块内部自适应变化!极大降低了维护工作量,且不易出错。


这里我就不以更多的示例来进行阐述,直接来看一个实际的应用场景:

目前需要设计一个内存池,且该内存池类型不固定(也就是说需要支持不同类型的内存池),且用户侧对于内存池的类型删减必须足够简便。很明显,这就是一个典型的 X-Macro 的用武之地了,一起来看。

我们先以一个默认的类型来入手,默认池采用每块512字节,供划分12块的配置。

1
2
3
4
5
6
/* This is default setting. */
#define DEFAULT_BLOCK_SIZE (512ul)
#define DEFAULT_BLOCK_NUM (12u)

#define MEM_POOL_LIST \
X(DEFAULT, DEFAULT_BLOCK_SIZE, DEFAULT_BLOCK_NUM)

X(DEFAULT, DEFAULT_BLOCK_SIZE, DEFAULT_BLOCK_NUM)

这就是我们的数据列表了,后续用户所有对于内存池配置的操作都在此进行。

列表定义好之后首先我们需要的就是再定义一个枚举,该枚举是后续我们访问不同类型内存池的关键(也可以理解为池索引),接下来我们使用 X-Macro 进行第一次展开!

1
2
3
4
5
6
7
// 第一次 X-Macro 展开.
#define X(name, size, depth) POOL_##name,
typedef enum {
MEM_POOL_LIST
POOL_MAX
} pool_id_t;
#undef X

展开后如下:

1
2
3
4
typedef enum {
POOL_DEFAULT,
POOL_MAX
} pool_id_t;

Ps:这里有个小细节要注意一下:POOL_MAX 是根据枚举型的递增特性,用来统计池类型数目的,后续用于循环计数。

现在索引有了,我们需要再定义物理数组,也就是这片内存池真正的空间来源。这是我们第二次 X-Macro 展开,注意看。

1
2
3
4
5
6
// 第二次 X-Macro 展开.
#define X(name, size, depth) \
static uint8_t __attribute__((section(POOL_SECTION))) __attribute__((aligned(4))) \
pool_##name##_heap[(size + sizeof(void *)) * (depth)];
MEM_POOL_LIST
#undef X

展开后,得到:

1
2
/* 这里 sizeof(void *)在32位下是4字节.一个小细节 */
static uint8_t pool_DEFAULT_heap[((512ul) + 4) * (12u)];

继续,要管理不同类型的池子,需要对应的描述头结构。这里我们描述头定义为:pool_desc_t,具体结构就不详细给出了。每个池子类型都要对应自己的描述头,因此描述头结构应该是一个对应结构数组中的表项。

这里我们第三次使用 X-Macro 展开:

1
2
3
4
5
6
7
// 第三次 X-Macro 展开.
static pool_desc_t pool_table[POOL_MAX] =
{
#define X(name, size, depth) { (size) + sizeof(void*), depth, NULL },
MEM_POOL_LIST
#undef X
};

展开后得到:

1
2
3
4
static pool_desc_t pool_table[POOL_MAX] = 
{
{ (512ul) + 4, (12u), NULL },
};

至此,三次展开已经全部结束。 我们来总览一下 X-Macro 给我们带来了什么。

1
2
3
4
5
6
7
8
9
10
typedef enum {
POOL_DEFAULT,
POOL_MAX
} pool_id_t;

static uint8_t pool_DEFAULT_heap[((512ul) + 4) * (12u)];
static pool_desc_t pool_table[POOL_MAX] =
{
{ (512ul) + 4, (12u), NULL },
};

看到了吧,一行数据列表给我们从结构声明到数据结构定义全部搞定。这就是 X-Macro 最大的意义:

零运行时开销,所有枚举、数组大小、地址排布在编译期就定了。且用户只需改一行列表,所有相关代码自动同步,不需要挨个模块去加实现。

假如我们目前需要加入 TEST 类型池子, 大小1024,深度15,怎么做呢?很简单,多加一个列表的事。

1
2
3
#define MEM_POOL_LIST   \
X(DEFAULT, DEFAULT_BLOCK_SIZE, DEFAULT_BLOCK_NUM) \
X(TEST, (1024ul), (15u))

X-Macro 将自动生成:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
typedef enum {
POOL_DEFAULT,
POOL_TEST,
POOL_MAX
} pool_id_t;

static uint8_t pool_DEFAULT_heap[((512ul) + 4) * (12u)];
static uint8_t pool_TEST_heap[((1024ul) + 4) * (15u)];

static pool_desc_t pool_table[POOL_MAX] =
{
{ (512ul) + 4, (12u), NULL },
{ (1024ul) + 4, (15u), NULL },
};

无需多言了,这就是 X-Macro 的魅力所在。


一些使用场景

我最早接触到 X-Macro 是在 LwIP 的内存管理算法源码中了解到的。LwIP 在内存池逻辑中就使用了 X-Macro 进行不同池类型的管理,其原理与我上述所说的本质是一样的:通过不断展开 X-Macro 来声明与定义数据结构,后续内存池类型的增删(比如 UDP/TCP 等等类型控制块其内存来源都是内存池)直接通过数据列表进行配置,无需改动内部逻辑。

这里我给一些 LwIP 源码中的具体实现,作为参考学习:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
LWIP_MEMPOOL(RAW_PCB,        1,   100, "RAW_PCB")
LWIP_MEMPOOL(TCP_PCB, 5, 200, "TCP_PCB")
LWIP_MEMPOOL(TCP_SEG, 10, 50, "TCP_SEG")
// ... 后面还有很多

#define LWIP_MEMPOOL(name,num,size,desc) MEMP_##name,

// 展开后.
typedef enum {
MEMP_RAW_PCB,
MEMP_TCP_PCB,
MEMP_TCP_SEG,
// ... 继续展开所有池
MEMP_MAX
} memp_t;

/* **************************************************** */
const struct memp_desc *const memp_pools[MEMP_MAX] = {
#define LWIP_MEMPOOL(name,num,size,desc) &memp_ ## name,
#include "lwip/priv/memp_std.h"
};

// 展开后.
const struct memp_desc *const memp_pools[] = {
&memp_RAW_PCB,
&memp_TCP_PCB,
&memp_TCP_SEG,
// ...
};

X-Macro 的适用场景其实很明确:同一份数据,需要在多个地方以不同形式出现——枚举、数组、字符串表、初始化代码。一旦用上,原来需要手工同步的重复代码全部可以删掉。

其技巧本身并不难,难在能不能够在具体业务场景中识别并运用。

音乐
0:000:00