Linux 下 I/O 多路复用技术 epoll

流？I/O操作？阻塞？
- 解决阻塞死等待的方法
- - 方法一：阻塞 + 多进程 / 多线程。
  - 方法二：非阻塞 + 忙轮询。
  - 方法三：select（与平台无关）
  - 方法四：epoll（Linux）
- epoll 的 API
- - 创建 epoll
  - 控制 epoll
  - 等待 epoll
  - epoll 编程主流骨架
- epoll 的触发模式
- - 水平触发
  - 边缘触发
- 简单的 epoll 服务器
- - 服务端
  - 客户端

致正在学习的你，无论前方会怎样？至少现在你是对的。

学习视频：深入理解 Linux 中网络 I/O 复用并发模型_哔哩哔哩

流？I/O操作？阻塞？

流：可以进行 I/O 操作的内核对象

文件、管道、套接字……
流的入口：文件描述符(fd)

I/O 操作：所有对流的读写操作，都可以称之为 IO 操作。

当一个流中，在没有数据进行 read 的时候，或者在流中已经写满了数据，再进行 write，我们的 IO 操作就会出现一种现象，就是阻塞现象，如下图。

阻塞等待：不占⽤ CPU 宝贵的时间⽚

非阻塞忙轮询：占用 CPU，系统资源

在处理数据的接收场景时，建议优先选择阻塞等待的⽅式，不浪费性能资源。

阻塞等待的缺点：

不能够很好的处理多个 I/O 请求的问题
同一个阻塞，同⼀时刻只能处理一个流的阻塞监听

解决阻塞死等待的方法

方法一：阻塞 + 多进程 / 多线程。

该方法需要开辟线程浪费资源。

方法二：非阻塞 + 忙轮询。

该方法中 CPU 在大量的做论询（while 和 for），CPU 利用率不高。

实现伪代码：

while true {for i in 流[] {if i has 数据 {读 或者 其他处理}}
}

方法三：select（与平台无关）

select 监听的 IO 数量有限，默认是 1024 个。它不会精准的告诉开发者，哪些 IO 可读可写，需要遍历。

实现伪代码：

while true {select(流[]); // 阻塞// 有消息抵达for i in 流[] {if i has 数据 {读 或者 其他处理}}
}

方法四：epoll（Linux）

epoll 与 select、poll 一样，是对 I/O 多路复用的技术

只关心 “活跃” 的链接，无需遍历全部描述符集合
能够处理大量的链接请求（系统可以打开的文件数目）

可以通过 /proc/sys/fd/file-max 查看

while true {可处理的流[] = epoll_wait(epoll_fd); // 阻塞// 有消息抵达，全部放在 “可处理的流[]”中for i in 可处理的流[] {读 或者 其他处理}
}

epoll 的 API

创建 epoll

/** * @param size 告诉内核监听的数目 * * @returns 返回一个epoll句柄（即一个文件描述符） */
int epoll_create(int size);

使用：

int epfd = epoll_create(1000);

创建一个 epoll 句柄，实际上是在内核空间，建立一颗红黑树的根节点，这个根节点的关系与 epfd 相对应。

控制 epoll

/**
* @param epfd 用epoll_create所创建的epoll句柄
* @param op 表示对epoll监控描述符控制的动作
*
* EPOLL_CTL_ADD(注册新的fd到epfd)
* EPOLL_CTL_MOD(修改已经注册的fd的监听事件)
* EPOLL_CTL_DEL(epfd删除一个fd)
*
* @param fd 需要监听的文件描述符
* @param event 告诉内核需要监听的事件
*
* @returns 成功返回0，失败返回-1, errno查看错误信息
*/
int epoll_ctl(int epfd, int op, int fd,
struct epoll_event *event);

struct epoll_event {__uint32_t events; /* epoll 事件 */epoll_data_t data; /* 用户传递的数据 */
}/** events : {EPOLLIN, EPOLLOUT, EPOLLPRI,EPOLLHUP, EPOLLET, EPOLLONESHOT}*/
typedef union epoll_data {void *ptr;int fd;uint32_t u32;uint64_t u64;
} epoll_data_t;

