I/O多路复用之epoll

ゝ一世哀愁。 2022-06-12 08:30 163阅读 0赞

epoll()是Linux特有的I/O复用函数。它在实现和使用上与select,poll有很大的差异。  
关于select和poll可以参考：[I/O多路复用之select][I_O_select] , [I/O多路复用之poll][I_O_poll]  
首先，epoll使用一组函数来完成任务，而不是单个函数。其次，epoll把用户关心的文件描述符上的事件放在内核里的一个事件表中，从而无须像select和poll那样每次调用都要重复传入文件描述符集或事件集。  
相比于select，epoll最大的好处在于它不会随着监听fd数目的增长而降低效率。因为在内核中的select实现中，它是采用轮询来处理的，轮询的fd数目越多，自然耗时越多。并且，在linux/posix\_types.h头文件有这样的声明：

`#define __FD_SETSIZE 1024`

表示select最多同时监听1024个fd，当然，可以通过修改头文件再重编译内核来扩大这个数目，但这并不治本。  
然而epoll则没有这个限制，它所支持的FD上限是最大可以打开文件的数目。具体数目可以`cat /proc/sys/fs/file-max`察看,一般来说这个数目和系统内存关系很大。我的系统的file-max值是392250。  
epoll在具有大量应用程序请求时能够获得较好的性能（ 此时被监视的文件描述符数目非常大，与旧的 select 和 poll 系统调用完成操作所需 O(n) 不同， epoll能在O(1)时间内完成操作，所以性能相当高）。

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

### **epoll的使用：** ###

#### **一.创建epoll实例：** ####

由于epoll把用户关心的文件描述符上的事件放在内核里的一个事件表中，所以epoll需要使用一个额外的文件描述符，来唯一标识内核中的这个事件表。  
这个文件描述符使用如下epoll\_create函数来创建：

#include<sys/epoll.h>
    int epoll_create(int size);

size参数现在并不起作用，只是给内核一个提示，告诉内核应该如何为内部数据结构划分初始大小。  
该函数返回的文件描述符将用作其他所有epoll系统调用的第一个参数，以指定要访问的内核事件表。  
需要注意的是，这个文件描述符也会占用一个fd值，在linux下如果查看/proc/进程id/fd/，能够看到这个fd，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。  
ps:

>  *  如果我们使用dup()或类似的函数复制一个epoll文件描述符，那么被复制的描述符所指代的epoll兴趣列表和就绪列表同原始的epoll文件描述符相同。
>  *  上一条观点同样也适合于fork()调用之后的情况。此时子进程通过继承复制了父进程的epoll文件描述符，而这个复制的文件描述符所指向的epoll数据结构同原始的描述符相同。

#### **二.修改epoll的兴趣列表：** ####

下面的函数用来操作epoll的内核事件表：

#include <sys/epoll.h>
    
    int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epfd是创建epoll实例时返回的文件描述符，fd参数是要操作的文件描述符。  
op参数则指定操作类型。操作类型有如下3种：

*  EPOLL\_CTL\_ADD，注册新的fd到epfd中；
 *  EPOLL\_CTL\_MOD，修改已注册fd的事件；
 *  EPOLL\_CTL\_DEL，从epfd中删除一个fd；  
    event参数指定事件，它是epoll\_event结构指针类型。epoll\_event的定义如下：

struct epoll_event {
      __uint32_t events;  /* Epoll 事件 */
      epoll_data_t data;  /* 用户数据 */
    };

其中events成员描述事件类型。epoll支持的事件类型和epoll基本相同。表示epoll事件类型的宏是在poll对应的宏前加上”E”，比如epoll的数据可读事件是EPOLLIN。但epoll有两个额外的事件类型——EPOLLET和EPOLLONESHOT。它们对于epoll的高效运作非常关键，我们一会将会具体讲解这两个事件类型。  
events事件类型：

