怎么做到高性能网络IO？

为什么要做高性能网络IO。主要是解决c10，c10M问题

最开始的时候我们走的内核协议栈，走内核协议栈其实性能比较低，因为我们之前介绍的时候需要拷贝两次

但是我们采用用户态协议栈可以少拷贝一次，可以大大提高效率，

步骤：

1）客户端请求数据，先经过网卡，服务器需要从网卡copy数据到内核协议栈（tcp/bsd）。

2）再从内核协议栈copy数据到应用程序。

由此可见，客户端与应用程序之间的数据交互，多了两次数据拷贝的操作，在大量数据并发的情况下，必将会严重影响性能。

优化思路：可以跳过内核协议栈，去除拷贝操作，数据直接从网卡到应用程序，这种方式称为零拷贝。

但是我们这个在做完用户态协议栈在配合reactor的时候，会出现一个问题，是什么问题呢？

问题是：我们的epoll并不会通知这个事件

那么我们是怎么看待这个问题的呢，我们如果不了解epoll的原理和底层的话，我们一下也不知道为什么epoll不会通知，其实是因为epoll的通知是由内核通知的，但是我们旁路之后，不走内核协议栈，那么内核协议栈就不会通知数据了

图例：(我们是在内核里创建了红黑树和fd的一些结构体信息，然后提供系统调用给用户)

协议栈解析出有数据来，通知到epoll中。应用程序操作epoll

 所以我们在走用户态协议栈的时候，就不能用系统自带的epoll了，需要自己再用户态实现一个

 epoll，进行管理

我们怎么设计epoll呢，我们采用红黑树结构是最好的，排除掉哈希表，优先队列，链表

就是红黑树最好了，为什么，因为红黑树这个数据结构更适合增删改查，效率也高，并且有

带有二叉搜索树的性质，所以很好用

需要查找性能高的数据结构，可选的数据结构有

• hash：fd 数量不确定，创建 hash 消耗大量的内存。若 fd 数量较少时，内存浪费多，性能低
• b/b+ 树：查找性能低于红黑树，降低树高，用于磁盘 io
• rbtree：查找性能高，效率稳定，这里选用红黑树

还有一点就是epoll 监听的是系统 fd。而在自定义用户态协议栈的过程中，我们定义的 fd 只是个 int 值，并不指向内核打开文件表中对应的 i-node 结点

epoll 通过 fd 检测协议栈中的 tcb 有无事件发生，并对这些 fd 进行管理。

 

自定义epoll 的主要结构体有

• epitem：存储每个 io 对应的事件，每个注册到 epoll 池的 fd 对应1个 epitem

// 自定义的 epitem
struct epitem {RB_ENTRY(epitem) rbn;		// 红黑树的结点LIST_ENTRY(epitem) rdlink;	// 就绪队列，双向链表结点int rdy; 				   // 是否在就绪队列中int sockfd;				   // 事件对应的sockfdstruct epoll_event event;	// 注册事件的类型 
};

eventpoll：用于管理1个 epoll 对象

// 自定义的 eventpoll
struct eventpoll {int fd;		    // epfdep_rb_tree rbr;	// 红黑树的根结点int rbcnt;	    LIST_HEAD( ,epitem) rdlist;	// 就绪队列头结点int rdnum;					int waiting;	// epoll_wait判断是否正在等待pthread_mutex_t mtx; 	 //rbtree updatepthread_spinlock_t lock; //rdlist updatepthread_cond_t cond; 	//block for event，用于epoll_wait的超时等待pthread_mutex_t cdmtx; 	//mutex for cond
};

 红黑树和双向链表共用结点 epitem。 

双向链表采用的是就绪队列，在处理事件的时候，可以按先来先服务策略进行处理时间

2、epoll 锁机制

考虑两个公共资源：红黑树和就绪队列。

• 红黑树：mutex，互斥锁
• 就绪队列：spinlock，采用自旋锁，避免 SMP 体系下，多核竞争。

 我们的红黑树的删除和修改和插入都是采用互斥锁的，因为不用锁的的话会发生线程安全问题，比如我们将epoll交给多个线程管理，那么当事件就绪的时候就会有惊群效应，如果此时

不加锁的话，那么多个线程会同时去处理这个事件，那么就会出现线程安全问题

3、epoll 用户态接口

epoll 为用户态提供的接口有：epoll_create, epoll_ctl, eoll_wait

 

3.1、epoll_create 的实现

功能： 创建 eventpoll 结构体

