I/O复用系统调用之epoll

I/O复用系统调用之epoll()

epoll API是linux系统特有的（在2.6内核新增），同I/O多路复用和信号驱动I/O功能类似，均可以监视多个文件描述符上的I/O就绪事件。 epoll()将用户关心的事件放入内核的事件表中，无须像select和poll那样每次调用都需要传入文件描述符集合或者事件集合，因为从用户态到内核态的切换很耗费资源，epoll每次管理一个文件描述符，该描述符对应内核中的事件表。

描述符准备就绪通知模式

在看epoll()函数之前先来看看区分两种文件描述符准备就绪的通知模式：水平触发(LT:Level Trigger)：被监控的套接字文件描述符有就绪事件发生时，epoll_wait通知应用进程去做相应的处理，若进程没有处理完本次事件(例如，读写缓冲区太小，一次不能处理完事件数据)，epoll会继续提醒通知。相对ET模式来说是低效模式，那些已就绪的描述符若不需要进行读写操作，而它却每次都返回提醒，这样会很大程度的降低自己处理关心就绪文件描述符的效率。边缘触发(ET:Edge Trigger)：被监听套接字文件描述符第一次就绪后，epoll_wait只提醒一次，即使是本次没有处理完事件（如读写缓冲区太小，未能都写完就绪事件），因而用户程序应该一次尽可能的将其相应就绪事件处理完。相比于LT模式来说是一个高效模式。这样讲两种触发模式，可能有些人还不太理解，可以举个例子来说明说明epoll的水平触发和边缘出发通知之间的区别： 假设我们使用epoll来监视一个套接字上的输入（EPOLLIN），接下来会发生如下事件：1.套接字上有输入到来2.我们调用一次epoll_wait()。无论采用水平触发还是边缘出发同知，该调用都会告诉套接字已经处于就绪状态。3.再次调用epoll_wait()。若采用的是水平触发通知，那么第二个epoll_wait()调用将告诉我们套接字处于就绪态。而若采用边缘触发通知，那么第二个epoll_wait()调用将阻塞，因为自从上一次调用epoll_wait()以来并没有新的输入到来。前边博文提到的select和poll只支持水平触发模式；信号驱动I/O支持边缘触发模式。只有epoll()两种模式都支持。

epoll函数定义

epoll通过一簇函数实现：epoll_create()、epoll_ctl()、epoll_wait()。

#include <sys/epoll.h>

int epoll_create(int size)；     //size指定内核中事件表的大小，即就是关心的事件个数

该函数返回一个文件描述符，它即就是epoll系统调用其他函数的第一个参数，用该描述符指定要访问的内核事件表。

系统调用epoll_ctl()操作同epoll实例相关联的进程中声明过感兴趣的文件描述符列表。通过epoll_ctl()可以增加新的描述符到列表，将已有的文件描述符从该列表中移除，以及修改代表文件描述符上事件类型的位掩码。

int epoll_ctl(

int epfd,    // 就是epoll_create(0函数返回的文件描述符；

int op,       //指定操作类型,如下表

int fd,        //要操作的文件描述符

struct epoll_event* event);

成功返回0，失败-1

 第四个参数为epoll_event类型的结构体：

struct epoll_event {

__uint32_t      event;  //epoll事件表

epoll_data_t   data;  //用户数据

}

/*epoll_data_t 定义*/

typedef union epoll_data {

void       *ptr;

int          fd;

uint32_t  u32;

uint64_t  u64;

} epoll_data_t;

系统调用epoll_wait()返回与epoll实例相关联的就绪列表中的成员。

int epoll_wait(int epfd,

struct epoll_event* events,  //指向的结构体数组中返回的是有关就绪态文件描述符的信息。

int maxevents,  //指定最多监听多少个事件，必须大于0

int timeout); //同poll

成功时返回就绪文件描述符的个数（也就是数组events中的元素个数），失败返回-1

epoll_wait()函数如果监听到事件，就将所有就绪的事件从内核事件表中复制到第二个参数events指向的数组中，这个系统调用返回时只从内核中拷贝就绪事件的描述符，这个数量是非常少的与select、poll完全不同，后两者返回的是监听的全部文件描述符。

epoll为何如此高效？

因为epoll内核中是靠链表和红黑树实现的，具体实现还需要继续研究。

实例

针对epoll()的水平触发模式和边缘触发模式来看看下边的实例。

/**服务器程序

*   gcc lt_et.c -o serv

*  ./serv ip port

*/

int main(int argc, char **argv)

{

/*服务器创建socket套接字...*/

struct epoll_event events[MAX_SOCKET_NUMBERS];

int epollfd = epoll_create(5);

assert( epollfd != -1);

addfd(epollfd,listenfd,true);

while(1) {

int ret = epoll_wait(epollfd, events,MAX_SOCKET_NUMBERS,-1);   //epoll_wait()监控事件发生，阻塞，有返回则有事件需要处理

lt(events,ret,epollfd,listenfd);  //使用LT模式，默认

//et(events,ret,epollfd,listenfd);

}

return 0;

}

 /*添加文件描述符事件到文件描述符列表*/