<table> 
 <thead> 
  <tr> 
   <th>事件</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>EPOLLIN</td> 
   <td>可读取非高优先级数据</td> 
  </tr> 
  <tr> 
   <td>EPOLLPRI</td> 
   <td>可读取高优先级数据</td> 
  </tr> 
  <tr> 
   <td>EPOLLOUT</td> 
   <td>普通数据可写</td> 
  </tr> 
  <tr> 
   <td>EPOLLHUP</td> 
   <td>出现挂断</td> 
  </tr> 
  <tr> 
   <td>EPOLLET</td> 
   <td>采用边沿触发事件通知</td> 
  </tr> 
  <tr> 
   <td>EPOLLONESHOT</td> 
   <td>在完成事件通知后禁用检查</td> 
  </tr> 
  <tr> 
   <td>EPOLLRDHUP</td> 
   <td>套接字对端关闭</td> 
  </tr> 
  <tr> 
   <td>EPOLLERR</td> 
   <td>有错误发生</td> 
  </tr> 
 </tbody> 
</table>

data成员用于存储用户数据，其类型epoll\_data\_t的定义如下：

typedef union epoll_data {    
            void *ptr;    
            int fd;    
            __uint32_t u32;    
            __uint64_t u64;    
        } epoll_data_t;

> epoll\_data\_t是一个共用体，其4个成员中使用最多的是fd，它指定事件所从属的目标文件描述符。ptr成员可以用来指定与fd相关的用户数据。但由于epoll\_data\_t是一个共用体，我们不能同时使用其ptr成员和fd成员，因此，如果要将文件描述符和用户数据关联起来，以实现快速的数据访问，只能放弃使用epoll\_data\_t的fd成员，而在ptr指向的用户数据中包含fd。

epoll\_ctl成功时返回0，失败时返回-1并设置errno。

#### **三.事件等待：** ####

epoll系列的系统调用的主要接口是epoll\_wait函数。它在一段超时时间内等待一组文件描述符上的事件，其原型如下：

#include <sys/epoll.h>
    
    int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

该函数成功时返回就绪的文件描述符个数，失败时返回-1并设置errno。  
timeout参数含义与poll相同，当timeout=-1时，阻塞直到有一个或多个文件描述符就绪。当timeout=0时，立即返回。当timeout值大于0时，等待timeout值的时间。  
maxevents参数指定最多监听多少个事件，它必须大于0。  
epoll\_wait函数如果检测到事件，就将所有就绪的事件从内核事件表中复制到它的第二个参数events指向的数组中。数组events的空间由调用者负责申请，所包含的元素个数在参数maxevents中指定。  
与poll的使用一样，我们只需要将输出数组中的元素的events数据成员与我们关心的事件就行&操作，结果大于0，则代表该事件已就绪。  
第二个参数events数组只用于输出epoll\_wait检测到的就绪事件，而不像select和poll的数组参数那样既用于传入用户注册事件，又用于输出内核检测到的就绪事件。这就极大地提高了应用程序索引就绪文件描述符的效率。  
我们可以看一个简单的代码：

//比较poll和epoll在使用上的差别
    
    //索引poll返回的就绪文件描述符
    int ret=poll(fds,MAX_EVENT_NUMBER,-1);
    //必须遍历所有已注册文件描述符并找到其中的就绪者
    for(int i=0;i<MAX_EVENT_NUMBER;i++)
    {
         if(fds[i].revents & POLLIN)  //判断第i个文件描述符是否就绪
         {  
             int sockfd=fds[i].fd;
             //处理sockfd.....
         }
    }
    
    
    //索引epoll返回的就绪文件描述符
    int ret = epoll_wait(epollfd,events,MAX_EVENT_NUMBER,-1);
    //仅需遍历就绪的ret个文件描述符
    for(int i=0;i<ret;i++)
    {
       int sockfd = events[i].data.fd;
       //sockfd肯定就绪，直接处理
    }

### **LT模式与ET模式：** ###

epoll对文件描述符的操作有两种模式：LT（水平触发）模式和ET（边沿触发）模式。LT模式是默认的工作模式，这种模式编程出错误可能性要小一点。传统的select/poll都是这种模型的代表。当往epoll内核事件表中注册一个文件描述符上的EPOLLET事件时，epoll将以ET模式来操作该文件描述符。ET模式是epoll的高效工作模式。  
关于什么是LT模式和ET模式，我们在这里就不详细介绍了。还不太清楚这两种工作方式的可以看我的这篇博客：[I/O多路复用之水平触发和边沿触发模式][I_O]  
我们在这里只简单的介绍一下，主要介绍如何使用epoll的LT模式和ET模式。

