网络缓冲区chainbuffer原理介绍以及编程入门
·
chainbuffer的作用
- 辅助解决 TCP 半包、黏包问题
缓存接收到的数据，配合应用层协议解析器（如 RESP），识别并提取完整的消息。
- 缓解数据生产与消费速度不匹配的问题
当数据接收速度大于处理速度时，临时缓存未处理的数据；如果持续积压，则需要配合背压机制，防止内存无限增长。
- 减少内存复制和数据搬移
通过多个 Chunk 动态扩容，避免整体缓冲区重新分配和复制；读取数据后通过移动读指针或释放 Chunk，减少 memmove() 操作。
- 提高网络 I/O 效率
配合 readv()、writev()，可以直接向多个 Chunk 读取数据或从多个 Chunk 发送数据，减少中间缓冲区的数据复制和系统调用开销。
chainbuffer的结构
解释:
假如我们需要接收的数据是abcdefghij，每个chunk用来接收数据，相当于链表中的节点，这里我们定义一个chunk的接收数据的容量为6（capacity=6），当新的数据来临的时候，比如abcdefghij，通过我们定义的chainbuffer的tail指针，指向对应的chunk，然后将数据copy到tail指向的chunk之中，如果tail指向的chunk无法容纳所有的数据，比如图中tail指向的chunk只能存储6个字节的数据，于是就将abcdef存储到chunk1之中，然后继续新建一个chunk2，将ghij放入到chunk2之中。一共10个字节的数据全部写入到chainbuffer之中，写入长度对应图中total_len=10。
对应的结构体定义如下：
- chunk的结构体定义
/*chainbuffer each chunk node*/
typedef struct cb_chunk
{
void *data; //存储数据
size_t read_pos; //读取当前chunk中的数据的索引
size_t write_pos; //将数据写入到当前chunk的索引
size_t capacity; //当前chunk可以容纳数据的大小
struct cb_chunk *next; //指向下一个chunk的地址的指针
} cb_chunk_t;
- chainbuffer的结构体定义
/*chainbuffer entry*/
typedef struct chainbuff
{
cb_chunk_t *head; //头结点
cb_chunk_t *tail; //尾节点
size_t total_len; //一共存储的数据的长度
size_t chunk_num; //存储数据用到的chunk的数量
size_t chunk_size; //每一个chunk可以存储的最大数据量
} chainbuff_t;
相关接口实现
初始化一个chainbuffer
主要是初始化这个chainbuffer的结构体对应变量，指定一个chunk的存储大小等信息。
int cb_create(chainbuff_t *c_buff, size_t chunk_size)
{
if (!c_buff)
return -1;
c_buff->head = NULL;
c_buff->tail = NULL;
c_buff->chunk_size = chunk_size;
c_buff->chunk_num = 0;
c_buff->total_len = 0;
return 0;
}
创建一个新的chunk
在堆空间上创建一个chunk，然后将其作为chainbuffer的尾节点，并将这个chunk的地址返回，为后续追加数据提供复用接口。
cb_chunk_t *cb_chunk_new(size_t capacity)
{
if (!capacity)
return NULL;
cb_chunk_t *chunk = (cb_chunk_t *)calloc(1, sizeof(cb_chunk_t));
if (!chunk)
{
fprintf(stderr, "cb_chunk_new chunk calloc failed: %s\n", strerror(errno));
return NULL;
}
chunk->capacity = capacity;
chunk->read_pos = 0;
chunk->write_pos = 0;
chunk->next = NULL;
chunk->data = calloc(1, chunk->capacity);
if (!chunk->data)
{
fprintf(stderr, "cb_chunk_new chunk data calloc failed: %s\n", strerror(errno));
return NULL;
}
return chunk;
}
将新的数据追加到chainbuffer
假如有了新的数据过来，通过const void *data指向这个数据的地址，将这个数据的对应的长度len追加到chainbuffer进行保存。我们需要循环的操作，直到将这个data中的对应长度的数据追加到chainbuffer中，如下所示。
int cb_data_append(chainbuff_t *c_buff, const void *data, size_t len)
{
if (!c_buff || !data || !len)
return -1;
/*move poiner*/
/*这里的src后续需要不断移动，在下面中的while循环汇总负责记录data已经追加数据到chainbuffer的索引*/
const void *src = data;
/*循环结束的条件所有的数据写入都chainbuffer中，后续需要写入的数据长度为0*/
while (len > 0)
{
/*
check chainbuffer tail chunk,if need create new chunk
case1:chainbuffer is null
case:there is no free space in the chainbuffer tail chunck
*/
/*需要扩容新增的条件：1.当前chainbuffer没有chunk，尾节点chunk的容量已满*/
if (!c_buff->tail || c_buff->tail->write_pos == c_buff->tail->capacity)
{
cb_chunk_t *new_chunk = cb_chunk_new(c_buff->chunk_size);
if (!new_chunk)
{
fprintf(stderr, "create chainbuffer chunk failed\n");
return -1;
}
if (!c_buff->tail && !c_buff->head)
c_buff->head = new_chunk;
else
c_buff->tail->next = new_chunk;
c_buff->tail = new_chunk;
c_buff->chunk_num++;
}
/*copy message from recv data --> chainbuffer*/
/*通过src将指定长度的数据copy到chainbuffer的最后一个chunk之中*/
size_t available = (c_buff->tail->capacity - c_buff->tail->write_pos);
size_t write_len = (available > len) ? len : available;
memcpy((c_buff->tail->data) + (c_buff->tail->write_pos),
src, write_len);
/*update need write len & write position*/
/*更新chainbuffer和最后一个chunk的相关信息*/
c_buff->tail->write_pos += write_len;
len -= write_len;
src += write_len; /*move data pointer*/
c_buff->total_len += write_len;
}
return 0;
}
从chainbuffer中读取数据
从chainbuffer中读取数据，需要指定需要读取的数据的长度len,同时需要指定将chainbuffer中的数据copy到指定位置的地址void *out。
ssize_t cb_read(chainbuff_t *c_buff, void *out, size_t len)
{
if (!c_buff || !out || !len)
return -1;
void *dst = out;
ssize_t total_len = 0;
/*继续循环读取数据的条件是:数据还没有读取完并且chunk->head不为空*/
while (total_len < len && c_buff->head != NULL)
{
cb_chunk_t *chunk = c_buff->head;
size_t reliable = chunk->write_pos - chunk->read_pos;
size_t read_len = len - total_len;
if (read_len > reliable)
read_len = reliable;
memcpy(dst + total_len, chunk->data + chunk->read_pos, read_len);
total_len += read_len;
chunk->read_pos += read_len;
c_buff->total_len -= read_len;
/*free read buffer*/
/*释放已经读取完数据的chunk*/
if (chunk->read_pos == chunk->write_pos)
{
c_buff->head = chunk->next;
if (!c_buff->head)
c_buff->tail = NULL;
free(chunk->data);
free(chunk);
}
}
return total_len;
}
完整代码以及测试用例
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include "chainbuff.h"
#define TESTSTR "Following my completion of undergraduate studies, I am driven by a strong passion to pursue further academic accomplishments in the field of finance."
#define TESTSTR_LEN (sizeof(TESTSTR) - 1)
int cb_create(chainbuff_t *c_buff, size_t chunk_size)
{
if (!c_buff)
return -1;
c_buff->head = NULL;
c_buff->tail = NULL;
c_buff->chunk_size = chunk_size;
c_buff->chunk_num = 0;
c_buff->total_len = 0;
return 0;
}
cb_chunk_t *cb_chunk_new(size_t capacity)
{
if (!capacity)
return NULL;
cb_chunk_t *chunk = (cb_chunk_t *)calloc(1, sizeof(cb_chunk_t));
if (!chunk)
{
fprintf(stderr, "cb_chunk_new chunk calloc failed: %s\n", strerror(errno));
return NULL;
}
chunk->capacity = capacity;
chunk->read_pos = 0;
chunk->write_pos = 0;
chunk->next = NULL;
chunk->data = calloc(1, chunk->capacity);
if (!chunk->data)
{
fprintf(stderr, "cb_chunk_new chunk data calloc failed: %s\n", strerror(errno));
return NULL;
}
return chunk;
}
int cb_data_append(chainbuff_t *c_buff, const void *data, size_t len)
{
if (!c_buff || !data || !len)
return -1;
/*move poiner*/
const void *src = data;
while (len > 0)
{
/*
check chainbuffer tail chunk,if need create new chunk
case1:chainbuffer is null
case:there is no free space in the chainbuffer tail chunck
*/
if (!c_buff->tail || c_buff->tail->write_pos == c_buff->tail->capacity)
{
cb_chunk_t *new_chunk = cb_chunk_new(c_buff->chunk_size);
if (!new_chunk)
{
fprintf(stderr, "create chainbuffer chunk failed\n");
return -1;
}
if (!c_buff->tail && !c_buff->head)
c_buff->head = new_chunk;
else
c_buff->tail->next = new_chunk;
c_buff->tail = new_chunk;
c_buff->chunk_num++;
}
/*copy message from recv data --> chainbuffer*/
size_t available = (c_buff->tail->capacity - c_buff->tail->write_pos);
size_t write_len = (available > len) ? len : available;
memcpy((c_buff->tail->data) + (c_buff->tail->write_pos),
src, write_len);
/*update need write len & write position*/
c_buff->tail->write_pos += write_len;
len -= write_len;
src += write_len; /*move data pointer*/
c_buff->total_len += write_len;
}
return 0;
}
ssize_t cb_read(chainbuff_t *c_buff, void *out, size_t len)
{
if (!c_buff || !out || !len)
return -1;
void *dst = out;
ssize_t total_len = 0;
while (total_len < len && c_buff->head != NULL)
{
cb_chunk_t *chunk = c_buff->head;
size_t reliable = chunk->write_pos - chunk->read_pos;
size_t read_len = len - total_len;
if (read_len > reliable)
read_len = reliable;
memcpy(dst + total_len, chunk->data + chunk->read_pos, read_len);
total_len += read_len;
chunk->read_pos += read_len;
c_buff->total_len -= read_len;
/*free read buffer*/
if (chunk->read_pos == chunk->write_pos)
{
c_buff->head = chunk->next;
if (!c_buff->head)
c_buff->tail = NULL;
free(chunk->data);
free(chunk);
}
}
return total_len;
}
int main(int argc, char *argv[])
{
chainbuff_t chain_buff = {0};
if (cb_create(&chain_buff, CHUNK_SIZE) != 0)
{
fprintf(stderr, "cb_create failed\n");
return -1;
}
cb_data_append(&chain_buff, TESTSTR, TESTSTR_LEN);
char read_buffer[4096] = {0};
cb_read(&chain_buff, &read_buffer, TESTSTR_LEN);
read_buffer[TESTSTR_LEN] = '\0';
printf("chunk num: %zu\n", chain_buff.chunk_num);
printf("read buffer: %s\n", read_buffer);
return 0;
}
后续优化
通过引入read_v和write_v减少数据的copy次数。
复现地址
GitHub - joey-systems/perf: linux c/ C ++ reproducible components · GitHub
其中有个chainbuffer的文件，里面的源文件可以直接编译。
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
1
0- 0
已为社区贡献1条内容
所有评论(0)