1 epoll模型简介

epoll可是当前在Linux下开发大规模并发网络程序的热门人选，epoll 在Linux2.6内核中正式引入，和select相似，其实都I/O多路复用技术而已，并没有什么神秘的。

其实在Linux下设计并发网络程序，向来不缺少方法，比如典型的Apache模型（Process Per Connection，简称PPC），TPC（Thread PerConnection）模型，以及select模型和poll模型，那为何还要再引入Epoll这个东东呢？那还是有得说说的…

2 常用模型的缺点

如果不摆出来其他模型的缺点，怎么能对比出Epoll的优点呢。

2.1 多进程PPC/多线程TPC模型

这两种模型思想类似，就是让每一个到来的连接一边自己做事去，别再来烦我。只是PPC是为它开了一个进程，而TPC开了一个线程。可是别烦我是有代价的，它要时间和空间啊，连接多了之后，那么多的进程/线程切换，这开销就上来了；

因此这类模型能接受的最大连接数都不会高，一般在几百个左右。

2.2 select模型-O(n)

多进程多线程的模型庞大而且繁琐，因此我们出现了select模型

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

       void FD_CLR(int fd, fd_set *set);
       int  FD_ISSET(int fd, fd_set *set);
       void FD_SET(int fd, fd_set *set);
       void FD_ZERO(fd_set *set);

select系统调用是用来让我们的程序监视多个文件句柄(file descrīptor)的状态变化的。通过select()系统调用来监视多个文件描述符的数组，当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位，使得进程可以获得这些文件描述符从而进行后续的读写操作。

select系统调用是用来让我们的程序监视多个文件描述符的状态变化的。程序会停在select这里等待，直到被监视的文件描述符有某一个或多个发生了状态改变。

select()的机制中提供一fd_set的数据结构，实际上是一long类型的数组，每一个数组元素都能与一打开的文件句柄建立联系，建立联系的工作由程序员完成，当调用select()时，由内核根据IO状态修改fd_set的内容，由此来通知执行了select()的进程哪些Socket或文件可读可写。

当某些描述符可以读写之后，select返回数据（没有数据读写时，select也会返回，因为select是同步）时就扫描一遍描述符fd_set来查询那些有数据请求的描述符，并进行处理。时间复杂度为O(n)

因此性能比那些阻塞的多进程或者多线程模型性能提高不少，但是仍然不够。因为select有很多限制

1. 最大并发数限制，因为一个进程所打开的FD（文件描述符）是有限制的，由FD_SETSIZE设置（可以查看深入解析为何select最多只能监听1024个），默认值是1024/2048，因此Select模型的最大并发数就被相应限制了。用户可以自己修改FD_SETSIZE，然后重新编译，但是其实，并不推荐这么做

   linux 下 fd_set 是个 1024 位的位图，每个位代表一个 fd 的值，返回后需要扫描位图，这也是效率低的原因。性能问题且不提，正确性问题则更值得重视。

   因为这是一个 1024 位的位图，因此当进程内的 fd 值 >= 1024 时，就会越界，可能会造成崩溃。对于服务器程序，fd >= 1024 很容易达到，只要连接数 + 打开的文件数足够大即可发生。

   include/linux/posix_types.h:
   
   #define __FD_SETSIZE         1024
   

2. 效率问题，select每次调用都会线性扫描全部的FD集合，这样效率就会呈现线性下降，把FD_SETSIZE改大的后果就是，大家都慢慢来，什么？都超时了？？！！

3. 内核/用户空间 内存拷贝问题，如何让内核把FD消息通知给用户空间呢？在这个问题上select采取了内存拷贝方法。

2.3 poll模型

poll的实现和select非常相似，只是描述fd集合的方式不同，poll使用pollfd结构而不是select的fd_set结构，其他的都差不多。

他通过注册一堆事件组，当有事件请求时返回，然后仍然需要轮询一遍pollfd才能知道查找到对应的文件描述符，数据也需要在内核空间和用户空间来回拷贝。时间复杂度为O(n)

因此他只解决了select的问题1，但是问题2，3仍然得不带解决。

3 epoll模型

3.1 epoll的性能提升

把其他模型逐个批判了一下，再来看看Epoll的改进之处吧，其实把select的缺点反过来那就是Epoll的优点了。