水平触发通知：如果文件描述符上可以非阻塞地执行I/O系统调用，此时认为它已经就绪，触发通知。  
边沿触发通知：如果文件描述符自上次状态检查以来有了新的I/O活动(比如新的输入)，此时需要触发通知。

二者的差异在于LT模式下只要某个socket处于readable/writable状态，无论什么时候进行epoll\_wait都会返回该socket；而ET模式下只有某个socket从unreadable变为readable或从unwritable变为writable时，epoll\_wait才会返回该socket，ET模式注重的是状态发生改变的时候才触发。下面两幅图清晰反映了二者区别。

![这里写图片描述][SouthEast]

![这里写图片描述][SouthEast 1]

对于采用LT工作模式的文件描述符，当epoll\_wait检测到其上有事件发生并将此事件通知应用程序后，应用程序可以不立即处理该事件。这样，当应用程序下一次调用epoll\_wait时，epoll\_wait还会再次向应用程序通告此事件，直到该事件被处理。而对于采用ET工作模式的文件描述符，当epoll\_wait检测到其上有事件发生并将此事件通知应用程序后，应用程序必须立即处理该事件，因为后续的epoll\_wait调用将不再向应用程序通知这件事。可见，ET模式在很大程度上降低了同一个epoll事件被重复触发的次数，因此效率要比LT模式高。

当我们使用ET模式的epoll时，我们应该按照以下规则设计：

*  在接收到一个I/O事件通知后，立即处理该事件。程序在某个时刻应该在相应的文件描述符上尽可能多地执行I/O。
 *  在ET模式下，在使用epoll\_ctl注册文件描述符的事件时，应该把描述符设置为非阻塞的。因为程序采用循环来对文件描述符执行尽可能多的I/O，而文件描述符又被设置为可阻塞的，那么最终当没有更多的I/O可执行时，I/O系统调用就会阻塞。基于这个原因，每个被检查的文件描述符通常应该置为非阻塞模式，在得到I/O事件通知后重复执行I/O操作，直到相应的系统调用(比如read(),write())以错误码EAGAIN或EWOULDBLOCK的形式失败。

说了这么多，我们来看一个简单的两种工作模式的代码，很容易就明白了：

