Linux2.6内核中epoll用法详解

epoll是linux2.6内核中才有的机制，其他版本内核中是没有的，是Linux2.6内核引入的多路复用IO的一种方式，用于提高网络IO性能的方法。在linux网络编程中，很长一段时间都是采用select来实现多事件触发处理的。Select存在如下几个方面的问题：一是每次调用时要重复地从用户态读入参数，二是每次调用时要重复地扫描文件描述符，三是每次在调用开始时，要把当前进程放入各个文件描述符的等待队列。在调用结束后，又把进程从各个等待队列中删除。Select采用轮询的方式来处理事件触发，当随着监听socket的文件描述符fd的数量增加时，轮询的时间也就越长，造成效率低下。而且linux/posix_types.h中有#define
__FD_SETSIZE 1024(也有说2048的)的定义，也就是说linux select能监听的最大fd数目是1024个，虽然能通过内核修改此参数，但这是治标不治本。
    epoll的出现可以有效的解决select效率低下的问题，epoll把参数拷贝到内核态，在每次轮询时不会重复拷贝。epoll有ET和LT两种工作模式，ET是高速模式只能以非阻塞方式进行，LT相当于快速的select，可以才有阻塞和非阻塞两种方式，epoll通过把操作拆分为epoll_create,epoll_ctl,epoll_wait三个步骤避免重复地遍历要监视的文件描述符。
EPOLL事件有两种模型：

Edge Triggered (ET)

Level Triggered (LT)

假如有这样一个例子：

1. 我们已经把一个用来从管道中读取数据的文件句柄(RFD)添加到epoll描述符

2. 这个时候从管道的另一端被写入了2KB的数据

3. 调用epoll_wait(2)，并且它会返回RFD，说明它已经准备好读取操作

4. 然后我们读取了1KB的数据

5. 调用epoll_wait(2)......

Edge Triggered 工作模式：
      如果我们在第1步将RFD添加到epoll描述符的时候使用了EPOLLET标志，那么在第5步调用epoll_wait(2)之后将有可能会挂起，因为剩余的数据还存在于文件的输入缓冲区内，而且数据发出端还在等待一个针对已经发出数据的反馈信息。只有在监视的文件句柄上发生了某个事件的时候 ET 工作模式才会汇报事件。因此在第5步的时候，调用者可能会放弃等待仍在存在于文件输入缓冲区内的剩余数据。在上面的例子中，会有一个事件产生在RFD句柄上，因为在第2步执行了一个写操作，然后，事件将会在第3步被销毁。因为第4步的读取操作没有读空文件输入缓冲区内的数据，因此我们在第5步调用epoll_wait(2)完成后，是否挂起是不确定的。epoll工作在ET模式的时候，必须使用非阻塞套接口，以避免由于一个文件句柄的阻塞读/阻塞写操作把处理多个文件描述符的任务饿死。最好以下面的方式调用ET模式的epoll接口，在后面会介绍避免可能的缺陷。
       i    基于非阻塞文件句柄
       ii   只有当read(2)或者write(2)返回EAGAIN时才需要挂起，等待。但这并不是说每次read()时都需要循环读，直到读到产生一个EAGAIN才认为此次事件处理完成，当read()返回的读到的数据长度小于请求的数据长度时，就可以确定此时缓冲中已没有数据了，也就可以认为此事读事件已处理完成。

      Level Triggered 工作模式
      相反的，以LT方式调用epoll接口的时候，它就相当于一个速度比较快的poll/select，在poll能用的地方epoll都可以用，因为他们具有同样的职能。即使使用ET模式的epoll，在收到多个数据包的时候仍然会产生多个事件。调用者可以设定EPOLLONESHOT标志，在 epoll_wait收到事件后epoll会与事件关联的文件句柄从epoll描述符中禁止掉。因此当EPOLLONESHOT设定后，使用带有 EPOLL_CTL_MOD标志的epoll_ctl处理文件句柄就成为调用者必须作的事情。

      以上是man手册对epoll中两种模式的简要介绍，这里有必要对两种模式进行详细的介绍：

LT是缺省的工作方式，并且同时支持block和no-block socket；在这种做法中，内核会告诉调用者一个文件描述符是否就绪了，然后调用者可以对这个就绪的fd进行IO操作。如果你不作任何操作，内核还是会继续通知调用者的，所以，这种模式编程出错误可能性要小一点。传统的select/poll都是这种模型的代表。LT模式跟select有一样的语义。就是如果可读就触发。比如某管道原来为空，如果有一个进程写入2k数据，就会触发。如果处理进程读取1k数据，下次轮询时继续触发。该模式下，默认不可读，只有epoll通知可读才是可读，否则不可读。

