一、Ring Buffer的核心结构与内存布局

上一篇已经知道，在网络程序中通常需要：

Socket
   ↓
Input Buffer
   ↓
协议解析
   ↓
业务逻辑
   ↓
Output Buffer
   ↓
Socket

那么 Buffer 内部到底应该怎样保存数据？

一种非常常见的方案就是：

Ring Buffer
环形缓冲区

最基本的数据结构可以定义为：

struct ringbuffer_s
{
    uint32_t size; // 缓冲区容量
    uint32_t tail; // 写入位置
    uint32_t head; // 读取位置
    uint8_t* buf;  // 真正保存数据的内存
};

然后：

typedef struct ringbuffer_s buffer_t;

于是外部仍然使用统一的：

buffer_t*

操作 Buffer，而不需要关心内部具体是不是 Ring Buffer。

这里最重要的四个成员是：

size
 ↓
整个缓冲区容量

head
 ↓
已经读取到哪里

tail
 ↓
已经写入到哪里

buf
 ↓
真正保存数据

例如创建一个容量为8的缓冲区：

物理数组：

下标
 0   1   2   3   4   5   6   7
┌───┬───┬───┬───┬───┬───┬───┬───┐
│   │   │   │   │   │   │   │   │
└───┴───┴───┴───┴───┴───┴───┴───┘

head = 0
tail = 0

此时：

head == tail

说明缓冲区为空。

如果写入：

A B C

那么：

 0   1   2   3   4   5   6   7
┌───┬───┬───┬───┬───┬───┬───┬───┐
│ A │ B │ C │   │   │   │   │   │
└───┴───┴───┴───┴───┴───┴───┴───┘
 ↑           ↑
head        tail

可以理解为：

head = 0
tail = 3

消费者读取两个字节以后：

A B

那么：

head = 2
tail = 3

此时真正有效的数据只剩：

C

需要注意：

head 和 tail 在这个实现中不是一直限制在 0 ~ size-1 范围内，而是作为不断递增的逻辑位置使用。
真正访问数组时，再把逻辑位置映射到：

0 ~ size-1

这一点非常重要。

创建 Buffer 的代码：

buffer_t* buffer_new(uint32_t sz)
{
    if (!is_power_of_two(sz)) sz = roundup_power_of_two(sz);

    buffer_t* buf = (buffer_t*)malloc(sizeof(buffer_t) + sz);
    if (!buf) return NULL;

    buf->size = sz;
    buf->head = 0;
    buf->tail = 0;

    // ringbuffer结构体后面的内存直接作为数据区
    buf->buf = (uint8_t*)(buf + 1);

    return buf;
}

这里：

malloc(sizeof(buffer_t) + sz);

一次申请了：

ringbuffer结构体
        +
真正的数据区

内存大概是：

malloc申请的一整块内存

┌──────────────────────┬──────────────────────────────┐
│   ringbuffer_s       │            buf               │
│                      │                              │
│ size                 │ 0 1 2 3 4 5 6 7 ...         │
│ head                 │                              │
│ tail                 │                              │
│ buf ────────────────────────────────┐               │
└──────────────────────┴──────────────┴───────────────┘

所以：

buf->buf = (uint8_t*)(buf + 1);

表示：

buffer_t 结构体后面的地址，就是实际数据缓冲区的起始地址。
释放时只需要：

void buffer_free(buffer_t* r)
{
    free(r);
}

一次就可以把：

结构体
+
数据区域

一起释放。

二、为什么容量使用2的幂，head和tail为什么不用直接取模

创建 Buffer 时有两个辅助函数：

static inline int is_power_of_two(uint32_t num)
{
    if (num < 2) return 0;
    return (num & (num - 1)) == 0;
}

它用来判断：

一个数是不是2的幂

例如：

2
4
8
16
32
64
1024
16384

都是2的幂。

例如：

8 = 1000
7 = 0111

执行：

1000
&
0111
----
0000

所以：

8 & (8 - 1)

结果为0。

如果传进来的容量不是2的幂：

if (!is_power_of_two(sz)) sz = roundup_power_of_two(sz);

就会向上调整。

例如：

sz = 10

最终调整成：

16

这样做的关键原因就是后面可以使用：

position & (size - 1)

代替：

position % size

实现环形下标映射。

假设：

size = 8

那么：

size - 1 = 7

逻辑位置：

0 1 2 3 4 5 6 7 8 9 10 11 ...

经过：

position & 7

以后：

0 1 2 3 4 5 6 7 0 1  2  3 ...