#include <sys/types.h>
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <assert.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <errno.h>
    #include <string.h>
    #include <fcntl.h>
    #include <stdlib.h>
    #include <sys/epoll.h>
    #include <pthread.h>
    
    #define MAX_EVENT_NUMBER 1024
    #define BUFFER_SIZE 10
    
    //将文件描述符设置成非阻塞的
    int setnonblocking( int fd )
    {
        int old_option = fcntl( fd, F_GETFL );
        int new_option = old_option | O_NONBLOCK;
        fcntl( fd, F_SETFL, new_option );
        return old_option;
    }
    
    //将文件描述符fd上的EPOLLIN注册到epollfd指示的epoll内核事件表中，参数enable_et指定是否对fd启用ET模式
    void addfd( int epollfd, int fd, bool enable_et )
    {
        epoll_event event;
        event.data.fd = fd;
        event.events = EPOLLIN;
        if( enable_et )
        {
            event.events |= EPOLLET;
        }
        epoll_ctl( epollfd, EPOLL_CTL_ADD, fd, &event );
        setnonblocking( fd );
    }
    
    
    //LT模式的工作流程
    void lt( epoll_event* events, int number, int epollfd, int listenfd )
    {
        char buf[ BUFFER_SIZE ];
        for ( int i = 0; i < number; i++ )
        {
            int sockfd = events[i].data.fd;
            if ( sockfd == listenfd )
            {
                struct sockaddr_in client_address;
                socklen_t client_addrlength = sizeof( client_address );
                int connfd = accept( listenfd, ( struct sockaddr* )&client_address, &client_addrlength );
                addfd( epollfd, connfd, false );
            }
            else if ( events[i].events & EPOLLIN )
            {
                //只要socket读缓存中还有未读出的数据，这段代码就被触发
                printf( "event trigger once\n" );
                memset( buf, '\0', BUFFER_SIZE );
                int ret = recv( sockfd, buf, BUFFER_SIZE-1, 0 );
                if( ret <= 0 )
                {
                    close( sockfd );
                    continue;
                }
                printf( "get %d bytes of content: %s\n", ret, buf );
            }
            else
            {
                printf( "something else happened \n" );
            }
        }
    }
    
    
    //ET模式的工作流程
    void et( epoll_event* events, int number, int epollfd, int listenfd )
    {
        char buf[ BUFFER_SIZE ];
        for ( int i = 0; i < number; i++ )
        {
            int sockfd = events[i].data.fd;
            if ( sockfd == listenfd )
            {
                struct sockaddr_in client_address;
                socklen_t client_addrlength = sizeof( client_address );
                int connfd = accept( listenfd, ( struct sockaddr* )&client_address, &client_addrlength );
                addfd( epollfd, connfd, true );
            }
            else if ( events[i].events & EPOLLIN )
            {
                //这段代码不会被重复触发，所以我们循环读取数据，以确保把socket读缓存中的所有数据读出
                printf( "event trigger once\n" );
                while( 1 )
                {
                    memset( buf, '\0', BUFFER_SIZE );
                    int ret = recv( sockfd, buf, BUFFER_SIZE-1, 0 );
                    if( ret < 0 )
                    {
                        //对于非阻塞IO，下面条件成立表示数据已经全部读取完毕。
                        //此后，epoll就能再次触发sockfd上的EPOLLIN事件，已驱动下一次读操作
                        if( ( errno == EAGAIN ) || ( errno == EWOULDBLOCK ) )
                        {
                            printf( "read later\n" );
                            break;
                        }
                        close( sockfd );
                        break;
                    }
                    else if( ret == 0 )
                    {
                        close( sockfd );
                    }
                    else
                    {
                        printf( "get %d bytes of content: %s\n", ret, buf );
                    }
                }
            }
            else
            {
                printf( "something else happened \n" );
            }
        }
    }
    
    int main( int argc, char* argv[] )
    {
        if( argc <= 2 )
        {
            printf( "usage: %s ip_address port_number\n", basename( argv[0] ) );
            return 1;
        }
        const char* ip = argv[1];
        int port = atoi( argv[2] );
    
        int ret = 0;
        struct sockaddr_in address;
        bzero( &address, sizeof( address ) );
        address.sin_family = AF_INET;
        inet_pton( AF_INET, ip, &address.sin_addr );
        address.sin_port = htons( port );
    
        int listenfd = socket( PF_INET, SOCK_STREAM, 0 );
        assert( listenfd >= 0 );
    
        ret = bind( listenfd, ( struct sockaddr* )&address, sizeof( address ) );
        assert( ret != -1 );
    
        ret = listen( listenfd, 5 );
        assert( ret != -1 );
    
        epoll_event events[ MAX_EVENT_NUMBER ];
        int epollfd = epoll_create( 5 );
        assert( epollfd != -1 );
        addfd( epollfd, listenfd, true );
    
        while( 1 )
        {
            int ret = epoll_wait( epollfd, events, MAX_EVENT_NUMBER, -1 );
            if ( ret < 0 )
            {
                printf( "epoll failure\n" );
                break;
            }
    
            lt( events, ret, epollfd, listenfd );     //使用LT模式
            //et( events, ret, epollfd, listenfd ); //使用ET模式
        }
    
        close( listenfd );
        return 0;
    }

### EPOLLONESHOT事件 ###

即使我们使用ET模式，一个socket上的某个事件还是可能被触发多次。这在并发程序中就会引起一个问题。比如一个线程（或进程）在读取完某个socket上的数据后开始处理这些数据，而在数据的处理过程中该socket上又有新的数据可读（EPOLLIN再次被触发），此时另外一个线程被唤醒来读取这些新的数据。于是就出现了两个线程同时操作一个socket的局面。这当然不是我们希望的，我们期望的是一个socket连接在任一时刻都只被一个线程处理。这一点可以使用epoll的EPOLLONESHOT事件实现。

