"前言打算输出一系列Netty源码分析与实践的文章，也作为后端开发学习过程中的沉淀，此文章为第一篇，从操作系统底层的IO讲起，为Netty 的出场做下知识准备。一些概念文件描述符文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向为每一个进程所维护的该进程打开文件的记录表。当程序打开 ...."

Fork me on GitHub

2037_nickname
网名真好吃咩
Netty select Epoll 高并发 • 0 回帖 • 3.3K 浏览 • 5 年前

深入剖析 Netty 源码设计（一）——深入理解 select poll epoll 机制

前言

打算输出一系列Netty源码分析与实践的文章，也作为后端开发学习过程中的沉淀，此文章为第一篇，从操作系统底层的IO讲起，为Netty 的出场做下知识准备。

一些概念

文件描述符

文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。读写文件也需要使用文件描述符来指定待读写的文件。

I/O模型分类

根据UNIX网络编程对I/O模型的分类，UNIX提供了5种IO模型：

阻塞I/O：

在进程空间中调用recvfrom，其系统调用直到数据包到达且被复制到应用进程的缓冲区中或者发生错误时才返回，在此期间会一直被阻塞

非阻塞I/O

recvfrom从应用层到内核的时候，如果该缓冲区没有数据的话，就直接返回一个EWOULDBLOCK错误，一般都对非阻塞I/O模型进行轮询检查这个状态，看内核是不是有数据到来。

I/O复用模型

将一个或者多个fd通过select/poll调用，然后阻塞在select上，一个进程用来监控多个fd，即所谓的I/O复用，细节后面会描述。

信号驱动I/O模型

首先开启套接字信号驱动I/O功能，并通过系统调用sigaction执行一个信号处理函数并立即返回（非阻塞）。当数据就绪时，就为该进程生成一个SIGIO信号，通过信号回调通知应用程序调用recvfrom来读取数据。

异步I/O

告知内核启动某个操作，内核会将数据从内核态自己copy到用户态缓冲区，然后通知我们。其和信号驱动模型的主要区别是：信号驱动I/O由内核通知我们何时可以开始一个I/O操作；异步I/O模型由内核通知我们I/O操作何时已经完成。

Linux 2.6+内核的wakeup callback机制

Linux通过socket的睡眠队列（sleep_list）来管理所有等待socket的某个事件的进程（task）， select、poll、epoll_wait 函数操作会陷入内核，判断监控的socket是否有关心的事件发生了，如果没，则为当前task构建一个wait_entry节点，然后插入到每个socket的sleep_list里，直到超时或事件发生，同时通过wakeup机制来异步唤醒整个睡眠队列上等待事件的task，通知task相关事件发生,每一个sleep_list上的wait_entry都拥有一个callback，wakeup逻辑在唤醒睡眠队列时，会遍历该队列链表上的每一个wait_entry，直到完成队列的遍历或遇到某个wait_entry节点是排他的才停止，调用每一个wait_entry的callback，并将当前task的wait_entry节点从socket的sleep_list中删除。

select

select 是一种同步IO，函数签名如下：

int  select(int  nfds,  fd_set  *readfds,  fd_set  *writefds,  fd_set  *exceptfds,  struct  timeval  *timeout);

nfds为最大的文件描述符值+1
readfds 某些文件描述符所指向的socket已经有数据可读或者数据EOF
writefds 某些文件描述符所指向的socket是否可写数据了
exceptfds 某些文件描述符所指向的socket出现异常

使用示例：

#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <wait.h>
#include <signal.h>
#include <errno.h>
#include <sys/select.h>
#include <sys/time.h>
#include <unistd.h>

#define MAXBUF 256

void child_process(void)
{
  sleep(2);
  char msg[MAXBUF];
  struct sockaddr_in addr = {0};
  int n, sockfd,num=1;
  srandom(getpid());
  /* Create socket and connect to server */
  sockfd = socket(AF_INET, SOCK_STREAM, 0);
  addr.sin_family = AF_INET;
  addr.sin_port = htons(2000);
  addr.sin_addr.s_addr = inet_addr("127.0.0.1");

  connect(sockfd, (struct sockaddr*)&addr, sizeof(addr));

  printf("child {%d} connected \n", getpid());
  while(1){
        int sl = (random() % 10 ) +  1;
        num++;
     	sleep(sl);
  	sprintf (msg, "Test message %d from client %d", num, getpid());
  	n = write(sockfd, msg, strlen(msg));	/* Send message */
  }

}