正好形成循环。

例如：

tail = 10

真正写入数组的位置：

tail & (size - 1)

就是：

10 & 7 = 2

所以：

逻辑位置10
    ↓
数组下标2

这也是这个 Ring Buffer 和最简单的：

tail = (tail + 1) % size;

写法不同的地方。

这里的：

head
tail

可以一直作为逻辑计数器递增。

真正访问数组时才：

head & (size - 1)
tail & (size - 1)

例如：

size = 8

head = 10
tail = 14

那么真实数组位置：

head：

10 & 7 = 2


tail：

14 & 7 = 6

但 Buffer 当前有效数据长度并不需要根据物理位置计算。

直接：

static uint32_t rb_len(buffer_t* r)
{
    return r->tail - r->head;
}

即可。

例如：

head = 10
tail = 14

那么：

len = 14 - 10
    = 4

当前有4字节有效数据。

判断空：

static uint32_t rb_isempty(buffer_t* r)
{
    return r->head == r->tail;
}

判断满：

static uint32_t rb_isfull(buffer_t* r)
{
    return r->size == (r->tail - r->head);
}

剩余空间：

static uint32_t rb_remain(buffer_t* r)
{
    return r->size - r->tail + r->head;
}

其实就是：

remain
=
size - 当前有效数据长度

也就是：

size - (tail - head)

例如：

size = 8
head = 10
tail = 14

当前：

len = 4

所以：

remain = 8 - 4 = 4

这种设计最大的好处就是：

逻辑位置和物理数组位置分开处理，长度计算非常直接，而真正访问内存时再利用 mask 映射到环形数组。

三、buffer_add和buffer_remove如何处理跨界数据

Ring Buffer 最关键的地方就是：

如果数据写到数组最后，但是还没有写完怎么办？
例如容量：

size = 8

当前写位置已经到了：

6

现在准备写入4字节：

A B C D

数组最后只剩：

6
7

两个位置。
显然不能连续写成：

6 7 8 9

因为：

8
9

已经超过数组范围。
正确方式应该是：

先写数组尾部：

6 → A
7 → B

然后回到数组开头：

0 → C
1 → D

形成：

 0   1   2   3   4   5   6   7
┌───┬───┬───┬───┬───┬───┬───┬───┐
│ C │ D │   │   │   │   │ A │ B │
└───┴───┴───┴───┴───┴───┴───┴───┘

这就是：

跨界写入

buffer_add() 正是在解决这个问题：

int buffer_add(buffer_t* r, const void* data, uint32_t sz)
{
    // 剩余空间不够，直接写入失败
    if (sz > rb_remain(r)) return -1;

    // 计算从当前位置到数组尾部最多可以连续写多少
    uint32_t i = min(sz, r->size - (r->tail & (r->size - 1)));

    // 第一段：写当前位置到数组尾部
    memcpy(r->buf + (r->tail & (r->size - 1)), data, i);

    // 第二段：剩余数据从数组开头继续写
    memcpy(r->buf, (const uint8_t*)data + i, sz - i);

    // tail只更新逻辑位置
    r->tail += sz;

    return 0;
}

最关键的是：

r->tail & (r->size - 1)

得到当前真正写入位置。

然后：

r->size - (r->tail & (r->size - 1))

得到：

当前写位置
到
数组末尾

还剩多少连续空间。

例如：

size = 8
tail物理位置 = 6
sz = 4

那么：

size - tail位置

= 8 - 6

= 2

所以：

i = min(4, 2);

得到：

i = 2

第一段：

memcpy(..., data, 2);

写：

A B

第二段：

memcpy(r->buf, data + 2, 2);

再写：

C D

于是完成一次跨界写入。

可以把 buffer_add() 总结成：

准备写入sz字节
      ↓
剩余空间够不够？
      ↓
找到tail真实位置
      ↓
计算尾部连续空间
      ↓
┌─────────────────────┐
│ 数据能一次写完吗？   │
└─────────────────────┘
       ↓        ↓
      能        不能
       ↓        ↓
一次memcpy   拆成两次memcpy
                 ↓
          尾部 + 数组开头
                 ↓
             tail += sz

读取操作完全类似：

