一、epoll函数接口

创建epoll实例

int epoll_create1(int flags);

函数参数：

flags: 当前版本只支持EPOLL_CLOEXEC标志(请注意不支持EPOLL_NONBLOCK标志)

其实我们也能够通过epoll_create(int size)这个函数来创建epoll实例，只不过这个函数中的size在2.6.27内核开始就不必要了，新的内核已经能够动态地管理所需的内存分配了。我们视之为废弃。

根据惯例，如果返回-1，则标志出现了问题，我们可以读取errno来定位错误，有如下errno会被设置：

EINVAL : 无效的标志

EMFILE : 用户打开的文件超过了限制

ENFILE : 系统打开的文件超过了限制

ENOMEM : 没有足够的内存完成当前操作

管理epoll事件

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

函数参数：

epfd : epoll实例的fd

op : 操作标志，下文会描述

fd : 监控对象的fd

event : 事件的内容，下文描述

op可以有3个值，分别为：

EPOLL_CTL_ADD : 添加监听的事件

EPOLL_CTL_DEL : 删除监听的事件

EPOLL_CTL_MOD : 修改监听的事件

event是一个如下结构体的一个实例：

typedef union epoll_data {

void *ptr;

int fd;

__uint32_t u32;

__uint64_t u64;

} epoll_data_t;

struct epoll_event {

__uint32_t events; /* Epoll events */

epoll_data_t data; /* User data variable */

};

其中，data是一个联合体，能够存储fd或其它数据，我们需要根据自己的需求定制。events表示监控的事件的集合，是一个状态值，通过状态位来表示，可以设置如下事件：

EPOLLERR : 文件上发上了一个错误。这个事件是一直监控的，即使没有明确指定

EPOLLHUP : 文件被挂断。这个事件是一直监控的，即使没有明确指定

EPOLLRDHUP : 对端关闭连接或者shutdown写入半连接

EPOLLET : 开启边缘触发，默认的是水平触发，所以我们并未看到EPOLLLT

EPOLLONESHOT : 一个事件发生并读取后，文件自动不再监控

EPOLLIN : 文件可读

EPOLLPRI : 文件有紧急数据可读

EPOLLOUT : 文件可写

EPOLLWAKEUP : 如果EPOLLONESHOT和EPOLLET清除了，并且进程拥有CAP_BLOCK_SUSPEND权限，那么这个标志能够保证事件在挂起或者处理的时候，系统不会挂起或休眠

注意一下，EPOLLHUP并不代表对端结束了连接，这一点需要和EPOLLRDHUP区分。通常情况下EPOLLHUP表示的是本端挂断，造成这种事件出现的原因有很多，其中一种便是出现错误，更加细致的应该是和RST联系在一起，不过目前相关文档并不是很全面，本文会进一步跟进。

根据惯例，如果返回-1，则标志出现了问题，我们可以读取errno来定位错误，有如下errno会被设置：

EBADF : epfd或者fd不是一个有效的文件描述符

EEXIST : op为EPOLL_CTL_ADD，但fd已经被监控

EINVAL : epfd是无效的epoll文件描述符

ENOENT : op为EPOLL_CTL_MOD或者EPOLL_CTL_DEL，并且fd未被监控

ENOMEM : 没有足够的内存完成当前操作

ENOSPC : epoll实例超过了/proc/sys/fs/epoll/max_user_watches中限制的监听数量

等待epoll事件

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

函数参数：

epfd : epoll实例的fd

events : 储存事件的数组首地址

maxevents : 最大事件的数量

timeout : 等待的最长时间

如果函数返回获得的时间的数量，如果返回-1，则标志出现了问题，我们可以读取errno来定位错误，有如下errno会被设置：

EBADF : epfd不是一个有效的文件描述符

EFAULT : events指向的内存无权访问

EINTR : 在请求事件发生或者过期之前，调用被信号打断

EINVAL : epfd是无效的epoll文件描述符

二、ET与LT模式读写示例