ET是高速工作方式，只支持no-block socket。在这种模式下，当描述符从未就绪变为就绪时，内核通过epoll告诉调用者，然后它会假设调用者知道文件描述符已经就绪，并且不会再为那个文件描述 符发送更多的就绪通知，直到调用者做了某些操作导致那个文件描述符不再为就绪状态了。但是请注意，如果一直不对这个fd作IO操作(从而导致它再次变成未就绪)，内核不会发送更多的通知。该模式与select有不同的语义，只有当从不可读变为可读时才触发。上面那种情况，还有1k可读，所以不会触发，当继续读，直到返回EAGAIN时，变为不可读，如果再次变为可读就触发。默认可读，调用者可以随便读，直到发生EAGAIN。可读时读和不读，怎么读都由调用者自己决定，中间epoll不管。EAGAIN后不可读了，等到再次可读，epoll会再通知一次。理解ET模式最重要的就是理解状态的变化，对于监听可读事件时,如果是socket是监听socket，那么当有新的主动连接到来为状态发生变化；对一般的socket而言，协议栈中相应的缓冲区有新的数据为状态发生变化。但是，如果在一个时间同时接收了N个连接(N>1)，但是监听socket只accept了一个连接，那么其它未 accept的连接将不会在ET模式下给监听socket发出通知，此时状态不发生变化；对于一般的socket，如果对应的缓冲区本身已经有了N字节的数据，而只取出了小于N字节的数据，那么残存的数据不会造成状态发生变化。

epoll的LT和ET的区别
LT：水平触发，效率会低于ET触发，尤其在大并发，大流量的情况下。但是LT对代码编写要求比较低，不容易出现问题。LT模式服务编写上的表现是：只要有数据没有被获取，内核就不断通知你，因此不用担心事件丢失的情况。
ET：边缘触发，效率非常高，在并发，大流量的情况下，会比LT少很多epoll的系统调用，因此效率高。但是对编程要求高，需要细致的处理每个请求，否则容易发生丢失事件的情况。

下面举一个列子来说明LT和ET的区别（都是非阻塞模式，阻塞就不说了，效率太低）：

采用LT模式下， 如果accept调用有返回就可以马上建立当前这个连接了，再epoll_wait等待下次通知，和select一样。

但是对于ET而言，如果accpet调用有返回，除了建立当前这个连接外，不能马上就epoll_wait还需要继续循环accpet，直到返回-1，且errno==EAGAIN，TAF里面的示例代码：

if(ev.events & EPOLLIN)

{

    do

    {

        struct sockaddr_in stSockAddr;

        socklen_t iSockAddrSize = sizeof(sockaddr_in);

        TC_Socket cs;

        cs.setOwner(false);

        //接收连接

        TC_Socket s;

        s.init(fd, false, AF_INET);

        int iRetCode = s.accept(cs, (struct sockaddr *) &stSockAddr,
iSockAddrSize);

        if (iRetCode > 0)

        {

            ...建立连接

        }

        else

        {

            //直到发生EAGAIN才不继续accept

            if(errno == EAGAIN)

            {

                break;

            }

        }

    }while(true);

}
ET模式仅当状态发生变化的时候才获得通知,这里所谓的状态的变化并不包括缓冲区中还有未处理的数据,也就是说,如果要采用ET模式, 需要一直read/write直到出错为止,很多人反映为什么采用ET模式只接收了一部分数据就再也得不到通知了,大多因为这样;而LT模式是只要有数据 没有处理就会一直通知下去的.

同样，recv/send等函数， 都需要到errno==EAGAIN

从本质上讲：与LT相比，ET模型是通过减少系统调用来达到提高并行效率的。

epoll的接口非 常简单，一共就三个函数：
1. int epoll_create(int size);

创建一个epoll的句柄，size用来告诉内 核这个监听的数目一共有多大。这个参数不同于select()中的第一个参数，给出最大监听的fd+1的值。需要注意的是，当创建好epoll句柄后，它 就是会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close() 关闭，否则可能导致fd被耗尽。

2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll的事件注册函数，它不同与select()是在监听事件时告诉内核要监听什么 类型的事件，而是在这里先注册要监听的事件类型。第一个参数是epoll_create()的返回值，第二个参数表示动作，用三个宏来表示：

EPOLL_CTL_ADD： 注册新的fd到epfd中；

EPOLL_CTL_MOD：修改已经注册的fd的监听事件；

EPOLL_CTL_DEL：从epfd中删除 一个fd；

第三个参数是需要监听的fd，第四个参数是告诉内核需要监听什么事，struct epoll_event结构如下：

typedef union epoll_data {

void *ptr;

int fd;

__uint32_t u32;

__uint64_t u64;

} epoll_data_t;

struct epoll_event {

__uint32_t events; /* Epoll events */

epoll_data_t data; /* User data variable */

};

events可以是以下几个宏的集合：

EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；

EPOLLOUT：表示对应的文件描述符可以写；

EPOLLPRI： 表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；

EPOLLERR：表示对应的文件描述符发生错误；

EPOLLHUP： 表示对应的文件描述符被挂断；

EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。

