Linux下epoll原理及封装的源代码

Linux 2.6内核中提高网络I/O性能的新方法epoll

I/O多路复用技术在比较多的TCP网络服务器中有使用，即比较多的用到select函数。

1、为什么select是落后的？
首先，在Linux内核中，select所用到的FD_SET是有限的，即内核中有个参数__FD_SETSIZE定义了每个FD_SET的句柄个数，在我用的2.6.15-25-386内核中，该值是1024，搜索内核源代码得到：
include/linux/posix_types.h:#define __FD_SETSIZE 1024
也就是说，如果想要同时检测1025个句柄的可读状态是不可能用select实现的。或者同时检测1025个句柄的可写状态也是不可能的。
其次，内核中实现select是用轮询方法，即每次检测都会遍历所有FD_SET中的句柄，显然，select函数执行时间与FD_SET中的句柄个数有一个比例关系，即select要检测的句柄数越多就会越费时。
当然，在前文中我并没有提及poll方法，事实上用select的朋友一定也试过poll，我个人觉得select和poll大同小异，个人偏好于用select而已。

(1)导言：

首先，我强烈建议大家阅读Richard Stevens著作《TCP/IP Illustracted Volume 1,2,3》和《UNIX Network Programming Volume 1,2》。虽然他离开我们大家已经5年多了，但是他的书依然是进入网络编程的最直接的道路。其中的3卷的《TCP/IP Illustracted》卷1是必读－如果你不了解tcp协议各个选项的详细定义，你就失去了优化程序重要的一个手段。卷2,3可以选读一下。比如卷2 讲解的是4.4BSD内核TCP/IP协议栈实现----这个版本的协议栈几乎影响了现在所有的主流os，但是因为年代久远，内容不一定那么vogue. 在这里我多推荐一本《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，reltime signal等等。最终，Davide Libenzi开发的epoll进入2.6内核成为正式的解决方案

(2)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网卡驱动架构。

(3)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/也有一个完整的例子，大家一看就知道如何使用了

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

.....未完成，主要是要避免过多的epoll_ctl调用,以及尝试使用EPOLLONESHOT加速......

(5)benchmark

.......未完成
/*********************************引用结束******************************/

3、epoll的使用方法
这是epoll的man手册提供的一个例子，这段代码假设一个非阻塞的socket监听listener被建立并且一个epoll句柄kdpfd已经提前用epoll_create建立了：

struct epoll_event ev, *events;

for(;;) {
nfds = epoll_wait(kdpfd, events, maxevents, -1);/*wait for an I/O event. All notes here added by zhoulifa(http://zhoulifa.bokee.com) on 2006-7-6 22:10:00*/

for(n = 0; n < nfds; ++n) {
if(events[n].data.fd == listener) {/*if listen socket has an I/O, accept the new connect*/
client = accept(listener, (struct sockaddr *) &local,
&addrlen);
if(client < 0){
perror("accept");
continue;
}
setnonblocking(client);
ev.events = EPOLLIN | EPOLLET;/*EPOLLIN-available for read*/
ev.data.fd = client;
if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, client, &ev) < 0) {/*add the new socket into the epoll file descriptors*/
fprintf(stderr, "epoll set insertion error: fd=%d/n",
client);
return -1;
}
}
else
do_use_fd(events[n].data.fd);/*read from a socket which has data come*/
}
}

4、epoll使用方法示意代码
以下代码由chinaunix.net上BBS用户safedead(http://bbs.chinaunix.net/viewpro.php?uid=407631)提供：

static int s_epfd;//epoll描述字

{//初始化epoll
&nsp; struct epoll_event ev;

//设置epoll
s_epfd = epoll_create(65535);

{//这个过程可以循环以便加入多个LISTEN套接字进入epoll事件集合
//服务器监听创建
rc = listen();//listen参数这里省略

//加入epoll事件集合
ev.events = EPOLLIN;
ev.data.fd = rc;
if (epoll_ctl(s_epfd, EPOLL_CTL_ADD, rc, &ev) < 0) {
fprintf(stderr, "epoll set insertion error: fd=%d", rc);
return(-1);
}
}
}

{//epoll事件处理
int i, nfds, sock_new;
struct epoll_event events[16384];
for( ; ; ) {
//等待epoll事件
nfds = epoll_wait(s_epfd, events, 16384, -1);
//处理epoll事件
for(i = 0; i < nfds; i++) {
//events[i].data.fd是epoll事件中弹出的套接字
//接收连接
sock_new = accept(events[i].data.fd);//accept其它参数这里省略了
if(0 > sock_new) {
fprintf(stderr, "接收客户端连接失败/n");
continue;
}
}
}
}

5、参考文档
Improving (network) I/O performance ...
http://www.xmailserver.org/linux-patches/nio-improve.html

#ifndef XEPOLL_H
#define XEPOLL_H

#include <iostream>
#include <sstream>
#include <vector>
#include <string>
#include <cstring>
#include <set>
#include <iterator>
#include <algorithm>
#include <map>
#include <queue>
#include <fstream>
#include <memory>
//system
#include <sys/uio.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/shm.h>
#include <sys/types.h>
#include <sys/sem.h>
#include <netdb.h>
#include <netinet/in.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <stdlib.h>
#include <time.h>
#include <stdarg.h>
#include <pthread.h>
#include <sys/wait.h>
#include <ctype.h>

/**
* @class XEpoll
*
* @brief epoll 操作的封装
*/
class XEpoll
{
public:
XEpoll()
{

//创建epoll描述符
m_Epoll = epoll_create(MAX_SOCKET_NUMBER);
if(m_Epoll == -1)
{
//抛出网络异常
}
pthread_mutex_init(&m_Lock,NULL);
}
~XEpoll()
{
close(m_Epoll);
pthread_mutex_destroy(&m_Lock);
}

//插入一个连接
intm_InsertLink(int nFd)
{
struct epoll_event tEv;

//事件设置成输入和边缘触发
tEv.events = EPOLLIN | EPOLLET;
tEv.data.fd = nFd;
return epoll_ctl(m_Epoll, EPOLL_CTL_ADD, nFd, &tEv) ;
}

//删除一个连接
intm_DeleteLink(int nFd)
{
struct epoll_event tEv;
tEv.events = EPOLLIN | EPOLLET;
tEv.data.fd = nFd;
close(nFd);
return epoll_ctl(m_Epoll, EPOLL_CTL_DEL, nFd, &tEv) ;
}
//等待接收,返回值是需要读取数据的描述符vector
vector<int>m_Wait()
{
vector<int>vFd;
struct epoll_event tEvents[MAX_EPOEVENTS];
int nFds = epoll_wait(m_Epoll, tEvents, MAX_EPOEVENTS,-1);
for (int i = 0; i < nFds; ++i)
{
if (tEvents[i].events & EPOLLIN)
{
vFd.push_back(tEvents[i].data.fd);
}
}
return vFd;
}
private:
pthread_mutex_t m_Lock; ///<缓存的读写锁
intm_Epoll; //epoll描述符
};

#endif