1. epoll没有最大并发连接的限制，上限是最大可以打开文件的数目，这个数字一般远大于2048, 一般来说这个数目和系统内存关系很大，具体数目可以cat /proc/sys/fs/file-max察看。

2. 效率提升，Epoll最大的优点就在于它只管你“活跃”的连接，而跟连接总数无关，因此在实际的网络环境中，Epoll的效率就会远远高于select和poll。

3. 内存拷贝，Epoll在这点上使用了“共享内存”，这个内存拷贝也省略了。

3.2 如何解决上述的3个缺点

epoll既然是对select和poll的改进，就避免上述的三个缺点。那epoll都是怎么解决的呢？

在此之前，我们先看一下epoll和select和poll的调用接口上的不同，select和poll都只提供了一个函数——select或者poll函数。

而epoll提供了三个函数，epoll_create,epoll_ctl和epoll_wait，

• epoll_create是创建一个epoll句柄；

• epoll_ctl是注册要监听的事件类型；

• epoll_wait则是等待事件的产生。

3.2.1 支持一个进程打开大数 目的socket描述符(FD)

对于第一个缺点并发数目限制

epoll没有这个限制，它所支持的FD上限是最大可以打开文件的数目，这个数字一般远大于2048,举个例子,在1GB内存的机器上大约是10万左右，具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。

select 最不能忍受的是一个进程所打开的FD是有一定限制的，由FD_SETSIZE设置，默认值是2048。对于那些需要支持的上万连接数目的IM服务器来说显 然太少了。这时候你一是可以选择修改这个宏然后重新编译内核，不过资料也同时指出这样会带来网络效率的下降，二是可以选择多进程的解决方案(传统的 Apache方案)，不过虽然linux上面创建进程的代价比较小，但仍旧是不可忽视的，加上进程间数据同步远比不上线程间同步的高效，所以也不是一种完 美的方案。不过 epoll则没有这个限制，它所支持的FD上限是最大可以打开文件的数目，这个数字一般远大于2048,举个例子,在1GB内存的机器上大约是10万左 右，具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。

3.2.2 IO效率不随FD数目增加而线性下降

对于第二个缺点轮询描述符的线性复杂度

epoll的解决方案不像select或poll一样每次都把current轮流加入fd对应的设备等待队列中，而只在epoll_ctl时把current挂一遍（这一遍必不可少）并为每个fd指定一个回调函数，当设备就绪，唤醒等待队列上的等待者时，就会调用这个回调函数，而这个回调函数会把就绪的fd加入一个就绪链表）。epoll_wait的工作实际上就是在这个就绪链表中查看有没有就绪的f

传统的select/poll另一个致命弱点就是当你拥有一个很大的socket集合，不过由于网络延时，任一时间只有部分的socket是”活跃”的， 但是select/poll每次调用都会线性扫描全部的集合，导致效率呈现线性下降。但是epoll不存在这个问题，它只会对”活跃”的socket进行 操作—这是因为在内核实现中epoll是根据每个fd上面的callback函数实现的。那么，只有”活跃”的socket才会主动的去调用 callback函数，其他idle状态socket则不会，在这点上，epoll实现了一个”伪”AIO，因为这时候推动力在os内核。在一些 benchmark中，如果所有的socket基本上都是活跃的—比如一个高速LAN环境，epoll并不比select/poll有什么效率，相 反，如果过多使用epoll_ctl,效率相比还有稍微的下降。但是一旦使用idle connections模拟WAN环境,epoll的效率就远在select/poll之上了。

3.2.3 使用mmap加速内核 与用户空间的消息传递

对于第三缺点数据在内核空间和用户空间的拷贝

epoll的解决方案在epoll_ctl函数中。每次注册新的事件到epoll句柄中时（在epoll_ctl中指定EPOLL_CTL_ADD），会把所有的fd拷贝进内核，而不是在epoll_wait的时候重复拷贝。epoll保证了每个fd在整个过程中只会拷贝一次。

这点实际上涉及到epoll的具体实现了。无论是select,poll还是epoll都需要内核把FD消息通知给用户空间，如何避免不必要的内存拷贝就 很重要，在这点上，epoll是通过内核于用户空间mmap同一块内存实现的。而如果你想我一样从2.5内核就关注epoll的话，一定不会忘记手工 mmap这一步的。

3.3 总结