int main()
{
  char buffer[MAXBUF];
  int fds[5];
  struct sockaddr_in addr;
  struct sockaddr_in client;
  int addrlen, n,i,max=0;;
  int sockfd, commfd;
  fd_set rset;
  
  //创建了5个子进程， 每个进程都向server发送了数据

  for(i=0;i<5;i++)
  {
  	if(fork() == 0)
  	{
  		child_process();
  		exit(0);
  	}
  }

  sockfd = socket(AF_INET, SOCK_STREAM, 0);
  memset(&addr, 0, sizeof (addr));
  addr.sin_family = AF_INET;
  addr.sin_port = htons(2000);
  addr.sin_addr.s_addr = INADDR_ANY;
  bind(sockfd,(struct sockaddr*)&addr ,sizeof(addr));
  listen (sockfd, 5); ///告诉内核服务端的一些信息 连接队列个数为5，大于5个socket连接，会出现延时
  
  for (i=0;i<5;i++) 
  {
    memset(&client, 0, sizeof (client));
    addrlen = sizeof(client);
    fds[i] = accept(sockfd,(struct sockaddr*)&client, &addrlen);
	
	//保留最大的 文件描述符值
    if(fds[i] > max)
    	max = fds[i];
  }
  
  while(1){ 
  //将文件描述符数组每一位全都置为0
	FD_ZERO(&rset);
	//每次while循环都要重新设置要监控的socket
  	for (i = 0; i< 5; i++ ) {
  		FD_SET(fds[i],&rset);
  	}

   	puts("round again");
	//一直阻塞直到有读事件已ready
	select(max+1, &rset, NULL, NULL, NULL);

	for(i=0;i<5;i++) {
	//循环判断是哪个socket可读
		if (FD_ISSET(fds[i], &rset)){
			memset(buffer,0,MAXBUF);
			read(fds[i], buffer, MAXBUF);
			puts(buffer);
		}
	}	
  }
  return 0;
}

为了要高效的处理网络IO数据，不可能为每个socket 创建一个进程task，进程创建是一种高昂的性能损耗，所以采用一个task来监控多个socket，但这一个task不可能去阻塞式的监控某一个socket的事件发生，我们应该block在关心的N个socket中一个或多个socket有数据可读的事件，意味着当block解除的时候，我们一定可以找到一个或多个socket上有可读的数据（至少一个可读），select将这个task放到每个 socket的sleep_list，等待任意一个socket可读事件发生而被唤醒，当task被唤醒的时候，其callback里面应该有个逻辑去检查具体哪些socket可读了。然后把这些事件反馈给用户程序，select为每个socket引入一个poll逻辑，该逻辑用于收集socket发生的事件。对于可读事件来说，简单伪码如下：

private int sk_event;

void poll() {
    //其他逻辑...
    when (receive queue is not empty) {
        sk_event |= POLL_IN；
    }
   //其他逻辑...
}

当receive queue不为空的时候，我们就给这个socket的sk_event添加一个POLL_IN事件，用来表示当前这个socket可读。将来task遍历到这个socket，发现其sk_event包含POLL_IN的时候，就说明这个socket已是可读的。当用户task调用select的时候，select会将需要监控的readfds集合拷贝到内核空间，然后遍历自己监控的socket，挨个调用socket的poll逻辑以便检查该socket是否有可读事件。

遍历完所有的socket后，如果没有任何一个sk可读，那么select会调用schedule，使得task进入睡眠。如果在timeout时间内某个socket上有数据可读了，或者等待timeout了，则调用select的task会被唤醒。唤醒后select就是遍历监控的socket集合，挨个收集可读事件并返回给用户了，相应的伪码如下：

for (socket in readfds) {
    sk_event.evt = socket.poll();
    sk_event.sk = socket;
    return_event_for_process;
}

就像示例代码一样while循环内的for循环，在select返回后，task需要遍历已ready的描述符集合，循环的次数就是之前记录的fd值。

select的问题：

每次select都需要将需要监控的文件描述符集合从用户态copy到内核态，内核并将ready的描述符集合再从内核态copy到用户态，如果socket很大，会有很大的上下文切换的损耗。
由于readfds是长度为32的整型数组，32*32=1024，bitmap机制来表示的fd最多可表示1024个，socket连接有上限
每次都是O(n)复杂度遍历所有socket收集有事件的socket。
每次都是O(n)复杂度（n是最大的fd值）遍历从内核态返回来的ready的fdset

poll

poll 实际上在Unix系统是不支持的，不像select使用bitmap集合来存储fd值，它通过一个大小为nfds的pollfd结构来表示需要监控的fd set，函数签名如下：

int poll (struct pollfd *fds, unsigned int nfds, int timeout);

pollfd的结构如下，每个fd都有对应的监听事件events，和就绪返回的事件revents，现在fd的大小是int最大值了。

