Linux c编程：I/O多路复用之epoll

前面介绍了select处理，这一章继续介绍另外一种I/O多路服用的机制：epoll。来比较下两种机制的不同点。
select: 调用过程如下：

（1）使用copy_from_user从用户空间拷贝fd_set到内核空间

（2）注册回调函数__pollwait

（3）遍历所有fd，调用其对应的poll方法（对于socket，这个poll方法是sock_poll，sock_poll根据情况会调用到tcp_poll,udp_poll或者datagram_poll）

（4）以tcp_poll为例，其核心实现就是__pollwait，也就是上面注册的回调函数。

（5）__pollwait的主要工作就是把current（当前进程）挂到设备的等待队列中，不同的设备有不同的等待队列，对于tcp_poll来说，其等待队列是sk->sk_sleep（注意把进程挂到等待队列中并不代表进程已经睡眠了）。在设备收到一条消息（网络设备）或填写完文件数据（磁盘设备）后，会唤醒设备等待队列上睡眠的进程，这时current便被唤醒了。

（6）poll方法返回时会返回一个描述读写操作是否就绪的mask掩码，根据这个mask掩码给fd_set赋值。

（7）如果遍历完所有的fd，还没有返回一个可读写的mask掩码，则会调用schedule_timeout是调用select的进程（也就是current）进入睡眠。当设备驱动发生自身资源可读写后，会唤醒其等待队列上睡眠的进程。如果超过一定的超时时间（schedule_timeout指定），还是没人唤醒，则调用select的进程会重新被唤醒获得CPU，进而重新遍历fd，判断有没有就绪的fd。

（8）把fd_set从内核空间拷贝到用户空间

总结：

select的几大缺点：

（1）每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大

（2）同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大

（3）select支持的文件描述符数量太小了，默认是1024

对于select的几个缺点。epoll的改进机制如下：

对于第一个缺点，epoll的解决方案在epoll_ctl函数中。每次注册新的事件到epoll句柄中时（在epoll_ctl中指定EPOLL_CTL_ADD），会把所有的fd拷贝进内核，而不是在epoll_wait的时候重复拷贝。epoll保证了每个fd在整个过程中只会拷贝一次。

　　对于第二个缺点，epoll的解决方案不像select或poll一样每次都把current轮流加入fd对应的设备等待队列中，而只在epoll_ctl时把current挂一遍（这一遍必不可少）并为每个fd指定一个回调函数，当设备就绪，唤醒等待队列上的等待者时，就会调用这个回调函数，而这个回调函数会把就绪的fd加入一个就绪链表）。epoll_wait的工作实际上就是在这个就绪链表中查看有没有就绪的fd（利用schedule_timeout()实现睡一会，判断一会的效果，和select实现中的第7步是类似的）。

　　对于第三个缺点，epoll没有这个限制，它所支持的FD上限是最大可以打开文件的数目，这个数字一般远大于2048,举个例子,在1GB内存的机器上大约是10万左右，具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。

总结：

（1）select，poll实现需要自己不断轮询所有fd集合，直到设备就绪，期间可能要睡眠和唤醒多次交替。而epoll其实也需要调用epoll_wait不断轮询就绪链表，期间也可能多次睡眠和唤醒交替，但是它是设备就绪时，调用回调函数，把就绪fd放入就绪链表中，并唤醒在epoll_wait中进入睡眠的进程。虽然都要睡眠和交替，但是select和poll在“醒着”的时候要遍历整个fd集合，而epoll在“醒着”的时候只要判断一下就绪链表是否为空就行了，这节省了大量的CPU时间。这就是回调机制带来的性能提升。

（2）select，poll每次调用都要把fd集合从用户态往内核态拷贝一次，并且要把current往设备等待队列中挂一次，而epoll只要一次拷贝，而且把current往等待队列上挂也只挂一次（在epoll_wait的开始，注意这里的等待队列并不是设备等待队列，只是一个epoll内部定义的等待队列）。这也能节省不少的开销

epoll的接口函数很简单，只有3个函数

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 结构如下：

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 表示已超时。 epoll有两种工作方式：

LT level triggered 水平触发模式，

同时支持阻塞和非阻塞的socket。在这种模式中，内核告诉你一个文件描述符是否就绪了，然后你可以对这个就绪的fd进行I/O操作，如果你不做任何操作，还是会继续通知你。（没处理这个流还是一直通知你）

ET edge triggered 边缘触发模式

只支持非阻塞的socket。效率比LT高。这种工作模式下，当从epoll_wait调用获取到事件后，如果没有把这次事件对应的套接字处理完，那么在这个套接字中没有心的时间再次到来时，ET模式下是无法再次从epoll_wait调用中获取这个事件的。而LT只要有数据就总可以获取。