1. select，poll实现需要自己不断轮询所有fd集合，直到设备就绪，期间可能要睡眠和唤醒多次交替。而epoll其实也需要调用epoll_wait不断轮询就绪链表，期间也可能多次睡眠和唤醒交替，但是它是设备就绪时，调用回调函数，把就绪fd放入就绪链表中，并唤醒在epoll_wait中进入睡眠的进程。虽然都要睡眠和交替，但是select和poll在“醒着”的时候要遍历整个fd集合，而epoll在“醒着”的时候只要判断一下就绪链表是否为空就行了，这节省了大量的CPU时间。这就是回调机制带来的性能提升。

2. select，poll每次调用都要把fd集合从用户态往内核态拷贝一次，并且要把current往设备等待队列中挂一次，而epoll只要一次拷贝，而且把current往等待队列上挂也只挂一次（在epoll_wait的开始，注意这里的等待队列并不是设备等待队列，只是一个epoll内部定义的等待队列）。这也能节省不少的开销。

4 Epoll的使用

4.1 epoll关键数据结构

前面提到Epoll速度快和其数据结构密不可分，其关键数据结构就是：

structepoll_event {

    __uint32_t events;      // Epoll events

    epoll_data_t data;      // User datavariable

};

typedef union epoll_data {

    void *ptr;

   int fd;

    __uint32_t u32;

    __uint64_t u64;

} epoll_data_t;

可见epoll_data是一个union结构体,借助于它应用程序可以保存很多类型的信息:fd、指针等等。有了它，应用程序就可以直接定位目标了。

4.2 使用Epoll

首先回忆一下select模型，当有I/O事件到来时，select通知应用程序有事件到了快去处理，而应用程序必须轮询所有的FD集合，测试每个FD是否有事件发生，并处理事件；代码像下面这样：
Epoll的高效和其数据结构的设计是密不可分的，这个下面就会提到。

首先回忆一下select模型，当有I/O事件到来时，select通知应用程序有事件到了快去处理，而应用程序必须轮询所有的FD集合，测试每个FD是否有事件发生，并处理事件；

代码像下面这样：

int res = select(maxfd+1, &readfds, NULL, NULL, 120);
if(res > 0)
{

    for(int i = 0; i < MAX_CONNECTION; i++)
    {
        if(FD_ISSET(allConnection[i],&readfds))
        {
            handleEvent(allConnection[i]);
        }
    }
}
// if(res == 0) handle timeout, res < 0 handle error

epoll不仅会告诉应用程序有I/0事件到来，还会告诉应用程序相关的信息，这些信息是应用程序填充的，因此根据这些信息应用程序就能直接定位到事件，而不必遍历整个FD集合。

intres = epoll_wait(epfd, events, 20, 120);

for(int i = 0; i < res;i++)
{
    handleEvent(events[n]);
}

首先通过create_epoll(int maxfds)来创建一个epoll的句柄，其中maxfds为你epoll所支持的最大句柄数。这个函数会返回一个新的epoll句柄，之后的所有操作 将通过这个句柄来进行操作。在用完之后，记得用close()来关闭这个创建出来的epoll句柄。之后在你的网络主循环里面，每一帧的调用epoll_wait(int epfd, epoll_event events, int max events, int timeout)来查询所有的网络接口，看哪一个可以读，哪一个可以写了。基本的语法为：

nfds = epoll_wait(kdpfd, events, maxevents, -1);

其中kdpfd为用epoll_create创建之后的句柄，events是一个 epoll_event*的指针，当epoll_wait这个函数操作成功之后，epoll_events里面将储存所有的读写事件。 max_events是当前需要监听的所有socket句柄数。最后一个timeout是 epoll_wait的超时，为0的时候表示马上返回，为-1的时候表示一直等下去，直到有事件范围，为任意正整数的时候表示等这么长的时间，如果一直没 有事件，则范围。一般如果网络主循环是单独的线程的话，可以用-1来等，这样可以保证一些效率，如果是和主逻辑在同一个线程的话，则可以用0来保证主循环 的效率。
既然epoll相比select这么好，那么用起来如何呢？会不会很繁琐啊…先看看下面的三个函数吧，就知道epoll的易用了。

intepoll_create(int size);

生成一个Epoll专用的文件描述符，其实是申请一个内核空间，用来存放你想关注的socket fd上是否发生以及发生了什么事件。size就是你在这个Epoll fd上能关注的最大socket fd数，大小自定，只要内存足够。

