Linux 环境下的 IO 复用编程（二十一）

心已赠人 2022-10-29 13:29 102阅读 0赞

今天分析Linux 环境下的 IO 复用编程原理

***一、简介***

select ， poll ， epoll  都是  IO  多路复用的机制。 I/O  多路复用就是通过一种机制，一个进程可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序

进行相应的读写操作。但  select ， poll ， epoll  本质上都是同步  I/O ，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，而异步 I/O  则无需自己负责

进行读写，异步  I/O  的实现会负责把数据从内核拷贝到用户空间。

1、select

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

select  函数监视的文件描述符分  3  类，分别是  writefds 、 readfds 、和  exceptfds 。调用后select 函数会阻塞，直到有描述副就绪（有数据 可读、可写、或者有  except ），或者超时

（ timeout  指定等待时间，如果立即返回设为  null  即可），函数返回。当  select  函数返回后，可以 通过遍历 fdset ，来找到就绪的描述符。select 目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点。 select  的一 个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux 上一般为 1024 ， 可以通过修改宏定义甚至重新编译内核的方式提升这一限制，但是这样也会造成效率的降低。

示例C语言代码：

#include <stdio.h>
    #include <string.h>
    #include <arpa/inet.h>
    #include <netinet/in.h>
    #include <sys/socket.h>
    #include <sys/select.h>
    
    const static int MAXLINE = 1024;
    const static int SERV_PORT = 10001;
    
    int main()
    {
        int i , maxi , maxfd, listenfd , connfd , sockfd ;
        /*nready 描述字的数量*/
        int nready ,client[FD_SETSIZE];
        int n ;
        /*创建描述字集合，由于select函数会把未有事件发生的描述字清零，所以我们设置两个集合*/
        fd_set rset , allset;
        char buf[MAXLINE];
        socklen_t clilen;
        struct sockaddr_in cliaddr , servaddr;
        /*创建socket*/
        listenfd = socket(AF_INET , SOCK_STREAM , 0);
        /*定义sockaddr_in*/
        memset(&servaddr , 0 ,sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        servaddr.sin_port = htons(SERV_PORT);
        servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    
        bind(listenfd, (struct sockaddr *) & servaddr , sizeof(servaddr));
        listen(listenfd , 100);
        /*listenfd 是第一个描述字*/
        /*最大的描述字，用于select函数的第一个参数*/
        maxfd = listenfd;
        /*client的数量，用于轮询*/
        maxi = -1;
        /*init*/
        for(i=0 ;i<FD_SETSIZE ; i++)
            client[i] = -1;
        FD_ZERO(&allset);
        FD_SET(listenfd, &allset);
    
        for (;;)
        {
            rset = allset;
            /*只select出用于读的描述字，阻塞无timeout*/
            nready = select(maxfd+1 , &rset , NULL , NULL , NULL);
            if(FD_ISSET(listenfd,&rset))
            {
                clilen = sizeof(cliaddr);
                connfd = accept(listenfd , (struct sockaddr *) & cliaddr , &clilen);
                /*寻找第一个能放置新的描述字的位置*/
                for (i=0;i<FD_SETSIZE;i++)
                {
                    if(client[i]<0)
                    {
                        client[i] = connfd;
                        break;
                    }
                }
                /*找不到，说明client已经满了*/
                if(i==FD_SETSIZE)
                {
                    printf("Too many clients , over stack .\n");
                    return -1;
                }
                FD_SET(connfd,&allset);//设置fd
                /*更新相关参数*/
                if(connfd > maxfd) maxfd = connfd;
                if(i>maxi) maxi = i;
                if(nready<=1) continue;
                else nready --;
            }
    
            for(i=0 ; i<=maxi ; i++)
            {
                if (client[i]<0) continue;
                sockfd = client[i];
                if(FD_ISSET(sockfd,&rset))
                {
                    n = read(sockfd , buf , MAXLINE);
                    if (n==0)
                    {
                        /*当对方关闭的时候，server关闭描述字，并将set的sockfd清空*/
                        close(sockfd);
                        FD_CLR(sockfd,&allset);
                        client[i] = -1;
                    }
                    else
                    {
                        buf[n]='\0';
                        printf("Socket %d said : %s\n",sockfd,buf);
                        write(sockfd,buf,n); //Write back to client
                    }
                    nready --;
                    if(nready<=0) break;
                }
            }
    
        }
        return 0;
    }

2、poll

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

不同与  select  使用三个位图来表示三个  fdset  的方式， poll  使用一个  pollfd  的指针实现。pollfd 结构包含了要监视的  event  和发生的  event ，不再使用  select “参数 \- 值”传递的方

式。同时， pollfd  并没有最大数量限制（但是数量过大后性能也是会下降）。 和  select  函数一样，poll  返回后，需要轮询  pollfd  来获取就绪的描述符。

示例C语言 代码:

#include <stdio.h>
    #include <string.h>
    #include <arpa/inet.h>
    #include <netinet/in.h>
    #include <sys/socket.h>
    #include <poll.h>
     
    /*环境为ubuntu10.04自带c环境，无法自动引入下列宏，所以自己写在前面了*/
    #define INFTIM -1
    #define POLLRDNORM  0x040       /* Normal data may be read.  */
    #define POLLRDBAND  0x080       /* Priority data may be read.  */
    #define POLLWRNORM  0x100       /* Writing now will not block.  */
    #define POLLWRBAND  0x200       /* Priority data may be written.  */
     
    #define MAXLINE  1024
    #define OPEN_MAX  16 //一些系统会定义这些宏
    #define SERV_PORT  10001
     
    int main()
    {
        int i , maxi ,listenfd , connfd , sockfd ;
        int nready;
        int n;
        char buf[MAXLINE];
        socklen_t clilen;
        struct pollfd client[OPEN_MAX];
     
        struct sockaddr_in cliaddr , servaddr;
        listenfd = socket(AF_INET , SOCK_STREAM , 0);
        memset(&servaddr,0,sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        servaddr.sin_port = htons(SERV_PORT);
        servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
     
        bind(listenfd , (struct sockaddr *) & servaddr, sizeof(servaddr));
        listen(listenfd,10);
        client[0].fd = listenfd;
        client[0].events = POLLRDNORM;
        for(i=1;i<OPEN_MAX;i++)
        {
            client[i].fd = -1;
        }
        maxi = 0;
     
        for(;;)
        {
            nready = poll(client,maxi+1,INFTIM);
            if (client[0].revents & POLLRDNORM)
            {
                clilen = sizeof(cliaddr);
                connfd = accept(listenfd , (struct sockaddr *)&cliaddr, &clilen);
                for(i=1;i<OPEN_MAX;i++)
                {
                    if(client[i].fd<0)
                    {
                        client[i].fd = connfd;
                        client[i].events = POLLRDNORM;
                        break;
                    }
                }
                if(i==OPEN_MAX)
                {
                    printf("too many clients! \n");
                }
                if(i>maxi) maxi = i;
                nready--;
                if(nready<=0) continue;
            }
     
            for(i=1;i<=maxi;i++)
            {
                if(client[i].fd<0) continue;
                sockfd = client[i].fd;
                if(client[i].revents & (POLLRDNORM|POLLERR))
                {
                    n = read(client[i].fd,buf,MAXLINE);
                    if(n<=0)
                    {
                        close(client[i].fd);
                        client[i].fd = -1;
                    }
                    else
                    {
                        buf[n]='\0';
                        printf("Socket %d said : %s\n",sockfd,buf);
                        write(sockfd,buf,n); //Write back to client
                    }
                    nready--;
                    if(nready<=0) break; //no more readable descriptors
                }
            }
        }
        return 0;
    }

3、epoll

epoll  是在  2.6  内核中提出的，是之前的  select  和  poll  的增强版本。相对于  select  和  poll 来说，epoll  更加灵活，没有描述符限制。

int epoll_create(int size)； 
    int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)； 
    int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

*int epoll\_create(int size);*

创建一个  epoll  的句柄， size  用来告诉内核这个监听的数目一共有多大，这个参数不同于 select() 中的第一个参数，给出最大监听的  fd+1  的值，参数  size  并不是限制了  epoll  所能

监听的描述符最大个数，只是对内核初始分配内部数据结构的一个建议。当创建好  epoll  句柄后，它就会占用一个 fd  值，在  linux  下如果查看 /proc/ 进程  id/fd/ ，是能够看到这个  fd  的，

所以在使用完  epoll  后，必须调用  close() 关闭，否则可能导致  fd  被耗尽。

*int epoll\_ctl(int epfd, int op, int fd, struct epoll\_event \*event)* *；*

函数是对指定描述符  fd  执行  op  操作。

epfd ：是  epoll\_create() 的返回值。

op ：表示  op  操作，用三个宏来表示：添加  EPOLL\_CTL\_ADD ，删除  EPOLL\_CTL\_DEL ，修改 EPOLL\_CTL\_MOD 。分别添加、删除和修改对  fd  的监听事件。

fd ：是需要监听的  fd （文件描述符）

epoll\_event ：是告诉内核需要监听什么事，有具体的宏可以使用，比如  EPOLLIN  ：表示对应的文件描述符可以读（包括对端 SOCKET  正常关闭）； EPOLLOUT ：表示对应的文件描述

符可以写；

*int epoll\_wait(int epfd, struct epoll\_event \* events, int maxevents, int timeout);*

等待  epfd  上的  io  事件，最多返回  maxevents  个事件。参数 events  用来从内核得到事件的集合， maxevents  告之内核这个  events  有多大，这个 maxevents  的值不能大于创建  epoll\_create() 时的  size ，参数  timeout  是超时时间（毫秒， 0 会立即返回，-1  将不确定，也有说法说是永久阻塞）。该函数返回需要处理的事件数目，如返回 0  表示已超时。

示例C语言 代码：

#include <stdio.h>
    #include <string.h>
    #include <arpa/inet.h>
    #include <netinet/in.h>
    #include <sys/socket.h>
    #include <poll.h>
    #include <stdlib.h>
    #include <errno.h>
    #include <sys/types.h>
    #include <sys/wait.h>
    #include <unistd.h>
    #include <arpa/inet.h>
    #include <fcntl.h>
    #include <sys/epoll.h>
    #include <sys/time.h>
    #include <sys/resource.h>
     
    #define MAXLINE  1024
    #define OPEN_MAX  16 //一些系统会定义这些宏
    #define SERV_PORT  10001
     
    int main()
    {
        int i , maxi ,listenfd , connfd , sockfd ,epfd, nfds;
        int n;
        char buf[MAXLINE];
        struct epoll_event ev, events[20];  
        socklen_t clilen;
        struct pollfd client[OPEN_MAX];
     
        struct sockaddr_in cliaddr , servaddr;
        listenfd = socket(AF_INET , SOCK_STREAM , 0);
        memset(&servaddr,0,sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        servaddr.sin_port = htons(SERV_PORT);
        servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
     
        bind(listenfd , (struct sockaddr *) & servaddr, sizeof(servaddr));
        listen(listenfd,10);
        
        epfd = epoll_create(256);
        ev.data.fd=listenfd; 
        ev.events=EPOLLIN|EPOLLET;
        epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);
        
        for(;;)
        {
            nfds=epoll_wait(epfd,events,20,500); 
            for(i=0; i<nfds; i++)
            {
                /*处理接收到的连接*/
    			if (listenfd == events[i].data.fd)
                {
                    clilen = sizeof(cliaddr);
                    connfd = accept(listenfd , (struct sockaddr *)&cliaddr, &clilen);
                    if(connfd < 0)  
                    {  
                        perror("connfd < 0");  
                        exit(1);  
                    }
                    ev.data.fd=connfd; 
                    ev.events=EPOLLIN|EPOLLET;
                    epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);                
                }
    			/*读处理*/
                else if (events[i].events & EPOLLIN)
                {
                    if ( (sockfd = events[i].data.fd) < 0)  
                        continue;  
                    n = recv(sockfd,buf,MAXLINE,0);
                    if (n <= 0)   
                    {    
                        close(sockfd);  
                        events[i].data.fd = -1;  
                    }
                    else
                    {
                        buf[n]='\0';
                        printf("Socket %d said : %s\n",sockfd,buf);
                        ev.data.fd=sockfd; 
                        ev.events=EPOLLOUT|EPOLLET;
                        epoll_ctl(epfd,EPOLL_CTL_MOD,connfd,&ev);
                    }
                }
    			/*写处理*/
                else if( events[i].events&EPOLLOUT )
                {
                    sockfd = events[i].data.fd;  
                    send(sockfd, "Hello!", 7, 0);  
                      
                    ev.data.fd=sockfd;  
                    ev.events=EPOLLIN|EPOLLET;  
                    epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); 
                }
                else 
                {
                    printf("This is not avaible!");
                }
            }
        }
        close(epfd);  
        return 0;
    }

