基本I/O模型与Epoll简介

5种基本的I/O模型：1)阻塞I/O ;2)非阻塞I/O; 3)I/O复用（select和poll）;4)信号驱动I/O（SIGIO）;5)异步I/O（POSIX.1的aio_系列函数）。

操作系统中一个输入操作一般有两个不同的阶段：

第一：等待数据准备好。第二：从内核到进程拷贝数据。
对于一个sockt上的输入操作，第一步一般是等待数据到达网络，当分组到达时，它被拷贝到内核中的某个缓冲区，第二步是将数据从内核缓冲区拷贝到应用程序缓冲区。

一、阻塞I/O模型

请求无法立即完成则保持阻塞。

整个过程分为两个阶段:

阶段一是等待数据就绪,网络I/O的情况就是等待远端数据陆续抵达；磁盘I/O的情况就是等待磁盘数据从磁盘上读取到内核态内存中。

阶段二是数据拷贝,出于系统安全,用户态的程序没有权限直接读取内核态内存,因此内核负责把内核态内存中的数据拷贝一份到用户态内存中。

理解这两个阶段非常重要,后续I/O模型的演变都是针对这两个阶段进行不同改造。

最流行的I/O模型是阻塞I/O模型，缺省时，所有sockt都是阻塞的,这意味着当一个sockt调用不能立即完成时，进程进入睡眠状态，等待操作完成。

                       

 在图1中，进程调用recvfrom,此调用直到数据报到达且拷贝到应用缓冲区或是出错才返回。最常见的错误是系统调用被信号中断，我们所说进程阻塞的整段时间是指从调用recvfrom开始到它返回的这段时间，当进程返回成功指示时，应用进程开始处理数据报。

二、非阻塞I/O模型

请求不会阻塞, 而是立即返回错误码(EWOULDBLOCK) ，(通过调用fcntl,参数置为O_NONBLOCK)，阶段一频繁轮询的话,也很耗费CPU时间，这种方式对单个I/O请求意义不大,但给I/O多路复用铺平了道路。

当我们把一个sockt设置成非阻塞放式时，即通知内核：当请求的I/O操作非得让进程睡眠才能完成时，不要让进程睡眠，而应该返回一个错误。如图:

 

如图2所示，前3次调用recvfrom时仍无数据返回，因此内核立即返回一个EWOULDBLOCK错误。第4次调用recvfrom时，数据报已经准备好了，被拷贝到应用缓冲区，recvfrom返回成功指示，接着就是我们处理数据报。
    当一个应用进程像这样对一个非阻塞sockt循环调用recvfrom时，我们称此过程为轮询（polling）.应用进程连续不断的查询内核，看看某操作是否准备好,这对CPU是极大的浪费，但这种模型只是偶尔才遇到。

三、I/O多路复用模型

调用 select / poll  该方法由一个用户态线程负责轮询多个sockets,直到某个阶段一的数据就绪,再通知实际的用户线程执行阶段二的拷贝.

通过一个专职的用户态线程执行非阻塞I/O轮询,模拟实现了阶段一的异步化。

I/O复用能让一个或多个I/O条件满足（例如，输入已经准备好被读，或者描述字可以承接更多的输出）时，我们就被通知到。I/O复用由select和poll支持，较新的Posix.1g也支持(pselect)。
I/O复用典型地用在下列网络应用场合：
    1.当客户处理多个描述字时，必须使用。
    2.一个客户同时处理多个sockt.
    3.如果一个服务器既要处理监听sockt,又要处理连接sockt,一般也用到。
    4.如果一个服务器既要处理TCP,又要处理UDP,一般也用到。
    5.如果一个服务器要处理多个服务或多个协议（例如inetd守护进程），一般也用到。

    I/O复用并非限于网络编程，许多正是应用程序也需要使用这项技术。

有了I/O复用，我们就可以调用select或poll，在这两个系统调用中的某一个上阻塞，而不阻塞于真正的I/O系统调用。图3是I/O复用模型的一个小结。

