目录
前置文章
阅读本文需要先阅读下面的文章：
unix网络编程1.1——TCP协议详解（一）
unix网络编程2.1——高并发服务器（一）基础——io与文件描述符、socket编程与单进程服务端客户端实现
unix网络编程2.2——高并发服务器（二）多进程与多线程实现
unix网络编程2.3——高并发服务器（三）多路IO复用之select
综述
- epoll是Linux下多路复用IO接口select/poll的增强版本，它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率，因为它会复用文件描述符集合来传递结果而不用迫使开发者每次等待事件之前都必须重新准备要被侦听的文件描述符集合，另一点原因就是获取事件的时候，它无须遍历整个被侦听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就行了。
命令
查看一个进程可以打开的socket描述符上限
cat /proc/sys/fs/file-max
修改socket描述符上限
sudo vi /etc/security/limits.conf
在文件尾部写入以下配置,soft软限制，hard硬限制。如下图所示。
* soft nofile 65536
* hard nofile 100000
sysctl 命令允许你查看并且修改 Linux 内核参数
参考：https://blog.csdn.net/shixin_0125/article/details/78943245
modprobe 从 Linux 内核中 添加或者移除模块
参考：https://cloud.tencent.com/developer/article/1647140
epoll 函数api
epoll_create————创建一个epoll句柄
int epoll_create(int size) size：监听数目, 大于1就行，返回epfd，为epoll维护的红黑树的头节点
epoll_ctl————控制某个epoll监控的文件描述符上的事件：注册、修改、删除
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
epfd： 为epoll_creat的句柄
op： 表示动作，用3个宏来表示：
EPOLL_CTL_ADD (注册新的fd到epfd)，
EPOLL_CTL_MOD (修改已经注册的fd的监听事件)，
EPOLL_CTL_DEL (从epfd删除一个fd)；
event： 告诉内核需要监听的事件
struct epoll_event {
__uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};
typedef union epoll_data {
void *ptr;
int fd;
uint32_t u32;
uint64_t u64;
} epoll_data_t;
EPOLLIN ： 表示对应的文件描述符可以读（包括对端SOCKET正常关闭）
EPOLLOUT： 表示对应的文件描述符可以写
EPOLLPRI： 表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）
EPOLLERR： 表示对应的文件描述符发生错误
EPOLLHUP： 表示对应的文件描述符被挂断；
EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)而言的
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里
epoll_wait————类似于select调用，等待所监控文件描述符上有事件的产生
#include <sys/epoll.h>
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout)
events： 用来存内核得到事件的集合，
maxevents： 告之内核这个events有多大，这个maxevents的值不能大于创建epoll_create()时的size，
timeout： 是超时时间
-1： 阻塞
0： 立即返回，非阻塞
>0： 指定毫秒
返回值： 成功返回有多少文件描述符就绪，时间到时返回0，出错返回-1
水平触发与边沿触发
事件模型
EPOLL事件有两种模型：
Edge Triggered (ET) 边缘触发只有数据到来才触发，不管缓存区中是否还有数据。
Level Triggered (LT) 水平触发只要有数据都会触发。
思考如下步骤：
1.假定我们已经把一个用来从管道中读取数据的文件描述符(RFD)添加到epoll描述符。
2.管道的另一端写入了2KB的数据
3.调用epoll_wait，并且它会返回RFD，说明它已经准备好读取操作
4.读取1KB的数据
5.调用epoll_wait……
在这个过程中，有两种工作模式：
ET模式
ET模式即Edge Triggered工作模式。
如果我们在第1步将RFD添加到epoll描述符的时候使用了EPOLLET标志，那么在第5步调用epoll_wait之后将有可能会挂起，因为剩余的数据还存在于文件的输入缓冲区内，而且数据发出端还在等待一个针对已经发出数据的反馈信息。只有在监视的文件句柄上发生了某个事件的时候 ET 工作模式才会汇报事件。因此在第5步的时候，调用者可能会放弃等待仍在存在于文件输入缓冲区内的剩余数据。epoll工作在ET模式的时候，必须使用非阻塞套接口，以避免由于一个文件句柄的阻塞读/阻塞写操作把处理多个文件描述符的任务饿死。最好以下面的方式调用ET模式的epoll接口，在后面会介绍避免可能的缺陷。
1)基于非阻塞文件句柄
2)只有当read或者write返回EAGAIN(非阻塞读，暂时无数据)时才需要挂起、等待。但这并不是说每次read时都需要循环读，直到读到产生一个EAGAIN才认为此次事件处理完成，当read返回的读到的数据长度小于请求的数据长度时，就可以确定此时缓冲中已没有数据了，也就可以认为此事读事件已处理完成。
LT模式
LT模式即Level Triggered工作模式。
与ET模式不同的是，以LT方式调用epoll接口的时候，它就相当于一个速度比较快的poll，无论后面的数据是否被使用。
LT(level triggered)：LT是缺省的工作方式，并且同时支持block和no-block socket。在这种做法中，内核告诉你一个文件描述符是否就绪了，然后你可以对这个就绪的fd进行IO操作。如果你不作任何操作，内核还是会继续通知你的，所以，这种模式编程出错误可能性要小一点。传统的select/poll都是这种模型的代表。
ET(edge-triggered)：ET是高速工作方式，只支持no-block socket。在这种模式下，当描述符从未就绪变为就绪时，内核通过epoll告诉你。然后它会假设你知道文件描述符已经就绪，并且不会再为那个文件描述符发送更多的就绪通知。请注意，如果一直不对这个fd作IO操作(从而导致它再次变成未就绪)，内核不会发送更多的通知(only once).
如果recv的buf小于客户端发来数据的大小
- 客户端发了len=4的数据 "123\n"
- 服务端的接受buf大小只有2
epollout和epollin如何判断可读可写？
- 水平触发：当buf里有数据时会一直触发，比如客户端发来的数据Len=4，但是服务端Buf len = 2，那么就要分两次去发，每次发之前要判断io是否可写，其实也是在判断buf中是否还有数据
当客户端数量 大于 MAX_EVENTS时怎么办？
- 比如客户端有1024个连接（假设支持的并发量大于1024），而maxevents数量为128（小于客户端数量），此时epoll还是可以处理这1024的，处理的方式是每次轮询处理128，然后在main loop里下次再去轮询下一个128，直到1024个都轮询完
server.c
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <strings.h>
#include <ctype.h>
#include <unistd.h>
#include <stdio.h>
// struct sockaddr_in对应的头文件 <arpa/inet.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <errno.h>
#define IP_STR_SIZE 20
#define SERVER_PORT 9999
#define MAX_EVENTS 5000
#define BUFF_SIZE 1024
extern int errno;
char r_buf[BUFF_SIZE];
char w_buf[BUFF_SIZE];
/*
* 待实现：
* 1.listen_fd和conn_fd的sock_item有没有区别？
* 2.假色有100w的sock_item,如何快速插入和查找？
*/
// TODO: listen_fd + conn_fd
typedef struct sock_item {
int fd;
char *r_buf;
int r_len;
char *w_buf;
int w_len;
int event; // fd对应的事件
void (*recv)(int fd, char *buf, int len);
void (*send)(int fd, char *buf, int len);
void (*accept)(int fd);
} sock_item;
// TODO
typedef struct reactor {
int epfd;
} reactor;
void setnonblocking()
{
/* TODO: 设置fd为non-blocking*/
}
int main(void)
{
int listen_fd, conn_fd, socket_fd;
int i, n, res;
int n_ready;
char ip_str[IP_STR_SIZE];
struct sockaddr_in server_addr, client_addr;
int client_addr_len = sizeof(client_addr);
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(SERVER_PORT);
listen_fd = socket(AF_INET, SOCK_STREAM, 0);
bind(listen_fd, (struct sockaddr *)&server_addr, sizeof(server_addr));
listen(listen_fd, 128);
/**********端口复用代码************/
#if 1
int opt = 1;
setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
#endif
memset(r_buf, 0, sizeof(r_buf));
memset(w_buf, 0, sizeof(w_buf));
/* epoll init */
struct epoll_event ev, events[MAX_EVENTS];
int epfd = epoll_create(1);
ev.data.fd = listen_fd;
ev.events = EPOLLIN;
res = epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev); // 监听 listen_fd的读事件
if (res == -1) {
perror("epoll_ctl eror");
exit(-1);
}
printf("res=%d, epfd=%d\n", res, epfd);
while (1) {
int n_ready = epoll_wait(epfd, events, MAX_EVENTS, -1); // -1: 阻塞监听io对应的fd读写事件; 0: 非阻塞; 其他：定时，单位ms
if (n_ready == -1) {
perror("epoll_wait eror");
exit(-1);
}
printf("n_ready=%d\n", n_ready);
for (i = 0; i < n_ready; i++) {
int sock_fd = events[i].data.fd;
/*************************************建立连接阶段***************************************/
if (sock_fd == listen_fd) {
// 新的客户端发起连接
conn_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &client_addr_len);
if (conn_fd == -1) {
perror("accept eror");
exit(-1);
}
printf("conn_fd=%d\n", conn_fd);
ev.data.fd = conn_fd;
ev.events = EPOLLIN;
res = epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &ev); // 将conn_fd添加读事件
if (res == -1) {
perror("while epoll_ctl eror");
exit(-1);
}
} else {
/*************************************数据传输阶段***************************************/
if (events[i].events & EPOLLIN) {
printf("recv......\n");
res = recv(sock_fd, r_buf, sizeof(r_buf), 0);
// 多个客户端可能共用缓冲区，最好的方式是，每个fd定义一个结构体，数据隔离，封装
// r_buf[res] = '\0'; // 防止脏数据，但是这样写不好，应该每个fd维护一个结构体，抽象维护独立的buf
if (res == 0) {
// 客户端断开连接
epoll_ctl(epfd, EPOLL_CTL_DEL, sock_fd, NULL);
close(sock_fd);
} else if (res > 0) {
printf("res=%d, r_buf=%s\n", res, r_buf);
memcpy(w_buf, r_buf, sizeof(r_buf));
ev.data.fd = sock_fd;
ev.events = EPOLLOUT;
epoll_ctl(epfd, EPOLL_CTL_MOD, sock_fd, &ev); // 读完数据--->改sock_fd属性为写数据
} else {
perror("recv eror");
exit(-1);
}
} else if (events[i].events & EPOLLOUT) {
printf("send......\n");
res = send(sock_fd, w_buf, sizeof(w_buf), 0);
printf("res=%d, w_buf=%s\n", res, w_buf);
if (res == -1) {
printf("errno=%d\n", errno);
perror("send eror");
exit(-1);
}
ev.data.fd = sock_fd;
ev.events = EPOLLIN;
epoll_ctl(epfd, EPOLL_CTL_MOD, sock_fd, &ev); // 写完数据--->改sock_fd属性为读数据
}
}
}
}
return 0;
}
select与epoll比较
数据结构对比
- select： r_set和w_set都是位图，可以理解成数组，用位图去管理fds，搜索效率是O(n)
- epoll: 红黑树，搜索效率是O(logn)
支持并发量
- select： 1024，或者修改内核文件，千级别
- epoll: 使用reactor，可以支持到百万级别
开发难度
- select： r_set与w_set每次都要备份，多次循环遍历，比较麻烦
- epoll: 结构体epoll_event中的联合体中有void *ptr，这种函数指针很适合写回调函数，可以使监听io读写事件与处理读写事件解耦，适合大型项目使用
struct epoll_event {
__uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};
typedef union epoll_data {
void *ptr;
int fd;
uint32_t u32;
uint64_t u64;
} epoll_data_t;
总结
- 本文server.c实现了一个基础班的水平触发的epoll模型，但是遗留了结构体sock_item与reactor的进一步开发与使用，下一篇将基于此实现reactor模型，并且学习记录如何测试服务端的并发量