***4、select*** ***、*** ***poll*** ***、*** ***epoll***  ***的区别？***

前面已经讲述过，[参考如下][Link 1] 。

***二、epoll***  ***高效原理和底层机制分析***

总述： 当某一进程调用 epoll\_create  方法时， Linux  内核会创建一个  eventpoll  结构体，在内核cache 里建了个红黑树用于存储以后  epoll\_ctl  传来的  socket  外，还会再建立一个  rdllist  双向 链表，用于存储准备就绪的事件，当 epoll\_wait  调用时，仅仅观察这个  rdllist  双向链表里有没有数据即可。有数据就返回，没有数据就 sleep ，等到  timeout  时间到后即使链表没数据

也返回同时，所有添加到 epoll  中的事件都会与设备 ( 如网卡 ) 驱动程序建立回调关系，也就是 说相应事件的发生时会调用这里的回调方法。这个回调方法在内核中叫做 ep\_poll\_callback ，

它会把这样的事件放到上面的  rdllist  双向链表中。当调用 epoll\_wait  检查是否有发生事件的连接时，只是检查  eventpoll  对象中的  rdllist 双向链表是否有 epitem  元素而已，如果  rdllist  链表不为空，则这里的事件复制到用户态内存（使用共享内存提高效率）中，同时将事件数量返回给用户。因此 epoll\_wait  效率非常高，可以轻易地处理百万级别的并发连接。