我们阻塞于select调用，等待数据报socket可读,当select返回socket可读条件时，我们调用recvfrom将数据报拷贝到应用缓存区中。
    将图3与图1比较，似乎没有显示什么优越性，实际上因使用了select,要求2此系统调用而不是一次，好像变的还有点差，但是select的好处在于我们可以等待多个描述字准备好。

      

四、 信号驱动I/O模型 （不经常使用）

阶段一演变为异步,由内核托管,应用程序只需告知内核,当阶段一数据就绪时向应用程序发出 SIGIO信号，至此为止,前述4种模型的阶段二仍是处于block状态的。

首先我们允许sockt进行信号驱动 I/O，并通过系统调用sigaction安装一个信号处理程序。此系统调用立即返回，进程继续工作，它是非阻塞的。当数据报准备好被读时，就为该进程生成个SIGIO信号。我们随即可以在信号处理程序中调用recvfrom来读取数据报，并通知主循环数据已准备好被处理，也可以通知主循环，让它来处理数据报。

     无论我们如何处理SIGIO信号，这种模型的好处是当等待数据报到达时，可以不阻塞。主循环可以继续执行，只是等待信号处理程序的通知：或者数据报已准备好被处理，或者数据报已准备好被读取。

     上层应用建立SIGIO信号处理程序。当缓冲区有数据到来，内核会发送信号告诉上层应用App; 当上层应用App接收到信号后，调用recv函数，因缓冲区有数据，recv函数一般不会阻塞。但是这种用于模型用的比较少，属于典型的“拉模式(上层应用被动的去Linux内核空间中拉数据)”。即：上层应用App,需要调用recv函数把数据拉进来,会有时间延迟,我们无法避免在延迟时,又有新的信号的产生，这也是他的缺陷。

     

五、异步I/O模型

告知内核,当整个过程(包括阶段一和阶段二)全部完成时,通知应用程序来读数据.

 异步I/O模型是Posix.1的1993版本中的新内容。我们让内核启动操作，并在整个操作完成后（包括将数据报从内核拷贝到我们自己的缓冲区）通知我们。这种模型与信号驱动模型的主要区别在于：信号驱动I/O是有内核通知我们何时可以启动一个I/O操作，而异步I/O模型是由内核通知我们I/O操作何时完成。图5给出了一个例子

        

我们调用aio_red(Posix异步I/O函数以aio_或lio_开头)，给内核传递描述字、缓冲区指针、缓冲区大小（与red相同的3个参数）、文件偏移（与lseek类似），并高书内核当整个操作完成时如何通知我们。此系统调用立即返回，我们的进程不阻塞于等待I/O操作的完成。在此例子中，我们假设要求内核在操作完成时产生一个信号，此信号直到数据已拷贝到应用程序缓冲区才产生，这一点是于信号驱动I/O模型不同的，linux3.5已经有针对C的异步I/O API。

六、各种I/O模型的比较

    
 

图6 各类I/O模型的比较： 图6给出了上述5中I/O模型的比较。它表明：前4种模型的区别都在第1阶段，因为前4种模型的第2阶段基本相同：在数据从内核拷贝到调用者的缓冲区时，进程阻塞于recvfrom调用。然而异步I/O处理的两个阶段都不同于前4个模型。
 同步I/O与异步I/O

 Posix.1定义这两个术语如下：

 1.同步I/O操作引起请求进程阻塞，直到I/O操作完成。

 2.异步 I/O操作不引起请求进程阻塞。

   根据上述定义，我们的前四个I/O模型都是同步I/O模型，因为真正的I/O操作（recvfrom）阻塞进程，只有异步I/O模型与异步I/O的定义相符合

