在平时read/recv时如果内核缓冲区有数据会拷贝到应用层缓冲区并返回但如果没有数据默认会阻塞直到有数据时再读上来write时也是一样只不过内核缓冲区很少会被写满因此IO不仅涉及拷贝更重要的是等待过程IO的本质为等 拷贝要想高效的IO只要减少【等】的比重即可五种IO模型阻塞IO默认的IO类型也是最常见的IO模型在内核将数据准备好之前系统调用会一直等待所有的套接字文件描述符默认都是阻塞方式非阻塞IO如果内核还未将数据准备好系统调用仍然会直接返回并且返回EWOULDBLOCK/EAGAIN错误码当采用非阻塞IO一般会用轮询的方式反复调用读取/写入接口信号驱动IO内核将数据准备好的时候使用SIGIO信号通知应用程序进行IO操作其余时间应用程序完全不管它IO多路转接/复用能够同时等待多个文件描述符的就绪状态就像多进程/线程同时等待。但多进程/线程的成本太高而多路转接成本很低由于多路转接原理是可以同时等待多个文件描述符就注定了之前的那些其他IO类接口不能直接使用因此操作系统为了满足我们同时等待多个文件描述符的需求必须单独设计其他的系统调用例如select/poll/epollselect接口只负责IO操作中【等】的部分当等到数据后会交给例如recvfrom进行数据拷贝工作异步IO由内核在数据拷贝完成时, 通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据)进程发起读请求后立即返回继续执行其他任务内核自动完成数据拷贝完成后通知进程不参与进程参与数据拷贝过程同步/异步通信虽然和线程/进程同步/异步一样都用的同步/异步这个词但两者完全没关系。线程/进程同步指的是协调任务执行顺序例如加锁保护临界区而线程/进程异步为多个任务并发执行而判断是同步通信还是异步通信的核心依据是数据从内核缓冲区复制到用户空间缓冲区时应用程序的线程是否亲自参与了等待或执行。同步通信例如阻塞IO、非阻塞IO、多路转接IO、信号驱动IO都参与了数据拷贝工作异步IO当发送方发出请求/消息后无需等待结果即可立即去处理其他事务直到接收方数据拷贝工作完成后再通知发送方非阻塞IO要想在IO时为非阻塞可以将例如recv/recvfrom/send接口的flags参数设为MSG_DONTWAIT或在打开文件时就将open接口的flags参数加上O_NONBLOCKssize_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下是CtrlD { 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() { vectorfunctionvoid () 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下是CtrlD { 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; }psprintlog、download、executeSql为函数名多路转接selectselect是多路转接的一种方式它只负责等待可以一次等待多个fdselect本身没有数据拷贝的能力拷贝要例如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);nfdsselect等待的多个文件描述符中最大的一个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_setFD_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.sizewritefds.sizeexceptfds.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) // 当接收到新连接时通过输出型参数返回客户端ipport { 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::functionstd::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_ptrSelectServer svr(new SelectServer(translation, port)); svr-init(); svr-start(); return 0; }select 缺点1. select能同时等待的文件fd是有上限的除非重新改内核否则无法解决2. 必须借助第三方数组来维护合法的fd集合3. select的大部分参数是输入输出型的调用select前要重新设置所有的fd调用之后还需要检查更新所有的fd遍历的成本很大4. select 采用位图输入用户告诉内核输出内核告诉用户来回的进行数据拷贝拷贝成本大pollpoll方案解决了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数据包括普通数据和优先数据可读是是POLLRDNORM普通数据可读是是POLLRDBAND优先级带数据可读Linux 不支持是是POLLPRI高优先级数据可读比如 TCP 带外数据是是POLLOUT数据包括普通数据和优先数据可写是是POLLWRNORM普通数据可写是是POLLWRBAND优先级带数据可写是是POLLRDHUPTCP 连接被对方关闭或者对方关闭了写操作。它由 GNU 引入是是POLLERR错误否是POLLHUP挂起。比如管道的写端被关闭后读端描述符上将收到 POLLHUP 事件否是POLLNVAL文件描述符没有打开否是常用的为POLLIN读、POLLOUT写、POLLERR错误每个宏都占用一个比特位因此可以通过按位操作包含多个事件fds通过将用户告诉内核与内核告诉用户的事件分开避免了每次都要重新设置fdnfds为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多了后会影响效率epollepoll是为了解决poll遍历问题的多路转接接口epoll_createint epoll_create(int size); int epoll_create1(int flags);创建一个epoll模型size 参数为预期要管理的fd个数现已被忽略只需要保证0即可返回值为epoll模型的fd现代写法推荐用epoll_create1flags填0和epoll_create(1)代表的意思一样epoll_create1(EPOLL_CLOEXEC)代表防止epoll文件描述符泄露给子进程例如fork()、exec()epoll_ctlint 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参数被忽略直接删除fdfd要操作的目标文件描述符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失败返回-1epoll_waitint epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);相当于pool()或select()用于等待文件描述符就绪epfd通过epoll_create创建的epoll模型events用于接收就绪事件输出型参数maxevents用于表示events的数组容量即一次最多返回多少个事件必须 0timeout语义与poll()时一致有就绪事件就返回就绪事件的个数若超时返回0若失败返回-1epoll模型epoll_create会创建一个epoll模型它主要有两个核心数据结构红黑树和就绪链表双向当调用epoll_ctl接口时会对红黑树进行增删改操作每个节点对应一个需要关心的fd及其事件并注册当该fd有数据时回调的接口ep_poll_callback用于将事件挂载到就绪队列中该红黑树的key为对应fdstruct 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/边缘触发ETepoll_wait将所有就绪的事件按照顺序放到用户传入的数组的中如果就绪队列很多数据节点一次拿不完下次epoll_wait时默认会继续通知也就是水平触发LT水平触发只要状态“处于就绪状态”例如缓冲区还有数据未读完、或者就绪队列里还有未处理完的事件epoll_wait就会不断通知。但如果将对应epitem节点改为边缘触发events对于ET 模式的节点内核在将事件拷贝给用户态数组后直接从就绪链表rdllist中移除该节点不再放回。下一次调用epoll_wait时即使底层缓冲区还有未读完的数据/就绪队列之前没处理完epoll_wait也不会通知你它会一直阻塞等待直到这个fd上再次有新的数据到达发生新的状态边沿变化。边缘触发只有当状态“发生变化/出现新边缘”例如数据从无到有、从不可写变为可写时epoll_wait才会通知一次要对某个fd设置边缘触发就将events | EPOLLETstruct 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 → 发送方可以一次发出更多数据 → 吞吐量提升ReactorReactor模式反应堆模式半同步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一直在通知写就绪... Linux高级IO
在平时read/recv时如果内核缓冲区有数据会拷贝到应用层缓冲区并返回但如果没有数据默认会阻塞直到有数据时再读上来write时也是一样只不过内核缓冲区很少会被写满因此IO不仅涉及拷贝更重要的是等待过程IO的本质为等 拷贝要想高效的IO只要减少【等】的比重即可五种IO模型阻塞IO默认的IO类型也是最常见的IO模型在内核将数据准备好之前系统调用会一直等待所有的套接字文件描述符默认都是阻塞方式非阻塞IO如果内核还未将数据准备好系统调用仍然会直接返回并且返回EWOULDBLOCK/EAGAIN错误码当采用非阻塞IO一般会用轮询的方式反复调用读取/写入接口信号驱动IO内核将数据准备好的时候使用SIGIO信号通知应用程序进行IO操作其余时间应用程序完全不管它IO多路转接/复用能够同时等待多个文件描述符的就绪状态就像多进程/线程同时等待。但多进程/线程的成本太高而多路转接成本很低由于多路转接原理是可以同时等待多个文件描述符就注定了之前的那些其他IO类接口不能直接使用因此操作系统为了满足我们同时等待多个文件描述符的需求必须单独设计其他的系统调用例如select/poll/epollselect接口只负责IO操作中【等】的部分当等到数据后会交给例如recvfrom进行数据拷贝工作异步IO由内核在数据拷贝完成时, 通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据)进程发起读请求后立即返回继续执行其他任务内核自动完成数据拷贝完成后通知进程不参与进程参与数据拷贝过程同步/异步通信虽然和线程/进程同步/异步一样都用的同步/异步这个词但两者完全没关系。线程/进程同步指的是协调任务执行顺序例如加锁保护临界区而线程/进程异步为多个任务并发执行而判断是同步通信还是异步通信的核心依据是数据从内核缓冲区复制到用户空间缓冲区时应用程序的线程是否亲自参与了等待或执行。同步通信例如阻塞IO、非阻塞IO、多路转接IO、信号驱动IO都参与了数据拷贝工作异步IO当发送方发出请求/消息后无需等待结果即可立即去处理其他事务直到接收方数据拷贝工作完成后再通知发送方非阻塞IO要想在IO时为非阻塞可以将例如recv/recvfrom/send接口的flags参数设为MSG_DONTWAIT或在打开文件时就将open接口的flags参数加上O_NONBLOCKssize_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下是CtrlD { 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() { vectorfunctionvoid () 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下是CtrlD { 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; }psprintlog、download、executeSql为函数名多路转接selectselect是多路转接的一种方式它只负责等待可以一次等待多个fdselect本身没有数据拷贝的能力拷贝要例如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);nfdsselect等待的多个文件描述符中最大的一个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_setFD_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.sizewritefds.sizeexceptfds.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) // 当接收到新连接时通过输出型参数返回客户端ipport { 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::functionstd::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_ptrSelectServer svr(new SelectServer(translation, port)); svr-init(); svr-start(); return 0; }select 缺点1. select能同时等待的文件fd是有上限的除非重新改内核否则无法解决2. 必须借助第三方数组来维护合法的fd集合3. select的大部分参数是输入输出型的调用select前要重新设置所有的fd调用之后还需要检查更新所有的fd遍历的成本很大4. select 采用位图输入用户告诉内核输出内核告诉用户来回的进行数据拷贝拷贝成本大pollpoll方案解决了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数据包括普通数据和优先数据可读是是POLLRDNORM普通数据可读是是POLLRDBAND优先级带数据可读Linux 不支持是是POLLPRI高优先级数据可读比如 TCP 带外数据是是POLLOUT数据包括普通数据和优先数据可写是是POLLWRNORM普通数据可写是是POLLWRBAND优先级带数据可写是是POLLRDHUPTCP 连接被对方关闭或者对方关闭了写操作。它由 GNU 引入是是POLLERR错误否是POLLHUP挂起。比如管道的写端被关闭后读端描述符上将收到 POLLHUP 事件否是POLLNVAL文件描述符没有打开否是常用的为POLLIN读、POLLOUT写、POLLERR错误每个宏都占用一个比特位因此可以通过按位操作包含多个事件fds通过将用户告诉内核与内核告诉用户的事件分开避免了每次都要重新设置fdnfds为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多了后会影响效率epollepoll是为了解决poll遍历问题的多路转接接口epoll_createint epoll_create(int size); int epoll_create1(int flags);创建一个epoll模型size 参数为预期要管理的fd个数现已被忽略只需要保证0即可返回值为epoll模型的fd现代写法推荐用epoll_create1flags填0和epoll_create(1)代表的意思一样epoll_create1(EPOLL_CLOEXEC)代表防止epoll文件描述符泄露给子进程例如fork()、exec()epoll_ctlint 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参数被忽略直接删除fdfd要操作的目标文件描述符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失败返回-1epoll_waitint epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);相当于pool()或select()用于等待文件描述符就绪epfd通过epoll_create创建的epoll模型events用于接收就绪事件输出型参数maxevents用于表示events的数组容量即一次最多返回多少个事件必须 0timeout语义与poll()时一致有就绪事件就返回就绪事件的个数若超时返回0若失败返回-1epoll模型epoll_create会创建一个epoll模型它主要有两个核心数据结构红黑树和就绪链表双向当调用epoll_ctl接口时会对红黑树进行增删改操作每个节点对应一个需要关心的fd及其事件并注册当该fd有数据时回调的接口ep_poll_callback用于将事件挂载到就绪队列中该红黑树的key为对应fdstruct 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/边缘触发ETepoll_wait将所有就绪的事件按照顺序放到用户传入的数组的中如果就绪队列很多数据节点一次拿不完下次epoll_wait时默认会继续通知也就是水平触发LT水平触发只要状态“处于就绪状态”例如缓冲区还有数据未读完、或者就绪队列里还有未处理完的事件epoll_wait就会不断通知。但如果将对应epitem节点改为边缘触发events对于ET 模式的节点内核在将事件拷贝给用户态数组后直接从就绪链表rdllist中移除该节点不再放回。下一次调用epoll_wait时即使底层缓冲区还有未读完的数据/就绪队列之前没处理完epoll_wait也不会通知你它会一直阻塞等待直到这个fd上再次有新的数据到达发生新的状态边沿变化。边缘触发只有当状态“发生变化/出现新边缘”例如数据从无到有、从不可写变为可写时epoll_wait才会通知一次要对某个fd设置边缘触发就将events | EPOLLETstruct 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 → 发送方可以一次发出更多数据 → 吞吐量提升ReactorReactor模式反应堆模式半同步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一直在通知写就绪...