1、从网卡接收数据说起

一个典型的计算机结构图，计算机由  CPU 、存储器（内存）、网络接口等部件组成。了解 epoll  本质的第一步，要从硬件的角度看计算机怎样接收网络数据。前面已经讲过这个内

容，我们这里快速复习一下。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

网卡接收数据的过程。网卡收到网线传来的数据；经过硬件电路的传输；最终将数据写入到内存中的某个地址上。这个过程涉及到 DMA  传输、 IO  通路选择等硬件有关的知识，

但我们只需知道：网卡会把接收到的数据写入内存。操作系统就可以去读取它们。

如何知道接收了数据？

CPU  的如何知道网络上有数据要接收？很简单，使用中断机制。

进程阻塞

了解  epoll  本质，要从操作系统进程调度的角度来看数据接收。阻塞是进程调度的关键一环，指的是进程在等待某事件（如接收到网络数据）发生之前的等待状态，recv 、 select

和  epoll  都是阻塞方法。了解“进程阻塞为什么不占用  cpu  资源？”，也就能够了解这一步。 为简单起见，我们从普通的 recv  接收开始分析，先看看下面代码：

// 创建  socket

int s = socket(AF\_INET, SOCK\_STREAM, 0);

// 绑定

bind(s, ...)

// 监听

