linux Epoll 介绍与应用

参考资料：http://www.cnblogs.com/bigwangdi/p/3182958.html

http://www.ccvita.com/515.html
/article/6409119.html

epoll简介

epoll是Linux内核为处理大批句柄而作改进的poll，是Linux下多路复用IO接口select/poll的增强版本，它能显著的减少程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。在说到epoll之前先了解一下select，首先看看为什么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而已。

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

[b]epoll两种工作方式[/b]

LT(level triggered)：水平触发，缺省方式，同时支持block和no-block socket，在这种做法中，内核告诉我们一个文件描述符是否被就绪了，如果就绪了，你就可以对这个就绪的fd进行IO操作。如果你不作任何操作，内核还是会继续通知你的，所以，这种模式编程出错的可能性较小。传统的select\poll都是这种模型的代表。

ET(edge-triggered)：边沿触发，高速工作方式，只支持no-block socket。在这种模式下，当描述符从未就绪变为就绪状态时，内核通过epoll告诉你(这才叫边沿触发)。然后它会假设你知道文件描述符已经就绪，并且不会再为那个描述符发送更多的就绪通知，直到你做了某些操作导致那个文件描述符不再为就绪状态了(比如：你在发送、接受或者接受请求，或者发送接受的数据少于一定量时导致了一个EWOULDBLOCK错误)。但请注意，如果一直不对这个fs做IO操作(从而导致它再次变成未就绪状态)，内核不会发送更多的通知。

区别：LT事件不会丢弃，而是只要读buffer里面有数据可以让用户读取，比如可能中间有一次读的时候返回了一个EWOULDBLOCK，但只要buffer还有数据，LT就会不断的通知你，希望能把数据拼接完整。而ET则只在事件发生之时（边沿）通知，上述情况就丢掉了数据。两种模式最重要的区别就在这里，一个如果buffer还有数据，就会不断通知，这就多花了时间，另一个是只通知了一次，后面不会再通知了，这个和后面的提到的就绪队列是一致的。

在一个非阻塞的socket上调用read/write函数, 返回EAGAIN或者EWOULDBLOCK，表示资源暂时不够，能read时，读缓冲区没有数据，或者write时，写缓冲区满了。遇到这种情况，如果是阻塞socket，read/write就要阻塞掉。而如果是非阻塞socket，read/write立即返回-1， 同时errno设置为EAGAIN。所以，对于阻塞socket，read/write返回-1代表网络出错了。但对于非阻塞socket，read/write返回-1不一定网络真的出错了。可能是Resource
temporarily unavailable。这时你应该再试，直到Resource available。综上，对于non-blocking的socket，正确的读写操作为:

读：忽略掉errno = EAGAIN的错误，下次继续读（epoll_wait回一次读一次，这样多次以后拼成一段完整的东西，水平触发）

写：忽略掉errno = EAGAIN的错误，下次继续写（遇到 errno = EAGAIN是要停止）

对于select和epoll的LT模式，这种读写方式是没有问题的。但对于epoll的ET模式，这种方式还有漏洞。（漏洞？？）epoll的两种模式LT和ET的差异在于level-trigger模式下只要某个socket处于readable/writable状态，无论什么时候进行epoll_wait都会返回该socket；而edge-trigger模式下只有某个socket从unreadable变为readable或从unwritable变为writable时，epoll_wait才会返回该socket。所以，在epoll的ET模式下，正确的读写方式为:

读：只要可读，就一直读，直到返回0，或者 errno = EAGAIN （//如果不用while抱住，读数据如果没有一次性读完，Epoll下ET模式就不能读数据了，除非下一次新的数据再次过来）

写: 只要可写，就一直写，直到数据发送完，或者 errno = EAGAIN

正确的读

