Linux高级IO

在平时read/recv时，如果内核缓冲区有数据，会拷贝到应用层缓冲区并返回，但如果没有数据，默认会阻塞，直到有数据时再读上来
write时也是一样，只不过内核缓冲区很少会被写满

因此，IO不仅涉及拷贝，更重要的是等待过程IO的本质为等 + 拷贝，要想高效的IO，只要减少【等】的比重即可

五种IO模型

阻塞IO：默认的IO类型，也是最常见的IO模型

在内核将数据准备好之前，系统调用会一直等待，所有的套接字（文件描述符）默认都是阻塞方式

非阻塞IO：如果内核还未将数据准备好，系统调用仍然会直接返回，并且返回EWOULDBLOCK/EAGAIN错误码
当采用非阻塞IO，一般会用轮询的方式反复调用读取/写入接口

信号驱动IO：内核将数据准备好的时候，使用SIGIO信号通知应用程序进行IO操作

其余时间应用程序完全不管它

IO多路转接/复用：能够同时等待多个文件描述符的就绪状态，就像多进程/线程同时等待。但多进程/线程的成本太高，而多路转接成本很低

由于多路转接原理是可以同时等待多个文件描述符，就注定了之前的那些其他IO类接口不能直接使用，因此操作系统为了满足我们同时等待多个文件描述符的需求，必须单独设计其他的系统调用，例如select/poll/epoll

select接口只负责IO操作中【等】的部分，当等到数据后，会交给例如recvfrom进行数据拷贝工作

异步IO：由内核在数据拷贝完成时, 通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据)

进程发起读请求后立即返回，继续执行其他任务，内核自动完成数据拷贝，完成后通知进程，不参与进程参与数据拷贝过程

同步/异步通信？

虽然和线程/进程同步/异步一样都用的同步/异步这个词，但两者完全没关系。

线程/进程同步指的是协调任务执行顺序，例如加锁保护临界区，而线程/进程异步为多个任务并发执行

而判断是同步通信还是异步通信的核心依据是：数据从内核缓冲区复制到用户空间缓冲区时，应用程序的线程是否亲自参与了等待或执行。同步通信例如阻塞IO、非阻塞IO、多路转接IO、信号驱动IO，都参与了数据拷贝工作；异步IO当发送方发出请求/消息后，无需等待结果即可立即去处理其他事务，直到接收方数据拷贝工作完成后再通知发送方

非阻塞IO

要想在IO时为非阻塞，可以将例如recv/recvfrom/send接口的flags参数设为MSG_DONTWAIT，或在打开文件时就将open接口的flags参数加上O_NONBLOCK

ssize_t ret = recv(sockfd, buf, len, MSG_DONTWAIT); int fd = open("/dev/mydevice", O_RDONLY | O_NONBLOCK);

这只能暂时设为非阻塞，要想将一个文件描述符永久设为非阻塞，可以用fcntl系统调用

int fcntl(int fd, int cmd, ... /* arg */ );

fd为要操作的文件描述符，cmd为要执行的操作，后面为可变参数

当cmd为F_GETFL时，代表获取fd的访问模式与状态标志

当cmd为F_SETFL时，代表修改fd的状态标志位，若要设为非阻塞，就设为O_NONBLOCK，但这样会覆盖掉fd原来的状态标志，因此要先获取原状态标志，再按位或O_NONBLOCK代表加上该状态标志

bool SetNonBlock(int fd) //为fd文件描述符设置非阻塞 { int fl = fcntl(fd, F_GETFL); // 获取原状态标志 if(fl < 0) // 获取失败 { std::cerr << "fcntl(F_GETFL): " << strerror(errno) << std::endl; return false; } // 给fd的状态标志加进非阻塞 if(fcntl(fd, F_SETFL, fl | O_NONBLOCK) < 0) // 若设置失败 { std::cerr << "fcntl(F_SETFL): " << strerror(errno) << std::endl; return false; } return true; }

非阻塞IO一般和轮询一起使用，通过循环调用read/write等系统调用，不断询问内核数据是否就绪。