使用：

struct epoll_event new_event;new_event.events = EPOLLIN | EPOLLOUT;
new_event.data.fd = 5;epoll_ctl(epfd, EPOLL_CTL_ADD, 5, &new_event);

创建一个用户态的事件，绑定到某个 fd 上，然后添加到内核中的 epoll 红黑树中。

等待 epoll

/**
*
* @param epfd 用epoll_create所创建的epoll句柄
* @param event 从内核得到的事件集合
* @param maxevents 告知内核这个events有多大,
* 注意: 值 不能大于创建epoll_create()时的size.
* @param timeout 超时时间
* -1: 永久阻塞
* 0: 立即返回，非阻塞
* >0: 指定微秒
*
* @returns 成功: 有多少文件描述符就绪,时间到时返回0
* 失败: -1, errno 查看错误
*/
int epoll_wait(int epfd, struct epoll_event *event,int maxevents, int timeout);

使用：

struct epoll_event my_event[1000];int event_cnt = epoll_wait(epfd, my_event, 1000, -1);

epoll_wait 是一个阻塞的状态，如果内核检测到 IO 的读写响应，会抛给上层的 epoll_wait，返回给用户态一个已经触发的事件队列，同时阻塞返回。开发者可以从队列中取出事件来处理，其中事件里就有绑定的对应 fd 是哪个（之前添加 epoll 事件的时候已经绑定）。

epoll 编程主流骨架

// 创建 epoll
int epfd = epoll_crete(1000);// 将 listen_fd 添加进 epoll 中
epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd,&listen_event);while (1) {// 阻塞等待 epoll 中的 fd 触发int active_cnt = epoll_wait(epfd, events, 1000, -1);for (i = 0 ; i < active_cnt; i++) {if (evnets[i].data.fd == listen_fd) {// accept. 并且将新 accept 的 fd 加进 epoll 中.}else if (events[i].events & EPOLLIN) {// 对此 fd 进行读操作}else if (events[i].events & EPOLLOUT) {// 对此 fd 进行写操作}}
}

epoll 的触发模式

水平触发

水平触发的主要特点是，如果用户在监听 epoll 事件，当内核有事件的时候，会拷贝给用户态事件，但是如果用户只处理了一次，那么剩下没有处理的会在下一次 epoll_wait 再次返回该事件。

如果用户永远不处理这个事件，就导致每次都会有该事件从内核到用户的拷贝，耗费性能，但是水平触发相对安全，最起码事件不会丢掉，除非用户处理完毕。

边缘触发

边缘触发，相对跟水平触发相反，当内核有事件到达，只会通知用户一次，至于用户处理还是不处理，以后将不会再通知。这样减少了拷贝过程，增加了性能，但是相对来说，如果用户马虎忘记处理，将会产生事件丢的情况。

简单的 epoll 服务器

服务端

