前言

前面讲了IO多路复用的API，select和poll的缺点是性能不够，客户端连接越多性能下降越明显，epoll的出现解决了这个问题，引用The Linux Programming Interface的一个统计对比如下：

fd数量 poll CPU时间(秒) select CPU时间(秒) epoll CPU时间(秒)

---------------------------------------------------------------------

10 0.61 0.73 0.41

100 2.9 3.0 0.42

1000 35 35 0.53

10000 990 930 0.66

---------------------------------------------------------------------

可以看出fd达到100个以后，select/poll就非常慢了，而epoll即使达到10000个也表现得非常好，因为：每次调用select/poll，内核必须检查所有传进来的描述符；而对于epoll，每次调用epoll_ctl，内核会把相关信息与底层的文件描述关联起来，当IO事件就绪时，内核把信息加到epoll的就绪列表里。随后调用epoll_wait，内核只需把就绪列表中的信息提取出来返回即可。

每次调用select/poll，都要把待监控的所有文件描述符传给内核，函数返回时，内核要把描述符返回并标识哪些就绪，得到结果后还要逐个判断所有描述符，才能确定哪些有事件；epoll在调用epoll_ctl时就已经维护着监控的列表，epoll_wait不需要传入任何信息，并且返回的结果只包含就绪的描述符，这样就不用去判断所有描述符。

从概念上理解epoll是这样的，把要监控的fd的IO事件注册给epoll(调用epoll_ctl)，然后调用epoll的API等待事件到达(调用epoll_wait)，内核可能对每个fd维护着一个读和写的缓冲区，那么：如果我监控读事件，并且读缓冲区有数据了，epoll_wait就会返回，此时我可以调用read读数据。

如果我监控写事件，并且写缓存区未满，epoll_wait也会返回，此时我可以调用write写数据。

如果fd发生了一些错误，epoll_wait也会返回，此时我根据返回的标志位，就可以知道。

如果我监控读事件 并且有客户端连接进来，epoll_wait就会返回，此时我可以调用accept接受客户端。

epoll的API介绍int epoll_create(int size);

创建一个epoll实例，返回代表实例的文件描述符(fd)，size自Linux 2.6.8以后忽略，但必须大于0.

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll控制接口，epfd就是epoll的文件描述符，fd是要操作的文件描述符，op有如下几种：EPOLL_CTL_ADD 注册fd的事件，事件类型在event指定。

EPOLL_CTL_MOD 修改已注册的fd事件。

EPOLL_CTL_DEL 删除fd的事件。

epoll_event有一个events成员，指定要注册的事件类型，比较重要的几个：EPOLLIN fd可读事件

EPOLLOUT fd可写事件

EPOLLERR fd发生错误，这个事件总是会被监控，不必手动增加

EPOLLHUP fd被挂起时，这个事件总是会被监控，不必手动增加，这通常发生在socket异常关闭时，此时read返回0，然后正常的清理socket资源。

epoll_event还有一个epoll_data_t成员，由外部设置自定义数据，以方便后续处理。int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

等待事件发生，如果没有事件发生，线程会被挂起，maxevents指定最大事件数，events外部传入的事件数组，长度应当等于maxevents，当事件发生时，epoll会把事件信息填到这里, timeout指定等待的最大时间, 0表示马上返回，-1表示无限等待。

epoll_wait返回等待到的事件数，返回时，遍历events对fd进行处理。 当epoll不再使用时，应该调用close关闭epollfd。

水平触发和边缘触发

epoll触发事件有两种模式，默认的叫水平触发(LT)，另一种叫边缘触发(ET)：LT模式：只要fd的读缓冲区不空，或写缓冲区不满，epoll_wait就会一直触发事件(也就是返回)。

ET模式：当被监控的fd状态变化时(从未就绪变成就绪状态)，事件触发一次。此后内核不再通知，除非有新的事件到来。

// 多谢 @黄蔚 的指正，原来ET模式的描述有误，仔细阅读过man文档后已修正过来。

LT处理起来比ET要简单得多，读事件触发，只需要read一次，如果数据没读完，下次epoll_wait还会返回，写也是一样的；ET模式就要求事件触发时，一直读一直写直到明确知道已经读写完毕(返回EAGIN或EWOULDBLOCK的错误码)。

水平触发的服务器程序大概是这样的流程：accept一个新连接，将这个新连接的fd加到epoll事件中，监听EPOLLIN事件。

EPOLLIN事件到达时，read该fd中的数据。

如果要向该fd写事件，向epoll增加EPOLLOUT事件。

EPOLLOUT事件到达，向fd write数据，如果数据太大无法一次写出，那么先保留EPOLLOUT事件，下次事件到达继续写；如果写出完毕，从epoll删除EPOLLOUT事件。