ET模型的逻辑：内核的读buffer有内核态主动变化时，内核会通知你， 无需再去mod。写事件是给用户使用的，最开始add之后，内核都不会通知你了，你可以强制写数据（直到EAGAIN或者实际字节数小于 需要写的字节数），当然你可以主动mod OUT，此时如果句柄可以写了（send buffer有空间），内核就通知你。

这里内核态主动的意思是：内核从网络接收了数据放入了读buffer（会通知用户IN事件，即用户可以recv数据）

并且这种通知只会通知一次，如果这次处理（recv）没有到刚才说的两种情况（EAGIN或者实际字节数小于 需要读写的字节数），则该事件会被丢弃，直到下次buffer发生变化。

与LT的差别就在这里体现，LT在这种情况下，事件不会丢弃，而是只要读buffer里面有数据可以让用户读，则不断的通知你。

另外对于ET而言，当然也不一定非send/recv到前面所述的结束条件才结束，用户可以自己随时控制，即用户可以在自己认为合适的时候去设置IN和OUT事件：

1 如果用户主动epoll_mod OUT事件，此时只要该句柄可以发送数据（发送buffer不满），则epoll

_wait就会响应（有时候采用该机制通知epoll_wai醒过来）。

2 如果用户主动epoll_mod IN事件，只要该句柄还有数据可以读，则epoll_wait会响应。

这种逻辑在普通的服务里面都不需要，可能在某些特殊的情况需要。 但是请注意，如果每次调用的时候都去epoll mod将显著降低效率，已经吃过几次亏了！

因此采用et写服务框架的时候，最简单的处理就是：

建立连接的时候epoll_add IN和OUT事件， 后面就不需要管了

每次read/write的时候，到两种情况下结束：

1 发生EAGAIN

2 read/write的实际字节数小于 需要读写的字节数

对于第二点需要注意两点：

A：如果是UDP服务，处理就不完全是这样，必须要recv到发生EAGAIN为止，否则就丢失事件了

因为UDP和TCP不同，是有边界的，每次接收一定是一个完整的UDP包，当然recv的buffer需要至少大于一个UDP包的大小

随便再说一下，一个UDP包到底应该多大？

对于internet，由于MTU的限制，UDP包的大小不要超过576个字节，否则容易被分包，对于公司的IDC环境，建议不要超过1472，否则也比较容易分包。

B 如果发送方发送完数据以后，就close连接，这个时候如果recv到数据是实际字节数小于读写字节数，根据开始所述就认为到EAGIN了从而直接返回，等待下一次事件，这样是有问题的，close事件丢失了！

因此如果依赖这种关闭逻辑的服务，必须接收数据到EAGIN为止，例如lb。

在一个非阻塞的socket上调用read/write函数, 返回EAGAIN或者EWOULDBLOCK(注: EAGAIN就是EWOULDBLOCK)

从字面上看, 意思是:

* EAGAIN: 再试一次

* EWOULDBLOCK: 如果这是一个阻塞socket, 操作将被block

* perror输出: Resource temporarily unavailable

总结:

这个错误表示资源暂时不够, 可能read时, 读缓冲区没有数据, 或者, write时,

写缓冲区满了.

遇到这种情况, 如果是阻塞socket, read/write就要阻塞掉.

而如果是非阻塞socket, read/write立即返回-1, 同 时errno设置为EAGAIN.

所以, 对于阻塞socket, read/write返回-1代表网络出错了.

但对于非阻塞socket, read/write返回-1不一定网络真的出错了.

可能是Resource temporarily unavailable. 这时你应该再试, 直到Resource available.

综上, 对于non-blocking的socket, 正确的读写操作为:

读: 忽略掉errno = EAGAIN的错误, 下次继续读 

写: 忽略掉errno = EAGAIN的错误, 下次继续写 

对于select和epoll的LT模式, 这种读写方式是没有问题的. 但对于epoll的ET模式, 这种方式还有漏洞.

epoll的两种模式 LT 和 ET

二者的差异在于 level-trigger 模式下只要某个 socket 处于 readable/writable 状态，无论什么时候

进行 epoll_wait 都会返回该 socket；而 edge-trigger 模式下只有某个 socket 从 unreadable 变为 readable 或从unwritable 变为 writable 时，epoll_wait 才会返回该 socket。

