网络缓冲区-阿里云开发者社区

windows下的体系，我不是特别了解。以下所有的内容都是在Linux下的理解，如果不对的地方，评论区欢迎留言。

Linux收发数据

接收数据

大体的流程如上图所示，接下来我们对图中一些名词进行解释。

-------------------------------------------------------------------------------------------------------------------------

DMA:

一种计算机系统中的技术，用于在外设和内存之间直接传输数据，而不需要CPU的干预。DMA技术通过在外设和内存之间建立直接的数据传输通路，绕过CPU直接进行数据传输。DMA控制器负责管理数据传输的过程，而CPU可以继续执行其他任务，提高了系统的并发性和效率。

扩展1：

在传统的计算机系统中，数据的传输通常需要通过CPU来完成。

扩展2：

DMA技术可以应用于多种外设。例如：硬盘、音频设备等，当外设需要进行数据传输时，DMA控制器会向CPU发送请求，CPU将数据传输的任务交给DMA控制器处理。DMA控制器会直接读取或写入数据而不需要CPU干预。数据传输完成后，DMA控制器会向CPU发送中断信号，通知操作完成。

-------------------------------------------------------------------------------------------------------------------------

硬中断:

向CPU发起硬中断的作用是通知CPU发生了一个重要的事件或需要CPU的处理。硬中断时由硬件设备触发的中断信号，用于引起CPU的注意并中断当前正在执行的程序。

当硬件设备需要CPU的处理时，它会向CPU发送一个中断请求。CPU会暂停当前正在执行的程序，并跳转至一个预定义的中断处理程序来处理中断。中断处理程序是操作系统或设备驱动程序提供的，用于对中断事件进行处理。

-------------------------------------------------------------------------------------------------------------------------

屏蔽硬中断:

处理器屏蔽硬中断是为了确保关键代码的执行不会被中断打断。

-------------------------------------------------------------------------------------------------------------------------

软中断：

CPU发起软中断的作用是触发操作系统中的中断处理程序，从而实现与操作系统进行交互和执行特定的系统功能。

-------------------------------------------------------------------------------------------------------------------------

总述:

网卡是计算机里的一个硬件，专门负责接收和发送网络包，当网卡接收到一个网络包后，会通过DNA技术将网络包写入到指定的内存地址，也就是写入到RingBuffer中，这是一个环形缓冲区。接着网卡向CPU发起硬中断，当CPU收到硬件中断请求后，根据中断注册表，调用已经注册的中断处理函数(中断处理函数:暂时屏蔽硬中断、发起软中断-->通知内核里的ksoftirqd线程进行轮询从ringbuffer中读数据、恢复中断)。ksoftirqd线程从ringbuffer中获取一个数据帧之后保存在sk_buffer中，并且交给网络协议栈进行数据帧处理，经过协议栈处理完后的数据被放到socket读缓冲区中，等待应用程序通过系统调用读取数据。

发送数据

这个过程没有好说的，跟接收数据的返过程。需要注意的是TCP保证可靠交付，需要超时重传，需要备份。UDP是不需要的，不需要备份。

用户层角度对数据包的处理流程

用户态网络缓冲区

为什么需要用户态网络缓冲区?(为什么需要为每条连接准备一个发送缓冲区和一个接收缓冲区?)

从业务层面来说，我们可以看到read和处理数据包，组成数据包和write写入之间存在处理速度上的差异，也就是说当需要产生数据的能力比处理数据的能力强时，防止数据丢失，需要将数据未处理的数据进行短暂的存储。(生产者的速度大于消费者的速度是)

从posix api接口(read、write)层面(粘包)，不能确定一次性接收数据，也不能保证一次性发送数据。

由于存在这样的问题，我们需要设计一个缓冲区解决这样的问题...

不同的网络模型会不会影响用户态网络缓冲区的设计？

显然不会，上面两个矛盾发生在读取数据之后和处理数据之间，而不同的网络模型是对IO的不同处理方式。两个过程不是在同一阶段。换句话说就是一个是作用于读数据的方式不同，一个是作用于处理数据和读取快慢的问题上，而不是读取方式的问题上。

UDP和TCP协议是否影响用户态网络缓冲区的设计？

显然不会，很明显可以看出，UDP和TCP已经完成了将数据包写入socket缓冲区，而上面两个矛盾发生在处理数据和读写数据之间。两个不在通过阶段。换句话说就是UDP和TCP协议都存在上面的两个矛盾。

用户态网络缓冲区的设计

这里的代码不要求完全自己实现，能看懂就行，知道其原理和缺点和优点就可以了。