对于注册了EPOLLONESHOT事件的文件描述符，操作系统最多触发其注册的一个可读，可写或者异常事件，且只触发一次，除非我们使用epoll\_ctl函数重置该文件描述符上注册的EPOLLONESHOT事件。这样，当一个线程在处理某个socket时，其他线程是不可能有机会操作该socket的。

让我们来看一个例子：

#include <sys/types.h>
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <assert.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <errno.h>
    #include <string.h>
    #include <fcntl.h>
    #include <stdlib.h>
    #include <sys/epoll.h>
    #include <pthread.h>
    
    #define MAX_EVENT_NUMBER 1024
    #define BUFFER_SIZE 1024
    struct fds
    {
       int epollfd;
       int sockfd;
    };
    
    //设置为非阻塞文件描述符
    int setnonblocking( int fd )
    {
        int old_option = fcntl( fd, F_GETFL );
        int new_option = old_option | O_NONBLOCK;
        fcntl( fd, F_SETFL, new_option );
        return old_option;
    }
    
    //将fd上的EPOLLIN和EPOLLET事件注册到epollfd指示的epoll内核事件表中，参数oneshot指定是否注册fd上的EPOLLONESHOT事件
    void addfd( int epollfd, int fd, bool oneshot )
    {
        epoll_event event;
        event.data.fd = fd;
        event.events = EPOLLIN | EPOLLET;
        if( oneshot )
        {
            event.events |= EPOLLONESHOT;
        }
        epoll_ctl( epollfd, EPOLL_CTL_ADD, fd, &event );
        setnonblocking( fd );
    }
    
    //重置fd上的事件。这样操作之后，尽管fd上的EPOLLONESHOT事件被注册，但是操作系统仍然会触发fd上的EPOLLIN事件，且只触发一次
    void reset_oneshot( int epollfd, int fd )
    {
        epoll_event event;
        event.data.fd = fd;
        event.events = EPOLLIN | EPOLLET | EPOLLONESHOT;
        epoll_ctl( epollfd, EPOLL_CTL_MOD, fd, &event );
    }
    
    void* worker( void* arg )
    {
        int sockfd = ( (fds*)arg )->sockfd;
        int epollfd = ( (fds*)arg )->epollfd;
        printf( "start new thread to receive data on fd: %d\n", sockfd );
        char buf[ BUFFER_SIZE ];
        memset( buf, '\0', BUFFER_SIZE );
        //循环读取sockfd上的数据，直到遇到EAGAIN错误
        while( 1 )
        {
            int ret = recv( sockfd, buf, BUFFER_SIZE-1, 0 );
            if( ret == 0 )
            {
                close( sockfd );
                printf( "foreiner closed the connection\n" );
                break;
            }
            else if( ret < 0 )
            {
                if( errno == EAGAIN )
                {
                    reset_oneshot( epollfd, sockfd );
                    printf( "read later\n" );
                    break;
                }
            }
            else
            {
                printf( "get content: %s\n", buf );
                //休眠5s,模拟数据处理过程
                sleep( 5 );
            }
        }
        printf( "end thread receiving data on fd: %d\n", sockfd );
    }
    
    int main( int argc, char* argv[] )
    {
        if( argc <= 2 )
        {
            printf( "usage: %s ip_address port_number\n", basename( argv[0] ) );
            return 1;
        }
        const char* ip = argv[1];
        int port = atoi( argv[2] );
    
        int ret = 0;
        struct sockaddr_in address;
        bzero( &address, sizeof( address ) );
        address.sin_family = AF_INET;
        inet_pton( AF_INET, ip, &address.sin_addr );
        address.sin_port = htons( port );
    
        int listenfd = socket( PF_INET, SOCK_STREAM, 0 );
        assert( listenfd >= 0 );
    
        ret = bind( listenfd, ( struct sockaddr* )&address, sizeof( address ) );
        assert( ret != -1 );
    
        ret = listen( listenfd, 5 );
        assert( ret != -1 );
    
        epoll_event events[ MAX_EVENT_NUMBER ];
        int epollfd = epoll_create( 5 );
        assert( epollfd != -1 );
        addfd( epollfd, listenfd, false );
    
        while( 1 )
        {
            int ret = epoll_wait( epollfd, events, MAX_EVENT_NUMBER, -1 );
            if ( ret < 0 )
            {
                printf( "epoll failure\n" );
                break;
            }
    
            for ( int i = 0; i < ret; i++ )
            {
                int sockfd = events[i].data.fd;
                if ( sockfd == listenfd )
                {
                    struct sockaddr_in client_address;
                    socklen_t client_addrlength = sizeof( client_address );
                    int connfd = accept( listenfd, ( struct sockaddr* )&client_address, &client_addrlength );
                    //对每个非监听文件描述符都注册EPOLLONESHOT事件
                    addfd( epollfd, connfd, true );
                }
                else if ( events[i].events & EPOLLIN )
                {
                    pthread_t thread;
                    fds fds_for_new_worker;
                    fds_for_new_worker.epollfd = epollfd;
                    fds_for_new_worker.sockfd = sockfd;
                    //新启动一个工作线程为sockfd服务
                    pthread_create( &thread, NULL, worker, ( void* )&fds_for_new_worker );
                }
                else
                {
                    printf( "something else happened \n" );
                }
            }
        }
    
        close( listenfd );
        return 0;
    }