如下两个示意图:

从socket读数据:

往socket写数据

所以, 在epoll的ET模式下, 正确的读写方式为:

读: 只要可读, 就一直读, 直到返回0, 或者 errno = EAGAIN

写: 只要可写, 就一直写, 直到数据发送完, 或者 errno = EAGAIN

正确的读:

n = 0;

while ((nread = read(fd, buf + n, BUFSIZ-1)) > 0) {

n += nread;

}

if (nread == -1 && errno != EAGAIN) {

perror("read error");

}

正确的写:

int nwrite, data_size = strlen(buf);

n = data_size;

while (n > 0) {

nwrite = write(fd, buf + data_size - n, n);

if (nwrite < n) {

if (nwrite == -1 && errno != EAGAIN) {

perror("write error");

}

break;

}

n -= nwrite;

}

正确的accept，accept 要考虑 2 个问题

(1) 阻塞模式 accept 存在的问题

考虑这种情况： TCP 连接被客户端夭折，即在服务器调用 accept 之前，客户端主动发送 RST 终止

连接，导致刚刚建立的连接从就绪队列中移出，如果套接口被设置成阻塞模式，服务器就会一直阻塞

在 accept 调用上，直到其他某个客户建立一个新的连接为止。但是在此期间，服务器单纯地阻塞在

accept 调用上，就绪队列中的其他描述符都得不到处理.

解决办法是把监听套接口设置为非阻塞，当客户在服务器调用 accept 之前中止某个连接时，accept 调用

可以立即返回 -1， 这时源自 Berkeley 的实现会在内核中处理该事件，并不会将该事件通知给 epool，

而其他实现把 errno 设置为 ECONNABORTED 或者 EPROTO 错误，我们应该忽略这两个错误。

(2) ET 模式下 accept 存在的问题

考虑这种情况：多个连接同时到达，服务器的 TCP 就绪队列瞬间积累多个就绪连接，由于是边缘触发模式，

epoll 只会通知一次，accept 只处理一个连接，导致 TCP 就绪队列中剩下的连接都得不到处理。

解决办法是用 while 循环抱住 accept 调用，处理完 TCP 就绪队列中的所有连接后再退出循环。如何知道

是否处理完就绪队列中的所有连接呢？ accept 返回 -1 并且 errno 设置为 EAGAIN 就表示所有连接都处理完。

综合以上两种情况，服务器应该使用非阻塞地 accept， accept 在 ET 模式下 的正确使用方式为：

while ((conn_sock = accept(listenfd,(struct sockaddr *) &remote,

(size_t *)&addrlen)) > 0) {

handle_client(conn_sock);

}

if (conn_sock == -1) {

if (errno != EAGAIN && errno != ECONNABORTED

&& errno != EPROTO && errno != EINTR)

perror("accept");

}

另 if (errno == EINTR) //可能被中断信号打断,,经过验证对非阻塞socket并未收到此错误,应该可以省掉该步判断

使用Linux epoll模型，水平触发模式；当socket可写时，会不停的触发 socket 可写的事件，如何处理？

第一种最普遍的方式：

需要向 socket 写数据的时候才把 socket 加入 epoll ，等待可写事件。

接受到可写事件后，调用 write 或者 send 发送数据。。。

当所有数据都写完后，把 socket 移出 epoll。

这种方式的缺点是，即使发送很少的数据，也要把 socket 加入 epoll，写完后在移出 epoll，有一定操作代价。

一种改进的方式：

开始不把 socket 加入 epoll，需要向 socket 写数据的时候，直接调用 write 或者 send 发送数据。

如果返回 EAGAIN，把 socket 加入 epoll，在 epoll 的驱动下写数据，全部数据发送完毕后，再移出 epoll。

这种方式的优点是：数据不多的时候可以避免 epoll 的事件处理，提高效率。

最后贴一个使用epoll, ET模式的简单HTTP服务器代码:

#include <sys/socket.h>

#include <sys/wait.h>

#include <netinet/in.h>

#include <netinet/tcp.h>

#include <sys/epoll.h>

#include <sys/sendfile.h>

#include <sys/stat.h>