定长buffer:

就是一个固定长度的数组，然后增加一个记录当前空闲的起始位置。

优点:

结构简单，易于实现。

缺点：

需要频繁的腾挪数据。

需要实现扩缩容机制。

ringBuffer:

就是一个环形队列，我们可以做到更好一点。就是采用2的能次幂大小的队列，这样可以把取余操作转成位操作，提高处理速度。

具体的情况这里就不做分析了，为了方便大家理解代码，我把各种会出现的情况图罗列出来，不明白的话，评论区见。

优点:

不需要频繁的腾挪数据。

缺点：

需要实现扩缩容机制，扩缩容时需要腾挪数据。

造成不连续的空间，可能引发多次系统调用。

chainBuffer：

就是通过队列的将一块块空间连在一起，说白了就是把数组的连续空间变成了链表的方式，然后记录第一个块位置和最后一个块的位置，错觉上看起来像个连续的空间。

优点:

不腾挪数据

动态扩缩容，不腾挪数据

缺点：

造成不连续的空间，可能引发多次系统调用

RingBuffer代码

#ifndef _ringbuffer_h
#define _ringbuffer_h
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
// #include <limits.h>  // for uint_max
#include <stdint.h>
#include <unistd.h>
typedef struct ringbuffer_s buffer_t;
buffer_t * buffer_new(uint32_t sz);
uint32_t buffer_len(buffer_t *r);
void buffer_free(buffer_t *r);
int buffer_add(buffer_t *r, const void *data, uint32_t sz);
int buffer_remove(buffer_t *r, void *data, uint32_t sz);
int buffer_drain(buffer_t *r, uint32_t sz);
int buffer_search(buffer_t *r, const char* sep, const int seplen);
uint8_t * buffer_write_atmost(buffer_t *r);
#endif

#include <stdbool.h>
#include <assert.h>
#include <stdint.h>
#include <stdatomic.h>
#include "buffer.h"
struct ringbuffer_s {
    uint32_t size;
    uint32_t tail;
    uint32_t head;
    uint8_t * buf;
};
#define min(lth, rth) ((lth)<(rth)?(lth):(rth))
static inline int is_power_of_two(uint32_t num) {
    if (num < 2) return 0;
    return (num & (num - 1)) == 0;
}
static inline uint32_t roundup_power_of_two(uint32_t num) {
    if (num == 0) return 2;
    int i = 0;
    for (; num != 0; i++)
        num >>= 1;
    return 1U << i;
}
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 = buf->tail = 0;
    buf->buf = (uint8_t *)(buf + 1);
    return buf;
}
void buffer_free(buffer_t *r) {
    free(r);
    r = NULL;
}
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_len(buffer_t *r) {
    return r->tail - r->head;
}
static uint32_t
rb_remain(buffer_t *r) {
    return r->size - r->tail + r->head;
}
int buffer_add(buffer_t *r, const void *data, uint32_t sz) {
    if (sz > rb_remain(r)) {
        return -1;
    }
    uint32_t i;
    i = min(sz, r->size - (r->tail & (r->size - 1)));
    memcpy(r->buf + (r->tail & (r->size - 1)), data, i);
    memcpy(r->buf, data+i, sz-i);
    r->tail += sz;
    return 0;
}
int buffer_remove(buffer_t *r, void *data, uint32_t sz) {
    assert(!rb_isempty(r));
    uint32_t i;
    sz = min(sz, r->tail - r->head);
    i = min(sz, r->size - (r->head & (r->size - 1)));
    memcpy(data, r->buf+(r->head & (r->size - 1)), i);
    memcpy(data+i, r->buf, sz-i);
    r->head += sz;
    return sz;
}
int buffer_drain(buffer_t *r, uint32_t sz) {
    if (sz > rb_len(r))
        sz = rb_len(r);
    r->head += sz;
    return sz;
}
// 找 buffer 中 是否包含特殊字符串（界定数据包的）
int buffer_search(buffer_t *r, const char* sep, const int seplen) {
    int i;
    for (i = 0; i <= rb_len(r)-seplen; i++) {
        int pos = (r->head + i) & (r->size - 1);
        if (pos + seplen > r->size) {
            if (memcmp(r->buf+pos, sep, r->size-pos))
                return 0;
            if (memcmp(r->buf, sep+r->size-pos, pos+seplen-r->size) == 0) {
                return i+seplen;
            }
        }
        if (memcmp(r->buf+pos, sep, seplen) == 0) {
            return i+seplen;
        }
    }
    return 0;
}
uint32_t buffer_len(buffer_t *r) {
    return rb_len(r);
}
uint8_t * buffer_write_atmost(buffer_t *r) {
    uint32_t rpos = r->head & (r->size - 1);
    uint32_t wpos = r->tail & (r->size - 1);
    if (wpos < rpos) {
        uint8_t* temp = (uint8_t *)malloc(r->size * sizeof(uint8_t));
        memcpy(temp, r->buf+rpos, r->size - rpos);
        memcpy(temp+r->size-rpos, r->buf, wpos);
        free(r->buf);
        r->buf = temp;
        return r->buf;
    }
    return r->buf + rpos;
}

