Linux网络篇:独有的I/O多路复用模型epoll(适用于大规模fd)

poll解决了fd上限问题,也将“入参”和“出参”区分了开来。但是,对于大规模并发系统poll的全拷贝操作,极大的影响了性能。为了解决该问题epoll模型被引进来了。

epoll模型是Linux 2.6以上版本的系统所独有的。它是一个比selectpoll模型使用起来更简单,却又更高效的模型,它只关注“活跃”着的fd,特别适用于大规模并发场景。

Unix-like中并没有epoll模型,但是有一个类似的模型叫做kqueue。 也就是说Mac OS X不支持epoll模型,也没有头文件<sys/epoll>,它支持kqueue
更多内容请参考《Unix网络篇(4)与epoll类似的kqueue模型——Unix-like系统》

1 epoll模型的改进

select模型开启了I/O复用的征程;poll模型解决了fd上限问题,同时也避免了入参、出参一把抓的混乱性;epoll模型不再copy全部的fd,它只关心“活跃”着,避免了对全部fd的轮询。

2 服务端

epoll模型使用起来比其他模型都要简单。总共就三个函数:epoll_createepoll_ctlepoll_wait

int epoll_create(int size);
int epoll_ctl(int epfd, int op, int fd, struct epoll_event* events);
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

epoll_create用来创建一个epoll句柄;epoll_ctl可以注册、修改或删除一个基于sockfd的事件;epoll_wait它用来等待事件的发生。更多内容(或者各参数的取值)可以参看《IO多路复用之epoll总结》

LT模式: (poll、epoll默认触发模式)当epoll_wait检测到描述符事件发生并将此事件通知应用程序,应用程序可以不立即处理该事件。下次调用epoll_wait时,会再次响应应用程序并通知此事件。
————Redis采用了该模式。

ET模式: 当epoll_wait检测到描述符事件发生并将此事件通知应用程序,应用程序必须立即处理该事件。如果不处理,下次调用epoll_wait时,不会再次响应应用程序并通知此事件。
————Nginx采用了该模式。

以下为epoll_server.c源代码

#include <sys/socket.h> 
#include <netinet/in.h> //sockaddr_in
#include <unistd.h> //fork,read,write
#include <stdlib.h> //exit(0)
#include <errno.h>
#include <string.h> //bzero
#include <sys/epoll.h> //epoll
#include <limits.h> //OPEN_MAX
#include <stdio.h>

const int SERV_PORT = 9527;
const int LISTENQ = 5;
const int MAXBUFSIZE = 1024;
const int OPEN_MAX = 10240;
const int INFTIM = -1; //TIMEOUT

int main(int argc, char* argv[]) {
int epollfd, i, nev, n, maxevents, listenfd, connfd, sockfd;
char buf[MAXBUFSIZE];
socklen_t clilen;
struct sockaddr_in cliaddr, servaddr;
struct epoll_event ev, events[OPEN_MAX];//#define OPEN_MAX 10240 (Mac OS X)

listenfd = socket(AF_INET, SOCK_STREAM, 0);

bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(SERV_PORT);

if ( bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0 )
exit_msg("bind error");

listen(listenfd, LISTENQ);

epollfd = epoll_create(OPEN_MAX);

ev.data.fd = listenfd;
ev.events = EPOLLIN | EPOLLET;// EPOLLET(边缘触发,一次), poll、epoll默认都是(水平触发,多次)
//1、注册所关注的事件
epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &ev);
maxevents = 1;
for ( ; ; ) {
//2、确定是否有所关注的事件发生
nev = epoll_wait(epollfd, events, maxevents, INFTIM);
for (i = 0; i < nev; ++i) {
if (events[i].data.fd == listenfd && maxevents < OPEN_MAX) {
clilen = sizeof(cliaddr);
connfd = accept(listenfd, (struct sockaddr*)&cliaddr, &clilen);
ev.data.fd = connfd;
ev.events = EPOLLIN | EPOLLET;
epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &ev);
++maxevents;
}
else {
sockfd = events[i].data.fd;
if ( (n = read(sockfd, buf, MAXBUFSIZE)) < 0 ) {
if (errno == ECONNRESET) {
close(sockfd);
epoll_ctl(epollfd, EPOLL_CTL_DEL, sockfd, &ev); //DELETE
--maxevents;
}
else {
exit_msg("read error");
}
}
else if (n == 0) {
close(sockfd);
epoll_ctl(epollfd, EPOLL_CTL_DEL, sockfd, &ev); //DELETE
--maxevents;
}
else {
writen(sockfd, buf, n);//DIY: writen = write + while 参见《Unix网络篇(1)一个典型的TCP Socket通信例子》
}
}
}
}