#include <unistd.h>

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <strings.h>

#include <fcntl.h>

#include <errno.h>

#define MAX_EVENTS 10

#define PORT 8080

//设置socket连接为非阻塞模式

void setnonblocking(int sockfd) {

int opts;

opts = fcntl(sockfd, F_GETFL);

if(opts < 0) {

perror("fcntl(F_GETFL)\n");

exit(1);

}

opts = (opts | O_NONBLOCK);

if(fcntl(sockfd, F_SETFL, opts) < 0) {

perror("fcntl(F_SETFL)\n");

exit(1);

}

}

int main(){

struct epoll_event ev, events[MAX_EVENTS];

int addrlen, listenfd, conn_sock, nfds, epfd, fd, i, nread, n;

struct sockaddr_in local, remote;

char buf[BUFSIZ];

//创建listen socket

if( (listenfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {

perror("sockfd\n");

exit(1);

}

setnonblocking(listenfd);

bzero(&local, sizeof(local));

local.sin_family = AF_INET;

local.sin_addr.s_addr = htonl(INADDR_ANY);;

local.sin_port = htons(PORT);

if( bind(listenfd, (struct sockaddr *) &local, sizeof(local)) < 0) {

perror("bind\n");

exit(1);

}

listen(listenfd, 20);

epfd = epoll_create(MAX_EVENTS);

if (epfd == -1) {

perror("epoll_create");

exit(EXIT_FAILURE);

}

ev.events = EPOLLIN;

ev.data.fd = listenfd;

if (epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev) == -1) {

perror("epoll_ctl: listen_sock");

exit(EXIT_FAILURE);

}

for (;;) {

nfds = epoll_wait(epfd, events, MAX_EVENTS, -1);

if (nfds == -1) {

perror("epoll_pwait");

exit(EXIT_FAILURE);

}

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

fd = events[i].data.fd;

if (fd == listenfd) {

while ((conn_sock = accept(listenfd,(struct sockaddr *) &remote,

(size_t *)&addrlen)) > 0) {

setnonblocking(conn_sock);

ev.events = EPOLLIN | EPOLLET;

ev.data.fd = conn_sock;

if (epoll_ctl(epfd, EPOLL_CTL_ADD, conn_sock,

&ev) == -1) {

perror("epoll_ctl: add");

exit(EXIT_FAILURE);

}

}

if (conn_sock == -1) {

if (errno != EAGAIN && errno != ECONNABORTED

&& errno != EPROTO && errno != EINTR)

perror("accept");

}

continue;

}

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

n = 0;

while ((nread = read(fd, buf + n, BUFSIZ-1)) > 0) {

n += nread;

}

if (nread == -1 && errno != EAGAIN) {

perror("read error");

}

ev.data.fd = fd;

ev.events = events[i].events | EPOLLOUT;

if (epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev) == -1) {

perror("epoll_ctl: mod");

}

}

if (events[i].events & EPOLLOUT) {

sprintf(buf, "HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\nHello World", 11);

int nwrite, data_size = strlen(buf);

n = data_size;

while (n > 0) {

nwrite = write(fd, buf + data_size - n, n);

if (nwrite < n) {

if (nwrite == -1 && errno != EAGAIN) {

perror("write error");

}

break;

}

n -= nwrite;

}

close(fd);

}

}

}

return 0;

}

#include <netdb.h>

#include <sys/epoll.h>

#include <string.h>

#define MAXEVENTS 64

int create_and_bind (int port) {

int sfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

if(sfd == -1) {

return -1;

}

struct sockaddr_in sa;

bzero(&sa, sizeof(sa));

sa.sin_family = AF_INET;

sa.sin_port = htons(port);

sa.sin_addr.s_addr = htonl(INADDR_ANY);

if(bind(sfd, (struct sockaddr*)&sa, sizeof(struct sockaddr)) == -1) {

return -1;

}

return sfd;

}

int make_socket_non_blocking (int sfd) {

int flags = fcntl (sfd, F_GETFL, 0);

if (flags == -1) {

return -1;

}

if(fcntl (sfd, F_SETFL, flags | O_NONBLOCK) == -1) {

return -1;

}

return 0;

}