listen(s, ...)

// 接受客户端连接

int c = accept(s, ...)

// 接收客户端数据

recv(c, ...);

// 将数据打印出来

printf(...)

这是一段最基础的网络编程代码，先新建  socket  对象，依次调用  bind 、 listen 、 accept ， 最后调用 recv  接收数据。 recv  是个阻塞方法，当程序运行到  recv  时，它会一直等待，直到

接收到数据才往下执行。

那么阻塞的原理是什么？

操作系统为了支持多任务，实现了进程调度的功能，会把进程分为“运行”和“等待” 等几种状态。运行状态是进程获得 cpu  使用权，正在执行代码的状态；等待状态是阻塞状态，

比如上述程序运行到  recv  时，程序会从运行状态变为等待状态，接收到数据后又变回运行状态。操作系统会分时执行各个运行状态的进程，由于速度很快，看上去就像是同时执行多

个任务。

下图中的计算机中运行着  A 、 B 、 C  三个进程，其中进程  A  执行着上述基础网络程序， 一开始，这 3  个进程都被操作系统的工作队列所引用，处于运行状态，会分时执行。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

当进程  A  执行到创建  socket  的语句时，操作系统会创建一个由文件系统管理的  socket 对象。这个 socket  对象包含了发送缓冲区、接收缓冲区、等待队列等成员。等待队列是个非常重要的结构，它指向所有需要等待该 socket  事件的进程。 当程序执行到 recv  时，操作系统会将进程  A  从工作队列移动到该  socket  的等待队列中 （如下图）。由于工作队列只剩下了进程 B  和  C ，依据进程调度， cpu  会轮流执行这两个进程的程序，不会执行进程 A  的程序。所以进程  A  被阻塞，不会往下执行代码，也不会占用 cpu 资源。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