n = 0; 
while ((nread =
read(fd,
buf +
n,
BUFSIZ-1)) > 0) { 
n +=
nread; 
} 
if (nread == -1 &&
errno !=
EAGAIN) { 
perror("read
error"); 
}

正确的写

int
nwrite,
data_size =
strlen(buf); 
n =
data_size; 
while (n > 0) { 
nwrite =
write(fd,
buf +
data_size -
n,
n); 
if (nwrite <
n) { 
if (nwrite == -1 &&
errno !=
EAGAIN) { 
perror("write
error"); 
} 
break; 
} 
n -=
nwrite; 
}

正确的accept (accept 要考虑 2 个问题)

(1) 阻塞模式 accept 存在的问题

考虑这种情况：TCP连接被客户端夭折，即在服务器调用accept之前，客户端主动发送RST终止连接，导致刚刚建立的连接从就绪队列中移出，如果套接口被设置成阻塞模式，服务器就会一直阻塞在accept调用上，直到其他某个客户建立一个新的连接为止。但是在此期间，服务器单纯地阻塞在accept调用上，就绪队列中的其他描述符都得不到处理。

解决办法是把监听套接口设置为非阻塞，当客户在服务器调用accept之前中止某个连接时，accept调用可以立即返回-1，这时源自Berkeley的实现会在内核中处理该事件，并不会将该事件通知给epool，而其他实现把errno设置为ECONNABORTED或者EPROTO错误，我们应该忽略这两个错误。

(2)ET模式下accept存在的问题

考虑这种情况：多个连接同时到达，服务器的TCP就绪队列瞬间积累多个就绪连接，由于是边缘触发模式，epoll只会通知一次，accept只处理一个连接，导致TCP就绪队列中剩下的连接都得不到处理。

等待epfd上的io事件，返回的结果是事件个数. (如有n个连接触发事件时，程序调用epoll_wait可以获得以上n个连接，但是在大量连接并发时，而且用的是ET模拟，只会通知一次，因此在处理前面的连接时可能会有新的连接事件再次到来，TCP就绪队列中就不止n个连接了，那么此时仅仅处理n个连接，后面来的连接就被丢掉了。因此在后面不仅仅是处理n次，而是用while抱住accept，将就绪队列中所有连接都处理掉。

解决办法是用while循环抱住accept调用，处理完TCP就绪队列中的所有连接后再退出循环。如何知道是否处理完就绪队列中的所有连接呢？accept返回-1并且errno设置为EAGAIN就表示所有连接都处理完。

综合以上两种情况，服务器应该使用非阻塞地accept，accept在ET模式下的正确使用方式为：

while ((conn_sock =
accept(listenfd,(struct
sockaddr *) &remote, (size_t *)&addrlen)) > 0) { 
handle_client(conn_sock); 
} 
if (conn_sock == -1) { 
if (errno !=
EAGAIN &&
errno !=
ECONNABORTED &&
errno !=
EPROTO &&
errno !=
EINTR) 
perror("accept"); 
}

epoll应用的框架

//============================================================================
// Name        : Epoll-test.cpp
// Author      : roger
// Version     :
// Copyright   : Your copyright notice
// Description : Hello World in C, Ansi-style
//============================================================================
#include <stdio.h>
#include <stdlib.h>
#include <iostream>
#include <fcntl.h>
#include <sys/param.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/stat.h>
#include <errno.h>
#include <syslog.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/wait.h>

using namespace std;
#define MAX_EPOLL_SIZE 51200		//最大文件描述符打开数

//BDSERVER_PORT
size_t g_server_port;
//MAX LENGTH_OF_LISTEN_QUEUE
size_t g_listen_size;
//MAX_EPOLL_SIZE   为需要监听的文件描述的数量，内核用该值去分分配相应的空间
size_t g_epoll_size;
//EPOLL
size_t kdpfd,nfds;
//MAX_FILE_NUM
size_t max_fd = 65536;

//声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要处理的事件
struct epoll_event ev, events[MAX_EPOLL_SIZE];

//启动为守护进程
void init_daemon()
{
int pid;
int i;
if((pid=fork())>0)
exit(0);//是父进程，结束父进程
else if(pid< 0)
exit(1);//fork失败，退出
//是第一子进程，后台继续执行
setsid();//第一子进程成为新的会话组长和进程组长
//并与控制终端分离
if((pid=fork())>0)
exit(0);//是第一子进程，结束第一子进程
else if(pid< 0)
exit(1);//fork失败，退出
//是第二子进程，继续
//第二子进程不再是会话组长

for(i=0;i< NOFILE;++i)//关闭打开的文件描述符
close(i);
chdir("/tmp");//改变工作目录到/tmp
umask(0);//重设文件创建掩模
return;
}

//设置非阻塞方式socket
int setnonblocking(int sockfd)
{
if (fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, 0)|O_NONBLOCK) == -1)
{
return -1;
}
return 0;
}

/*
*初始化BDServer, 生成服务套接字,启动接收服务
*/
int initServerSocket(int port)
{
struct sockaddr_in server_addr;

server_addr.sin_family = AF_INET;
server_addr.sin_addr.s_addr = htons(INADDR_ANY);
server_addr.sin_port = htons(port);

int server_socket = socket(AF_INET,SOCK_STREAM,0);
setnonblocking(server_socket);
if( server_socket < 0)
{
syslog(LOG_ERR,"Create Socket Failed! - %m\n");
exit(1);
}

if( bind(server_socket,(sockaddr *)&server_addr,sizeof(server_addr)))
{
syslog(LOG_ERR,"Server Bind Port : %d Failed! - %m\n", port);
exit(1);
}

//N connection requests will be queued before further requests are refused
if (listen(server_socket, g_listen_size) )
{
syslog(LOG_ERR,"Server Listen Failed! - %m\n");
exit(1);
}

struct rlimit rt;
rt.rlim_max = rt.rlim_cur = max_fd;   //！！attention： 打开的最大文件数量,ulimit 设置
if (setrlimit(RLIMIT_NOFILE, &rt) == -1)
{
syslog(LOG_ERR,"setrlimit - %m");
exit(1);
}
else
{
syslog(LOG_NOTICE,"设置系统资源参数成功！\n");
}

return server_socket;
}

// 处理函数
void BD_RecvData(void* data){
cout<<"epoll接收消息后用于后续事件处理的函数"<<endl;
}

/*
* @author: roger
* @version ：2014-05-15 10:13:12
* epoll 运行基本框架编写     ET模式
*/
int main(int argc, char **argv) {
char ch;
int d = 0;
//处理argv命令行参数
if(argc<7)
{
printf("Useage: -p [BD_PORT|8083] -l [MAX_LENGTH_OF_LISTEN_QUEUE|4096] -e [MAX_EPOLL_SIZE|51200] -d (RUN AS DAEMON.)\n");
exit(1);
}
while( ( ch = getopt( argc, argv, "p:l:e:d?" ) ) != EOF )
{
switch(ch)
{
case 'p':
printf("BDSERVER_PORT =>%s ", optarg);
g_server_port = atoi(optarg);
break;
case 'l':
printf("MAX_LENGTH_OF_LISTEN_QUEUE => %s. ",optarg); //socket wait queue
g_listen_size = atol(optarg);
break;
case 'e':
printf("MAX_EPOLL_SIZE => %s. ",optarg);
g_epoll_size = atol(optarg);
break;
case 'd':
printf("RUN AS DAEMON. ");
d = 1;
break;
default:
printf("Not support option :%c\n",ch);
exit(2);
}
}
if(d == 1)
init_daemon();

//初始化系统日志：syslog
const char *ident = "BDServer";
int logopt = LOG_PID | LOG_CONS | LOG_PERROR;
int facility = LOG_USER;
openlog(ident, logopt, facility);
setlogmask(LOG_UPTO(LOG_DEBUG)); //LOG_DEBUG之前的全部记录到/var/log/user.log中

int server_socket = initServerSocket(g_server_port);

//加入epoll,进行高效异步io处理，共有三步，分别对应下方的 ##1～##3
// 生成一个epoll专用的文件描述符，其中的参数是指定生成描述符的最大范围。       ##1
kdpfd = epoll_create(g_epoll_size);
ev.events = EPOLLIN | EPOLLET;  //ET模式
ev.data.fd = server_socket;
//控制某个文件描述符上的事件，可以注册事件、修改事件、删除事件                                  ##2
if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, server_socket, &ev) < 0) {
cout<<"epoll set insertion error: fd="+server_socket<<endl;
return -1;
}