int epoll_ctl(int epfd, intop, int fd, structepoll_event *event);

控制某个Epoll文件描述符上的事件：注册、修改、删除。其中参数epfd是epoll_create()创建Epoll专用的文件描述符。相对于select模型中的FD_SET和FD_CLR宏。

int epoll_wait(int epfd,structepoll_event * events,int maxevents,int timeout);

等待I/O事件的发生，返回发生事件数；

功能类似与select函数

参数说明：

4.3 epoll的工作模式

4.3.1 LT和ET模式

令人高兴的是，2.6内核的epoll比其2.5开发版本的/dev/epoll简洁了许多，所以，大部分情况下，强大的东西往往是简单的。

唯一有点麻烦

epoll对文件描述符的操作有2种模式: LT和ET

LT(level triggered)是缺省的工作方式，并且同时支持block和no-block socket.在这种做法中，内核告诉你一个文件描述符是否就绪了，然后你可以对这个就绪的fd进行IO操作。如果你不作任何操作，内核还是会继续通知你的，所以，这种模式编程出错误可能性要小一点。传统的select/poll都是这种模型的代表. 这种模式相当于一个效率高的poll

对于采用LT模式工作的文件描述符, 当epoll_wait检测到其上有事件发生并将此事件通知应用程序后, 应用程序可以不用立即处理该事件. 这样，　当应用程序下次调用epoll_wait时, epoll_wait还会再次向应用程序通告此事件,直到该事件被处理.

ET (edge-triggered)是高速工作方式，只支持no-block socket。在这种模式下，当描述符从未就绪变为就绪时，内核通过epoll告诉你。然后它会假设你知道文件描述符已经就绪，并且不会再为那个文件描述 符发送更多的就绪通知，直到你做了某些操作导致那个文件描述符不再为就绪状态了(比如，你在发送，接收或者接收请求，或者发送接收的数据少于一定量时导致 了一个EWOULDBLOCK 错误）。但是请注意，如果一直不对这个fd作IO操作(从而导致它再次变成未就绪)，内核不会发送更多的通知(only once),不过在TCP协议中，ET模式的加速效用仍需要更多的benchmark确认。

而当往epoll内核事件表中注册一个文件描述符上的EPOLLET事件时, epoll将以ET模式来操作该文件描述符, ET模式是epoll的高效工作模式. 对于采用ET模式工作的文件描述符, 当epoll_wait检测到其上有事件发生并将此事件通知应用程序后, 应用程序必须立即处理该事件, 因为后续的epoll_wait调用将不再向应用程序通知该事件.

可见, ET模式在很大程序上降低了同一个epoll事件被重复触发的次数, 因此效率比LT模式高.

注意

每个使用ET模式的文件描述符都应该是非阻塞的.

如果文件描述符是阻塞的, 那么读写操作将会因为没有后续的事件而一直处于阻塞状态(饥渴状态)

4.3.2 示例代码

//  代码清单9-3 LT和ET模式
//  参见Linux高性能服务器编程
#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


#define DEFAULT_SERVER_PORT 6666



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

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(Level Trigger, 电平触发)
 *  相当于一个效率较高的poll
 *  对于采用LT工作模式的文件描述符
 *  当epoll_wait检测到其上时间发生并将此事件通知应用程序后
 *  应用程序可以不用立即处理
 *  这样下次调用时, 还会再次向应用程序通知此时间
 */
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 )
        {
            printf( "LT-event trigger oncen" );
            memset( buf, '', BUFFER_SIZE );
            int ret = recv( sockfd, buf, BUFFER_SIZE-1, 0 );
            if( ret <= 0 )
            {
                close( sockfd );
                continue;
            }
            printf( "get %d bytes of content: %sn", ret, buf );
        }
        else
        {
            printf( "something else happened n" );
        }
    }
}