操作系统添加等待队列只是添加了对这个“等待中”进程的引用，以便在接收到数据时获取进程对象、将其唤醒，而非直接将进程管理纳入自己之下。上图为了方便说明，直接将 进程挂到等待队列之下。当 socket 接收到数据后，操作系统将该  socket  等待队列上的进程重新放回到工作队列，该进程变成运行状态，继续执行代码。也由于 socket  的接收缓冲区已经有了数据， recv  可以返回接收到的数据。

2、内核接收网络数据全过程

进程在  recv  阻塞期间，计算机收到了对端传送的数据（步骤①）。数据经由网卡传送 到内存（步骤②），然后网卡通过中断信号通知 cpu  有数据到达， cpu  执行中断程序（步骤 ③）。此处的中断程序主要有两项功能，先将网络数据写入到对应  socket  的接收缓冲区里面（步骤④），再唤醒进程 A （步骤⑤），重新将进程  A  放入工作队列中。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

3、思考下，操作系统如何知道网络数据对应于哪个  socket ？

因为一个 socket  对应着一个端口号，而网络数据包中包含了  ip  和端口的信息，内核可以通过端口号找到对应的 socket 。当然，为了提高处理速度，操作系统会维护端口号到  socket

的索引结构，以快速读取。

4、思考下，如何同时监视多个 socket  的数据？

同时监视多个  socket  的简单方法

服务端需要管理多个客户端连接，而  recv  只能监视单个  socket ，这种矛盾下，人们开始寻找监视多个 socket  的方法。 epoll  的要义是高效的监视多个  socket 。从历史发展角度看， 必然先出现一种不太高效的方法，人们再加以改进。只有先理解了不太高效的方法，才能够理解 epoll 的本质。假如能够预先传入一个 socket  列表，如果列表中的  socket  都没有数据，挂起进程，直到有一个 socket  收到数据，唤醒进程。这种方法很直接，也是  select  的设计思想。为方便理解，我们先理解select  的用法。在如下的代码中，先准备一个数组（下面代码

中的  fds ），让  fds  存放着所有需要监视的  socket 。然后调用  select ，如果  fds  中的所有  socket 都没有数据，select  会阻塞，直到有一个  socket  接收到数据， select  返回，唤醒进程。用户可以遍历 fds ，通过  FD\_ISSET  判断具体哪个  socket  收到数据，然后做出处理。