从理论上说，AIO似乎是最高效的，你的IO操作可以立即返回，然后等待os告诉你IO操作完成。但是一直以来，如何实现就没有一个完美的方 案。最著名的windows完成端口实现的AIO,实际上也是内部用线程池实现的罢了，最后的结果是IO有个线程池，你应用也需要一个线程池...... 很多文档其实已经指出了这带来的线程context-switch带来的代价。在linux 平台上，关于网络AIO一直是改动最多的地方，2.4的年代就有很多AIO内核patch,最著名的应该算是SGI那个。但是一直到2.6内核发布，网络 模块的AIO一直没有进入稳定内核版本(大部分都是使用用户线程模拟方法，在使用了NPTL的linux上面其实和windows的完成端口基本上差不多 了)。2.6内核所支持的AIO特指磁盘的AIO---支持io_submit(),io_getevents()以及对Direct IO的支持(就是绕过VFS系统buffer直接写硬盘，对于流服务器在内存平稳性上有相当帮助)。
所以，剩下的select模型基本上就是我们在linux上面的唯一选择，其实，如果加上no-block socket的配置，可以完成一个"伪"AIO的实现，只不过推动力在于你而不是os而已。不过传统的select/poll函数有着一些无法忍受的缺 点，所以改进一直是2.4-2.5开发版本内核的任务，包括/dev/poll，realtime signal等等。最终，Davide Libenzi开发的epoll进入2.6内核成为正式的解决方案。

七、I/O多路复用模型

Epoll 可是当前在 Linux 下开发大规模并发网络程序的热门人选， Epoll 在 Linux2.6 内核中正式引入，和 select 相似，其实都 I/O 多路复用技术而已 ，并没有什么神秘的。其实在 Linux 下设计并发网络程序，向来不缺少方法，比如典型的 Apache 模型（ Process Per Connection ，简称 PPC ）， TPC （ Thread Per Connection ）模型，以及 select 模型和 poll 模型，那为何还要再引入 Epoll 这个东东呢？那还是有得说说的 …

(一)  常用模型的缺点

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

1)     PPC/TPC 模型

这两种模型思想类似，就是让每一个到来的连接一边自己做事去，别再来烦我 。只是 PPC 是为它开了一个进程，而 TPC 开了一个线程。可是别烦我是有代价的，它要时间和空间啊，连接多了之后，那么多的进程 / 线程切换，这开销就上来了；因此这类模型能接受的最大连接数都不会高，一般在几百个左右。

2)     select 模型

a)      最大并发数限制，因为一个进程所打开的 FD （文件描述符）是有限制的，由 FD_SETSIZE 设置，默认值是 1024/2048 ，因此 Select 模型的最大并发数就被相应限制了。自己改改这个 FD_SETSIZE ？想法虽好，可是先看看下面吧.。

b)     效率问题，内核中实 现select是用轮询方法，即每次检测都会遍历所有FD_SET中的句柄，显然，select函数执行时间与FD_SET中的句柄个数有一个比例关系， 即select要检测的句柄数越多就会越费时。select 每次调用都会线性扫描全部的 FD 集合，这样效率就会呈现线性下降，把 FD_SETSIZE 改大的后果就是，大家都慢慢来，什么？都超时了。

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

总结为：1.连接数受限  2.查找配对速度慢 3.数据由内核拷贝到用户态

3)     poll 模型

基本上效率和 select 是相同的， select 缺点的 2 和 3 它都没有改掉。

(二)  Epoll 的提升

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

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

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

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

(三) Epoll 为什么高效

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

    

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

　　

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

int res = epoll_wait(epfd, events, 20, 120);  
for (int i = 0; i < res;i++)  
{  
   handleEvent(events[n]);  
}  

　

　(四)  Epoll 关键数据结构

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

  

struct epoll_event {  
    __uint32_t events;      // Epoll events  
    epoll_data_t data;      // User data variable  
};  
typedef union epoll_data {  
     void *ptr;  
     int fd;  
     __uint32_t u32;  
     __uint64_t u64;   
 } epoll_data_t;  

　　(五) 使用 Epoll