/*  ET模式的工作流程
 *  对于采用ET模式的文件描述符号
 *  当epoll_wait检测到其上由事件发生并将此时间通知应用程序后
 *  应用程序应该立即处理该事件
 *  因为后续的epoll_wait不会再向应用程序通知这一事件
 */
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( "ET-event trigger only oncen" );
            while( 1 )
            {
                memset( buf, '', BUFFER_SIZE );
                int ret = recv( sockfd, buf, BUFFER_SIZE-1, 0 );
                if( ret < 0 )
                {
                    if( ( errno == EAGAIN ) || ( errno == EWOULDBLOCK ) )
                    {
                        printf( "read latern" );
                        break;
                    }
                    close( sockfd );
                    break;
                }
                else if( ret == 0 )
                {
                    close( sockfd );
                }
                else
                {
                    printf( "get %d bytes of content: %sn", ret, buf );
                }
            }
        }
        else
        {
            printf( "something else happened n" );
        }
    }
}

int main( int argc, char* argv[] )
{
    int     port = DEFAULT_SERVER_PORT;
    char    *ip = NULL;

    if( argc > 3 )
    {
        printf( "usage: %s [port_number [ip_address]]n", basename( argv[0] ) );
        return 1;
    }
    else if (argc == 2)
    {
        port = atoi( argv[1] );
    }
    else if(argc == 3)
    {
        port = atoi( argv[1] );
        ip = argv[2];
    }

    int ret = 0;
    struct sockaddr_in address;
    bzero( &address, sizeof( address ) );
    address.sin_family = AF_INET;
    if(ip != NULL)
    {
        inet_pton( AF_INET, ip, &address.sin_addr );
    }
    else
    {
        address.sin_addr.s_addr = AF_INET;
    }
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    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 failuren" );
            break;
        }

        //lt( events, ret, epollfd, listenfd );
        et( events, ret, epollfd, listenfd );
    }

    close( listenfd );
    return 0;
}

4.4 EPOLLONESHOT事件

4.4.1 EPOLLONESHOT事件简介

在前面说过，epoll有两种触发的方式即LT（水平触发）和ET（边缘触发）两种，在前者，只要存在着事件就会不断的触发，直到处理完成，而后者只触发一次相同事件或者说只在从非触发到触发两个状态转换的时候儿才触发。

这会出现一种情况，就是即使我们使用ET模式, 一个socket上的某个事件还是可能被触发多次, 这在并发程序中会引起一个问题.

比如, 一个线程(或者进程)在读取完某个socket上的数据后开始处理这些数据, 而在处理数据的过程中该socket上又有新数据可读(EPOLLIN再次被触发), 此时如果应用程序调度另外一个线程来读取这些数据, 就会出现两个线程同时操作一个socket的局面, 这会使程序的健壮性大降低而编程的复杂度大大增加. 这显然不是我们所期望的.

解决这种现象有两种方法

• 一种是在单独的线程或进程里解析数据，也就是说，接收数据的线程接收到数据后立刻将数据转移至另外的线程

*第二种方法就是使用EPOLLONESHOT事件. 对于注册了EPOLLONESHOT事件的文件描述符, 操作系统最多触发其上注册的一个可读, 可写或者异常事件, 且只触发一次, 除非我们使用epoll_ctl和函数重置该文件描述符上注册的EPOLLONESHOT事件.

这样, 当一个线程在处理某个socket的时候, 其他线程就不可能有机会操作该socket

但是反过来思考, 注册了EPOLLONESHOT事件的socket一旦被某个线程处理完毕, 该线程就有责任立即重置这个socket上的EPOLLONESHOT事件, 以确保这个socket下一次可读的时候, 其EPOLLIN事件能被再次触发, 进而让其他线程有机会处理这个socket.

4.4.2 EPOLLONESHOT事件示例程序

//  代码清单9-4 使用EPOLLONESHOT事件
//  参见Linux高性能服务器编程
#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

#define DEFAULT_SERVER_PORT 6666