int fds[] = 存放需要监听的 socket
    while(1){
    int n = select(..., fds, ...)
    for(int i=0; i < fds.count; i++){
    if(FD_ISSET(fds[i], ...)){
    //fds[i]的数据处理}
    }
    }

select  的实现思路很直接。假如程序同时监视  sock1 、 sock2  和  sock3  三个  socket ，那么在调用 select  之后，操作系统把进程  A  分别加入这三个  socket  的等待队列中。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

当任何一个  socket  收到数据后，中断程序将唤起进程。所谓唤起进程，就是将进程从所有的等待队列中移除，加入到工作队列里面。经由这些步骤，当进程 A  被唤醒后，它知道至少有一个  socket  接收了数据。程序只需遍历一遍 socket  列表，就可以得到就绪的  socket 。这种简单方式行之有效，在几乎所有操作系统都有对应的实现。

但是简单的方法往往有缺点，主要是：

其一，每次调用  select  都需要将进程加入到所有被监视  socket  的等待队列，每次唤醒都需要从每个队列中移除，都必须要进行遍历。而且每次都要将整个 fds  列表传递给内核，有

一定的开销。正是因为遍历操作开销大，出于效率的考量，才会规定  select  的最大监视数量，默认只能监视 1024  个  socket 。

其二，进程被唤醒后，程序并不知道哪些  socket  收到数据，还需要遍历一次。那么，有没有减少遍历的方法？有没有保存就绪 socket  的方法？这两个问题便是  epoll 技术要解决的。 当然，当程序调用 select 时，内核会先遍历一遍  socket ，如果有一个以上的  socket  接收缓冲区有数据，那么 select  直接返回，不会阻塞。这也是为什么  select  的返回值有可能大于 1 的原因之一。如果没有  socket  有数据，进程才会阻塞。

4、epoll  的设计思路

epoll  是在  select  出现  N  多年后才被发明的，是  select  和  poll  的增强版本。 epoll  通过以下一些措施来改进效率。

措施一：功能分离

select  低效的原因之一是将“维护等待队列”和“阻塞进程”两个步骤合二为一。每次调用 select  都需要这两步操作，然而大多数应用场景中，需要监视的  socket  相对固定，并不需要每次都修改。epoll  将这两个操作分开，先用  epoll\_ctl  维护等待队列，再调用  epoll\_wait 阻塞进程。显而易见的，效率就能得到提升。相比 select ， epoll  拆分了功能 为方便理解后续的内容，我们先理解下  epoll  的用法。如下的代码中，先用  epoll\_create 创建一个 epoll  对象  epfd ，再通过  epoll\_ctl  将需要监视的  socket  添加到  epfd  中，最后调用epoll\_wait 等待数据。

int epfd = epoll_create(...);
    epoll_ctl(epfd, ...); //将所有需要监听的 socket 添加到 epfd 中
    while(1){
    int n = epoll_wait(...)
    for(接收到数据的 socket){
    //处理
    }
    }

功能分离，使得  epoll  有了优化的可能。

措施二：就绪列表

select  低效的另一个原因在于程序不知道哪些  socket  收到数据，只能一个个遍历。如果内核维护一个“就绪列表”，引用收到数据的 socket ，就能避免遍历。

5、epoll  的原理和流程

当某个进程调用  epoll\_create  方法时，内核会创建一个  eventpoll  对象（也就是程序中 epfd 所代表的对象）。 eventpoll  对象也是文件系统中的一员，和  socket  一样，它也会有等

待队列。创建 epoll  对象后，可以用  epoll\_ctl  添加或删除所要监听的  socket 。以添加  socket  为例，如下图，如果通过 epoll\_ctl  添加  sock1 、 sock2  和  sock3  的监视，内核会将  eventpoll  添加到这三个 socket  的等待队列中。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5][]

当  socket  收到数据后，中断程序会操作  eventpoll  对象，而不是直接操作进程。中断程序会给 eventpoll  的“就绪列表”添加  socket  引用。如下图展示的是  sock2  和  sock3  收到数据

后，中断程序让  rdlist  引用这两个  socket 。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6][]

eventpoll  对象相当于是  socket  和进程之间的中介， socket  的数据接收并不直接影响进程，而是通过改变 eventpoll  的就绪列表来改变进程状态。 当程序执行到 epoll\_wait  时，如果  rdlist  已经引用了  socket ，那么  epoll\_wait  直接返回，如果 rdlist  为空，阻塞进程。 假设计算机中正在运行进程 A  和进程  B ，在某时刻进程  A  运行到了  epoll\_wait  语句。如下图所示，内核会将进程 A  放入  eventpoll  的等待队列中，阻塞进程。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7][]

当  socket  接收到数据，中断程序一方面修改  rdlist ，另一方面唤醒  eventpoll  等待队列中的进程，进程 A  再次进入运行状态。也因为  rdlist  的存在，进程  A  可以知道哪些  socket  发生

了变化。

6、epoll  的实现细节

现在对  epoll  的本质已经有一定的了解。但我们还留有一个问题， eventpoll  的数据结构是什么样子？

思考两个问题，就绪队列应该应使用什么数据结构？ eventpoll  应使用什么数据结构来管理通过 epoll\_ctl  添加或删除的  socket ？

linux源码：