既然 Epoll 相比 select 这么好，那么用起来如何呢？会不会很繁琐啊 … 先看看下面的三个函数吧，就知道 Epoll 的易用了。 

int epoll_create(int size);  

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

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

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

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

等待 I/O 事件的发生；参数说明：

epfd: 由 epoll_create() 生成的 Epoll 专用的文件描述符；

epoll_event: 用于回传代处理事件的数组；

maxevents: 每次能处理的事件数；

timeout: 等待 I/O 事件发生的超时值；

返回发生事件数。

相对于 select 模型中的 select 函数

// a simple echo server using epoll in linux  
	// 2009-11-05       
	#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>  

	#include <iostream>  

	 using namespace std;  

	 #define MAX_EVENTS 500  

     struct myevent_s  
	 {  
	     int fd;  
	     void (*call_back)(int fd, int events, void *arg);  
	     int events;  
	     void *arg;  
	     int status; // 1: in epoll wait list, 0 not in  
	     char buff[128]; // recv data buffer  
	     int len;  
	     long last_active; // last active time  
	 };  

	 // set event  
    void EventSet(myevent_s *ev, int fd, void (*call_back)(int, int, void*), void *arg)  
     {  
	    ev->fd = fd;  
        ev->call_back = call_back;  
        ev->events = 0;  
        ev->arg = arg;  
	    ev->status = 0;  
	    ev->last_active = time(NULL);  
	 }  
     // add/mod an event to epoll  

	 void EventAdd(int epollFd, int events, myevent_s *ev)  
     {  
	     struct epoll_event epv = {0, {0}};  
	     int op;  
	     epv.data.ptr = ev;  
	     epv.events = ev->events = events;  
	     if(ev->status == 1){  
	         op = EPOLL_CTL_MOD;  
         }  
	     else{  
	        op = EPOLL_CTL_ADD;  
	        ev->status = 1;  
	     }  
	     if(epoll_ctl(epollFd, op, ev->fd, &epv) < 0)  
	       printf("Event Add failed[fd=%d]/n", ev->fd);  
	    else  
	         printf("Event Add OK[fd=%d]/n", ev->fd);  
	}  
     // delete an event from epoll  
    void EventDel(int epollFd, myevent_s *ev)  
    {  
	   struct epoll_event epv = {0, {0}};  
	   if(ev->status != 1) return;  
        epv.data.ptr = ev;  
	    ev->status = 0;  
	    epoll_ctl(epollFd, EPOLL_CTL_DEL, ev->fd, &epv);  
    }  
    int g_epollFd;  
    myevent_s g_Events[MAX_EVENTS+1]; // g_Events[MAX_EVENTS] is used by listen fd  
    void RecvData(int fd, int events, void *arg);  
    void SendData(int fd, int events, void *arg);  
    // accept new connections from clients  
    void AcceptConn(int fd, int events, void *arg)  
    {  
	      struct sockaddr_in sin;  
	   socklen_t len = sizeof(struct sockaddr_in);  
       int nfd, i;  
	     // accept  
	     if((nfd = accept(fd, (struct sockaddr*)&sin, &len)) == -1)  
	     {  
	         if(errno != EAGAIN && errno != EINTR)  
	         {  
	             printf("%s: bad accept", __func__);  
	         }  
	         return;  
	     }  
	     do  
	     {  
	        for(i = 0; i < MAX_EVENTS; i++)  
	         {  
	             if(g_Events[i].status == 0)  
	             {  
	                 break;  
             }  
         }  
         if(i == MAX_EVENTS)  
         {  
          printf("%s:max connection limit[%d].", __func__, MAX_EVENTS);
	             break;  
         }  
	         // set nonblocking  
	        if(fcntl(nfd, F_SETFL, O_NONBLOCK) < 0) break;  
	        // add a read event for receive data  
	     EventSet(&g_Events[i], nfd, RecvData, &g_Events[i]);  
       EventAdd(g_epollFd, EPOLLIN|EPOLLET, &g_Events[i]);  
	       printf("new conn[%s:%d][time:%d]/n", inet_ntoa(sin.sin_addr), ntohs(sin.sin_port), g_Events[i].last_active);  
	   }while(0);  
	}  
	// receive data  
	void RecvData(int fd, int events, void *arg)  
	{  
	    struct myevent_s *ev = (struct myevent_s*)arg;  
	    int len;  
	    // receive data  
	   len = recv(fd, ev->buff, sizeof(ev->buff)-1, 0);    
	   EventDel(g_epollFd, ev);  
      if(len > 0)  
	    {  
	        ev->len = len;  
	        ev->buff[len] = '/0';  
	        printf("C[%d]:%s/n", fd, ev->buff);  
	        // change to send event  
	        EventSet(ev, fd, SendData, ev);  
	        EventAdd(g_epollFd, EPOLLOUT|EPOLLET, ev);  
	   }  
	   else if(len == 0)  
       {  
        close(ev->fd);  
        printf("[fd=%d] closed gracefully./n", fd);  
	    }  
	    else  
	   {  
	        close(ev->fd);  
	        printf("recv[fd=%d] error[%d]:%s/n", fd, errno, strerror(errno));  
	    }  
	}  
	// send data  
	void SendData(int fd, int events, void *arg)  
	{  
	    struct myevent_s *ev = (struct myevent_s*)arg;  
	   int len;  
	    // send data  
	   len = send(fd, ev->buff, ev->len, 0);  
	    ev->len = 0;  
	    EventDel(g_epollFd, ev);  
	   if(len > 0)  
	    {  
	       // change to receive event  
	        EventSet(ev, fd, RecvData, ev);  
	        EventAdd(g_epollFd, EPOLLIN|EPOLLET, ev);  
	   }  
	    else  
	    {    close(ev->fd);  
	      printf("recv[fd=%d] error[%d]/n", fd, errno);  
	   }  
	}  
	void InitListenSocket(int epollFd, short port)  
	{  
	   int listenFd = socket(AF_INET, SOCK_STREAM, 0);  
       fcntl(listenFd, F_SETFL, O_NONBLOCK); // set non-blocking  
       printf("server listen fd=%d/n", listenFd);  
       EventSet(&g_Events[MAX_EVENTS], listenFd, AcceptConn, &g_Events[MAX_EVENTS]);  
	    // add listen socket  
	    EventAdd(epollFd, EPOLLIN|EPOLLET, &g_Events[MAX_EVENTS]);  
	    // bind & listen  
	    sockaddr_in sin;  
	    bzero(&sin, sizeof(sin));  
	    sin.sin_family = AF_INET;  
	    sin.sin_addr.s_addr = INADDR_ANY;  
	    sin.sin_port = htons(port);  
	    bind(listenFd, (const sockaddr*)&sin, sizeof(sin));  
	    listen(listenFd, 5);  
	}  
	int main(int argc, char **argv)  
	{  
	    short port = 12345; // default port  
	    if(argc == 2){  
        port = atoi(argv[1]);  
    }  
	    // create epoll  
	    g_epollFd = epoll_create(MAX_EVENTS);  
	    if(g_epollFd <= 0) printf("create epoll failed.%d/n", g_epollFd);  
	   // create & bind listen socket, and add to epoll, set non-blocking 
	    InitListenSocket(g_epollFd, port);  
	   // event loop  
	    struct epoll_event events[MAX_EVENTS];  
	    printf("server running:port[%d]/n", port);  
	    int checkPos = 0;  
	   while(1){  
	        // a simple timeout check here, every time 100, better to use a mini-heap, and add timer event  
	       long now = time(NULL);  
           for(int i = 0; i < 100; i++, checkPos++) // doesn't check listen fd  
	        {  
               if(checkPos == MAX_EVENTS) checkPos = 0; // recycle  
	           if(g_Events[checkPos].status != 1) continue;  
	            long duration = now - g_Events[checkPos].last_active;  
	            if(duration >= 60) // 60s timeout  
	            {  
	                close(g_Events[checkPos].fd);  
	                printf("[fd=%d] timeout[%d--%d]./n", g_Events[checkPos].fd, g_Events[checkPos].last_active, now);  
	                EventDel(g_epollFd, &g_Events[checkPos]);  
	            }  
	       }  
	        // wait for events to happen  
           int fds = epoll_wait(g_epollFd, events, MAX_EVENTS, 1000);  
          if(fds < 0){  
	            printf("epoll_wait error, exit/n");  
	            break;  
          }  
         for(int i = 0; i < fds; i++){  
	            myevent_s *ev = (struct myevent_s*)events[i].data.ptr;  
	            if((events[i].events&EPOLLIN)&&(ev->events&EPOLLIN)) // read event  
	            {  
	                ev->call_back(ev->fd, events[i].events, ev->arg);  
	            }  
	            if((events[i].events&EPOLLOUT)&&(ev->events&EPOLLOUT)) // write event  
	            {  
	                ev->call_back(ev->fd, events[i].events, ev->arg);  
	           }  
	        }  
	    }  
	    // free resource  
	   return 0;  
	}  

 