return 0;
}

3 客户端

以下是epoll_client.c源代码

#include <sys/socket.h> 
#include <netinet/in.h> //sockaddr_in
#include <arpa/inet.h> //inet_pon
#include <unistd.h> //fork,read,write
#include <stdlib.h> //exit(0)
#include <errno.h>
#include <sys/types.h>
#include <sys/epoll.h> //epoll
#include <sys/time.h>
#include <string.h> //bzero
#include <stdio.h>
#include <sys/stat.h> //fstat

const int SERV_PORT = 9527;
const int LISTENQ = 5;
const int MAXBUFSIZE = 1024;
const int OPEN_MAX = 10240;
const int INFTIM = -1; //TIMEOUT

void str_cli(FILE* fp, int sockfd) {
int epollfd, i, n, nev, maxevents, stdineof=0, isfile;
char buf[MAXBUFSIZE];
struct epoll_event ev, events[2];
struct stat st;

isfile = ((fstat(fileno(fp), &st) == 0) && (st.st_mode & S_IFMT) == S_IFREG);

epollfd = epoll_create(2);

ev.data.fd = fileno(fp);
ev.events = EPOLLIN | EPOLLET;// EPOLLET(边缘触发,一次), EPOLLLT(水平触发,多次)
epoll_ctl(epollfd, EPOLL_CTL_ADD, fileno(fp), &ev);

ev.data.fd = sockfd;
epoll_ctl(epollfd, EPOLL_CTL_ADD, sockfd, &ev);

maxevents = 2;
for ( ; ; ) {
//2、确定是否有所关注的事件发生
nev = epoll_wait(epollfd, events, maxevents, INFTIM);
for (i = 0; i < nev; ++i) {
if (events[i].data.fd == sockfd) {
if ( (n = read(sockfd, buf, MAXBUFSIZE)) == 0 ) {
if (stdineof == 1)
return;
else
exit_msg("str_cli:server terminated prematurely");
}
write(fileno(stdout), buf, n); //stdout使用write即可,不必writen
}
if (events[i].data.fd == fileno(fp)) {
n = read(fileno(fp), buf, MAXBUFSIZE);
if (n > 0)
writen(sockfd, buf, n);//DIY: 参见《Unix网络篇(1)一个典型的TCP Socket通信例子》
if (n == 0 || (isfile && n == events[i].data.u32)) {
stdineof = 1;
shutdown(sockfd, SHUT_WR);
//3、DELETE事件
epoll_ctl(epollfd, EPOLL_CTL_DEL, sockfd, &ev); //DELETE
--maxevents;
}
}
}
}
}

int main(int argc, char* argv[]) {
int sockfd;
struct sockaddr_in servaddr;

if (argc != 2)
exit_msg("argv != 2");

sockfd = socket(AF_INET, SOCK_STREAM, 0);

bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_port = htons(SERV_PORT);
inet_pton(AF_INET, argv[1], &servaddr.sin_addr);

if (connect(sockfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0)
exit_msg("conn err");//警告: connect请勿捕获EINTR

str_cli(stdin, sockfd);

return 0;
}

References:

[1] Rabbit Dale,《IO多路复用之epoll总结》
[2] 23云恋49枫, 《I/O多路复用之epoll实战》
[3] 《Unix网络篇(1)一个典型的TCP Socket通信例子》
[4] 《Unix网络篇(2)使用select复用I/O》
[5] 《Unix网络篇(3)poll模型突破fd上限》
[6] 《Unix网络篇(4)与epoll类似的kqueue模型——Unix-like系统》
[7] ka__ka__, 《Linux网络编程:基于epoll的IO多路复用并发模型》