拿read举例，读取成功时返回读取的字节数（>0），若读取到文件结尾返回0，若读取失败返回-1，并且errno被设置
但是非阻塞情况下的read,如果没检测到数据，返回后，它的返回值也是-1，虽然不是错误，但以错误的形式被返回了，它的errno会被设置为11，代表资源暂时不可用，意思就是资源还没就绪

errno: 11,Resource temporarily unavailable

因此对于返回值为-1的情况需要先判断是否为非阻塞正常返回，虽然直接判断errno是否为11也可以，但Linux提供了对应的宏，可以不用硬编码：EAGAIN / EWOULDBLOCK，这两个宏的值在Linux中完全一致：

// <errno.h> #define EAGAIN 11 #define EWOULDBLOCK EAGAIN // 直接别名

除此之外，EINTR错误代表被信号打断，并不清楚数据到底有没有，因此需要重来

完整代码：

#include "util.hpp" #include <unistd.h> using namespace std; int main() { SetNonBlock(0); char buffer[1024] = {0}; while (true) { printf(">>> "); fflush(stdout); ssize_t s = read(0, buffer, sizeof(buffer) - 1); if (s > 0) // 读到数据 { buffer[s - 1] = 0; // 把换行符去掉 printf("echo# %s\n", buffer); } else if (s == 0) // 读到文件结尾,Linux下是Ctrl+D { printf("read end\n"); break; } else if (s < 0) // 出错 { if (errno == EWOULDBLOCK || errno == EINTR)//EWOULDBLOCK代表只是非阻塞的正常返回，EINTR代表被信号打断 { sleep(1); continue; } else // 真出错了 { cerr << "errno: " << errno << "," << strerror(errno) << endl; break; } } sleep(1); } return 0; }

也可以在轮询检测时干点别的事，例如使用回调函数

完整代码：

#include "util.hpp" #include <unistd.h> #include <vector> #include <functional> using namespace std; //初始化 #define INIT(v) do{\ v.push_back(printlog);\ v.push_back(download);\ v.push_back(executeSql);\ }while(0) //遍历任务 #define EXEC_OTHER(cbs) do{\ for(auto cb : cbs)\ cb();\ }while(0) int main() { vector<function<void ()>> cbs; // 任务集合 INIT(cbs); SetNonBlock(0); char buffer[1024] = {0}; while (true) { printf(">>> "); fflush(stdout); ssize_t s = read(0, buffer, sizeof(buffer) - 1); if (s > 0) // 读到数据 { buffer[s - 1] = 0; // 把换行符去掉 printf("echo# %s\n", buffer); } else if (s == 0) // 读到文件结尾,Linux下是Ctrl+D { printf("read end\n"); break; } else if (s < 0) // 出错 { if (errno == EWOULDBLOCK)//EWOULDBLOCK/EAGAIN代表只是非阻塞的正常返回 { EXEC_OTHER(cbs); } else if(errno == EINTR) // EINTR代表被信号打断，需要重来一次 { continue; } else // 真出错了 { cerr << "errno: " << errno << "," << strerror(errno) << endl; break; } } sleep(1); } return 0; }

ps：printlog、download、executeSql为函数名

多路转接

select

select是多路转接的一种方式，它只负责等待，可以一次等待多个fd，select本身没有数据拷贝的能力，拷贝要例如read/write来完成

int select(int nfds, fd_set *_Nullable restrict readfds, fd_set *_Nullable restrict writefds, fd_set *_Nullable restrict exceptfds, struct timeval *_Nullable restrict timeout);

nfds：select等待的多个文件描述符中最大的一个+1
告诉内核只需要在0到nfds - 1的范围内检查描述符的状态，避免内核进行不必要的全量检查，提升效率

timeout：输入输出型参数，输入时通过传入timeval结构体决定阻塞的时间，当超过该时间还没有就绪的文件描述符时就超时返回，返回值被置0，例如struct timeval timeout = {5, 0}代表5秒、timeout = {0, 0}代表非阻塞、timeout = nullptr代表阻塞式调用

struct timeval { time_t tv_sec; /* 秒 */ suseconds_t tv_usec; /* 微秒 */ };

当select返回时，timeout参数会返回剩余未等待的时间。如果输入时timeout = {5, 0}，等待了3秒后有文件描述符就绪返回，还剩余2秒，输出时timeout = {2, 0}