(六) Epoll详解说明

poll(select)的限制
      Poll函数起源于SVR3，最初局限于流设备，SVR4取消了这种限制。总是来说，poll比select要高效一些，但是，它有可移植性问题，例如，windows就只支持select。
一个poll的简单例子：

Epoll的优点：

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

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

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

传统的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.使用mmap加速内核与用户空间的消息传递。

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

4.内核微调

 这一点其实不算epoll的优点了，而是整个linux平台的优点。也许你可以怀疑linux平台，但是你无法回避linux平台赋予你微调内核的能力。比如，内核TCP/IP协议栈使用内存池管理sk_buff结构，那么可以在运行时期动态调整这个内存pool(skb_head_pool)的大小--- 通过echo XXXX>/proc/sys/net/core/hot_list_length完成。再比如listen函数的第2个参数(TCP完成3次握手的数据包队列长度)，也可以根据你平台内存大小动态调整。更甚至在一个数据包面数目巨大但同时每个数据包本身大小却很小的特殊系统上尝试最新的NAPI网卡驱动架构

5.select/epoll的特点

select的特点：select 选择句柄的时候，是遍历所有句柄，也就是说句柄有事件响应时，select需要遍历所有句柄才能获取到哪些句柄有事件通知，因此效率是非常低。但是如果连接很少的情况下， select和epoll的LT触发模式相比， 性能上差别不大。这里要多说一句，select支持的句柄数是有限制的， 同时只支持1024个，这个是句柄集合限制的，如果超过这个限制，很可能导致溢出，而且非常不容易发现问题， TAF就出现过这个问题， 调试了n天，才发现：）当然可以通过修改linux的socket内核调整这个参数。
epoll的特点：epoll对于句柄事件的选择不是遍历的，是事件响应的，就是句柄上事件来就马上选择出来，不需要遍历整个句柄链表，因此效率非常高，内核将句柄用红黑树保存的。相比于select，epoll最大的好处在于它不会随着监听fd数目的增长而降低效率。因为在内核中的select实现中，它是采用轮询来处理的，轮询的fd数目越多，自然耗时越多。并且，在linux/posix_types.h头文件有这样的声明：
#define __FD_SETSIZE    1024
表示select最多同时监听1024个fd，当然，可以通过修改头文件再重编译内核来扩大这个数目，但这似乎并不治本。

