五种IO模型、非阻塞IO与多路转接

目录重新理解IO五种IO模型高级IO重要概念同步通信 vs 异步通信阻塞 vs 非阻塞非阻塞IOselect认识 select 接口编写 SelectServerselect 的特点poll认识 poll 接口编写 PollServerpoll 的特点epoll核心机制认识 epoll 接口编写 EpollServerepoll 的两种工作模式重新理解IO我们在Linux系统部分已经学过基础IO了IO就是 Input/Output 输入和输出我们之前学过的 read/write/recvfrom/sendto/recv/send 都是IO接口当读/写条件不满足时会阻塞住等读/写条件满足了才会将数据从内核读取(拷贝)到用户空间中因此IO 等 拷贝高效的IO指的是单位时间内拷贝更多的数据如何做到呢减少等的比重即可因此等的时间越短IO效率越高五种IO模型举一个钓鱼的例子张三全神贯注的盯着鱼漂鱼漂不动张三不动直到有鱼上钩阻塞式IO李四不会一直盯着鱼漂而是看一眼鱼漂发现鱼漂没动去刷刷抖音打打游戏做做其他事情过一会又看一眼鱼漂循环往复非阻塞式IO王五鱼竿顶部加铃铛自己不去看鱼漂了直接做其他事情等铃铛响即可信号驱动IO赵六拉了一卡车鱼竿和鱼漂全部用来钓鱼轮询检测所有鱼漂是否动了只要有一个鱼漂动了就是有鱼上钩了多路复用IO田七是个大老板只吃鱼不钓鱼老板只是发起了钓鱼的动作直接派小王去钓鱼异步IO结论1赵六的效率是最高的由于赵六使用了多个鱼竿只要有1个鱼竿上有鱼上钩了就是钓到鱼了单位时间内等的比重是所有钓鱼方式里面最低的因此效率是最高的结论2阻塞式IO和非阻塞式IO的IO效率本质是一样的IO 等 拷贝张三和李四等和拷贝的时间都是一样的只是张三在等期间啥都没有做而李四将等的时间利用起来了因此李四做了更多的事情但就单论IO来说两者效率是一样的结论3只有田七是异步IO其他都是同步IO没有参与IO的任何过程(没有等也没有拷贝)只是最后拿取IO的结果就是异步IO只要参与了IO的任何过程(无论是等还是拷贝)都是同步IO阻塞IO: 在内核将数据准备好之前系统调用会一直等待所有的套接字默认都是阻塞方式非阻塞IO: 如果内核还未将数据准备好系统调用仍然会直接返回并且返回EWOULDBLOCK错误码。非阻塞IO往往需要程序员循环的方式反复尝试读写文件描述符这个过程称为轮询这对CPU来说是较大的浪费一般只有特定场景下才使用。信号驱动IO内核将数据准备好的时候使用SIGIO信号通知应用程序进行IO操作。IO多路转接虽然从流程图上看起来和阻塞IO类似实际上最核心在于IO多路转接能够同时等待多个文件描述符的就绪状态。异步IO由内核在数据拷贝完成时通知应用程序而信号驱动是告诉应用程序何时可以开始拷贝数据。总结任何IO过程中都包含两个步骤。第一是等待第二是拷贝。而且在实际的应用场景中等待消耗的时间往往都远远高于拷贝的时间。让IO更高效最核心的办法就是让等待的时间尽量少。高级IO重要概念同步通信 vs 异步通信所谓同步就是在发出一个调用时在没有得到结果之前该调用就不返回。但是一旦调用返回就得到返回值了换句话说就是由调用者主动等待这个调用的结果。异步则是相反调用在发出之后这个调用就直接返回了所以没有返回结果换句话说当一个异步过程调用发出后调用者不会立刻得到结果而是在调用发出后被调用者通过状态、通知来通知调用者或通过回调函数处理这个调用。注意同步通信、异步通信 与 线程的同步与互斥没有任何关系阻塞 vs 非阻塞阻塞和非阻塞关注的是程序在等待调用结果消息返回值时的状态。阻塞调用是指调用结果返回之前当前线程会被挂起。调用线程只有在得到结果之后才会返回。非阻塞调用指在不能立刻得到结果之前该调用不会阻塞当前线程。非阻塞IO之前使用 scanf 时如果键盘没有数据那么 scanf 就阻塞住了从今天开始我们叫做 读事件没有就绪而之前学习的 /recv/recvfrom等函数的诸多读选项中都有一个参数就叫做flagsflags选项决定了是阻塞式读取还是非阻塞式读取而我们今天学习一种更加通用的非阻塞式的方式能直接将内核 struct file 结构体 中的 flags 标志设置为非阻塞这就是 fcntlfcntl 函数有五种功能复制一个现有的描述符cmdF_DUPFD。获得/设置文件描述符标记cmdF_GETFD或F_SETFD。获得/设置文件状态标记cmdF_GETFL或F_SETFL。获得/设置异步I/O所有权cmdF_GETOWN或F_SETOWN。获得/设置记录锁cmdF_GETLK,F_SETLK或F_SETLKW。我们此处只是用第三种功能获取/设置文件状态标记就可以将一个文件描述符设置为非阻塞。#include iostream #include unistd.h #include fcntl.h void SetNoBlock(int fd) { int fl fcntl(fd, F_GETFL); //获取flags标志位 if (fl 0) { perror(fcntl); return; } fcntl(fd, F_SETFL, fl | O_NONBLOCK); //设置flags为非阻塞 }下面这段代码就是正常的阻塞式读取键盘不输入read一直阻塞直到有输入int main() { char buffer[1024]; while(true) { ssize_t n read(0, buffer, sizeof(buffer) - 1); if(n 0) { buffer[n] 0; printf(#%s, buffer); } else if(n 0) { printf(read end of file); } else { perror(read); } } return 0; }当我们将读取方式设置为非阻塞时read 发现读取缓冲区中没有数据读条件不就绪就直接返回了并且根据打印结果可以看到 read 返回值 0全局错误码是11错误信息是资源没就绪#include iostream #include unistd.h #include fcntl.h #include errno.h #include cstring void SetNoBlock(int fd) { int fl fcntl(fd, F_GETFL); if (fl 0) { perror(fcntl); return; } fcntl(fd, F_SETFL, fl | O_NONBLOCK); } int main() { SetNoBlock(0); //设置非阻塞 char buffer[1024]; while(true) { ssize_t n read(0, buffer, sizeof(buffer) - 1); if(n 0) { buffer[n] 0; printf(#%s, buffer); } else if(n 0) { printf(read end of file); } else { // perror(read); printf(read error, read read: %ld, errno: %d, err message: %s\n, n, errno, strerror(errno)); } } return 0; }read 0 其实分了两种情况一种是真的出错了发生了不可恢复的错误不能再进行读取了一种是发生了需要再次读取的软错误需要重新发起 read 调用到底是哪种情况需要根据 全局错误码 errno 来区别发生了需要再次读取的软错误通常是两种情况第一种 errno 11宏定义是EAGAIN / EWOULDBLOCK表示读取的资源没有就绪第二种 errno EINTR表示 read 在阻塞等待数据时进程收到了一个信号如 SIGALRM 或 SIGINT导致内核提前返回#include iostream #include unistd.h #include fcntl.h #include errno.h #include cstring void SetNoBlock(int fd) { int fl fcntl(fd, F_GETFL); if (fl 0) { perror(fcntl); return; } fcntl(fd, F_SETFL, fl | O_NONBLOCK); } int main() { SetNoBlock(0); //设置非阻塞 char buffer[1024]; while(true) { ssize_t n read(0, buffer, sizeof(buffer) - 1); if(n 0) { buffer[n] 0; printf(#%s, buffer); } else if(n 0) { printf(read end of file); break; } else { if(errno EAGAIN || errno EWOULDBLOCK) { printf(你的数据没有准备好, 下次再来吧\n); sleep(1); continue; } else if(errno EINTR) { sleep(1); continue; } else { perror(read); break; } } } return 0; }我们上面讲解非阻塞 IO 是以输入函数 read 为例的 为什么没有用 标准输出去讲呢因为标准输出在大多数情况下写条件都是满足的select认识 select 接口select 是一个多路转接/多路复用的函数也就是说 select 是实现多路转接/多路复用的一种具体方案IO 等 拷贝select 专门用于IO中的等动作一次可以等待多个 fd任意一个 fd 就绪就会通知用户哪些fd就绪了本质是一种就绪事件通知机制参数nfds等待的最大文件描述符的值1readfds /writefds / exceptfds这三个参数的类型都是 fd_setfd_set 是文件描述符集文件描述符集是一种位图结构这三个参数都是输入输出型参数作为输入型参数表示要关心哪些文件描述符上的读/写/异常事件作为输出型参数表示哪些文件描述符上的读/写/异常事件已经就绪以 readfds 为例输入的时候用户告诉内核内核你要帮我关心readfds 中哪些 fd 上的读事件比特位的位置表示文件描述符的编号比特位的内容表示是否关心例如 0000 0101 作为输入表示关心 0号 和 3 号文件描述符上的读事件输出的时候内核告诉用户用户你让我关心的 readfds 中的哪些fd上的读事件已经就绪了比特位的位置表示文件描述符的编号比特位的内容表示该文件描述符上的读事件是否就绪了例如 0000 0101 作为输出表示 0 号 和 3号 文件描述符上的读事件已经就绪了细节1某个比特位只在 readfds 中被设置为1了表示只关心该fd上的读事件不关心写事件和异常事件某个比特位既在 readfds 中被设置了也在 writefds 中被设置了表示既关心读事件也关心写事件细节2fd_set 是一种数据类型既然是数据类型那么其大小就是固定的意味着位图中比特位的个数是固定的也就是说 select 能管理的 fd 是有上限的#include iostream #include sys/select.h using namespace std; int main() { printf(%ld\n, sizeof(fd_set) * 8); //输出1024 return 0; }细节3fd_set 底层是位图结构但是不推荐我们自己对位图做操作系统已经给我们提供了一批操作 fd_set 集合的接口timeouttimeout 是一个 struct timeval* 类型的结构体指针 用于设置 select 函数的超时行为。它决定了 select 在等待文件描述符就绪时的等待策略是一个输入输出型参数。结构体定义struct timeval 定义在 sys/times.h 头文件中包含两个成员struct timeval { long tv_sec; // 秒 long tv_usec; // 微秒 };• NULL阻塞式等待等待的多个 fd 中至少有 1 个 fd 就绪才能返回• {00}非阻塞式等待等待的多个 fd 没有一个就绪直接返回• {x秒y微妙}先阻塞等待一定的时间如果在该时间内没有任何 fd 就绪会超时一次然后返回下一次再调用 select 时继续等待一定的时间...而这个参数是输入输出型参数作为输入型参数表示要等多长时间作为输出型参数表示还剩多长时间比如设置 {50}表示5s内阻塞式等待如果第2s的时候有fd就绪了那么 timeout 最终的结果就是 {30}返回值• 0表示有多少个 fd 就绪了• 0超时了指定时间内没有任何 fd 就绪了继续循环检测• 0函数调用出错当timeout 参数设置为NULL时是阻塞等待那么返回值只会有 0 和 0 两种情况而其他两种情况三种返回值都可能出现编写 SelectServerSelectServer.hpp#pragma once #include iostream #include string #include memory #include sys/select.h #include Logger.hpp #include InetAddr.hpp #include Socket.hpp class SelectServer { public: SelectServer(uint16_t port) :_listensock(std::make_uniqueTcpSocket()) { _listensock-BuildListenSocketMethod(port); } void HandlerEvents() { LOG(LogLevel::DEBUG) fd就绪, 有新事件到来; } void Run() { while(true) { //不能直接accept, 会阻塞, 获取连接也是IO, 交给select int maxfd _listensock-SockFd(); fd_set rfds; FD_ZERO(rfds); FD_SET(_listensock-SockFd(), rfds); struct timeval timeout {3, 0}; int n select(maxfd 1, rfds, nullptr, nullptr, /*timeout*/nullptr); if(n 0) { LOG(LogLevel::ERROR) select error; } else if(n 0) { LOG(LogLevel::ERROR) timeout... timeout.tv_sec : timeout.tv_usec; } else { HandlerEvents(); } } } ~SelectServer(){} private: std::unique_ptrSocket _listensock; };select 上确实检测到了 fd 上面有读事件就绪了但是一直循环打印 fd 就绪有新事件到来因为我们并没有处理该事件那么连接就一直存在因此我们需要将到来的连接进行获取void HandlerEvents() { LOG(LogLevel::DEBUG) fd就绪, 有新事件到来; InetAddr clientaddr; int sockfd _listensock-Accept(clientaddr); if(sockfd 0) { LOG(LogLevel::INFO) get new sockfd: sockfd , client addr: clientaddr.ToString(); } }获取上来连接之后我们能直接进行 recv 读取 sockfd 吗肯定是不能的因为 sockfd 不一定有数据客户端可能一直不发数据那服务器就一直阻塞在这里了就无法处理其他客户端请求了因此我们需要将 sockfd 托管给 select因为 select 的作用就是等并且一次性可以等多个文件描述符如何托管给 select 呢select 一次可以等多个 fd但有可能我们让 select 关心的 fd 有些没有就绪而 readfds / writefds / exceptfds 是输入输出型参数内核返回给用户的时候会将没有就绪的 fd 清0但是下次 读取时我们依旧要关心上次没有就绪的 fd因此我们需要将历史的合法 fd 记录下来就需要借助其他数据结构了此处我们使用辅助数组用其他容器可以嘛绝对是可以滴因为 select 的主要矛盾是遍历不需要插入删除等操作而根据局部性原理数组效率其实是更高的构造函数const static int gsize sizeof(fd_set) * 8; const static int gdefaultfd -1; class SelectServer { public: SelectServer(uint16_t port) :_listensock(std::make_uniqueTcpSocket()) { _listensock-BuildListenSocketMethod(port); for(int i 0; i gsize; i) { fd_array[i] gdefaultfd; } fd_array[0] _listensock-SockFd(); } private: std::unique_ptrSocket _listensock; int fd_array[gsize]; };Run 函数void Run() { while(true) { //不能直接accept, 会阻塞, 获取连接也是IO, 交给select fd_set rfds; int maxfd gdefaultfd; for(int i 0; i gsize; i) { if(fd_array[i] gdefaultfd) continue; FD_SET(fd_array[i], rfds); if(maxfd fd_array[i]) maxfd fd_array[i]; LOG(LogLevel::DEBUG) 添加fd: fd_array[i]; } int n select(maxfd 1, rfds, nullptr, nullptr, nullptr); if(n 0) { LOG(LogLevel::ERROR) select error; } else if(n 0) { LOG(LogLevel::ERROR) timeout...; } else { HandlerEvents(); } } }HandlerEvents 函数void HandlerEvents() { LOG(LogLevel::DEBUG) fd就绪, 有新事件到来; InetAddr clientaddr; int sockfd _listensock-Accept(clientaddr); if(sockfd 0) { LOG(LogLevel::INFO) get new sockfd: sockfd , client addr: clientaddr.ToString(); } //将新的fd托管给select, 只需要将新的fd添加到辅助数组即可 int pos 0; for(; pos gsize; pos) { if(fd_array[pos] gdefaultfd) { fd_array[pos] sockfd; break; } } if(pos gsize) { LOG(LogLevel::WARNING) server is full; close(sockfd); } }我们今天的代码里只关心了读事件实际可能还有写事件、异常事件因此 HandlerEvents 函数需要带上参数表明要处理什么事件而且 fd 就绪了不一定就是 listensockfd还可能是其他普通的读事件因此在 HandlerEvents 函数内部我们还需要分类讨论判断void Accepter() { InetAddr clientaddr; int sockfd _listensock-Accept(clientaddr); if(sockfd 0) { LOG(LogLevel::INFO) get new sockfd: sockfd , client addr: clientaddr.ToString(); } //将新的fd托管给select, 只需要将新的fd添加到辅助数组即可 int pos 0; for(; pos gsize; pos) { if(fd_array[pos] gdefaultfd) { fd_array[pos] sockfd; break; } } if(pos gsize) { LOG(LogLevel::WARNING) server is full; close(sockfd); } } void Recver(int index) { char buffer[1024]; int n recv(fd_array[index], buffer, sizeof(buffer) - 1, 0); if(n 0) { buffer[n] 0; LOG(LogLevel::INFO) client say: buffer; std::string echo_string server echo#; echo_string buffer; send(fd_array[index], echo_string.c_str(), echo_string.size(), 0); } else if(n 0) { LOG(LogLevel::INFO) client quit, me too: fd_array[index]; close(fd_array[index]); fd_array[index] gdefaultfd; } else { LOG(LogLevel::WARNING) recv error; close(fd_array[index]); fd_array[index] gdefaultfd; } } void HandlerEvents(fd_set rfds) { LOG(LogLevel::DEBUG) fd就绪, 有新事件到来; for(int i 0; i gsize; i) { if(fd_array[i] gdefaultfd) continue; if(FD_ISSET(fd_array[i], rfds)) //读事件就绪 { if(fd_array[i] _listensock-SockFd()) { Accepter(); } else { Recver(i); } } } }连接到来属于新事件到来连接断开也是新事件到来fd上有数据了也是新事件就绪了而 新事件就绪后我们根据事件的不同去调用不同的函数本质就是用不同的模块去处理不同的事件因此 HandlerEvents 应该叫做 Eventdispatcher 更合适而 Accepter 是连接管理器Recver 是 IO处理器...select 的特点1. 能监控的 fd 是有上限的取决于 sizeof(fd_set) 的值2. 将fd加入select监控集的同时还要再使用一个数据结构array保存放到select监控集中的fd一是用于在select返回后array作为源数据和fd_set进行FD_ISSET判断。二是select返回后会把以前加入的但并无事件发生的fd清空则每次开始select前都要重新从array取得fd逐一加入(FD_ZERO最先)扫描array的同时取得fd最大值maxfd用于select的第一个参数select 的缺点1. 每次调用select, 都需要手动设置fd集合, 从接口使用角度来说也非常不便.2. 每次调用select, 都需要把fd集合从用户态拷贝到内核态这个开销在fd很多时会很大3. 同时每次调用select都需要在内核遍历传递进来的所有fd这个开销在fd很多时也很大4. select支持的文件描述符数量太小由于上述问题因此有了新的多路转接方案: poll那为啥还要学习 select主要是因为 select 是 POSIX 标准中最早定义的 I/O 多路复用接口几乎存在于所有支持网络编程的操作系统上因此可以很好的跨平台poll认识 poll 接口poll 的定位和 select 完全一样是多路转接/多路复用的一种具体方案主要解决 IO 中 等的问题参数fds nfds这两个参数可以当作一个整体来看nfds 表示结构体数组元素个数而 fds 是该数组首元素的地址fds 依旧是输入输出型参数输入时表示用户让内核关心哪些 fd 上的哪些事件看的是 struct pollfd 的 fd 和 events而 输出时表示用户让内核关心的哪些 fd 上的哪些事件已经就绪了看的是 fd 和 revents因此 poll 将输入参数和输出参数 分开了解决了 select 每次需要对参数进行重置的问题。但是 events 和 revents 都是 short 类型怎么表示具体的事件采用的是宏定义需要表示什么事件只需要将 short 类型的某一个比特位进行设置即可细节poll 的第2个参数不就是等待的文件描述符的个数吗那不也是有上限的吗?select 的文件描述符上限是由 fd_set 类型大小决定的是由内核决定的不由用户决定要改变该上限就需要重新编译内核了而 poll 的文件描述符上限由用户传参决定只要内存足够大上限就会很高并且结构体数组不止是静态数组还可以是 malloc 动态分配的但是一个进程能够打开的文件本身就是有上限的是的刚开始的OS能够打开的fd是比较少的就是 struct file 中的一个静态数组大小写死了后来 struct file 中支持对 fd 进行动态扩展了timeout超时时间单位是毫秒返回值三种返回值同select编写 PollServerPollServer.hpp#pragma once #include iostream #include string #include memory #include sys/select.h #include poll.h #include Logger.hpp #include InetAddr.hpp #include Socket.hpp const static int gsize sizeof(fd_set) * 8; const static int gdefaultfd -1; class PollServer { public: PollServer(uint16_t port) :_listensock(std::make_uniqueTcpSocket()) { _listensock-BuildListenSocketMethod(port); for(int i 0; i gsize; i) { fd_array[i].fd gdefaultfd; fd_array[i].events fd_array[i].revents 0; } fd_array[0].fd _listensock-SockFd(); fd_array[0].events POLLIN; } void Accepter() { InetAddr clientaddr; int sockfd _listensock-Accept(clientaddr); if(sockfd 0) { LOG(LogLevel::INFO) get new sockfd: sockfd , client addr: clientaddr.ToString(); } //将新的fd托管给select, 只需要将新的fd添加到辅助数组即可 int pos 0; for(; pos gsize; pos) { if(fd_array[pos].fd gdefaultfd) { fd_array[pos].fd sockfd; fd_array[pos].events POLLIN; break; } } if(pos gsize) { LOG(LogLevel::WARNING) server is full; close(sockfd); //也可以选择动态扩容! } } void Recver(int index) { int sockfd fd_array[index].fd; char buffer[1024]; int n recv(sockfd, buffer, sizeof(buffer) - 1, 0); if(n 0) { buffer[n] 0; LOG(LogLevel::INFO) client say: buffer; std::string echo_string server echo#; echo_string buffer; send(sockfd, echo_string.c_str(), echo_string.size(), 0); } else if(n 0) { LOG(LogLevel::INFO) client quit, me too: fd_array[index].fd; close(sockfd); fd_array[index].fd gdefaultfd; fd_array[index].events fd_array[index].revents gdefaultfd; } else { LOG(LogLevel::WARNING) recv error; close(sockfd); fd_array[index].fd gdefaultfd; fd_array[index].events fd_array[index].revents gdefaultfd; } } void HandlerEvents() { LOG(LogLevel::DEBUG) fd就绪, 有新事件到来; for(int i 0; i gsize; i) { if(fd_array[i].fd gdefaultfd) continue; if(fd_array[i].revents POLLIN) //读事件就绪 { if(fd_array[i].fd _listensock-SockFd()) { Accepter(); } else { Recver(i); } } } } void Run() { while(true) { int n poll(fd_array, gsize, -1); //阻塞式等待 if(n 0) { LOG(LogLevel::ERROR) select error; } else if(n 0) { LOG(LogLevel::ERROR) timeout...; } else { HandlerEvents(); } } } ~PollServer(){} private: std::unique_ptrSocket _listensock; struct pollfd fd_array[gsize]; //可以malloc动态开辟, 后续动态扩容 };poll 的特点Poll 解决了 select 中两个问题一个是每次 select 前都要进行参数重置的问题一个是等待的文件描述符有上限的问题Poll 的底层本身也要对 fd_array 进行遍历因此只要 fd 增多那么 poll 遍历一遍周期就会变长开销依旧比较大epollepoll 的定位 和 poll 与 select 完全一样并且比 select 和 poll 都高效被认为是性能最好的多路转接/多路复用方案核心机制epoll 模型底层主要是三部分红黑树就绪队列(双链表)回调机制当我们创建了一个 epoll 模型底层会创建一颗空的红黑树用户要关心哪些文件描述符上的哪些事件就给红黑树新增节点并且注册对应的回调机制当某个节点上的fd有事件就绪了自动触发回调机制将节点插入到就绪队列中用户只需要检测就绪队列是否为空即可只要不为空就一定有事件就绪了时间复杂度是O(1)细节一个结构体可以既属于A数据结构也属于B数据结构epoll 中的每个fd事件构成的一个个节点既可以属于红黑树也可以属于就绪队列因此当事件就绪了直接链到就绪队列中即可不需要从红黑树中删除认识 epoll 接口epoll_create 用于创建一个 epoll 模型返回值也是文件描述符内核中可能是同时存在多个 epoll 模型的他们之间要相互独立互不影响Linux 下一切皆文件每创建一个 epoll 模型都会创建一个 struct file 对象该对象中有一个 void* private 成员指向了一个 struct eventpoll 结构体该结构体中相关成员保存了红黑树的根节点地址和就绪队列的头节点地址最终向上层用户返回一个文件描述符 epfd 因此后续通过一个 epfd 就可以操作一个 epoll 模型了epoll_create 的参数 size 表示红黑树节点的上限现在已经废弃使用了随便填写即可epoll_ctl 用于向红黑树中插入、修改、删除节点根据参数 op 决定操作的具体动作struct epoll_event 是内核提供的一个结构体该结构体中的 events 表示关心的事件而 daat 是一个联合体我们此处只使用里面的 fd因此该结构体表示用户让内核关心哪些 fd 上的哪些事件当就绪队列中有事件就绪了用户可以调用 epoll_wait 将就绪的事件获取上来第二个参数传递一个数组存放从就绪队列中拷贝上来的就绪的事件maxevents 表述 存放就绪事件的数组大小必须 0timeout 含义与 poll 完全一样返回值同 select 与 poll细节1.就绪的事件会被按照顺序依次拷贝到 events 数组中而返回值表示的是就绪的事件个数那么我们在处理就绪事件时就只会访问就绪的事件而不会访问到其他事件2.就绪事件可能有很多有可能一次读取不完不用担心如果一次没有读取完epoll_wait 还会循环读取就绪事件的编写 EpollServer#pragma once #include iostream #include string #include memory #include sys/epoll.h #include Logger.hpp #include InetAddr.hpp #include Socket.hpp const static int gsize 128; class EpollServer { public: EpollServer(uint16_t port) :_listensock(std::make_uniqueTcpSocket()), _epfd(-1) { _listensock-BuildListenSocketMethod(port); _epfd epoll_create(128); if(_epfd 0) { LOG(LogLevel::FATAL) create epoll fail; return; } LOG(LogLevel::INFO) listen sockfd: _listensock-SockFd() epfd: _epfd; //将 listensockfd 添加到 红黑树中 struct epoll_event ev; ev.events EPOLLIN; ev.data.fd _listensock-SockFd(); int n epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock-SockFd(), ev); if(n) { LOG(LogLevel::INFO) epoll_ctl error; return; } } void Accepter() { InetAddr client; int sockfd _listensock-Accept(client); if(sockfd 0) { LOG(LogLevel::INFO) 获取到一个新连接, fd: sockfd 客户端地址: client.ToString(); //不能直接读取, sockfd 上不一定有数据了, 必须交给 epoll 处理 struct epoll_event ev; ev.events EPOLLIN; ev.data.fd sockfd; epoll_ctl(_epfd, EPOLL_CTL_ADD, sockfd, ev); LOG(LogLevel::INFO) 添加新连接到epoll中, fd: sockfd; } } void Recver(int fd) { char buffer[1024]; ssize_t n recv(fd, buffer, sizeof(buffer) - 1, 0); if(n 0) { LOG(LogLevel::INFO) client say buffer; buffer[n] 0; std::string echo_string server echo# ; echo_string buffer; send(fd, echo_string.c_str(), echo_string.size(), 0); } else if(n 0) { LOG(LogLevel::INFO) client quit, me too!; //坑:epoll_ctl必须对合法fd进行操作因此必须先调用epoll_ctl从红黑树中删除节点, 再关闭fd epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr); close(fd); } else { LOG(LogLevel::INFO) recv error; epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr); close(fd); } } void Dispatcher(int n) { LOG(LogLevel::INFO) 新事件到来, fd就绪; for(int i 0; i n; i) { int fd revs[i].data.fd; uint32_t events revs[i].events; if(events EPOLLIN) { if(fd _listensock-SockFd()) { Accepter(); } else { Recver(fd); } } // if(revs[i].events EPOLLOUT) // { // } } } void Run() { int timeout -1; while(true) { //不能直接accept, 必须交给 epoll 来等待 int n epoll_wait(_epfd, revs, gsize, timeout); switch(n) { case 0: LOG(LogLevel::INFO) timeout...; break; case -1: LOG(LogLevel::FATAL) epoll_wait error; break; default: Dispatcher(n); break; } } } ~EpollServer(){} private: std::unique_ptrSocket _listensock; int _epfd; struct epoll_event revs[gsize]; };epoll 的两种工作模式LT(Level Triggered)水平触发是epoll的默认工作模式行为机制和 select / poll 类似只要 fd 上还有未处理完成的事件例如缓冲区中还有数据每次调用 epoll_wait 时内核都会通知 fd 就绪也就是说如果你一次性没有将缓冲区数据读取完没有关系下次 epoll_wait 时 fd 依旧是 就绪的内核会通知你继续读因此LT模式下fd 可以是阻塞的也可以是非阻塞的ET(Egde Triggered)边缘触发epoll_ctrl 时需要给事件携带 EPOLLET 选项该工作模式下内核只在 fd 状态发生变化时才触发一次通知比如当缓冲区数据从无到有或者缓冲区又增多了对端新发来的数据才会触发一次通知如果你没有一次性将缓冲区数据读取完内核不会再通知你了直到 fd 上有新的数据到来因为 ET模式下事件就绪只通知一次这就要求程序员一次性(一轮)必须将数据读完而用户缓冲区的大小是有限的因此程序员必须循环读取缓冲区并且我们也不知道哪一次就将缓冲区的数据读完了比如缓冲区共有500字节数据每次读取100字节循环读取了5次因为不知道读取完了没第6次依旧要读取但如果 fd 是阻塞状态那么第六次读取就因为缓冲区没有数据读就阻塞了服务器就被挂起了无法再继续提供服务因此 ET模式下必须将 fd 设置为非阻塞循环读取 fd直到 读取错误errno 为 EAGAIN 或 EWOULDBLOCK 时确保缓冲区数据读完了ET本质是倒逼程序员一次性将缓冲区数据读完也就是应用层会尽快读走缓冲区数据那么 ACK 时就可以通告发送端一个更大的16位窗口大小发送方发送数据时滑动窗口就可以更大一次性就可以发送更多的数据就提高了通信的效率下面代码是验证 LT模式下 只要 fd 就绪就会一直通知而 ET 模式下 fd 就绪只会通知一次epoll 的默认工作模式是 LT我们在代码中不处理新到来的事件那么事件就一直处于就绪状态每次调用 epoll_wait 都会进入 Dispatcher 就会一直通知#pragma once #include iostream #include string #include memory #include sys/epoll.h #include Logger.hpp #include InetAddr.hpp #include Socket.hpp const static int gsize 128; class EpollServer { public: EpollServer(uint16_t port) :_listensock(std::make_uniqueTcpSocket()), _epfd(-1) { _listensock-BuildListenSocketMethod(port); _epfd epoll_create(128); if(_epfd 0) { LOG(LogLevel::FATAL) create epoll fail; return; } LOG(LogLevel::INFO) listen sockfd: _listensock-SockFd() epfd: _epfd; //将 listensockfd 添加到 红黑树中 struct epoll_event ev; ev.events EPOLLIN; ev.data.fd _listensock-SockFd(); int n epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock-SockFd(), ev); if(n) { LOG(LogLevel::INFO) epoll_ctl error; return; } } void Dispatcher(int n) { LOG(LogLevel::INFO) 新事件到来, fd就绪; } void Run() { int timeout -1; while(true) { //不能直接accept, 必须交给 epoll 来等待 int n epoll_wait(_epfd, revs, gsize, timeout); switch(n) { case 0: LOG(LogLevel::INFO) timeout...; break; case -1: LOG(LogLevel::FATAL) epoll_wait error; break; default: Dispatcher(n); break; } } } ~EpollServer(){} private: std::unique_ptrSocket _listensock; int _epfd; struct epoll_event revs[gsize]; };调用 epoll_ctl 时给事件添加 EPOLLET并且在 Socket.hpp 中创建出 listen 套接字后将其设置为非阻塞最终连接到来后就通知了一次本次没有处理到来事件下次再也不会通知了#pragma once #include iostream #include string #include memory #include sys/epoll.h #include Logger.hpp #include InetAddr.hpp #include Socket.hpp const static int gsize 128; class EpollServer { public: EpollServer(uint16_t port) :_listensock(std::make_uniqueTcpSocket()), _epfd(-1) { _listensock-BuildListenSocketMethod(port); _epfd epoll_create(128); if(_epfd 0) { LOG(LogLevel::FATAL) create epoll fail; return; } LOG(LogLevel::INFO) listen sockfd: _listensock-SockFd() epfd: _epfd; //将 listensockfd 添加到 红黑树中 struct epoll_event ev; ev.events EPOLLIN | EPOLLET; ev.data.fd _listensock-SockFd(); int n epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock-SockFd(), ev); if(n) { LOG(LogLevel::INFO) epoll_ctl error; return; } } void Dispatcher(int n) { LOG(LogLevel::INFO) 新事件到来, fd就绪; } void Run() { int timeout -1; while(true) { //不能直接accept, 必须交给 epoll 来等待 int n epoll_wait(_epfd, revs, gsize, timeout); switch(n) { case 0: LOG(LogLevel::INFO) timeout...; break; case -1: LOG(LogLevel::FATAL) epoll_wait error; break; default: Dispatcher(n); break; } } } ~EpollServer(){} private: std::unique_ptrSocket _listensock; int _epfd; struct epoll_event revs[gsize]; };