readfds：读位图，输入输出型参数，输入时通过传入fd_set结构体（位图）指针告知select需要关注可读事件的文件描述符集合，若不关心可传nullptr

当select返回时，readfds会返回读就绪的文件描述符集合（改变原位图）

writefds：写位图，输入输出型参数，输入时通过传入fd_set结构体指针告知select需要关注可写事件的文件描述符集合，若不关心可传nullptr
当select返回时，writefds会返回写就绪的文件描述符集合（改变原位图）

exceptfds：异常位图，输入输出参数，输入时通过传入fd_set结构体指针告知select需要关注带外数据（Out-of-band / OOB，即带外异常）的文件描述符集合，若不关心可传nullptr
当select返回时，exceptfds会返回异常就绪的文件描述符集合（改变原位图）

fd_set：本质上是位图，用bit位1代表被监视

//内核版fd_set typedef struct { /* XPG4.2 requires this member name. Otherwise avoid the name from the global namespace. */ #ifdef __USE_XOPEN __fd_mask fds_bits[__FD_SETSIZE / __NFDBITS]; # define __FDS_BITS(set) ((set)->fds_bits) #else __fd_mask __fds_bits[__FD_SETSIZE / __NFDBITS]; # define __FDS_BITS(set) ((set)->__fds_bits) #endif } fd_set; // 简化版fd_set typedef struct { unsigned long fds_bits[1024 / (8 * sizeof(long))]; // 1024 位 } fd_set;

由于直接操作位图非常繁琐且不安全，系统提供了4 个宏来操作fd_set

FD_ZERO(fd_set *set) // 清空集合（将所有 bit 位置 0） FD_SET(int fd, fd_set *set) // 将指定的 fd 加入集合（位置 1） FD_CLR(int fd, fd_set *set) // 将指定的 fd 从集合中移除（位置 0） FD_ISSET(int fd, fd_set *set) // 检查 fd 是否在集合中（常用于 select 返回后判断就绪状态）

当因为有就绪文件描述符而返回时，select的返回值为就绪文件描述符总数（readfds.size+writefds.size+exceptfds.size）；当调用失败时返回值被置-1，且errno被设置

select多路转接服务器