while(true) {
struct sockaddr_in local;
socklen_t length = sizeof(local);
int client;
int nfds;

//epoll_wait 用于轮询I/O事件的发生 ,它实现了阻塞,而不是busy loop，返回发生事件数    ## 3
nfds = epoll_wait(kdpfd, events, g_epoll_size, -1);

for(int n = 0; n < nfds; ++n) {
if(events
.data.fd == server_socket) {//新客户

//while循环抱住accept调用，处理完TCP就绪队列中的所有连接后再退出循环      dxt
while((client = accept(server_socket, (struct sockaddr *) &local,&length))>0){
setnonblocking(client);

ev.events = EPOLLIN | EPOLLET;// ET 模式
ev.data.fd = client;
//加入vip客户列表
if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, client, &ev) < 0) {
cout<<"epoll set insertion error: fd="+client<<endl;
return -1;
}
}//end while
}
else {//老客户
BD_RecvData(&events
.data.fd);
}
}//end for loop
}//end while

//关闭server socket
close(server_socket);
return 0;
}

一道腾讯后台开发的面试题

使用Linuxepoll模型，水平触发模式；当socket可写时，会不停的触发socket可写的事件，如何处理？

第一种最普遍的方式：

需要向socket写数据的时候才把socket加入epoll，等待可写事件。

接受到可写事件后，调用write或者send发送数据。

当所有数据都写完后，把socket移出epoll。

这种方式的缺点是，即使发送很少的数据，也要把socket加入epoll，写完后在移出epoll，有一定操作代价。

一种改进的方式：

开始不把socket加入epoll，需要向socket写数据的时候，直接调用write或者send发送数据。如果返回EAGAIN，把socket加入epoll，在epoll的驱动下写数据，全部数据发送完毕后，再移出epoll。

