前言
I/O多路复用有很多种实现。在linux上,2.4内核前主要是select和poll,自Linux 2.6内核正式引入epoll以来,epoll已经成为了目前实现高性能网络服务器的必备技术。尽管他们的使用方法不尽相同,但是本质上却没有什么区别。本文将重点探讨将放在EPOLL的实现与使用详解。
为什么会有EPOLL?
select的缺陷
高并发的核心解决方案是1个线程处理所有连接的“等待消息准备好”,这一点上epoll和select是无争议的。但select预估错误了一件事,当数十万并发连接存在时,可能每一毫秒只有数百个活跃的连接,同时其余数十万连接在这一毫秒是非活跃的。select采用轮询的方法遍历所有的FD_SET句柄
返回的活跃连接 ==select(全部待监控的连接)。
什么时候会调用select方法呢?在你认为需要找出有报文到达的活跃连接时,就应该调用。所以,调用select在高并发时是会被频繁调用的。这样,这个频繁调用的方法就很有必要看看它是否有效率,因为,它的轻微效率损失都会被“频繁”二字所放大。它有效率损失吗?显而易见,全部待监控连接是数以十万计的,返回的只是数百个活跃连接,这本身就是无效率的表现。被放大后就会发现,处理并发上万个连接时,select就完全力不从心了。
此外,在Linux内核中,select所用到的FD_SET是有限的,即内核中有个参数__FD_SETSIZE定义了每个FD_SET的句柄个数。一般为1024(fd也不能大于1024,否则可能core dump)
/linux/posix_types.h:
#define __FD_SETSIZE 1024
其次,内核中实现 select是用轮询方法,即每次检测都会遍历所有FD_SET中的句柄,显然,select函数执行时间与FD_SET中的句柄个数有一个比例关系,即 select要检测的句柄数越多就会越费时。看到这里,您可能要要问了,你为什么不提poll?笔者认为select与poll在内部机制方面并没有太大的差异。相比于select机制,poll只是取消了最大监控文件描述符数限制,并没有从根本上解决select存在的问题。
接下来我们看张图,当并发连接为较小时,select与epoll似乎并无多少差距。可是当并发连接上来以后,select就显得力不从心了。
图 1.主流I/O复用机制的benchmark
epoll接口
int epoll_create(int size);
创建一个epoll的句柄,size用来告诉内核这个监听的数目一共有多大。创建好epoll句柄后它是会占用一个fd值,所以记住使用完要调用close()关闭,否则可能fd耗尽。
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
epoll的事件注册函数。它不同与select()是在监听事件时(epoll使用epoll_wait监听)告诉内核要监听什么类型的事件,而是在这里先注册要监听的事件类型。
epfd:epoll_create返回的句柄
op:操作动作
EPOLL_CTL_ADD:注册新的fd到epfd中;
EPOLL_CTL_MOD:修改已经注册的fd的监听事件;
EPOLL_CTL_DEL:从epfd中删除一个fd;
fd:需要监听的fd
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
用来等待所监听文件描述符上有事件发生
epfd:epoll_create返回的句柄
timeout:是超时事件,-1为阻塞,0为立即返回,非阻塞,大于0是指定的微妙
events:是一个 传入传出参数,它是epoll_event结构的指针,用来从内核得到事件的集合
maxevents:告知内核events的大小,但不能大于epoll_create()时创建的size
LT和ET模式
采用LT模式的文件描述符,当epoll_wait检测到其上有事件发生并将此事件通知应用程序后,应用程序可以不立即处理此事件,当下一次调用epoll_wait是,epoll_wait还会将此事件通告应用程序。
对于采用ET模式的文件描述符,当epoll_wait检测到其上有事件发生并将此通知应用程序后,应用程序必须立即处理该事件,因为后续的epoll_wait调用将不在向应用程序通知这一事件。ET模式降低了同意epoll事件被触发的次数,效率比LT模式高。
select、poll和epoll三种I/O复用模式的比较
系统调用 | select | poll | |
事件集合 | 通过3个参数分别传入感兴趣的可读,可写及异常等事件 内核通过对这些参数的在线修改来反馈其中的就绪事件 这使得用户每次调用select都要重置这3个参数 | 统一处理所有事件类型,因此只需要一个事件集参数。 用户通过pollfd.events传入感兴趣的事件,内核通过 修改pollfd.revents反馈其中就绪的事件 | 内核通过一个事件表直接管理用户感兴趣的所有事件。 因此每次调用epoll_wait时,无需反复传入用户感兴趣 的事件。epoll_wait系统调用的参数events仅用来反馈就绪的事件 |
应用程序索引就绪文件 描述符的时间复杂度 | O(n) | O(n) | O(1) |
最大支持文件描述符数 | 一般有最大值限制 | 65535 | 65535 |
工作模式 | LT | LT | 支持ET高效模式 |
内核实现和工作效率 | 采用轮询方式检测就绪事件,时间复杂度:O(n) | 采用轮询方式检测就绪事件,时间复杂度:O(n) | 采用回调方式检测就绪事件,时间复杂度:O(1) |
epoll框架
for( ; ; )
{
nfds = epoll_wait(epfd,events,20,500);
for(i=0;i<nfds;++i)
{
if(events[i].data.fd==listenfd) //有新的连接
{
connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen); //accept这个连接
ev.data.fd=connfd;
ev.events=EPOLLIN|EPOLLET;
epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev); //将新的fd添加到epoll的监听队列中
}
else if( events[i].events&EPOLLIN ) //接收到数据,读socket
{
n = read(sockfd, line, MAXLINE)) < 0 //读
ev.data.ptr = md; //md为自定义类型,添加数据
ev.events=EPOLLOUT|EPOLLET;
epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);//修改标识符,等待下一个循环时发送数据,异步处理的精髓
}
else if(events[i].events&EPOLLOUT) //有数据待发送,写socket
{
struct myepoll_data* md = (myepoll_data*)events[i].data.ptr; //取数据
sockfd = md->fd;
send( sockfd, md->ptr, strlen((char*)md->ptr), 0 ); //发送数据
ev.data.fd=sockfd;
ev.events=EPOLLIN|EPOLLET;
epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); //修改标识符,等待下一个循环时接收数据
}
else
{
//其他的处理
}
}
}
完整的epoll服务端示例:
#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[n] = '/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;
}