EPOLLONESHOT：只监听一次事件，当监听完 这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

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

下面给出一个完整的服务器端例子：
#include <iostream> 
#include <sys/socket.h> 
#include <sys/epoll.h> 
#include <netinet/in.h> 
#include <arpa/inet.h> 
#include <fcntl.h> 
#include <unistd.h> 
#include <stdio.h> 
#include <errno.h> 

using namespace std; 

#define MAXLINE 5 
#define OPEN_MAX 100 
#define LISTENQ 20 
#define SERV_PORT 5000 
#define INFTIM 1000 

void setnonblocking(int sock) 
{ 
int opts; 
opts=fcntl(sock,F_GETFL); 
if(opts<0) 
{ 
perror("fcntl(sock,GETFL)"); 
exit(1); 
} 
opts = opts|O_NONBLOCK; 
if(fcntl(sock,F_SETFL,opts)<0) 
{ 
perror("fcntl(sock,SETFL,opts)"); 
exit(1); 
} 
} 

int main(int argc, char* argv[]) 
{ 
int i, maxi, listenfd, connfd, sockfd,epfd,nfds, portnumber; 
ssize_t n; 
char line[MAXLINE]; 
socklen_t clilen; 

if ( 2 == argc ) 
{ 
if( (portnumber = atoi(argv[1])) < 0 ) 
{ 
fprintf(stderr,"Usage:%s portnumber\a\n",argv[0]); 
return 1; 
} 
} 
else 
{ 
fprintf(stderr,"Usage:%s portnumber\a\n",argv[0]); 
return 1; 
} 

//声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要 处理的事件 
struct epoll_event ev,events[20]; 
//生成用于处理accept的epoll专用的文件描述符 
epfd=epoll_create(256); 
struct sockaddr_in clientaddr; 
struct sockaddr_in serveraddr; 
listenfd = socket(AF_INET, SOCK_STREAM, 0); 
//把socket设 置为非阻塞方式 
//setnonblocking(listenfd); 
//设置与要处理的事件相关的文件描述符 
ev.data.fd=listenfd; 
//设置要处理的事件类型 
ev.events=EPOLLIN|EPOLLET; 
//ev.events=EPOLLIN; 

//注册epoll事件 
epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev); 
bzero(&serveraddr, sizeof(serveraddr)); 
serveraddr.sin_family = AF_INET; 
char *local_addr="127.0.0.1"; 
inet_aton(local_addr,&(serveraddr.sin_addr));//htons(portnumber); 

serveraddr.sin_port=htons(portnumber); 
bind(listenfd,(sockaddr *)&serveraddr, sizeof(serveraddr)); 
listen(listenfd, LISTENQ); 
maxi = 0; 
for ( ; ; ) { 
//等待epoll事件的发生 
nfds=epoll_wait(epfd,events,20,500); 
//处理所发生的所有事件 
for(i=0;i<nfds;++i) 
{ 
if(events[i].data.fd==listenfd)
//如果新监测到一个SOCKET用户连接到了绑定的SOCKET端口，建立新的 连接。 
{ 
connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen); 
if(connfd<0){ 
perror("connfd<0"); 
exit(1); 
} 
//setnonblocking(connfd); 
char *str = inet_ntoa(clientaddr.sin_addr); 
cout << "accapt a connection from " << str << endl; 
//设置用于读操作的文件描述符 
ev.data.fd=connfd; 
//设置用于注测的读操作事件 
ev.events=EPOLLIN|EPOLLET; 
//ev.events=EPOLLIN; 
//注册ev 
epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev); 
} 
else if(events[i].events&EPOLLIN)//如果是已经连 接的用户，并且收到数据，那么进行读入。 
{ 
cout << "EPOLLIN" << endl; 
if ( (sockfd = events[i].data.fd) < 0) 
continue; 
if ( (n = read(sockfd, line, MAXLINE)) < 0) { 
if (errno == ECONNRESET) { 
close(sockfd); 
events[i].data.fd = -1; 
} else 
std::cout<<"readline error"<<std::endl; 
} else if (n == 0) { 
close(sockfd); 
events[i].data.fd = -1; 
} 
line
 = '\0'; 
cout << "read " << line << endl; 
//设置用于写操作的文件描述符 
ev.data.fd=sockfd; 
//设置用于注测的写操作事件 
ev.events=EPOLLOUT|EPOLLET; 
//修改sockfd上要处理的事件为EPOLLOUT 
//epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); 
} 
else if(events[i].events&EPOLLOUT) // 如果有数据发送 
{ 
sockfd = events[i].data.fd; 
write(sockfd, line, n); 
//设置用于读操作的文件描述符 
ev.data.fd=sockfd; 
//设置用于注测的读操作事件 
ev.events=EPOLLIN|EPOLLET; 
//修改sockfd上要处理的事件为EPOLIN 
epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); 
} 
} 
} 
return 0; 
}