一个实用的echo程序：

这一次我们要使用epoll和非阻塞socket来写一个真正实用的echo服务器，调用fcntl函数，设置O_NONBLOCK标志位，即可让socket的文件描述符变成非阻塞模式。非阻塞模式处理起来比阻塞模式要复杂一些：read, write, accept这些函数不会阻塞，要么成功，要么返回-1失败，errno记录了失败的原因，有几个错误码要关注：EAGAIN 或 EWOULDBLOCK 只有非阻塞的fd才会发生，表示没数据可读，或没空间可写，或没有客户端可接受，下次再来吧。这两个值可能相同也可能不同，最好一起判断。

EINTR 表示被信号中断，这种情况可以再一次尝试调用。

其他错误表示真的出错了。

向一个fd写数据比较麻烦，我们没法保证一次性把所有数据都写完，所以需要先保存在缓冲里，然后向epoll增加写事件，事件触发时再向fd写数据。等数据都写完了，再把事件从epoll移除。这个程序把写数据保存在链表中。

我们把监听fd保留为阻塞模式，因为epoll_wait返回，可以确定一定有客户端连接进来，所以accept一般可以成功，并不用担心会阻塞。客户端连接使用的是非阻塞模式，确保读写未完成时不会阻塞。

下面是这个程序的代码，关键地方加了一些注释，仔细看代码比看文字描述更有用：)

#include "socket_lib h"#include #include #include #include #include

#define MAX_CLIENT 10000#define MIN_RSIZE 124#define BACKLOG 128#define EVENT_NUM 64

// 缓存结点struct twbuffer {

struct twbuffer *next; // 下一个缓存 void *buffer; // 缓存 char *ptr; // 当前未发送的缓存，buffer != ptr表示只发送了一部分 int size; // 当前未发送的缓存大小};

// 缓存列表struct twblist {

struct twbuffer *head;

struct twbuffer *tail;

};

// 客户端连接信息struct tclient {

int fd; // 客户端fd int rsize; // 当前读的缓存区大小 int wbsize; // 还未写完的缓存大小 struct twblist wblist; // 写缓存链表};

// 服务器信息struct tserver {

int listenfd; // 监听fd int epollfd; // epollfd struct tclient clients[MAX_CLIENT]; // 客户端结构数组};

// epoll增加读事件void epoll_add(int efd, int fd, void *ud) {

struct epoll_event ev;

ev.events = EPOLLIN;

ev.data.ptr = ud;

epoll_ctl(efd, EPOLL_CTL_ADD, fd, &ev);

}

// epoll修改写事件void epoll_write(int efd, int fd, void *ud, int enabled) {

struct epoll_event ev;

ev.events = EPOLLIN | (enabled ? EPOLLOUT : 0);

ev.data.ptr = ud;

epoll_ctl(efd, EPOLL_CTL_MOD, fd, &ev);

}

// epoll删除fdvoid epoll_del(int efd, int fd) {

epoll_ctl(efd, EPOLL_CTL_DEL, fd, NULL);

}

// 设置socket为非阻塞void set_nonblocking(int fd) {

int flag = fcntl(fd, F_GETFL, 0);

if (flag >= 0) {

fcntl(fd, F_SETFL, flag | O_NONBLOCK);

}

}

// 增加写缓存void add_wbuffer(struct twblist *list, void *buffer, int sz) {

struct twbuffer *wb = malloc(sizeof(*wb));

wb->buffer = buffer;

wb->ptr = buffer;

wb->size = sz;

wb->next = NULL;

if (!list->head) {

list->head = list->tail = wb;

} else {

list->tail->next = wb;

list->tail = wb;

}

}

// 释放写缓存void free_wblist(struct twblist *list) {

struct twbuffer *wb = list->head;

while (wb) {

struct twbuffer *tmp = wb;

wb = wb->next;

free(tmp);

}

list->head = NULL;

list->tail = NULL;

}

// 创建客户端信息struct tclient* create_client(struct tserver *server, int fd) {

int i;

struct tclient *client = NULL;

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

if (server->clients[i].fd < 0) {

client = &server->clients[i];

break;

}

}

if (client) {

client->fd = fd;

client->rsize = MIN_RSIZE;

set_nonblocking(fd); // 设为非阻塞模式 epoll_add(server->epollfd, fd, client); // 增加读事件 return client;

} else {

fprintf(stderr, "too many client: %d\n", fd);

close(fd);

return NULL;

}

}

// 关闭客户端void close_client(struct tserver *server, struct tclient *client) {

assert(client->fd >= 0);