// socket.hpp #pragma once #include <iostream> #include <string> #include <cstring> #include <cerrno> #include <sys/socket.h> #include <arpa/inet.h> #include "log.hpp" const static int backlog = 32; class Socket { public: static int create_socket() // 创建套接字 { int sockfd = socket(AF_INET, SOCK_STREAM, 0); if (sockfd == -1) { LogMessage(FATAL, (char *)"socket创建监听套接字失败, 错误码: %d, 错误描述：%s", errno, strerror(errno)); exit(SOCKET_ERR); } LogMessage(DEBUG, (char *)"socket创建监听套接字成功"); // 设置端口复用 int opt = 1; setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)); return sockfd; } static void bind_socket(int sockfd, uint16_t port) // bind绑定自己的网络信息 { struct sockaddr_in local; memset(&local, 0, sizeof(local)); local.sin_family = AF_INET; local.sin_port = htons(port); local.sin_addr.s_addr = INADDR_ANY; if (bind(sockfd, (struct sockaddr *)&local, sizeof(local)) != 0) { LogMessage(FATAL, (char *)"bind绑定失败, 错误码: %d, 错误描述：%s", errno, strerror(errno)); exit(BIND_ERR); } LogMessage(DEBUG, (char *)"bind绑定成功"); } static void listen_socket(int sockfd) // 开启监听状态 { if (listen(sockfd, backlog) != 0) { LogMessage(FATAL, (char *)"listen监听状态开启失败, 错误码: %d, 错误描述：%s", errno, strerror(errno)); exit(LISTEN_ERR); } LogMessage(DEBUG, (char *)"listen监听状态开启成功"); } static int accpet_socket(int listenfd, std::string &clientIp, uint16_t &clientPort) // 当接收到新连接时，通过输出型参数返回客户端ip+port { struct sockaddr_in ClientAddr; memset(&ClientAddr, 0, sizeof(ClientAddr)); socklen_t socklen = sizeof(ClientAddr); int sockfd = accept(listenfd, (struct sockaddr *)&ClientAddr, &socklen); if (sockfd == -1) { LogMessage(WARNING, (char *)"accept建立新连接失败, 错误码: %d, 错误描述：%s", errno, strerror(errno)); // exit(ACCEPT_ERR); } else { LogMessage(DEBUG, (char *)"accept建立新连接成功,sockfd = %d", sockfd); // 将IP传给输出参数clientIp char ip_str[16] = {0}; if (inet_ntop(AF_INET, &(ClientAddr.sin_addr.s_addr), ip_str, sizeof(ip_str)) == nullptr) { LogMessage(FATAL, (char *)"inet_ntop()失败，错误码：%d，错误信息：%s", errno, strerror(errno)); exit(INETPN_ERR); } else { clientIp = ip_str; } // 将port传给输出参数clientPort clientPort = ntohs(ClientAddr.sin_port); } return sockfd; } }; // SelectServer.hpp: #pragma once #include <iostream> #include <string> #include <functional> #include "socket.hpp" namespace select_ns { static const uint16_t defaultport = 8080; // 默认端口 static const int fdnum = sizeof(fd_set) * 8; // fd_set可以存储的文件描述符数量 static const int defaultfd = -1; // 非法文件描述符的标志 using func_t = std::function<std::string(const std::string &)>; class SelectServer { public: SelectServer(func_t func, int port = defaultport) : _func(func), _port(port), _listensockfd(-1) { } void init() { _listensockfd = Socket::create_socket(); Socket::bind_socket(_listensockfd, _port); Socket::listen_socket(_listensockfd); // 初始化_fdarray _fdarray = new int[fdnum]; for (int i = 0; i < fdnum; i++) _fdarray[i] = defaultfd; _fdarray[0] = _listensockfd; // 先把listensock放进去 } void Accepter() { // accept接收新连接 std::string ClientIp; uint16_t ClientPort; int sockfd = Socket::accpet_socket(_listensockfd, ClientIp, ClientPort); if (sockfd < 0) return; // accept失败 // 更新读位图 int i; for (i = 0; i < fdnum; i++) { if (_fdarray[i] != defaultfd) continue; // 跳过合法fd break; } if (i == fdnum) // 读位图已满 { LogMessage(WARNING, (char *)"读位图被占满"); close(sockfd); } else // 将新获取的sockfd添加到文件描述符集合_fdarray中 { _fdarray[i] = sockfd; } } void Recver(int sockfd, int pos /*在_fdarray中的位置*/) { // 读取 char request[1024]; ssize_t s = recv(sockfd, request, sizeof(request) - 1, 0); // 不考虑读不完的情况 if (s > 0) // 读到数据 { request[s] = 0; std::cout << "client# " << request << std::endl; } else if (s == 0) // 文件被关闭 { close(sockfd); _fdarray[pos] = defaultfd; LogMessage(DEBUG, (char *)"客户端退出"); } else // recv报错 { close(sockfd); _fdarray[pos] = defaultfd; LogMessage(ERROR, (char *)"客户端退出，错误码：%d, 错误描述：%s", errno, strerror(errno)); } // 请求转响应 std::string response = _func(request); // 写入 write(sockfd, response.c_str(), response.size()); } void HandleEvent(fd_set &fds) { for (int i = 0; i < fdnum; i++) { if (_fdarray[i] == defaultfd) continue; // 过滤非法fd if ((_fdarray[i] == _listensockfd) && (FD_ISSET(_fdarray[i], &fds))) // accept就绪 Accepter(); else if (FD_ISSET(_fdarray[i], &fds)) // 普通fd读就绪 Recver(_fdarray[i], i); } // Print std::cout << "_fdarray: "; for (int i = 0; i < fdnum; i++) { if (_fdarray[i] == defaultfd) continue; std::cout << _fdarray[i] << " "; } fflush(stdout); } void start() { while (true) { // std::string ClientIp; // uint16_t ClientPort; // int sock = Socket::accpet_socket(_listensockfd, ClientIp, ClientPort); // if(sock < 0) continue; // 初始化rfds与maxfd fd_set rfds; FD_ZERO(&rfds); // 初始化 int maxfd = _fdarray[0]; for (int i = 0; i < fdnum; i++) { if (_fdarray[i] == defaultfd) continue; // 非法fd跳过 FD_SET(_fdarray[i], &rfds); // 将合法fd加进读位图 if (maxfd < _fdarray[i]) maxfd = _fdarray[i]; // 更新maxfd } // int n = select(_listensockfd + 1, &rfds, nullptr, nullptr, &timeout); int n = select(maxfd + 1, &rfds, nullptr, nullptr, nullptr); switch (n) { case 0: // 超时返回 std::cout << "timeout...\n"; break; case -1: // 错误返回 LogMessage(WARNING, (char *)"select失败，错误码：%d，错误描述：%s", errno, strerror(errno)); break; default: // 就绪返回 LogMessage(DEBUG, (char *)"读就绪"); HandleEvent(rfds); break; } } } ~SelectServer() { if (_listensockfd == -1) close(_listensockfd); if (_fdarray) delete[] _fdarray; } private: uint16_t _port; // 服务端口号 int _listensockfd; // 监听套接字 int *_fdarray; // 文件描述符集合 func_t _func; // 用于将请求转响应的函数 }; } // SelectServer.cpp: #include <iostream> #include <memory> #include <string> #include "SelectServer.hpp" #include "log.hpp" using namespace std; using namespace select_ns; static void usage(string proc) // 使用手册 { cout << RED << "\nUsage：\n\t" << proc << " " << "port\n\n " << ED; } string translation(const string &request) // [TODO] 请求转响应 { return request; } int main(int argc, char *argv[]) { if (argc != 2) { usage(argv[0]); exit(USAGE_ERR); } uint16_t port = atoi(argv[1]); unique_ptr<SelectServer> svr(new SelectServer(translation, port)); svr->init(); svr->start(); return 0; }