参考下面这个图：

实现代码如下：

//网络编程服务端

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <unistd.h>

#include <errno.h>

#include <sys/types.h>

#include <sys/socket.h>

#include <arpa/inet.h>//htons()函数头文件

#include <netinet/in.h>//inet_addr()头文件

#include <fcntl.h>

#include <sys/epoll.h>

#include "pub.h"

#define MAXSOCKET 20

int main(int arg, char *args[])

{

if (arg < 2)

{

printf("please print one param!\n");

return -1;

}

//create server socket

int listen_st = server_socket(atoi(args[1]));

if (listen_st < 0)

{

return -1;

}

* 声明epoll_event结构体变量ev，变量ev用于注册事件，

* 数组events用于回传需要处理的事件

struct epoll_event ev, events[100];

//生成用于处理accept的epoll专用文件描述符

int epfd = epoll_create(MAXSOCKET);

//把socket设置成非阻塞方式 setnonblock(listen_st);

//设置需要放到epoll池里的文件描述符

ev.data.fd = listen_st;

//设置这个文件描述符需要epoll监控的事件

* EPOLLIN代表文件描述符读事件

*accept,recv都是读事件

ev.events = EPOLLIN | EPOLLERR | EPOLLHUP;

* 注册epoll事件

* 函数epoll_ctl中&ev参数表示需要epoll监视的listen_st这个socket中的一些事件

epoll_ctl(epfd, EPOLL_CTL_ADD, listen_st, &ev);

while (1)

{

* 等待epoll池中的socket发生事件，这里一般设置为阻塞的

* events这个参数的类型是epoll_event类型的数组

* 如果epoll池中的一个或者多个socket发生事件，

* epoll_wait就会返回，参数events中存放了发生事件的socket和这个socket所发生的事件

* 这里强调一点，epoll池存放的是一个个socket，不是一个个socket事件

* 一个socket可能有多个事件，epoll_wait返回的是有消息的socket的数目

* 如果epoll_wait返回事件数组后，下面的程序代码却没有处理当前socket发生的事件

* 那么epoll_wait将不会再次阻塞，而是直接返回，参数events里面的就是刚才那个socket没有被处理的事件

int nfds = epoll_wait(epfd, events, MAXSOCKET, -1);

if (nfds == -1)

{

printf("epoll_wait failed ! error message :%s \n", strerror(errno));

break;

}

int i = 0;

for (; i < nfds; i++)

{

if (events[i].data.fd < 0)

continue;

if (events[i].data.fd == listen_st)

{

//接收客户端socket

int client_st = server_accept(listen_st);

* 监测到一个用户的socket连接到服务器listen_st绑定的端口

if (client_st < 0)

{

continue;

}

//设置客户端socket非阻塞 setnonblock(client_st);

//将客户端socket加入到epoll池中

struct epoll_event client_ev;

client_ev.data.fd = client_st;

client_ev.events = EPOLLIN | EPOLLERR | EPOLLHUP;

epoll_ctl(epfd, EPOLL_CTL_ADD, client_st, &client_ev);

* 注释：当epoll池中listen_st这个服务器socket有消息的时候

* 只可能是来自客户端的连接消息

* recv,send使用的都是客户端的socket，不会向listen_st发送消息的

continue;

}

//客户端有事件到达

if (events[i].events & EPOLLIN)

{

//表示服务器这边的client_st接收到消息

if (socket_recv(events[i].data.fd) < 0)

{

close_socket(events[i].data.fd);

//接收数据出错或者客户端已经关闭

events[i].data.fd = -1;

/*这里continue是因为客户端socket已经被关闭了，

* 但是这个socket可能还有其他的事件，会继续执行其他的事件，

* 但是这个socket已经被设置成-1

* 所以后面的close_socket()函数都会报错

continue;

}

* 此处不能continue，因为每个socket都可能有多个事件同时发送到服务器端

* 这也是下面语句用if而不是if-else的原因，

}

//客户端有事件到达

if (events[i].events & EPOLLERR)

{

printf("EPOLLERR\n");

//返回出错事件，关闭socket，清理epoll池，当关闭socket并且events[i].data.fd=-1,epoll会自动将该socket从池中清除 close_socket(events[i].data.fd);

events[i].data.fd = -1;

continue;

}

//客户端有事件到达

if (events[i].events & EPOLLHUP)

{

printf("EPOLLHUP\n");

//返回挂起事件，关闭socket，清理epoll池 close_socket(events[i].data.fd);

events[i].data.fd = -1;

continue;

}

//close epoll close(epfd);

//close server socket close_socket(listen_st);

return 0;

}

巴特西

Linux c编程：I/O多路复用之epoll

最新文章

热门文章