6.epoll的工作模式简介

对于epoll而言还有ET和LT的区别，LT表示水平触发，ET表示边缘触发，两者在性能以及代码实现上差别也是非常大的。

epoll的LT和ET的区别

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

7.epoll相关ＡＰＩ：

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表示已超时。

8、epoll历史

epoll是什么？按照man手册的说法：是为处理大批量句柄而作了改进的poll。要使用epoll只需要这三个系统调用：epoll_create(2)， epoll_ctl(2)， epoll_wait(2)。
Linux2.6内核epoll介绍
先介绍2本书《The Linux Networking Architecture--Design and Implementation of Network Protocols in the Linux Kernel》，以2.4内核讲解Linux TCP/IP实现，相当不错.作为一个现实世界中的实现，很多时候你必须作很多权衡，这时候参考一个久经考验的系统更有实际意义。举个例子,linux内 核中sk_buff结构为了追求速度和安全，牺牲了部分内存，所以在发送TCP包的时候，无论应用层数据多大,sk_buff最小也有272的字节.其实 对于socket应用层程序来说，另外一本书《UNIX Network Programming Volume 1》意义更大一点.2003年的时候，这本书出了最新的第3版本，不过主要还是修订第2版本。其中第6章《I/O Multiplexing》是最重要的。Stevens给出了网络IO的基本模型。在这里最重要的莫过于select模型和Asynchronous I/O模型.从理论上说，AIO似乎是最高效的，你的IO操作可以立即返回，然后等待os告诉你IO操作完成。但是一直以来，如何实现就没有一个完美的方 案。最著名的windows完成端口实现的AIO,实际上也是内部用线程池实现的罢了，最后的结果是IO有个线程池，你应用也需要一个线程池...... 很多文档其实已经指出了这带来的线程context-switch带来的代价。在linux 平台上，关于网络AIO一直是改动最多的地方，2.4的年代就有很多AIO内核patch,最著名的应该算是SGI那个。但是一直到2.6内核发布，网络 模块的AIO一直没有进入稳定内核版本(大部分都是使用用户线程模拟方法，在使用了NPTL的linux上面其实和windows的完成端口基本上差不多 了)。2.6内核所支持的AIO特指磁盘的AIO---支持io_submit(),io_getevents()以及对Direct IO的支持(就是绕过VFS系统buffer直接写硬盘，对于流服务器在内存平稳性上有相当帮助)。
所以，剩下的select模型基本上就是我们在linux上面的唯一选择，其实，如果加上no-block socket的配置，可以完成一个"伪"AIO的实现，只不过推动力在于你而不是os而已。不过传统的select/poll函数有着一些无法忍受的缺 点，所以改进一直是2.4-2.5开发版本内核的任务，包括/dev/poll，realtime signal等等。最终，Davide Libenzi开发的epoll进入2.6内核成为正式的解决方案。