select 缺点：

1. select能同时等待的文件fd是有上限的，除非重新改内核，否则无法解决
2. 必须借助第三方数组，来维护合法的fd集合
3. select的大部分参数是输入输出型的，调用select前，要重新设置所有的fd，调用之后，还需要检查更新所有的fd，遍历的成本很大
4. select 采用位图，输入：用户告诉内核，输出：内核告诉用户，来回的进行数据拷贝，拷贝成本大

poll

poll方案解决了select中fd有上限、每次调用都要重新设置所有关心的fd的问题

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

fds为struct pollfd类型的动态数组，用于存储需要关心的fd及事件

每个元素所包含的信息如下：

struct pollfd { int fd; /* 文件描述符 */ short events; /* 请求的事件 */ short revents; /* 返回的事件 */ };

fd为文件描述符

events为想要关心的事件，revents为返回时就绪的事件，可以填的值如下：

常用的为POLLIN（读）、POLLOUT（写）、POLLERR（错误）

每个宏都占用一个比特位，因此可以通过按位操作包含多个事件

fds通过将用户告诉内核与内核告诉用户的事件分开，避免了每次都要重新设置fd

nfds为fds的长度，也就是要关心的文件描述符的个数

timeout参数单位是ms（毫秒），>0代表阻塞timeout ms，超时则返回；=0代表非阻塞；<0代表阻塞；

有就绪事件就返回每个事件就绪的fd相加的值，若超时返回0，若失败返回-1

当调用poll接口时，用户通过fds[i].events告诉内核需要关心的fd及其事件，当poll返回时，内核通过fds[i].revents告诉用户每个关心的fd关心的事件中已就绪的事件

但poll为了知道哪些fd的哪些事件就绪，需要遍历struct pollfd数组，当fd多了后会影响效率

epoll

epoll是为了解决poll遍历问题的多路转接接口

epoll_create

int epoll_create(int size); int epoll_create1(int flags);

创建一个epoll模型，size 参数为预期要管理的fd个数，现已被忽略，只需要保证>0即可
返回值为epoll模型的fd

现代写法推荐用epoll_create1，flags填0和epoll_create(1)代表的意思一样
epoll_create1(EPOLL_CLOEXEC)代表防止epoll文件描述符泄露给子进程（例如fork()、exec()）