chainBuffer代码

#include <string.h>
#include <stdbool.h>
#include <string.h>
#include <stdlib.h>
#include "buffer.h"
struct buf_chain_s {
    struct buf_chain_s *next;
    uint32_t buffer_len;
    uint32_t misalign;
    uint32_t off;
    uint8_t *buffer;
};
struct buffer_s {
    buf_chain_t *first;
    buf_chain_t *last;
    buf_chain_t **last_with_datap;
    uint32_t total_len;
    uint32_t last_read_pos; // for sep read
};
#define CHAIN_SPACE_LEN(ch) ((ch)->buffer_len - ((ch)->misalign + (ch)->off))
#define MIN_BUFFER_SIZE 1024
#define MAX_TO_COPY_IN_EXPAND 4096
#define BUFFER_CHAIN_MAX_AUTO_SIZE 4096
#define MAX_TO_REALIGN_IN_EXPAND 2048
#define BUFFER_CHAIN_MAX 16*1024*1024  // 16M
#define BUFFER_CHAIN_EXTRA(t, c) (t *)((buf_chain_t *)(c) + 1)
#define BUFFER_CHAIN_SIZE sizeof(buf_chain_t)
uint32_t
buffer_len(buffer_t *buf) {
    return buf->total_len;
}
buffer_t *
buffer_new(uint32_t sz) {
    (void)sz;
    buffer_t * buf = (buffer_t *) malloc(sizeof(buffer_t));
    if (!buf) {
        return NULL;
    }
    memset(buf, 0, sizeof(*buf));
    buf->last_with_datap = &buf->first;
    return buf;
}
static buf_chain_t *
buf_chain_new(uint32_t size) {
    buf_chain_t *chain;
    uint32_t to_alloc;
    if (size > BUFFER_CHAIN_MAX - BUFFER_CHAIN_SIZE)
        return (NULL);
    size += BUFFER_CHAIN_SIZE;
    if (size < BUFFER_CHAIN_MAX / 2) {
        to_alloc = MIN_BUFFER_SIZE;
        while (to_alloc < size) {
            to_alloc <<= 1;
        }
    } else {
        to_alloc = size;
    }
    if ((chain = malloc(to_alloc)) == NULL)
        return (NULL);
    memset(chain, 0, BUFFER_CHAIN_SIZE);
    chain->buffer_len = to_alloc - BUFFER_CHAIN_SIZE;
    chain->buffer = BUFFER_CHAIN_EXTRA(uint8_t, chain);
    return (chain);
}
static void 
buf_chain_free_all(buf_chain_t *chain) {
    buf_chain_t *next;
    for (; chain; chain = next) {
        next = chain->next;
        free(chain);
    }
}
void
buffer_free(buffer_t *buf) {
    buf_chain_free_all(buf->first);
}
static buf_chain_t **
free_empty_chains(buffer_t *buf) {
    buf_chain_t **ch = buf->last_with_datap;
    while ((*ch) && (*ch)->off != 0)
        ch = &(*ch)->next;
    if (*ch) {
        buf_chain_free_all(*ch);
        *ch = NULL;
    }
    return ch;
}
static void
buf_chain_insert(buffer_t *buf, buf_chain_t *chain) {
    if (*buf->last_with_datap == NULL) {
        buf->first = buf->last = chain;
    } else {
        buf_chain_t **chp;
        chp = free_empty_chains(buf);
        *chp = chain;
        if (chain->off)
            buf->last_with_datap = chp;
        buf->last = chain;
    }
    buf->total_len += chain->off;
}
static inline buf_chain_t *
buf_chain_insert_new(buffer_t *buf, uint32_t datlen) {
    buf_chain_t *chain;
    if ((chain = buf_chain_new(datlen)) == NULL)
        return NULL;
    buf_chain_insert(buf, chain);
    return chain;
}
static int
buf_chain_should_realign(buf_chain_t *chain, uint32_t datlen)
{
    return chain->buffer_len - chain->off >= datlen &&
        (chain->off < chain->buffer_len / 2) &&
        (chain->off <= MAX_TO_REALIGN_IN_EXPAND);
}
static void
buf_chain_align(buf_chain_t *chain) {
    memmove(chain->buffer, chain->buffer + chain->misalign, chain->off);
    chain->misalign = 0;
}
int buffer_add(buffer_t *buf, const void *data_in, uint32_t datlen) {
    buf_chain_t *chain, *tmp;
    const uint8_t *data = data_in;
    uint32_t remain, to_alloc;
    int result = -1;
    if (datlen > BUFFER_CHAIN_MAX - buf->total_len) {
        goto done;
    }
    if (*buf->last_with_datap == NULL) {
        chain = buf->last;
    } else {
        chain = *buf->last_with_datap;
    }
    if (chain == NULL) {
        chain = buf_chain_insert_new(buf, datlen);
        if (!chain)
            goto done;
    }
    remain = chain->buffer_len - chain->misalign - chain->off;
    if (remain >= datlen) {
        memcpy(chain->buffer + chain->misalign + chain->off, data, datlen);
        chain->off += datlen;
        buf->total_len += datlen;
        // buf->n_add_for_cb += datlen;
        goto out;
    } else if (buf_chain_should_realign(chain, datlen)) {
        buf_chain_align(chain);
        memcpy(chain->buffer + chain->off, data, datlen);
        chain->off += datlen;
        buf->total_len += datlen;
        // buf->n_add_for_cb += datlen;
        goto out;
    }
    to_alloc = chain->buffer_len;
    if (to_alloc <= BUFFER_CHAIN_MAX_AUTO_SIZE/2)
        to_alloc <<= 1;
    if (datlen > to_alloc)
        to_alloc = datlen;
    tmp = buf_chain_new(to_alloc);
    if (tmp == NULL)
        goto done;
    if (remain) {
        memcpy(chain->buffer + chain->misalign + chain->off, data, remain);
        chain->off += remain;
        buf->total_len += remain;
        // buf->n_add_for_cb += remain;
    }
    data += remain;
    datlen -= remain;
    memcpy(tmp->buffer, data, datlen);
    tmp->off = datlen;
    buf_chain_insert(buf, tmp);
    // buf->n_add_for_cb += datlen;
out:
    result = 0;
done:
    return result;
}
static uint32_t
buf_copyout(buffer_t *buf, void *data_out, uint32_t datlen) {
    buf_chain_t *chain;
    char *data = data_out;
    uint32_t nread;
    chain = buf->first;
    if (datlen > buf->total_len)
        datlen = buf->total_len;
    if (datlen == 0)
        return 0;
    nread = datlen;
    while (datlen && datlen >= chain->off) {
        uint32_t copylen = chain->off;
        memcpy(data,
            chain->buffer + chain->misalign,
            copylen);
        data += copylen;
        datlen -= copylen;
        chain = chain->next;
    }
    if (datlen) {
        memcpy(data, chain->buffer + chain->misalign, datlen);
    }
    return nread;
}
static inline void
ZERO_CHAIN(buffer_t *dst) {
    dst->first = NULL;
    dst->last = NULL;
    dst->last_with_datap = &(dst)->first;
    dst->total_len = 0;
}
int buffer_drain(buffer_t *buf, uint32_t len) {
    buf_chain_t *chain, *next;
    uint32_t remaining, old_len;
    old_len = buf->total_len;
    if (old_len == 0)
        return 0;
    if (len >= old_len) {
        len = old_len;
        for (chain = buf->first; chain != NULL; chain = next) {
            next = chain->next;
            free(chain);
        }
        ZERO_CHAIN(buf);
    } else {
        buf->total_len -= len;
        remaining = len;
        for (chain = buf->first; remaining >= chain->off; chain = next) {
            next = chain->next;
            remaining -= chain->off;
            if (chain == *buf->last_with_datap) {
                buf->last_with_datap = &buf->first;
            }
            if (&chain->next == buf->last_with_datap)
                buf->last_with_datap = &buf->first;
            free(chain);
        }
        buf->first = chain;
        chain->misalign += remaining;
        chain->off -= remaining;
    }
    // buf->n_del_for_cb += len;
    return len;
}
int buffer_remove(buffer_t *buf, void *data_out, uint32_t datlen) {
    uint32_t n = buf_copyout(buf, data_out, datlen);
    if (n > 0) {
        if (buffer_drain(buf, n) < 0)
            n = -1;
    }
    return (int)n;
}
static bool
check_sep(buf_chain_t * chain, int from, const char *sep, int seplen) {
    for (;;) {
        int sz = chain->off - from;
        if (sz >= seplen) {
            return memcmp(chain->buffer + chain->misalign + from, sep, seplen) == 0;
        }
        if (sz > 0) {
            if (memcmp(chain->buffer + chain->misalign + from, sep, sz)) {
                return false;
            }
        }
        chain = chain->next;
        sep += sz;
        seplen -= sz;
        from = 0;
    }
}
int buffer_search(buffer_t *buf, const char* sep, const int seplen) {
    buf_chain_t *chain;
    int i;
    chain = buf->first;
    if (chain == NULL)
        return 0;
    int bytes = chain->off;
    while (bytes <= buf->last_read_pos) {
        chain = chain->next;
        if (chain == NULL)
            return 0;
        bytes += chain->off;
    }
    bytes -= buf->last_read_pos;
    int from = chain->off - bytes;
    for (i = buf->last_read_pos; i <= buf->total_len - seplen; i++) {
        if (check_sep(chain, from, sep, seplen)) {
            buf->last_read_pos = 0;
            return i+seplen;
        }
        ++from;
        --bytes;
        if (bytes == 0) {
            chain = chain->next;
            from = 0;
            if (chain == NULL)
                break;
            bytes = chain->off;
        }
    }
    buf->last_read_pos = i;
    return 0;
}
uint8_t * buffer_write_atmost(buffer_t *p) {
    buf_chain_t *chain, *next, *tmp, *last_with_data;
    uint8_t *buffer;
    uint32_t remaining;
    int removed_last_with_data = 0;
    int removed_last_with_datap = 0;
    chain = p->first;
    uint32_t size = p->total_len;
    if (chain->off >= size) {
        return chain->buffer + chain->misalign;
    }
    remaining = size - chain->off;
    for (tmp=chain->next; tmp; tmp=tmp->next) {
        if (tmp->off >= (size_t)remaining)
            break;
        remaining -= tmp->off;
    }
    if (chain->buffer_len - chain->misalign >= (size_t)size) {
        /* already have enough space in the first chain */
        size_t old_off = chain->off;
        buffer = chain->buffer + chain->misalign + chain->off;
        tmp = chain;
        tmp->off = size;
        size -= old_off;
        chain = chain->next;
    } else {
        if ((tmp = buf_chain_new(size)) == NULL) {
            return NULL;
        }
        buffer = tmp->buffer;
        tmp->off = size;
        p->first = tmp;
    }
    last_with_data = *p->last_with_datap;
    for (; chain != NULL && (size_t)size >= chain->off; chain = next) {
        next = chain->next;
        if (chain->buffer) {
            memcpy(buffer, chain->buffer + chain->misalign, chain->off);
            size -= chain->off;
            buffer += chain->off;
        }
        if (chain == last_with_data)
            removed_last_with_data = 1;
        if (&chain->next == p->last_with_datap)
            removed_last_with_datap = 1;
        free(chain);
    }
    if (chain != NULL) {
        memcpy(buffer, chain->buffer + chain->misalign, size);
        chain->misalign += size;
        chain->off -= size;
    } else {
        p->last = tmp;
    }
    tmp->next = chain;
    if (removed_last_with_data) {
        p->last_with_datap = &p->first;
    } else if (removed_last_with_datap) {
        if (p->first->next && p->first->next->off)
            p->last_with_datap = &p->first->next;
        else
            p->last_with_datap = &p->first;
    }
    return tmp->buffer + tmp->misalign;
}

网络缓冲区

Linux收发数据

接收数据

发送数据

用户层角度对数据包的处理流程

用户态网络缓冲区

为什么需要用户态网络缓冲区?(为什么需要为每条连接准备一个发送缓冲区和一个接收缓冲区?)

不同的网络模型会不会影响用户态网络缓冲区的设计？

UDP和TCP协议是否影响用户态网络缓冲区的设计？

用户态网络缓冲区的设计

RingBuffer代码

chainBuffer代码

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

网络缓冲区

Linux收发数据

接收数据

发送数据

用户层角度对数据包的处理流程

用户态网络缓冲区

为什么需要用户态网络缓冲区?(为什么需要为每条连接准备一个发送缓冲区和一个接收缓冲区?)

不同的网络模型会不会影响用户态网络缓冲区的设计？

UDP和TCP协议是否影响用户态网络缓冲区的设计？

用户态网络缓冲区的设计

RingBuffer代码

chainBuffer代码

热门文章

最新文章

相关课程

相关电子书

相关实验场景