int epoll_create(int size) {if (size <= 0) return -1;// 从位图中获取新的fd，fd从3开始依次递增	int epfd =get_fd_frombitmap();struct eventpoll *ep = (struct eventpoll*)rte_calloc("eventpoll",1, sizeof(struct eventpoll), 0);if (!ep) {// 创建失败，将fd从位图中删除set_fd_frombitmap(epfd);return -1;}// 初始化红黑树和就绪队列ep->rbcnt = 0;RB_INIT(&ep->rbr);LIST_INIT(&ep->rdlist);if (pthread_mutex_init(&ep->mtx, NULL)) {rte_free(ep);set_fd_frombitmap(epfd);return -2;}if (pthread_mutex_init(&ep->cdmtx, NULL)) {pthread_mutex_destroy(&ep->mtx);rte_free(ep);set_fd_frombitmap(epfd);return -2;}if (pthread_cond_init(&ep->cond, NULL)) {pthread_mutex_destroy(&ep->cdmtx);pthread_mutex_destroy(&ep->mtx);rte_free(ep);set_fd_frombitmap(epfd);return -2;}if (pthread_spin_init(&ep->lock, PTHREAD_PROCESS_SHARED)) {pthread_cond_destroy(&ep->cond);pthread_mutex_destroy(&ep->cdmtx);pthread_mutex_destroy(&ep->mtx);rte_free(ep);set_fd_frombitmap(epfd);return -2;}return epfd;
}

 

3.2、epoll_ctl 的实现

功能：对红黑树进行增添，修改、删除。

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) {// 通过fd查找到协议栈中对应的tcb连接，返回 eventpoll 对象，fd -> hoststruct eventpoll *ep = (struct eventpoll *)get_hostinfo_fromfd(epfd);// 若ep对象为空，或没有要设置的事件（del除外）if (!ep || (!event && op != EPOLL_CTL_DEL)) {errno = -EINVAL;return -1;}///1、ADD 操作if (op == EPOLL_CTL_ADD) {pthread_mutex_lock(&ep->mtx);struct epitem tmp;tmp.sockfd = fd;// 在红黑树查找该结点 struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);// 若红黑树已经存在该结点，返回if (epi) {pthread_mutex_unlock(&ep->mtx);return -1;}// 不存在，则创建 epitem 结点，并为其添加sockfd和事件epi = (struct epitem*)rte_calloc("epitem",1, sizeof(struct epitem), 0);if (!epi) {pthread_mutex_unlock(&ep->mtx);errno = -ENOMEM;return -1;}	epi->sockfd = fd;memcpy(&epi->event, event, sizeof(struct epoll_event));// 插入到红黑树中epi = RB_INSERT(_epoll_rb_socket, &ep->rbr, epi);assert(epi == NULL);// 红黑树结点数量增加ep->rbcnt ++;pthread_mutex_unlock(&ep->mtx);} // 2、DEL 操作else if (op == EPOLL_CTL_DEL) {pthread_mutex_lock(&ep->mtx);struct epitem tmp;tmp.sockfd = fd;struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);// 若红黑树中不存在该结点，直接返回if (!epi) {pthread_mutex_unlock(&ep->mtx);return -1;}// 存在该结点，则从红黑树中删除epi = RB_REMOVE(_epoll_rb_socket, &ep->rbr, epi);if (!epi) {pthread_mutex_unlock(&ep->mtx);return -1;}// 红黑树结点数量减少ep->rbcnt --;// 释放结点空间rte_free(epi);pthread_mutex_unlock(&ep->mtx);} // 3、MOD 操作else if (op == EPOLL_CTL_MOD) {struct epitem tmp;tmp.sockfd = fd;struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);// 该结点存在，则修改if (epi) {epi->event.events = event->events;epi->event.events |= EPOLLERR | EPOLLHUP;} // 不存在，返回-1else {errno = -ENOENT;return -1;}} // 4、非法操作else {assert(0);}return 0;
}

3.3、epoll_wait 的实现
功能：等待 fd 就绪，监控就绪队列，若有数据，从内核拷贝数据到用户空间；若没有数据，阻塞。

等待的实现方法