epoll_ctl

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

控制指定epoll模型中需要关心的fd及其事件

• epfd：通过epoll_create创建的epoll模型
• op：有三种选项：EPOLL_CTL_ADD（增）、EPOLL_CTL_DEL（删）、EPOLL_CTL_MOD（改），代表要增/删/改指定fd及其事件（若为删，event参数被忽略，直接删除fd）
• fd：要操作的目标文件描述符
• event：描述感兴趣的事件和携带的用户数据

其中events参数为关心的事件类型，值如下：

• EPOLLIN : 表示对应的文件描述符可以读 (包括对端SOCKET正常关闭);
• EPOLLOUT : 表示对应的文件描述符可以写;
• EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来);
• EPOLLERR : 表示对应的文件描述符发生错误;
• EPOLLHUP : 表示对应的文件描述符被挂断;
• EPOLLET : 将EPOLL设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(Level Triggered)来说的.
• EPOLLONESHOT：只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个socket的话, 需要再次把这个socket加入到EPOLL队列里.

data为用户数据：

里面的fd成员用于表明该事件所属的文件描述符，epoll_wait拿取事件时只有struct epoll_event结构体，只能通过.data.fd确定文件描述符

成功返回0，失败返回-1

epoll_wait

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

相当于pool()或select()，用于等待文件描述符就绪

• epfd：通过epoll_create创建的epoll模型
• events：用于接收就绪事件，输出型参数
• maxevents：用于表示events的数组容量，即一次最多返回多少个事件，必须 > 0
• timeout：语义与poll()时一致

有就绪事件就返回就绪事件的个数，若超时返回0，若失败返回-1

epoll模型

epoll_create会创建一个epoll模型，它主要有两个核心数据结构：红黑树和就绪链表（双向）

当调用epoll_ctl接口时，会对红黑树进行增删改操作，每个节点对应一个需要关心的fd及其事件，并注册当该fd有数据时回调的接口ep_poll_callback，用于将事件挂载到就绪队列中
该红黑树的key为对应fd+struct file*指针组成

这其中epitem.event就是epoll_ctl时传入的event结构体，里面包含了events（事件）和data（用户数据，包含fd、指针等 ）

当底层的网卡驱动接收到数据后，传给协议栈解析并放在Socket的接收缓冲区中后，就会调用属于该fd的回调方法，而由于该fd的回调方法是ep_poll_callback， 就会将epoll模型中红黑树的对应节点挂载到就绪队列（将epitem.rdlink插入到就绪双向链表的末尾）

当调用epoll_wait()时，就会从就绪队列中拿取事件

epitem结构：