struct eventpoll {
    	/*
    	 * This mutex is used to ensure that files are not removed
    	 * while epoll is using them. This is held during the event
    	 * collection loop, the file cleanup path, the epoll file exit
    	 * code and the ctl operations.
    	 */
    	struct mutex mtx;
    
    	/* Wait queue used by sys_epoll_wait() */
    	wait_queue_head_t wq;
    
    	/* Wait queue used by file->poll() */
    	wait_queue_head_t poll_wait;
    
    	/* List of ready file descriptors */
    	struct list_head rdllist;
    
    	/* Lock which protects rdllist and ovflist */
    	rwlock_t lock;
    
    	/* RB tree root used to store monitored fd structs */
    	struct rb_root_cached rbr;
    
    	/*
    	 * This is a single linked list that chains all the "struct epitem" that
    	 * happened while transferring ready events to userspace w/out
    	 * holding ->lock.
    	 */
    	struct epitem *ovflist;
    
    	/* wakeup_source used when ep_scan_ready_list is running */
    	struct wakeup_source *ws;
    
    	/* The user that created the eventpoll descriptor */
    	struct user_struct *user;
    
    	struct file *file;
    
    	/* used to optimize loop detection check */
    	u64 gen;
    	struct hlist_head refs;

就绪列表引用着就绪的 socket，所以它应能够快速的插入数据。程序可能随时调用 epoll\_ctl 添加监视 socket，也可能随时删除。当删除时，若该 socket 已经存放在就绪列表中，它也应该被移除。 所以就绪列表应是一种能够快速插入和删除的数据结构。双向链表就是这样一种数据结构，epoll 使用双向链表来实现就绪队列，也就是 Linux 源码中的

/* List of ready file descriptors */
    	struct list_head rdllist;

既然 epoll 将“维护监视队列”和“进程阻塞”分离，也意味着需要有个数据结构来保存监视的 socket。至少要方便的添加和移除，还要便于搜索，以避免重复添加。红黑树是一 种自平衡二叉查找树，搜索、插入和删除时间复杂度都是 O(log(N))，效率较好。epoll 使用了红黑树作为索引结构，也就是 Linux 源码中的

/* RB tree root used to store monitored fd structs */
    	struct rb_root_cached rbr;

因为操作系统要兼顾多种功能，以及由更多需要保存的数据，rdlist 并非直接引用 socket， 而是通过 epitem 间接引用，红黑树的节点也是 epitem 对象。

struct epitem {
    	union {
    		/* RB tree node links this structure to the eventpoll RB tree */
    		struct rb_node rbn;
    		/* Used to free the struct epitem */
    		struct rcu_head rcu;
    	};
    
    	/* List header used to link this structure to the eventpoll ready list */
    	struct list_head rdllink;
    
    	/*
    	 * Works together "struct eventpoll"->ovflist in keeping the
    	 * single linked chain of items.
    	 */
    	struct epitem *next;
    
    	/* The file descriptor information this item refers to */
    	struct epoll_filefd ffd;
    
    	/* List containing poll wait queues */
    	struct eppoll_entry *pwqlist;
    
    	/* The "container" of this item */
    	struct eventpoll *ep;
    
    	/* List header used to link this item to the "struct file" items list */
    	struct hlist_node fllink;
    
    	/* wakeup_source used when EPOLLWAKEUP is set */
    	struct wakeup_source __rcu *ws;
    
    	/* The structure that describe the interested events and the source fd */
    	struct epoll_event event;
    };

如图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 8][]

到此，Linux 环境下的 IO 复用编程分析完毕，下篇分析JDK 中 NIO 的实现源码分析，敬请期待！

[Link 1]: https://blog.csdn.net/nandao158/article/details/113746804
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/342dfa70dc0c45d09769978c42ec079b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/cc2bdfe8a39846ecab8987d46b900033.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/dff4958d36164dee82de92f907863cc0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221024/92317a92f7ae4543948d3cc1c7fbc4df.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221024/a6db3a43b1d34156b99e711963039d14.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 5]: /images/20221024/6651fb00971041a1951e8eafa3264f5f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 6]: /images/20221024/bd60a0db8044425590fd9567a6ab361f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 7]: /images/20221024/9634fffc486b43d3834ff8a3ffb1ee20.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 8]: /images/20221024/2088d3aab7b14b4f8579e338e668bacb.png