void addfd(int epollfd, int fd, int flag) {

struct epoll_event event;

memset(&event,0x00,sizeof(event));

event.data.fd = fd;

event.events = EPOLLIN;

if(flag) {

event.events |= EPOLLET;

}

epoll_ctl(epollfd,EPOLL_CTL_ADD,fd,&event);

setnonblocking(fd);   //将fd设置为非阻塞，因为epoll_wait()时已经阻塞，返回必定有就绪事件发生，此处就不需要继续阻塞

}

 /*LT水平触发模式*/

void lt(struct epoll_event *events, int number, int epollfd, int listenfd) {

char buf[BUF_SIZE];    //缓冲区大小为BUF_SIZE

int i;

for(i=0; i<number; i++) {

int sockfd = events[i].data.fd;

if(sockfd == listenfd) {   //监听套接字就绪，说明有client连接

/*接收客户连接accept*/

if(connfd < 0) {

printf("accept error\n");

exit(EXIT_FAILURE);

}

addfd(epollfd,connfd,0);

} else if(events[i].events & EPOLLIN) {  //连接描述符就绪

printf("LT once\n");

memset(buf,0x00,sizeof(buf));

int ret = recv(sockfd,buf,sizeof(buf)-1,0);

if(ret <= 0) {

printf("recv 0 \n");

close(sockfd);

continue;

}

send(sockfd,buf,strlen(buf),0);

printf("recv data form %d buf is %s\n",sockfd,buf);

}else {

printf("somthing else happen\n");

}

}               

}

/*ET边缘触发模式*/

void et(struct epoll_event * events, int number, int epollfd, int listenfd) {

char buf[BUF_SIZE];

int i;

for(i=0; i<number;i++) {

int sockfd = events[i].data.fd;

if(sockfd == listenfd) {

/*接收客户连接accept*/

}else if(events[i].events & EPOLLIN) {

printf("ET once\n");

//  while(1) {     //ET模式使用while循环读写完就绪描述符事件，因为只通知一次

    memset(buf,0x00,sizeof(buf));

    int ret = recv(sockfd,buf,sizeof(buf)-1,0);

if(ret < 0){

if((errno == EAGAIN) || (errno == EWOULDBLOCK)){

printf("read later\n");

break;

}

    close(sockfd);

break;

}else if(ret == 0) {

    close(sockfd);

}else {

        send(sockfd,buf,strlen(buf),0);

printf("recv data from %d buf is %s\n",sockfd,buf);

}

//   }

}else {

printf("somthing else happen\n");

}

}

}

/*客户端程序

* 回射输入字符串

*/

/*调用select监听socket套接字和标准输入*/

void str_cli(FILE *fp, int sockfd)

{

...

    while(1) {

FD_ZERO(&reset);

FD_SET(fileno(fp),&reset);

FD_SET(sockfd,&reset);

maxfd = fileno(fp) > sockfd ? fileno(fp) : sockfd;

select(maxfd+1,&reset,NULL,NULL,NULL);

if(FD_ISSET(fileno(fp),&reset)) {

if(fgets(sendline,BUFSIZ,fp) != NULL)

write(sockfd,sendline,strlen(sendline));

}                                                                                                                                                                                   

if(FD_ISSET(sockfd,&reset)) {

memset(recvline,0x00,sizeof(recvline));

if(read(sockfd,recvline,BUFSIZ) == 0)

printf("readline:%m\n"),exit(1);

fputs(recvline,stdout);

}

}

}

int main(int argc, char **argv)

{

/*创建socket，连接server*/

str_cli(stdin,sockfd);

exit(0);

}

程序测试：

当接收缓冲区BUF_SIZE设置的比较小时，客户端发送的数据大于缓冲区大小，两种触发模式结果如下：BUF_SIZE设置为10，Client发送两次数据，第一次发送“test”（小于缓冲区大小），第二次发送“testtesttest”（大于缓冲区大小）

[ty@tiany I_O]$ ./clint

test

test

testtesttest

testtesttest

服务器端显示：使用LT模式

[ty@tiany I_O]$ ./serv 192.168.234.129 12000

LT once

recv data form 5 buf is test

LT once

recv data form 5 buf is testtestt

LT once

recv data form 5 buf is est

可以看到服务器端使用三次正常输出了client端的输入，client第二次的输入，服务器分两次接收。使用ET模式

[ty@tiany I_O]$ ./serv 192.168.234.129 12000

ET once

recv data from 5 buf is test

ET once

recv data from 5 buf is testtestt

使用ET模式只通知一次，可以发现client第二次发送超出缓冲区大小的部分丢失，接收的数据不完整；为了防止这种情况，可以在ET模式接收数据时加入循环，直到数据接收结束。（本次代码中只需要去掉ET模式化中while循环处的注释符）本文完整代码地址：https://github.com/ty92/epoll_LT_ET