struct epitem { /* 红黑树节点：用于将当前 epitem 挂载到 epoll 实例的红黑树中 */ struct rb_node rbn; /* 双向链表节点：用于将当前 epitem 挂载到就绪队列 (rdllist) 或 overflow 队列中 */ struct list_head rdlink; /* 指向当前 epitem 所属的 epoll 实例 (struct eventpoll) */ struct eventpoll *ep; /* 包含用户态传入的 fd 和 event（用户感兴趣的事件掩码及用户自定义数据 epoll_data） */ struct epoll_filefd ffd; /* 用户注册的关注事件掩码（如 EPOLLIN | EPOLLOUT | EPOLLET） */ struct epoll_event event; /* 等待队列头：用于管理当前 epitem 挂载在底层 Socket (sk_sleep) 上的等待项 */ struct eppoll_entry *pwqlist; /* 指向该 fd 在 VFS 层的内核文件结构体 struct file */ struct file *ffd_file; // ... 其他内核同步/统计相关的内部字段 };

因此epoll并不需要遍历数据结构，直接用O(1)的复杂度就可以知道哪些fd就绪

epoll_create返回的是epol模型抽象成文件的fd

水平触发（LT）/边缘触发（ET）

epoll_wait将所有就绪的事件，按照顺序放到用户传入的数组的中，如果就绪队列很多数据节点，一次拿不完，下次epoll_wait时默认会继续通知，也就是水平触发（LT）
水平触发：只要状态“处于就绪状态”（例如缓冲区还有数据未读完、或者就绪队列里还有未处理完的事件），epoll_wait就会不断通知。

但如果将对应epitem节点改为边缘触发（events），对于ET 模式的节点，内核在将事件拷贝给用户态数组后，直接从就绪链表（rdllist）中移除该节点，不再放回。下一次调用epoll_wait时，即使底层缓冲区还有未读完的数据/就绪队列之前没处理完，epoll_wait也不会通知你，它会一直阻塞等待，直到这个fd上再次有新的数据到达（发生新的状态边沿变化）。
边缘触发：只有当状态“发生变化/出现新边缘”（例如数据从无到有、从不可写变为可写）时，epoll_wait才会通知一次

要对某个fd设置边缘触发，就将events |= EPOLLET

struct epoll_event ev; // 设置为 读事件(EPOLLIN) + 边缘触发(EPOLLET) ev.events = EPOLLIN | EPOLLET; ev.data.fd = sockfd; // 将该 fd 注册到 epoll 实例中 epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);

ET模式的优点

虽然在LT模式下也可以通过尽快将数据读上来而达到和ET一样的效果，但ET模式下强制约束了代码要高效的IO数据，否则就跑不对，适用于高并发场景

在ET模式下，为防止服务器因fd阻塞而挂起，必须要将fd设为非阻塞。由于直到read/recv返回EWOULDBLOCK/EAGAIN时才可以确认数据读完，但在阻塞模式下会导致整个事件循环卡死（因为只会通知一次）。但只要是非阻塞，可以一直读取，直到返回EWOULDBLOCK/EAGAIN

并且由于ET模式下就算有数据没有读完，也不会一直通知，减少了重复通知，可以将这个名额用到其他新事件上（因为epoll_wait()传入的事件数组是有大小限制的！）

由于ET模式的机制，应用层会及时读取，那么就会给TCP接收缓冲区腾出更大的空间，从而让TCP可以通告更大的窗口大小，发送方就可以一次发出更多数据

应用层及时 read() → 内核接收缓冲区 (sk_rmem) 腾出空间 → TCP 通告更大的 Window Size → 发送方可以一次发出更多数据 → 吞吐量提升

Reactor

Reactor模式（反应堆模式），半同步（IO同步）半异步（事件分发异步），通过让主线程监听，分线程负责执行IO和业务。与之对应的还有Proactor模式，这里不做介绍

Reactor实现代码已放出，下面主要讲设计中的一些问题（只实现了单线程Reactor）

为什么要用Connection来管理一个fd，每个fd都有自己的IO缓冲区？

ET模式下的epoll，由于阻塞模式下可能会出现一次读取读不完的情况（例如被信号中断、应用层缓冲区过小等原因），因此需要设置非阻塞。
但如果其中一个fd的报文只来了一半，多个连接的数据是交错到达的，共用缓冲区会互相覆盖。每个fd一个Connection，本质上是给每个连接一个独立的"记忆"，让它记住自己读到哪了、发到哪了、状态是什么。

写事件也是一样，若一个报文只发送了一半，而此时返回了EWOULDBLOCK，就要将数据存到写就绪时再发送。但等到写就绪时，不一定是这个fd先执行，如果共用一个缓冲区，就会被覆盖掉

一个 fd 对应一个 Connection，是为了给每个网络连接建立独立的上下文环境。由于 TCP 数据可能分批到达、非阻塞 IO 可能部分读写、多个连接事件交错发生，因此每个 fd 必须独立保存自己的输入输出缓冲区和状态，否则数据会互相覆盖，连接状态会混乱。

epoll中对于IO就绪事件的不同设计

对于读取事件，会将EPOLLIN常驻在epoll中，只要有数据到来就通知
对于写入事件，会先尝试直接写入，如果没有写完或返回了EWOULDBLOCK（因为是非阻塞），再让epoll关注写就绪事件（将EPOLLOUT添加到epoll），当底层的输出缓冲区空出来，就会通知。并且在确定写完数据后再将EPOLLOUT从epoll中移除！

这种读取与写入的不同处理，是因为大部分情况下，写总是就绪的，如果epoll一直关注写就绪，可能我们还没有想写入的数据，但epoll一直在通知写就绪...