9、epoll的工作模式详细

令人高兴的是，2.6内核的epoll比其2.5开发版本的/dev/epoll简洁了许多，所以，强大的东西往往是简单的。唯一有点麻烦是epoll有2种工作方式:LT和ET。
LT(level triggered)是缺省的工作方式，并且同时支持block和no-block socket.在这种做法中，内核告诉你一个文件描述符是否就绪了，然后你可以对这个就绪的fd进行IO操作。如果你不作任何操作，内核还是会继续通知你 的，所以，这种模式编程出错误可能性要小一点。传统的select/poll都是这种模型的代表．
ET (edge-triggered)是高速工作方式，只支持no-block socket。在这种模式下，当描述符从未就绪变为就绪时，内核通过epoll告诉你。然后它会假设你知道文件描述符已经就绪，并且不会再为那个文件描述 符发送更多的就绪通知，直到你做了某些操作导致那个文件描述符不再为就绪状态了(比如，你在发送，接收或者接收请求，或者发送接收的数据少于一定量时导致 了一个EWOULDBLOCK 错误）。但是请注意，如果一直不对这个fd作IO操作(从而导致它再次变成未就绪)，内核不会发送更多的通知(only once),不过在TCP协议中，ET模式的加速效用仍需要更多的benchmark确认。
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的配合。

9、epoll的使用方法

首先通过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_wait范围之后应该是一个循环，遍利所有的事件： 
for(n = 0; n < nfds; ++n) { 
                if(events[n].data.fd == listener) { //如果是主socket的事件的话，则表示有新连接进入了，进行新连接的处理。 
                    client = accept(listener, (struct sockaddr *) &local, 
                                    &addrlen); 
                    if(client < 0){ 
                        perror("accept"); 
                        continue; 
                    } 
                    setnonblocking(client); // 将新连接置于非阻塞模式 
                    ev.events = EPOLLIN | EPOLLET; // 并且将新连接也加入EPOLL的监听队列。 
注意，这里的参数EPOLLIN | EPOLLET并没有设置对写socket的监听，如果有写操作的话，这个时候epoll是不会返回事件的，如果要对写操作也监听的话，应该是EPOLLIN | EPOLLOUT | EPOLLET 
                    ev.data.fd = client; 
                    if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, client, &ev) < 0) { 
// 设置好event之后，将这个新的event通过epoll_ctl加入到epoll的监听队列里面，这里用EPOLL_CTL_ADD来加一个新的 epoll事件，通过EPOLL_CTL_DEL来减少一个epoll事件，通过EPOLL_CTL_MOD来改变一个事件的监听方式。 
                        fprintf(stderr, "epoll set insertion error: fd=%d0, 
                                client); 
                        return -1; 
                    } 
                } 
                else // 如果不是主socket的事件的话，则代表是一个用户socket的事件，则来处理这个用户socket的事情，比如说read(fd,xxx)之类的，或者一些其他的处理。
   do_use_fd(events[n].data.fd); 
}