#include <stdio.h>
#include <stdlib.h>
#include <ctype.h>
#include <string.h>#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>#include <sys/epoll.h>#define SERVER_PORT (7778)
#define EPOLL_MAX_NUM (2048)
#define BUFFER_MAX_LEN (4096)char buffer[BUFFER_MAX_LEN];void str_toupper(char *str)
{int i;for (i = 0; i < strlen(str); i ++) {str[i] = toupper(str[i]);}
}int main(int argc, char **argv)
{int listen_fd = 0;int client_fd = 0;struct sockaddr_in server_addr;struct sockaddr_in client_addr;socklen_t client_len;int epfd = 0;struct epoll_event event, *my_events;/ socketlisten_fd = socket(AF_INET, SOCK_STREAM, 0);// bindserver_addr.sin_family = AF_INET;server_addr.sin_addr.s_addr = htonl(INADDR_ANY);server_addr.sin_port = htons(SERVER_PORT);bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr));// listenlisten(listen_fd, 10);// epoll createepfd = epoll_create(EPOLL_MAX_NUM);if (epfd < 0) {perror("epoll create");goto END;}// listen_fd -> epollevent.events = EPOLLIN;event.data.fd = listen_fd;if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event) < 0) {perror("epoll ctl add listen_fd ");goto END;}my_events = malloc(sizeof(struct epoll_event) * EPOLL_MAX_NUM);while (1) {// epoll waitint active_fds_cnt = epoll_wait(epfd, my_events, EPOLL_MAX_NUM, -1);int i = 0;for (i = 0; i < active_fds_cnt; i++) {// if fd == listen_fdif (my_events[i].data.fd == listen_fd) {//acceptclient_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &client_len);if (client_fd < 0) {perror("accept");continue;}char ip[20];printf("new connection[%s:%d]\n", inet_ntop(AF_INET, &client_addr.sin_addr, ip, sizeof(ip)), ntohs(client_addr.sin_port));event.events = EPOLLIN | EPOLLET;event.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &event);}else if (my_events[i].events & EPOLLIN) {printf("EPOLLIN\n");client_fd = my_events[i].data.fd;// do readbuffer[0] = '\0';int n = read(client_fd, buffer, 5);if (n < 0) {perror("read");continue;}else if (n == 0) {epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &event);close(client_fd);}else {printf("[read]: %s\n", buffer);buffer[n] = '\0';
#if 1str_toupper(buffer);write(client_fd, buffer, strlen(buffer));printf("[write]: %s\n", buffer);memset(buffer, 0, BUFFER_MAX_LEN);
#endif/*event.events = EPOLLOUT;event.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event);*/}}else if (my_events[i].events & EPOLLOUT) {printf("EPOLLOUT\n");/*client_fd = my_events[i].data.fd;str_toupper(buffer);write(client_fd, buffer, strlen(buffer));printf("[write]: %s\n", buffer);memset(buffer, 0, BUFFER_MAX_LEN);event.events = EPOLLIN;event.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event);*/}}}END:close(epfd);close(listen_fd);return 0;
}

客户端

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>#define MAX_LINE (1024)
#define SERVER_PORT (7778)void setnoblocking(int fd)
{int opts = 0;opts = fcntl(fd, F_GETFL);opts = opts | O_NONBLOCK;fcntl(fd, F_SETFL);
}int main(int argc, char **argv)
{int sockfd;char recvline[MAX_LINE + 1] = {0};struct sockaddr_in server_addr;if (argc != 2) {fprintf(stderr, "usage ./client <SERVER_IP>\n");exit(0);}// 创建socketif ( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {fprintf(stderr, "socket error");exit(0);}// server addr 赋值bzero(&server_addr, sizeof(server_addr));server_addr.sin_family = AF_INET;server_addr.sin_port = htons(SERVER_PORT);if (inet_pton(AF_INET, argv[1], &server_addr.sin_addr) <= 0) {fprintf(stderr, "inet_pton error for %s", argv[1]);exit(0);}// 链接服务端if (connect(sockfd, (struct sockaddr*) &server_addr, sizeof(server_addr)) < 0) {perror("connect");fprintf(stderr, "connect error\n");exit(0);}setnoblocking(sockfd);char input[100];int n = 0;int count = 0;// 不断的从标准输入字符串while (fgets(input, 100, stdin) != NULL){printf("[send] %s\n", input);n = 0;// 把输入的字符串发送 到 服务器中去n = send(sockfd, input, strlen(input), 0);if (n < 0) {perror("send");}n = 0;count = 0;// 读取 服务器返回的数据while (1){n = read(sockfd, recvline + count, MAX_LINE);if (n == MAX_LINE){count += n;continue;}else if (n < 0){perror("recv");break;}else {count += n;recvline[count] = '\0';printf("[recv] %s\n", recvline);break;}}}return 0;
}