从工作线程函数worker来看，如果一个工作线程处理完某个socket上的一次请求（我们用休眠5s来模拟这个过程）之后，又接收到该socket上新的客户请求，则该线程将继续为这个socket服务。并且因为该socket上注册了EPOLLONESHOT事件，其他线程没有机会接触这个socket，如果工作线程等待5s后仍没有收到该socket上的下一批客户端数据，则它将放弃为该socket服务。同时，它调用reset\_oneshot函数来重置该socket上的注册事件，这会使epoll有机会再次检测到该socket上的EPOLLIN事件，进而使得其他线程有机会为该socket服务。

#### **epoll的优点：** ####

1.  支持一个进程打开大数目的socket描述符(FD)
2.  IO效率不随FD数目增加而线性下降  
    传统的select/poll另一个致命弱点就是当你拥有一个很大的socket集合，不过由于网络延时，任一时间只有部分的socket是”活跃”的，但是select/poll每次调用都会线性扫描全部的集合，导致效率呈现线性下降。但是epoll不存在这个问题，它只会对”活跃”的socket进行操作。这是因为在内核实现中epoll是根据每个fd上面的callback函数实现的。那么，只有”活跃”的socket才会主动的去调用 callback函数。
3.  使用mmap加速内核与用户空间的消息传递  
    这点实际上涉及到epoll的具体实现了。无论是select,poll还是epoll都需要内核把FD消息通知给用户空间，如何避免不必要的内存拷贝就很重要，在这点上，epoll是通过内核于用户空间mmap同一块内存实现的。

需要我们注意的是，虽然epoll相比select和poll有了很多的改进，但是我们只能说通常使用epoll的效率比select和poll都高，但这并不是绝对的，要根据实际的应用情况来判断使用哪种I/O。  
如果已连接套接字数目不太大并且这些套接字一直处于活跃的状态，那么epoll会不停地调用callback函数可能会造成低效率，也就是说低于一次性遍历，此时epoll的效率就可能低于select和poll。总的来说，epoll在需要监听大量的文件描述符，但是这些文件描述符不是很活跃的情况下，是效率最高的。

[I_O_select]: http://blog.csdn.net/fengxinlinux/article/details/75268914
[I_O_poll]: http://blog.csdn.net/fengxinlinux/article/details/75303969
[I_O]: http://blog.csdn.net/fengxinlinux/article/details/75331567
[SouthEast]: /images/20220612/0c5a13d06f96479c9fbbb128a7575ed4.png
[SouthEast 1]: /images/20220612/60ad3a7cb3bf40cca07cee24ccae2822.png