对，epoll的操作就这么简单，总共不过4个API：epoll_create, epoll_ctl, epoll_wait和close。以前公司的服务器都是使用HTTP连接，但是这样的话，在手机目前的网络情况下不但显得速度较慢，而且不稳定。因此大家一致同意用SOCKET来进行连 接。虽然使用SOCKET之后，对于用户的费用可能会增加(由于是用了CMNET而非CMWAP)，但是，秉着用户体验至上的原则，相信大家还是能够接受 的(希望那些玩家月末收到帐单不后能够保持克制...)。
这次的服务器设计中，最重要的一个突破，是使用了EPOLL模型，虽然对之也是一知半解，但是既然在各大PC网游中已经经过了如此严酷的考验，相信他不会让我们失望，使用后的结果，确实也是表现相当不错。在这里，我还是主要大致介绍一下这个模型的结构。

10、Linux下epoll编程实例

EPOLL模型似乎只有一种格式，所以大家只要参考我下面的代码，就能够对EPOLL有所了解了，代码的解释都已经在注释中：while (TRUE)
{
int nfds = epoll_wait (m_epoll_fd, m_events, MAX_EVENTS, EPOLL_TIME_OUT);//等待EPOLL时间的发生，相当于监听，至于相关的端口，需要在初始化EPOLL的时候绑定。
if (nfds <= 0)
continue;
m_bOnTimeChecking = FALSE;
G_CurTime = time(NULL);
for (int i=0; i
{try
{if (m_events[i].data.fd == m_listen_http_fd)//如果新监测到一个HTTP用户连接到绑定的HTTP端口，建立新的连接。由于我们新采用了SOCKET连接，所以基本没用。
{OnAcceptHttpEpoll ();
}
else if (m_events[i].data.fd == m_listen_sock_fd)//如果新监测到一个SOCKET用户连接到了绑定的SOCKET端口，建立新的连接。
{OnAcceptSockEpoll ();}
else if (m_events[i].events & EPOLLIN)//如果是已经连接的用户，并且收到数据，那么进行读入。
{
OnReadEpoll (i);
}OnWriteEpoll (i);//查看当前的活动连接是否有需要写出的数据。
}
catch (int)
{
PRINTF ("CATCH捕获错误 ");continue;
}
}
m_bOnTimeChecking = TRUE;
OnTimer ();//进行一些定时的操作，主要就是删除一些短线用户等。
}