epoll_del(server->epollfd, client->fd);

if (close(client->fd) < 0) perror("close: ");

client->fd = -1;

client->wbsize = 0;

free_wblist(&client->wblist);

}

// 初始化服务信息struct tserver* create_server(const char *host, const char *port) {

struct tserver *server = malloc(sizeof(*server));

memset(server, 0, sizeof(*server));

for (int i = 0; i < MAX_CLIENT; ++i) {

server->clients[i].fd = -1;

}

server->epollfd = epoll_create(MAX_CLIENT);

server->listenfd = tcpListen(host, port, BACKLOG);

epoll_add(server->epollfd, server->listenfd, NULL);

return server;

}

// 释放服务器void release_server(struct tserver *server) {

for (int i = 0; i < MAX_CLIENT; ++i) {

struct tclient *client = &server->clients[i];

if (client->fd >= 0) {

close_client(server, client);

}

}

epoll_del(server->epollfd, server->listenfd);

close(server->listenfd);

close(server->epollfd);

free(server);

}

// 处理接受void handle_accept(struct tserver *server) {

struct sockaddr_storage claddr;

socklen_t addrlen = sizeof(struct sockaddr_storage);

for (;;) {

int cfd = accept(server->listenfd, (struct sockaddr*)&claddr, &addrlen);

if (cfd < 0) {

int no = errno;

if (no == EINTR)

continue;

perror("accept: ");

exit(1); // 出错 }

char host[NI_MAXHOST];

char service[NI_MAXSERV];

if (getnameinfo((struct sockaddr *)&claddr, addrlen, host, NI_MAXHOST, service, NI_MAXSERV, 0) == 0)

printf("client connect: fd=%d, (%s:%s)\n", cfd, host, service);

else

printf("client connect: fd=%d, (?UNKNOWN?)\n", cfd);

create_client(server, cfd);

break;

}

}

// 处理读void handle_read(struct tserver *server, struct tclient *client) {

int sz = client->rsize;

char *buf = malloc(sz);

ssize_t n = read(client->fd, buf, sz);

if (n < 0) { // error free(buf);

int no = errno;

if (no != EINTR && no != EAGAIN && no != EWOULDBLOCK) {

perror("read: ");

close_client(server, client);

}

return;

}

if (n == 0) { // client close free(buf);

printf("client close: %d\n", client->fd);

close_client(server, client);

return;

}

// 确定下一次读的大小 if (n == sz)

client->rsize >>= 1;

else if (sz > MIN_RSIZE && n *2 < sz)

client->rsize <<= 1;

// 加入写缓存 add_wbuffer(&client->wblist, buf, n);

// 增加写事件 epoll_write(server->epollfd, client->fd, client, 1);

}

// 处理写void handle_write(struct tserver *server, struct tclient *client) {

struct twblist *list = &client->wblist;

while (list->head) {

struct twbuffer *wb = list->head;

for (;;) {

ssize_t sz = write(client->fd, wb->ptr, wb->size);

if (sz < 0) {

int no = errno;

if (no == EINTR) // 信号中断，继续 continue;

else if (no == EAGAIN || no == EWOULDBLOCK) // 内核缓冲满了，下次再来 return;

else { // 其他错误 perror("write: ");

close_client(server, client);

return;

}

}

client->wbsize -= sz;

if (sz != wb->size) { // 未完全发送出去，下次再来 wb->ptr += sz;

wb->size -= sz;

return;

}

break;

}

list->head = wb->next;

free(wb);

}

list->tail = NULL;

// 到这里写全部完成，关闭写事件 epoll_write(server->epollfd, client->fd, client, 0);

}

// 先处理错误void handle_error(struct tserver *server, struct tclient *client) {

perror("client error: ");

close_client(server, client);

}

int main() {

signal(SIGPIPE, SIG_IGN);

struct tserver *server = create_server("127.0.0.1", "3459");

struct epoll_event events[EVENT_NUM];

for (;;) {

int nevent = epoll_wait(server->epollfd, events, EVENT_NUM, -1);

if (nevent <= 0) {

if (nevent < 0 && errno != EINTR) {

perror("epoll_wait: ");

return 1;

}

continue;

}

int i = 0;

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

struct epoll_event ev = events[i];

if (ev.data.ptr == NULL) { // accept handle_accept(server);

} else {

if (ev.events & (EPOLLIN | EPOLLHUP)) { // read handle_read(server, ev.data.ptr);

}

if (ev.events & EPOLLOUT) { // write handle_write(server, ev.data.ptr);

}

if (ev.events & EPOLLERR) { // error handle_error(server, ev.data.ptr);

}

}

}

}

release_server(server);

return 0;

}