这种方式的优点是：数据不多的时候可以避免epoll的事件处理，提高效率。

最后贴一个使用epoll,ET模式的简单HTTP服务器代码:

#include <sys/socket.h> 
#include <sys/wait.h> 
#include <netinet/in.h> 
#include <netinet/tcp.h> 
#include <sys/epoll.h> 
#include <sys/sendfile.h> 
#include <sys/stat.h> 
#include <unistd.h> 
#include <stdio.h> 
#include <stdlib.h> 
#include <string.h> 
#include <strings.h> 
#include <fcntl.h> 
#include <errno.h> 
#define
MAX_EVENTS 10 
#define
PORT 8080 
//设置socket连接为非阻塞模式 
void
setnonblocking(int
sockfd) { 
int
opts; 
opts =
fcntl(sockfd,
F_GETFL); 
if(opts < 0) { 
perror("fcntl(F_GETFL)\n"); 
exit(1); 
} 
opts = (opts |
O_NONBLOCK); 
if(fcntl(sockfd,
F_SETFL,
opts) < 0) { 
perror("fcntl(F_SETFL)\n"); 
exit(1); 
} 
} 
int
main(){ 
struct
epoll_event ev,
events[MAX_EVENTS]; 
int
addrlen,
listenfd,
conn_sock,
nfds,
epfd,
fd,
i,
nread,
n; 
struct
sockaddr_in local,
remote; 
char
buf[BUFSIZ]; 
//创建listen socket 
if( (listenfd =
socket(AF_INET,
SOCK_STREAM, 0)) < 0) { 
perror("sockfd\n"); 
exit(1); 
} 
setnonblocking(listenfd); 
bzero(&local, sizeof(local)); 
local.sin_family =
AF_INET; 
local.sin_addr.s_addr =
htonl(INADDR_ANY);; 
local.sin_port =
htons(PORT); 
if(
bind(listenfd, (struct
sockaddr *) &local, sizeof(local)) < 0) { 
perror("bind\n"); 
exit(1); 
} 
listen(listenfd, 20); 
epfd =
epoll_create(MAX_EVENTS); 
if (epfd == -1) { 
perror("epoll_create"); 
exit(EXIT_FAILURE); 
} 
ev.events =
EPOLLIN; 
ev.data.fd =
listenfd; 
if (epoll_ctl(epfd,
EPOLL_CTL_ADD,
listenfd, &ev) == -1) { 
perror("epoll_ctl:
listen_sock"); 
exit(EXIT_FAILURE); 
} 
for (;;) { 
nfds =
epoll_wait(epfd,
events,
MAX_EVENTS, -1); 
if (nfds == -1) { 
perror("epoll_pwait"); 
exit(EXIT_FAILURE); 
} 
for (i = 0;
i <
nfds; ++i) { 
fd =
events[i].data.fd; 
if (fd ==
listenfd) { 
while ((conn_sock =
accept(listenfd,(struct
sockaddr *) &remote, 
(size_t *)&addrlen)) > 0) { 
setnonblocking(conn_sock); 
ev.events =
EPOLLIN |
EPOLLET; 
ev.data.fd =
conn_sock; 
if (epoll_ctl(epfd,
EPOLL_CTL_ADD,
conn_sock, 
&ev) == -1) { 
perror("epoll_ctl:
add"); 
exit(EXIT_FAILURE); 
} 
} 
if (conn_sock == -1) { 
if (errno !=
EAGAIN &&
errno !=
ECONNABORTED 
&&
errno !=
EPROTO &&
errno !=
EINTR) 
perror("accept"); 
} 
continue; 
}
if (events[i].events &
EPOLLIN) { 
n = 0; 
while ((nread =
read(fd,
buf +
n,
BUFSIZ-1)) > 0) { 
n +=
nread; 
} 
if (nread == -1 &&
errno !=
EAGAIN) { 
perror("read
error"); 
} 
ev.data.fd =
fd; 
ev.events =
events[i].events |
EPOLLOUT; 
if (epoll_ctl(epfd,
EPOLL_CTL_MOD,
fd, &ev) == -1) { 
perror("epoll_ctl:
mod"); 
} 
} 
if (events[i].events &
EPOLLOUT) { 
sprintf(buf, "HTTP/1.1
200 OK\r\nContent-Length: %d\r\n\r\nHello World", 11); 
int
nwrite,
data_size =
strlen(buf); 
n =
data_size; 
while (n > 0) { 
nwrite =
write(fd,
buf +
data_size -
n,
n); 
if (nwrite <
n) { 
if (nwrite == -1 &&
errno !=
EAGAIN) { 
perror("write
error"); 
} 
break; 
} 
n -=
nwrite; 
} 
close(fd); 
} 
} 
} 
return 0; 
}