struct  pollfd  {

      int  fd;

      short  events;

      short  revents;

};

代码示例：

  for (i=0;i<5;i++) 
  {
    memset(&client, 0, sizeof (client));
    addrlen = sizeof(client);
    pollfds[i].fd = accept(sockfd,(struct sockaddr*)&client, &addrlen);
    pollfds[i].events = POLLIN;
  }
  sleep(1);
  while(1){
  	puts("round again");
	poll(pollfds, 5, 50000);

	for(i=0;i<5;i++) {
		if (pollfds[i].revents & POLLIN){
			pollfds[i].revents = 0;
			memset(buffer,0,MAXBUF);
			read(pollfds[i].fd, buffer, MAXBUF);
			puts(buffer);
		}
	}
  }

select VS poll:

poll不需要每次都重新构建需要监控的fd set，但还是会有引起上下文切换的内存copy
poll不需要像select那样需要用户计算fd的最大值+1，作为select函数的第一个参数
poll减少了fd的遍历，在select中监控的某socket所对应的fd值为1000，那么需要做1000次循环
poll 解除了select对于fd数量1024的限制
poll在unix下不支持

epoll

细看select和poll的函数原型，我们会发现，每次调用select或poll都在重复地准备整个需要监控的fds集合。我们需要监控三个socket，就要准备一个readfds，然后新增监控一个socket，就要再准备一个readfds（包含旧的和新的socket的readfds）。然而对于频繁调用的select或poll而言，fds集合的变化频率要低得多，我们没必要每次都重新准备整个fds集合。

于是，epoll引入了epoll_ctl系统调用，将高频调用的epoll_wait和低频的epoll_ctl隔离开。epoll_ctl是epoll的事件注册函数，它不同与select()是在监听事件时，告诉内核要监听什么类型的事件，而是在这里先注册要监听的事件类型。到了有变化才变更，将select或poll高频、大块内存拷贝变成epoll_ctl的低频、小块内存的拷贝，避免了大量的内存拷贝。

同时，对于高频epoll_wait的可读就绪的fd集合返回的拷贝问题，epoll通过内核与用户空间mmap同一块内存来解决。mmap将用户空间的一块地址和内核空间的一块地址同时映射到相同的一块物理内存地址（不管是用户空间还是内核空间都是虚拟地址，最终要通过地址映射映射到物理地址），使得这块物理内存对内核和对用户均可见，减少用户态和内核态之间的数据交换。

另外，epoll通过epoll_ctl来对监控的fds集合来进行增、删、改，那么必须涉及到fd的快速查找问题。于是在linux 2.6.8以后的内核中采用了红黑树的结构来组织fds。
示例代码：

 struct epoll_event events[5];
  int epfd = epoll_create(10);
  ...
  ...
  for (i=0;i<5;i++) 
  {
    static struct epoll_event ev;
    memset(&client, 0, sizeof (client));
    addrlen = sizeof(client);
    ev.data.fd = accept(sockfd,(struct sockaddr*)&client, &addrlen);
    ev.events = EPOLLIN;
    epoll_ctl(epfd, EPOLL_CTL_ADD, ev.data.fd, &ev); 
  }
  
  while(1){
  	puts("round again");
  	nfds = epoll_wait(epfd, events, 5, 10000);
	
	for(i=0;i<nfds;i++) {
			memset(buffer,0,MAXBUF);
			read(events[i].data.fd, buffer, MAXBUF);
			puts(buffer);
	}
  }

遍历就绪的fds集合
通过上面的socket的睡眠队列唤醒逻辑我们知道，socket唤醒睡眠在其睡眠队列的wait_entry的时候会调用wait_entry的回调函数callback，并且，我们可以在callback中做任何事情。为了做到只遍历就绪的fd，我们需要有个地方来组织那些已经就绪的fd。

为此，epoll引入了一个中间层，一个双向链表ready_list，一个单独的睡眠队列single_epoll_wait_list，并且，与select或poll不同的是，epoll的task不需要同时插入到多路复用的socket集合的所有睡眠队列中，相反task只是插入到中间层的epoll的单独睡眠队列中（即single_epoll_wait_list），task睡眠在epoll的单独队列上，等待事件的发生。同时，引入一个中间的wait_entry_sk，它与某个socket密切相关，wait_entry_sk睡眠在socket的睡眠队列上，其callback函数逻辑是将当前socket排入到epoll的ready_list中，并唤醒epoll的single_epoll_wait_list。而single_epoll_wait_list上睡眠的task的回调函数就明朗了：遍历ready_list上的所有socket，挨个调用socket的poll函数收集事件，然后唤醒task从epoll_wait返回。