等待规定的时间，条件变量 + pthread_cond_timedwait
一直等待（阻塞），条件变量 + pthread_cond_wait
 

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) {// 通过fd查找到协议栈中对应的tcb连接，返回 eventpoll 对象，fd -> hoststruct eventpoll *ep = (struct eventpoll *)get_hostinfo_fromfd(epfd);// 若ep对象为空，或没有要设置的事件（del除外）if (!ep || (!event && op != EPOLL_CTL_DEL)) {errno = -EINVAL;return -1;}///1、ADD 操作if (op == EPOLL_CTL_ADD) {pthread_mutex_lock(&ep->mtx);struct epitem tmp;tmp.sockfd = fd;// 在红黑树查找该结点 struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);// 若红黑树已经存在该结点，返回if (epi) {pthread_mutex_unlock(&ep->mtx);return -1;}// 不存在，则创建 epitem 结点，并为其添加sockfd和事件epi = (struct epitem*)rte_calloc("epitem",1, sizeof(struct epitem), 0);if (!epi) {pthread_mutex_unlock(&ep->mtx);errno = -ENOMEM;return -1;}	epi->sockfd = fd;memcpy(&epi->event, event, sizeof(struct epoll_event));// 插入到红黑树中epi = RB_INSERT(_epoll_rb_socket, &ep->rbr, epi);assert(epi == NULL);// 红黑树结点数量增加ep->rbcnt ++;pthread_mutex_unlock(&ep->mtx);} // 2、DEL 操作else if (op == EPOLL_CTL_DEL) {pthread_mutex_lock(&ep->mtx);struct epitem tmp;tmp.sockfd = fd;struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);// 若红黑树中不存在该结点，直接返回if (!epi) {pthread_mutex_unlock(&ep->mtx);return -1;}// 存在该结点，则从红黑树中删除epi = RB_REMOVE(_epoll_rb_socket, &ep->rbr, epi);if (!epi) {pthread_mutex_unlock(&ep->mtx);return -1;}// 红黑树结点数量减少ep->rbcnt --;// 释放结点空间rte_free(epi);pthread_mutex_unlock(&ep->mtx);} // 3、MOD 操作else if (op == EPOLL_CTL_MOD) {struct epitem tmp;tmp.sockfd = fd;struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);// 该结点存在，则修改if (epi) {epi->event.events = event->events;epi->event.events |= EPOLLERR | EPOLLHUP;} // 不存在，返回-1else {errno = -ENOENT;return -1;}} // 4、非法操作else {assert(0);}return 0;
}

4、epoll 回调

4.1、epoll 回调函数的实现

当内核 io 准备就绪的时候，执行 epoll 回调函数，将 epitem 添加到 rdlist 中，唤醒 epoll_wait。当 epoll_wait 被激活重新运行的时候，将 rdlist 的 epitem 逐一拷贝到 events 中，同时删除 rdlist 中对应的结点。换句话说， epoll_callback 是生产者，放入结点，唤醒 epoll_wait；epoll_wait 是消费者，消费结点。

// 从协议栈回调到epoll，把fd和对应的事件拷贝到应用程序
static int nepoll_event_callback(struct eventpoll *ep, int sockid, uint32_t event) {struct epitem tmp;tmp.sockfd = sockid;// 在红黑树中查找 epitemstruct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);if (!epi) {return -1;}// 已经在就绪队列中，只添加事件if (epi->rdy) {epi->event.events |= event;return 1;} // 不在就绪队列，则将结点加入到就绪队列pthread_spin_lock(&ep->lock);epi->rdy = 1;LIST_INSERT_HEAD(&ep->rdlist, epi, rdlink);ep->rdnum ++;pthread_spin_unlock(&ep->lock);pthread_mutex_lock(&ep->cdmtx);// 就绪队列中增加结点，唤醒epoll_waitpthread_cond_signal(&ep->cond);pthread_mutex_unlock(&ep->cdmtx);return 0;}

4.2、epoll 回调的时机

触发 epoll 回调4个时机，需要在这些地方添加 epoll 回调函数，使得 epoll 可以正常接收数据。

三次握手中，在 syn-rcvd 状态，对端返回 ack 后，tcb 结点放入到全连接队列，将对应的 sockfd 的置为 EPOLLIN 状态，等待 accept 取出，触发 epoll 回调。
 

if (stream->status == TCP_SYN_RCVD) {// 进入到 ESTABLISHED 状态stream->status = TCP_STATUS_ESTABLISHED;// 设置 epoll 回调函数，等待 accept
}

在 established 状态，收到数据后，将 sockfd 置为 EPOLLIN 状态，等待读取数据，触发epoll 回调

if (tcphdr->tcp_flags & TCP_PSH_FLAG) {// 建立连接后，push 接收数据，设置 epoll 回调函数
} 

在 established 状态，收到 fin 时，进入到 close_wait 状态。将 sockfd 的 event 置为 EPOLLIN，读取断开信息，触发 epoll 回调

if (tcphdr->tcp_flags & TCP_FIN_FLAG) {// 收到 fin，进入到 CLOSE_WAIT 状态stream->status = TCP_STATUS_CLOSE_WAIT; // 设置 epoll 回调函数，读取断开信息
}

• 检测 socket 的 send 状态，如果对端 cwnd>0， 可以发送数据，将 sockfd 置为 EPOLLOUT，等待发送数据

5、epoll 事件通知机制

水平触发(LT)，有事件，则一直触发；边缘触发(ET)，只触发一次，关注的是 io 状态的变化。

实现的关键是内核 io 就绪时，epoll 回调函数的执行次数。

LT，检测 recvbuffer 有数据则调用 epoll 回调函数
ET，从协议栈中检测到recvbuffer中接收数据就调用 epoll 回调函数
 

我们后面还可以用io_uring来处理，先不介绍了