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; }