typedef struct fds_pthread_args
{
   int epollfd;
   int sockfd;
}fds_pthread_args;

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;
    /*  对于注册了EPOLLONESHOT时间的文件描述符
     *  操作系统最多触发其上注册的一个可读, 可写或者异常事件
     *  且只触发一次  */
    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* args )
{
    fds_pthread_args *_args = (fds_pthread_args *)args;

    int sockfd = _args->sockfd;
    int epollfd = _args->epollfd;
    printf( "start new thread to receive data on fd: %dn", sockfd );

    char buf[ BUFFER_SIZE ];
    memset( buf, '', BUFFER_SIZE );

    /*  循环读取sockfd上的数据, 直到遇见EAGAIN错误  */
    while( 1 )
    {
        int ret = recv( sockfd, buf, BUFFER_SIZE - 1, 0 );
        if( ret == 0 )
        {
            close( sockfd );
            printf( "foreiner closed the connectionn" );
            break;
        }
        else if( ret < 0 )
        {
            /*  首先我们看看recv的返回值：
             *  EAGAIN、EWOULDBLOCK、EINTR与非阻塞 长连接
             *  EWOULDBLOCK     用于非阻塞模式，不需要重新读或者写
             *  EINTR           指操作被中断唤醒，需要重新读/写
             *  在Linux环境下开发经常会碰到很多错误(设置errno)，
             *  其中EAGAIN是其中比较常见的一个错误(比如用在非阻塞操作中)
             *  从字面上来看, 是提示再试一次.
             *  这个错误经常出现在当应用程序进行一些非阻塞(non-blocking)操作
             *  (对文件或socket)的时候
             *  例如，以 O_NONBLOCK的标志打开文件/socket/FIFO,
             *  如果你连续做read操作而没有数据可读.
             *  此时程序不会阻塞起来等待数据准备就绪返回,
             *  read函数会返回一个错误EAGAIN，
             *  提示你的应用程序现在没有数据可读请稍后再试重新读数据,
             *  对非阻塞socket而言, EAGAIN不是一种错误。在VxWorks和Windows上,
             *  EAGAIN的名字叫做EWOULDBLOCK
             */
            if( errno == EAGAIN )
            {
                reset_oneshot( epollfd, sockfd );
                printf( "read latern" );
                break;
            }
        }
        else
        {
            printf( "get content: %sn", buf );
            /*  休眠5s, 模拟数据处理过程    */
            sleep( 5 );
        }
    }
    printf( "end thread receiving data on fd: %dn", sockfd );

    return NULL;
}

int main( int argc, char* argv[] )
{
    int port = DEFAULT_SERVER_PORT;
    char *ip = NULL;

    if( argc > 3)
    {
        printf( "usage: %s port_number ip_addressn", basename( argv[0] ) );
        return 1;
    }
    else if( argc == 2 )
    {
        port = atoi(argv[1]);
    }
    else if(argc == 3)
    {
        port = atoi(argv[1]);
        ip = argv[2];
    }

    int ret = 0;
    struct sockaddr_in address;
    bzero( &address, sizeof( address ) );
    address.sin_family = AF_INET;
    if(ip != NULL)
    {
        inet_pton( AF_INET, ip, &address.sin_addr );
    }
    else
    {
        address.sin_addr.s_addr = INADDR_ANY;
    }
    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 );

    /*   注意, 监听套接字listen上不能注册EPOLLONESHOT事件,
     *   否则应用程序只能处理一个客户端连接
     *   因为由于EPOLLONESHOT被设置
     *   后续的客户端连接请求将不再触发listenfd的EPOLLIN事件
     */
    addfd( epollfd, listenfd, false );

    while( 1 )
    {
        int ret = epoll_wait( epollfd, events, MAX_EVENT_NUMBER, -1 );
        if ( ret < 0 )
        {
            printf( "epoll failuren" );
            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 );

                /*  对每个非监听文件描述符都注册EPOLLONEHOT事件  */
                addfd( epollfd, connfd, true );
            }
            else if ( events[i].events & EPOLLIN )
            {
                pthread_t           thread;
                fds_pthread_args    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;
}

5 参考

epoll只有epoll_create,epoll_ctl,epoll_wait 3个系统调用

具体用法请参考http://www.xmailserver.org/linux-patches/nio-improve.html

在http://www.kegel.com/rn/也有一个完整的例子

Leader/follower模式线程 pool实现，以及和epoll的配合。

Epoll的高效和其数据结构的设计是密不可分的，这个下面就会提到。

我读过最好的Epoll模型讲解

Epoll模型详解

通过完整示例来理解如何使用 epol

epoll 使用详解

最后

以上就是潇洒白开水为你收集整理的Linux下套接字详解（十）---epoll模式下的IO多路复用服务器1 epoll模型简介2 常用模型的缺点3 epoll模型4 Epoll的使用5 参考的全部内容，希望文章能够帮你解决Linux下套接字详解（十）---epoll模式下的IO多路复用服务器1 epoll模型简介2 常用模型的缺点3 epoll模型4 Epoll的使用5 参考所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。