int buffer_remove(buffer_t* r, void* data, uint32_t sz)
{
    assert(!rb_isempty(r));

    // 最多只能读取当前已有的数据
    sz = min(sz, r->tail - r->head);

    // 计算从head到数组尾部可以连续读取多少
    uint32_t i = min(sz, r->size - (r->head & (r->size - 1)));

    // 第一段：读取head到数组末尾
    memcpy(data, r->buf + (r->head & (r->size - 1)), i);

    // 第二段：如果发生环绕，再从数组开头继续读取
    memcpy((uint8_t*)data + i, r->buf, sz - i);

    // head向后移动
    r->head += sz;

    return sz;
}

假设有效数据是：

 0   1   2   3   4   5   6   7
┌───┬───┬───┬───┬───┬───┬───┬───┐
│ C │ D │   │   │   │   │ A │ B │
└───┴───┴───┴───┴───┴───┴───┴───┘
                         ↑
                        head

真正的逻辑顺序其实是：

A B C D

所以读取4字节时：

第一段：

6 → A
7 → B

第二段：

0 → C
1 → D

最终用户拿到的仍然是连续的：

A B C D

所以 Ring Buffer 虽然内部的数据可能分成：

数组尾部一段
+
数组开头一段

但是通过两次 memcpy()，上层仍然可以拿到一块正常的连续数据。

四、buffer_drain和buffer_search如何处理网络数据

除了普通：

buffer_add();
buffer_remove();

网络 Buffer 还需要两个非常实用的接口：

buffer_drain();
buffer_search();

buffer_drain() 很简单：

int buffer_drain(buffer_t* r, uint32_t sz)
{
    if (sz > rb_len(r)) sz = rb_len(r);

    r->head += sz;

    return sz;
}

它和：

buffer_remove();

最大的区别是：

buffer_remove

Buffer
  ↓
拷贝给用户
  ↓
head移动


buffer_drain

Buffer
  ↓
不拷贝
  ↓
直接head移动

例如 Buffer 中：

hello\nworld\n

如果前面的：

hello\n

已经处理完成，并且不需要复制出来，可以：

buffer_drain(buf, 6);

直接让：

head += 6

逻辑上前面6字节就已经从 Buffer 中删除了。

在网络协议处理中，更重要的是：

buffer_search();

例如使用：

\n

作为消息结束符，就可以查找：

int len = buffer_search(buf, "\n", 1);

但是 Ring Buffer 的搜索不能简单写：

strstr((char*)buf, "\n");

原因是数据可能发生环绕。

例如逻辑数据：

hello\n

实际可能存成：

数组：

 0   1   2   3   4   5   6   7
┌───┬───┬───┬───┬───┬───┬───┬───┐
│ o │ \n│   │   │   │ h │ e │ l │
└───┴───┴───┴───┴───┴───┴───┴───┘

逻辑顺序：

5 → h
6 → e
7 → l
0 → o
1 → \n

所以搜索时也必须：

逻辑位置
   ↓
转换为物理位置

核心就是：

int pos = (r->head + i) & (r->size - 1);

其中：

head + i

表示当前正在检查的逻辑位置。
再通过：

& (size - 1)

转换成数组下标。

一个更安全、容易理解的搜索实现可以写成：

int buffer_search(buffer_t* r, const char* sep, int seplen)
{
    uint32_t len = rb_len(r);

    if (!sep || seplen <= 0 || len < (uint32_t)seplen) return 0;

    for (uint32_t i = 0; i <= len - (uint32_t)seplen; ++i)
    {
        int matched = 1;

        // 一个字符一个字符比较，即使分隔符跨越数组尾部也没有问题
        for (int j = 0; j < seplen; ++j)
        {
            uint32_t pos = (r->head + i + j) & (r->size - 1);

            if (r->buf[pos] != (uint8_t)sep[j])
            {
                matched = 0;
                break;
            }
        }

        if (matched) return i + seplen;
    }

    return 0;
}

例如：

Buffer逻辑数据：

hello\nworld\n

执行：

int len = buffer_search(buf, "\n", 1);

搜索过程：

h
↓
不是\n

e
↓
不是\n

l
↓
不是\n

l
↓
不是\n

o
↓
不是\n

\n
↓
匹配成功

于是返回：

6

上层再：

char data[1024] = {0};
buffer_remove(buf, data, len);

得到：

hello\n

所以整个 TCP 半包处理就变成：

read()
  ↓
buffer_add()
  ↓
Ring Buffer积累数据
  ↓
buffer_search()
  ↓
找到完整协议分隔符？
   ↓             ↓
  没有           找到
   ↓             ↓
继续read     buffer_remove()
                 ↓
              业务处理

这也是 Ring Buffer 真正应用到网络编程中的关键一步。