/* 此函数用于读取参数或者错误提示 */

int read_param(int argc, char *argv[]) {

if (argc != 2) {

fprintf (stderr, "Usage: %s [port]\n", argv[0]);

exit (EXIT_FAILURE);

}

return atoi(argv[1]);

}

int main (int argc, char *argv[]) {

int sfd, s;

int efd;

struct epoll_event event;

struct epoll_event *events;

int port = read_param(argc, argv);

/* 创建并绑定socket */

sfd = create_and_bind (port);

if (sfd == -1) {

perror("create_and_bind");

abort ();

}

/* 设置sfd为非阻塞 */

s = make_socket_non_blocking (sfd);

if (s == -1) {

perror("make_socket_non_blocking");

abort ();

}

/* SOMAXCONN 为系统默认的backlog */

s = listen (sfd, SOMAXCONN);

if (s == -1) {

perror ("listen");

abort ();

}

efd = epoll_create1 (0);

if (efd == -1) {

perror ("epoll_create");

abort ();

}

event.data.fd = sfd;

/* 设置ET模式 */

event.events = EPOLLIN | EPOLLET;

s = epoll_ctl (efd, EPOLL_CTL_ADD, sfd, &event);

if (s == -1) {

perror ("epoll_ctl");

abort ();

}

/* 创建事件数组并清零 */

events = calloc (MAXEVENTS, sizeof event);

/* 开始事件循环 */

while (1) {

int n, i;

n = epoll_wait (efd, events, MAXEVENTS, -1);

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

if (events[i].events & (EPOLLERR | EPOLLHUP)) {

/* 监控到错误或者挂起 */

fprintf (stderr, "epoll error\n");

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

continue;

}

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

if (sfd == events[i].data.fd) {

/* 处理新接入的socket */

while (1) {

struct sockaddr_in sa;

socklen_t len = sizeof(sa);

char hbuf[INET_ADDRSTRLEN];

int infd = accept (sfd, (struct sockaddr*)&sa, &len);

if (infd == -1) {

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

/* 资源暂时不可读，再来一遍 */

break;

} else {

perror ("accept");

break;

}

}

inet_ntop(AF_INET, &sa.sin_addr, hbuf, sizeof(hbuf));

printf("Accepted connection on descriptor %d "

"(host=%s, port=%d)\n", infd, hbuf, sa.sin_port);

/* 设置接入的socket为非阻塞 */

s = make_socket_non_blocking (infd);

if (s == -1) abort ();

/* 为新接入的socket注册事件 */

event.data.fd = infd;

event.events = EPOLLIN | EPOLLET;

s = epoll_ctl (efd, EPOLL_CTL_ADD, infd, &event);

if (s == -1) {

perror ("epoll_ctl");

abort ();

}

}

//continue;

} else {

/* 接入的socket有数据可读 */

while (1) {

ssize_t count;

char buf[512];

count = read (events[i].data.fd, buf, sizeof buf);

if (count == -1) {

if (errno != EAGAIN) {

perror ("read");

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

}

break;

} else if (count == 0) {

/* 数据读取完毕，结束 */

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

printf ("Closed connection on descriptor %d\n", events[i].data.fd);

break;

}

/* 输出到stdout */

s = write (1, buf, count);

if (s == -1) {

perror ("write");

abort ();

}

event.events = EPOLLOUT | EPOLLET;

epoll_ctl(efd, EPOLL_CTL_MOD, events[i].data.fd, &event);

}

}

} else if((events[i].events & EPOLLOUT) && (events[i].data.fd != sfd)) {

/* 接入的socket有数据可写 */

write(events[i].data.fd, "it's echo man\n", 14);

event.events = EPOLLET | EPOLLIN;

epoll_ctl(efd, EPOLL_CTL_MOD, events[i].data.fd, &event);

}

}

}

free (events);

close (sfd);

return EXIT_SUCCESS;

}

发布时间: 9年前【底层开发】118人已围观【返回】【回到顶端】

转载：个人转载请注明文章出处 "www.itarticle.cc-IT技术类文章收集&分享"。

很赞哦！ (1)