五、连续数据获取与Ring Buffer整体流程总结

Ring Buffer 还有一个问题：

数据发生环绕以后，怎样获得一块连续的数据用于 write()？
例如当前逻辑数据：

A B C D

但是物理内存：

 0   1   2   3   4   5   6   7
┌───┬───┬───┬───┬───┬───┬───┬───┐
│ C │ D │   │   │   │   │ A │ B │
└───┴───┴───┴───┴───┴───┴───┴───┘

如果直接：

write(fd, r->buf + 6, 4);

就会访问：

6
7
8
9

显然不正确。

因此可以提供：

uint8_t* buffer_write_atmost(buffer_t* r);

用来获得尽可能连续的一块有效数据。

更稳妥的一种做法，是在发生环绕时先将有效数据整理到数组开头：

uint8_t* buffer_write_atmost(buffer_t* r)
{
    uint32_t len = rb_len(r);
    if (len == 0) return NULL;

    uint32_t rpos = r->head & (r->size - 1);
    uint32_t wpos = r->tail & (r->size - 1);

    // 数据没有发生环绕，直接返回head位置
    if (rpos < wpos) return r->buf + rpos;

    // 数据发生环绕，临时整理成连续数据
    uint8_t* temp = (uint8_t*)malloc(len);
    if (!temp) return NULL;

    uint32_t first = r->size - rpos;

    memcpy(temp, r->buf + rpos, first);
    memcpy(temp + first, r->buf, len - first);

    // 再复制回原Buffer开头
    memcpy(r->buf, temp, len);
    free(temp);

    // 重新调整逻辑位置
    r->head = 0;
    r->tail = len;

    return r->buf;
}

这样原来：

 0   1   2   3   4   5   6   7
┌───┬───┬───┬───┬───┬───┬───┬───┐
│ C │ D │   │   │   │   │ A │ B │
└───┴───┴───┴───┴───┴───┴───┴───┘

就可以整理成：

 0   1   2   3   4   5   6   7
┌───┬───┬───┬───┬───┬───┬───┬───┐
│ A │ B │ C │ D │   │   │   │   │
└───┴───┴───┴───┴───┴───┴───┴───┘
 ↑               ↑
head            tail

然后网络层就可以直接：

uint8_t* data = buffer_write_atmost(evbuf_out(e));
int len = buffer_len(evbuf_out(e));

write(fd, data, len);

当然，在更加追求性能的实现中，不一定非要为了连续数据进行一次整理拷贝，也可以分别处理：

数组尾部一段
+
数组开头一段

甚至使用：

writev()

进行分散写。

但作为 Ring Buffer 的基础实现，先把：

环形数据
      ↓
必要时整理成连续数据

这个思路理解清楚即可。

到这里，一个基础网络 Ring Buffer 的核心接口已经串起来：

buffer_new()
     ↓
创建固定大小Ring Buffer

buffer_add()
     ↓
向tail追加数据

buffer_len()
     ↓
获取有效数据长度

buffer_search()
     ↓
寻找完整协议消息

buffer_remove()
     ↓
从head读取并删除数据

buffer_drain()
     ↓
只删除，不复制

buffer_write_atmost()
     ↓
获得连续可写数据

buffer_free()
     ↓
释放缓冲区

其中最核心的数据结构始终只有：

struct ringbuffer_s
{
    uint32_t size;
    uint32_t tail;
    uint32_t head;
    uint8_t* buf;
};

真正需要掌握的是下面几组关系：

tail - head
=
当前有效数据长度

size - (tail - head)
=
当前剩余空间

以及：

position & (size - 1)

负责把：

不断增长的逻辑位置

转换成：

Ring Buffer中的真实数组下标

而当一次读写跨越数组末尾时：

第一段
数组当前位置 → 数组末尾

第二段
数组开头 → 剩余部分

通过两次：

memcpy();

完成数据搬运。

所以整个 Ring Buffer 可以概括成：

固定连续内存
      +
head / tail逻辑位置
      +
2的幂容量
      +
& mask环形映射
      +
跨界两段读写

相比普通动态字符串或不断扩容的数组，它最大的特点就是：

预先申请固定内存，并不断循环复用这一块空间，非常适合网络数据这种持续到达、持续消费的场景。

下一篇继续写 Chain Buffer 链式缓冲区，重点分析它为什么不需要固定容量、如何使用多个内存块连接数据，以及 buffer_node、链表扩容、跨节点读取和 Ring Buffer 相比到底有什么区别。

0voice · GitHub