【Linux高级IO】select、poll、epoll

toc

作者：爱写代码的刚子

时间：2024.6.5

前言：本篇博客将会介绍面试重点考察的select、poll、epoll

IO: input && Output

read && write

应用层read&&write的时候，本质把数据从用户层写给OS —— 本质就是拷贝函数IO = 等 + 拷贝（要进行拷贝，必须先判断条件成立，即等读写事件就绪）

什么叫做高效IO？单位时间内，IO过程中，等的比重越小，IO效率越高！几乎所有的提高IO效率的策略，本质就是这个！

同步IO：参与了等或者参与了拷贝就是同步IO，信号驱动IO也是同步IO因为它参与了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

非阻塞IO，记录锁，系统V流机制，I/O多路转接（也叫I/O多路复用），readv和writev函数以及存储映射IO（mmap），这些统称为高级IO.

非阻塞IO

fcntl

一个文件描述符，默认都是阻塞IO

#include <unistd.h>#include <fcntl.h>int fcntl(int fd, int cmd, ... /* arg */ ); 通过这个接口可以设置文件描述符属性为非阻塞

传入的cmd的值不同，后面追加的参数也不相同。

fcntl函数有5种功能：

复制一个现有的描述符(cmd=F_DUPFD).获得/设置文件描述符标记(cmd=F_GETFD或F_SETFD).获得/设置文件状态标记(cmd=F_GETFL或F_SETFL).获得/设置异步I/O所有权(cmd=F_GETOWN或F_SETOWN).获得/设置记录锁(cmd=F_GETLK,F_SETLK或F_SETLKW).

实现函数SetNoBlock

基于fcntl，我们实现一个SetNoBlock函数，将文件描述符设置为非阻塞.

void SetNoBlock(int fd) { int fl = fcntl(fd, F_GETFL); if (fl < 0) { perror("fcntl");return; } fcntl(fd, F_SETFL, fl | O_NONBLOCK);}

实验：

#include <iostream>#include <unistd.h>#include <fcntl.h>#include <cstdio>#include <cerrno>#include <cstring>using namespace std;void SetNonBlock(int fd){ int fl = fcntl(fd, F_GETFL); if (fl < 0) { perror("fcntl"); return; } fcntl(fd, F_SETFL, fl | O_NONBLOCK); cout << " set " << fd << " nonblock done" << endl;}int main(){ char buffer[1024]; SetNonBlock(0); sleep(1); while (true) { // printf("Please Enter# "); // fflush(stdout); ssize_t n = read(0, buffer, sizeof(buffer) - 1); if (n > 0) { buffer[n - 1] = 0; cout << "echo : " << buffer << endl; } else if (n == 0) { cout << "read done" << endl; break; } else { // 1. 设置成为非阻塞，如果底层fd数据没有就绪，recv/read/write/send, 返回值会以出错的形式返回 // 2. a. 真的出错 b. 底层没有就绪 // 3. 我怎么区分呢？通过errno区分！！！ if (errno == EWOULDBLOCK) { cout << "0 fd data not ready, try again!" << endl; // do_other_thing(); sleep(1); } else { cerr << "read error, n = " << n << "errno code: " << errno << ", error str: " << strerror(errno) << endl; } } } return 0;} 如果出错，errno也会被设置

使用F_GETFL将当前的文件描述符的属性取出来(这是一个位图).然后再使用F_SETFL将文件描述符设置回去. 设置回去的同时, 加上一个O_NONBLOCK参数.

I/O多路转接之select

初识select

系统提供select函数来实现多路复用输入/输出模型.

select系统调用是用来让我们的程序监视多个文件描述符的状态变化的;程序会停在select这里等待，直到被监视的文件描述符有一个或多个发生了状态改变;

select函数原型

select的函数原型如下: #include <sys/select.h>

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

参数解释:

nfds = maxfd +1;rdset,wrset,exset分别对应于需要检测的可读文件描述符的集合，可写文件描述符的集合及异常文件描述符的集合;参数timeout为结构timeval，用来设置select()的等待时间

返回值：

n > 0：有n个fd就绪了

n==0：超时，没有错误，但是也没有fd就绪

n<0：等待出错了

获取时间的接口：

其中struct timeval结构体：

参数timeout取值：

NULL:则表示select()没有timeout，select将一直被阻塞，直到某个文件描述符上发生了事件;0:仅检测描述符集合的状态，然后立即返回，并不等待外部事件的发生。特定的时间值:如果在指定的时间段里没有事件发生，select将超时返回。

如果设置了timeout参数，就是输入输出型参数

关于fd_set结构（位图）：

比特位的位置表示文件描述符编号（从右向左:0，1，2，3…），

比特位的内容0或者是1，表示是否需要内核关心相应的事件。

参数作为返回值返回时：比特位的内容0或者是1，表示上面相应的事件是否就绪

fd_set是一张位图，让用户<->内核传递fd是否就绪的信息的！

*fd_set readfds：fd_set内核提供的一种数据类型，它是位图

输入时：用户告诉内核，我给你的一个或者多个fd，内核要关心fd上的读事件，读事件就绪了，要通知用户。

输出时：内核告诉用户，用户让内核关心的多个fd中，有哪些已经就绪了，让用户来读取。

一组操作fd_set的接口,用来方便操作位图.

void FD_CLR(int fd, fd_set *set);// 用来清除描述词组set中相关fd 的位int FD_ISSET(int fd, fd_set *set);// 用来测试描述词组set中相关fd 的位是否为真 void FD_SET(int fd, fd_set *set);// 用来设置描述词组set中相关fd 的位void FD_ZERO(fd_set *set);// 用来清除描述词组set 的全部位

关于timeval结构

timeval结构用于描述一段时间长度，如果在这个时间内，需要监视的描述符没有事件发生则函数返回，返回值为 0。

函数返回值:

执行成功则返回文件描述词状态已改变的个数如果返回0代表在描述词状态改变前已超过timeout时间，没有返回当有错误发生时则返回-1，错误原因存于errno，此时参数readfds，writefds, exceptfds和timeout的 值变成不可预测。

错误值可能为:

EBADF 文件描述词为无效的或该文件已关闭EINTR 此调用被信号所中断EINVAL 参数n 为负值。ENOMEM 核心内存不足

常见的程序片段如下:

fs_set readset;FD_SET(fd,&readset); select(fd+1,&readset,NULL,NULL,NULL); if(FD_ISSET(fd,readset)){ ......}

理解select执行过程

理解select模型的关键在于理解fd_set,为说明方便，取fd_set长度为1字节，fd_set中的每一bit可以对应一个文件描述符fd。则1字节长的fd_set最大可以对应8个fd.

执行fd_set set; FD_ZERO(&set);则set用位表示是0000,0000。若fd=5,执行FD_SET(fd,&set); 后set变为0001,0000(第5位置为1)若再加入fd=2，fd=1,则set变为0001,0011执行 select(6,&set,0,0,0)阻塞等待若fd=1,fd=2上都发生可读事件，则select返回，此时set变为 0000,0011。注意:没有事件发生的fd=5被清空。

编写select服务器

Main.cc

#include "SelectServer.hpp"#include <memory>int main(){ // std::cout <<"fd_set bits num : " << sizeof(fd_set) * 8 << std::endl; std::unique_ptr<SelectServer> svr(new SelectServer()); svr->Init(); svr->Start(); return 0;}

SelectServer.hpp

#pragma once#include <iostream>#include <sys/select.h>#include <sys/time.h>#include "Socket.hpp"using namespace std;static const uint16_t defaultport = 8888;static const int fd_num_max = (sizeof(fd_set) * 8);int defaultfd = -1;class SelectServer{ public: SelectServer(uint16_t port = defaultport) : _port(port) { for (int i = 0; i < fd_num_max; i++) { fd_array[i] = defaultfd; // std::cout << "fd_array[" << i << "]" << " : " << fd_array[i] << std::endl; } } bool Init() { _listensock.Socket(); _listensock.Bind(_port); _listensock.Listen(); return true; } void Accepter() { // 我们的连接事件就绪了 std::string clientip; uint16_t clientport = 0; int sock = _listensock.Accept(&clientip, &clientport); // 会不会阻塞在这里？不会 if (sock < 0) return; lg(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sock); // sock -> fd_array[] int pos = 1; for (; pos < fd_num_max; pos++) // 第二个循环 { if (fd_array[pos] != defaultfd) continue; else break; } if (pos == fd_num_max) { lg(Warning, "server is full, close %d now!", sock); close(sock); } else { fd_array[pos] = sock; PrintFd(); // TODO } } void Recver(int fd, int pos) { // demo char buffer[1024]; ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug? if (n > 0) { buffer[n] = 0; cout << "get a messge: " << buffer << endl; } else if (n == 0) { lg(Info, "client quit, me too, close fd is : %d", fd); close(fd); fd_array[pos] = defaultfd; // 这里本质是从select中移除 } else { lg(Warning, "recv error: fd is : %d", fd); close(fd); fd_array[pos] = defaultfd; // 这里本质是从select中移除 } } void Dispatcher(fd_set &rfds) { for (int i = 0; i < fd_num_max; i++) // 这是第三个循环 { int fd = fd_array[i]; if (fd == defaultfd) continue; if (FD_ISSET(fd, &rfds)) { if (fd == _listensock.Fd()) { Accepter(); // 连接管理器 } else // non listenfd { Recver(fd, i); } } } } void Start() { int listensock = _listensock.Fd(); fd_array[0] = listensock; for (;;) { fd_set rfds; FD_ZERO(&rfds); int maxfd = fd_array[0]; for (int i = 0; i < fd_num_max; i++) // 第一次循环 { if (fd_array[i] == defaultfd) continue; FD_SET(fd_array[i], &rfds); if (maxfd < fd_array[i]) { maxfd = fd_array[i]; lg(Info, "max fd update, max fd is: %d", maxfd); } } // accept?不能直接accept！检测并获取listensock上面的事件，新连接到来，等价于读事件就绪 // struct timeval timeout = {1, 0}; // 输入输出型参数，要进行周期的重复设置 struct timeval timeout = { 0, 0}; // 输入输出，可能要进行周期的重复设置 // 如果事件就绪，上层不处理，select会一直通知！ // select告诉用户就绪了，接下来的一次读取，我们读取fd的时候，不会被阻塞 // rfds: 输入输出型参数。 1111 1111 -> 0000 0000 int n = select(maxfd + 1, &rfds, nullptr, nullptr, /*&timeout*/ nullptr); switch (n) { case 0: cout << "time out, timeout: " << timeout.tv_sec << "." << timeout.tv_usec << endl; break; case -1: cerr << "select error" << endl; break; default: // 有事件就绪了，TODO cout << "get a new link!!!!!" << endl; Dispatcher(rfds); // 就绪的事件和fd break; } } } void PrintFd() { cout << "online fd list: "; for (int i = 0; i < fd_num_max; i++) { if (fd_array[i] == defaultfd) continue; cout << fd_array[i] << " "; } cout << endl; } ~SelectServer() { _listensock.Close(); }private: Sock _listensock; uint16_t _port; int fd_array[fd_num_max]; // 数组, 来用户维护 // int wfd_array[fd_num_max];};

Socket.hpp

#pragma once#include <iostream>#include <string>#include <unistd.h>#include <cstring>#include <sys/types.h>#include <sys/stat.h>#include <sys/socket.h>#include <arpa/inet.h>#include <netinet/in.h>#include "Log.hpp"enum{ SocketErr = 2, BindErr, ListenErr,};// TODOconst int backlog = 10;class Sock{ public: Sock() { } ~Sock() { }public: void Socket() { sockfd_ = socket(AF_INET, SOCK_STREAM, 0); if (sockfd_ < 0) { lg(Fatal, "socker error, %s: %d", strerror(errno), errno); exit(SocketErr); } int opt = 1; setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)); } void Bind(uint16_t port) { 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) { lg(Fatal, "bind error, %s: %d", strerror(errno), errno); exit(BindErr); } } void Listen() { if (listen(sockfd_, backlog) < 0) { lg(Fatal, "listen error, %s: %d", strerror(errno), errno); exit(ListenErr); } } int Accept(std::string *clientip, uint16_t *clientport) { struct sockaddr_in peer; socklen_t len = sizeof(peer); int newfd = accept(sockfd_, (struct sockaddr*)&peer, &len); if(newfd < 0) { lg(Warning, "accept error, %s: %d", strerror(errno), errno); return -1; } char ipstr[64]; inet_ntop(AF_INET, &peer.sin_addr, ipstr, sizeof(ipstr)); *clientip = ipstr; *clientport = ntohs(peer.sin_port); return newfd; } bool Connect(const std::string &ip, const uint16_t &port) { struct sockaddr_in peer; memset(&peer, 0, sizeof(peer)); peer.sin_family = AF_INET; peer.sin_port = htons(port); inet_pton(AF_INET, ip.c_str(), &(peer.sin_addr)); int n = connect(sockfd_, (struct sockaddr*)&peer, sizeof(peer)); if(n == -1) { std::cerr << "connect to " << ip << ":" << port << " error" << std::endl; return false; } return true; } void Close() { close(sockfd_); } int Fd() { return sockfd_; }private: int sockfd_;};

socket就绪条件

读就绪

socket内核中, 接收缓冲区中的字节数, 大于等于低水位标记SO_RCVLOWAT. 此时可以无阻塞的读该文件 描述符, 并且返回值大于0;socket TCP通信中, 对端关闭连接, 此时对该socket读, 则返回0;监听的socket上有新的连接请求;socket上有未处理的错误;

写就绪

socket内核中, 发送缓冲区中的可用字节数(发送缓冲区的空闲位置大小), 大于等于低水位标记 SO_SNDLOWAT, 此时可以无阻塞的写, 并且返回值大于0;socket的写操作被关闭(close或者shutdown). 对一个写操作被关闭的socket进行写操作, 会触发SIGPIPE 信号;socket使用非阻塞connect连接成功或失败之后;socket上有未读取的错误;

异常就绪

有关TCP中的紧急指针

select的特点

可监控的文件描述符个数取决与sizeof(fd_set)的值. 我这边服务器上sizeof(fd_set)=512，每bit表示一个文件 描述符，则我服务器上支持的最大文件描述符是512*8=4096.

将fd加入select监控集的同时，还要再使用一个数据结构array保存放到select监控集中的fd，

一是用于再select 返回后，array作为源数据和fd_set进行FD_ISSET判断。二是select返回后会把以前加入的但并无事件发生的fd清空，则每次开始select前都要重新从array取得 fd逐一加入(FD_ZERO最先)，扫描array的同时取得fd最大值maxfd，用于select的第一个参数。

备注: fd_set的大小可以调整，可能涉及到重新编译内核.

select缺点

每次调用select, 都需要手动设置fd集合, 从接口使用角度来说也非常不便.每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大（数据拷贝频率比较高）同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大select支持的文件描述符数量太小.等待的fd是有上限的( fd_set 结构体位图就限制了select的上限)

I/O多路转接之poll

poll函数接口

#include <poll.h>int poll(struct pollfd *fds, nfds_t nfds, int timeout);// pollfd结构 struct pollfd { int fd;/* file descriptor */ short events;/* requested events */ short revents; /* returned events */};

参数说明

fds是一个poll函数监听的结构列表. 每一个元素中, 包含了三部分内容: 文件描述符, 监听的事件集合, 返 回的事件集合.nfds表示fds数组的长度.timeout表示poll函数的超时时间, 单位是毫秒(ms).

events和revents的取值:

返回结果

返回值小于0, 表示出错;返回值等于0, 表示poll函数等待超时;返回值大于0, 表示poll由于监听的文件描述符就绪而返回.

socket就绪条件

同select

poll的优点

不同于select使用三个位图来表示三个fdset的方式，poll使用一个pollfd的指针实现

pollfd结构包含了要监视的event和发生的event，不再使用select“参数-值”传递的方式。接口使用比select更方便。poll并没有最大数量限制 (但是数量过大后性能也是会下降).

poll的缺点

poll中监听的文件描述符数目增多时

和select函数一样，poll返回后，需要轮询pollfd来获取就绪的描述符.每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中.同时连接的大量客户端在一时刻可能只有很少的处于就绪状态, 因此随着监视的描述符数量的增长, 其效 率也会线性下降.

poll和select有一样的缺点：都需要遍历，在内核中需要检测哪些文件描述符就绪，在用户层检测哪些事件就绪

编写poll服务器

PollServer.hpp:

#pragma once#include <iostream>#include <poll.h>#include <sys/time.h>#include "Socket.hpp"using namespace std;static const uint16_t defaultport = 8888;static const int fd_num_max = 64;int defaultfd = -1;int non_event = 0;class PollServer{ public: PollServer(uint16_t port = defaultport) : _port(port) { for (int i = 0; i < fd_num_max; i++) { _event_fds[i].fd = defaultfd; _event_fds[i].events = non_event; _event_fds[i].revents = non_event; // std::cout << "fd_array[" << i << "]" << " : " << fd_array[i] << std::endl; } } bool Init() { _listensock.Socket(); _listensock.Bind(_port); _listensock.Listen(); return true; } void Accepter() { // 我们的连接事件就绪了 std::string clientip; uint16_t clientport = 0; int sock = _listensock.Accept(&clientip, &clientport); // 会不会阻塞在这里？不会 if (sock < 0) return; lg(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sock); // sock -> fd_array[] int pos = 1; for (; pos < fd_num_max; pos++) // 第二个循环 { if (_event_fds[pos].fd != defaultfd) continue; else break; } if (pos == fd_num_max) { lg(Warning, "server is full, close %d now!", sock); close(sock); // 扩容 } else { // fd_array[pos] = sock; _event_fds[pos].fd = sock; _event_fds[pos].events = POLLIN; _event_fds[pos].revents = non_event; PrintFd(); // TODO } } void Recver(int fd, int pos) { // demo char buffer[1024]; ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug? if (n > 0) { buffer[n] = 0; cout << "get a messge: " << buffer << endl; } else if (n == 0) { lg(Info, "client quit, me too, close fd is : %d", fd); close(fd); _event_fds[pos].fd = defaultfd; // 这里本质是从select中移除 } else { lg(Warning, "recv error: fd is : %d", fd); close(fd); _event_fds[pos].fd = defaultfd; // 这里本质是从select中移除 } } void Dispatcher() { for (int i = 0; i < fd_num_max; i++) // 这是第三个循环 { int fd = _event_fds[i].fd; if (fd == defaultfd) continue; if (_event_fds[i].revents & POLLIN) { if (fd == _listensock.Fd()) { Accepter(); // 连接管理器 } else // non listenfd { Recver(fd, i); } } } } void Start() { _event_fds[0].fd = _listensock.Fd(); _event_fds[0].events = POLLIN; int timeout = 3000; // 3s for (;;) { int n = poll(_event_fds, fd_num_max, timeout); switch (n) { case 0: cout << "time out... " << endl; break; case -1: cerr << "poll error" << endl; break; default: // 有事件就绪了，TODO cout << "get a new link!!!!!" << endl; Dispatcher(); break; } } } void PrintFd() { cout << "online fd list: "; for (int i = 0; i < fd_num_max; i++) { if (_event_fds[i].fd == defaultfd) continue; cout << _event_fds[i].fd << " "; } cout << endl; } ~PollServer() { _listensock.Close(); }private: Sock _listensock; uint16_t _port; struct pollfd _event_fds[fd_num_max]; // 数组, 用户维护的！ // struct pollfd *_event_fds; // int fd_array[fd_num_max]; // int wfd_array[fd_num_max];};

Main.cc:

#include "PollServer.hpp"#include <memory>int main(){ // std::cout <<"fd_set bits num : " << sizeof(fd_set) * 8 << std::endl; // std::unique_ptr<SelectServer> svr(new SelectServer()); std::unique_ptr<PollServer> svr(new PollServer()); svr->Init(); svr->Start(); return 0;}

I/O多路转接之epoll

epoll初识

按照man手册的说法: 是为处理大批量句柄而作了改进的poll. 它是在2.5.44内核中被引进的(epoll(4) is a new API introduced in Linux kernel 2.5.44) 它几乎具备了之前所说的一切优点，被公认为Linux2.6下性能最好的多路I/O就绪通知方法.

epoll的相关系统调用

epoll的三个相关系统调用:

epoll_create

#include <sys/epoll.h>int epoll_create(int size);

创建一个epoll的句柄

自从linux2.6.8之后，size参数是被忽略的.用完之后, 必须调用close()关闭.

epoll_ctl（关心红黑树）

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

epoll的事件注册函数.

它不同于select()是在监听事件时告诉内核要监听什么类型的事件, 而是在这里先注册要监听的事件类型.第一个参数是epoll_create()的返回值(epoll的句柄).第二个参数表示动作，用三个宏来表示.第三个参数是需要监听的fd.第四个参数是告诉内核需要监听什么事.

第二个参数的取值：

EPOLL_CTL_ADD :注册新的fd到epfd中;EPOLL_CTL_MOD :修改已经注册的fd的监听事件;EPOLL_CTL_DEL :从epfd中删除一个fd;

struct epoll_event结构如下:

events可以是以下几个宏的集合:

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

epoll_wait（关心就绪队列）

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

收集在epoll监控的事件中已经发送的事件.

参数events是分配好的epoll_event结构体数组.（输出型参数）epoll将会把发生的事件赋值到events数组中 (events不可以是空指针，内核只负责把数据复制到这个 events数组中，不会去帮助我们在用户态中分配内存).maxevents告之内核这个events有多大，这个 maxevents的值不能大于创建epoll_create()时的size.参数timeout是超时时间 (毫秒，0会立即返回，-1是永久阻塞).如果函数调用成功，返回对应I/O上已准备好的文件描述符数目，如返回0表示已超时, 返回小于0表示函 数失败.

epoll工作原理

epoll的整个工作流程：

当某一进程调用epoll_create方法时，Linux内核会创建一个eventpoll结构体,这个结构体中有两个成员与epoll的使用方式密切相关。

struct eventpoll{ ..../*红黑树的根节点，这颗树中存储着所有添加到epoll中的需要监控的事件*/ struct rb_root rbr; /*双链表中则存放着将要通过epoll_wait返回给用户的满足条件的事件*/ struct list_head rdlist;.... };

每个epoll对象都有一个独立的eventpoll结构体，用于存放通过epoll_ctl方法向epoll对象中添加进来的事件这些事件都会挂载在红黑树中，如此，重复添加的事件就可以通过红黑树而高效的识别出来（红黑树的插入时间效率是lgn，其中n为树的高度）.而所有添加到epoll中的事件都会与设备（网卡）驱动程序建立回调关系，也就是说，当响应的事件发生时会调用这个回调方法。这个回调方法在内核中叫ep_poll_callback，它会将发生的事件添加到rdlist双链表中。在epoll中，对于每一个事件，都会建立一个epitem结构体

struct epitem{ struct rb_node rbn;//红黑树节点struct list_head rdllink;//双向链表节点struct epoll_filefd ffd; //事件句柄信息struct eventpoll *ep; //指向其所属的eventpoll对象 struct epoll_event event; //期待发生的事件类型}

epoll模型

操作系统中可能有多个epoll模型（也是一个地址，会被操作系统管理起来）

【问题】：进程是怎么找到epoll模型的呢？

进程中会为epoll模型创建struct file结构体，进程可以通过文件描述符（fd）来找到，从而可以找到所有的信息

事实上callback会和每个节点对应，更高效。

epoll和poll、select差别非常大

当调用epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否有epitem 元素即可.如果rdlist不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户. 这个操作的时间复杂度 是 O(1).

总结一下，epoll的使用过程：

调用epoll_create创建一个epoll句柄;调用epoll_ctl, 将要监控的文件描述符进行注册;调用epoll_wait, 等待文件描述符就绪;

epoll的优点（和select的缺点对应）

接口使用方便: 虽然拆分成了三个函数, 但是反而使用起来更方便高效. 不需要每次循环都设置关注的文 件描述符, 也做到了输入输出参数分离开数据拷贝轻量: 只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中, 这个操作并不频 繁(而select/poll都是每次循环都要进行拷贝)事件回调机制: 避免使用遍历, 而是使用回调函数的方式, 将就绪的文件描述符结构加入到就绪队列中, epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪. 这个操作时间复杂度O(1)( 检测事件就绪是O(1)的). 即使文件描述 符数目很多, 效率也不会受到影响.没有数量限制: 文件描述符数目无上限.(fd,event没有上限（由红黑树的大小决定，也就是计算机的内存决定的）)，其实这颗红黑树其实相当于poll里面自己维护的数组，而epoll让用户不再自己维护事件描述符及其关心的事件，直接提供系统调用即可

epoll的返回值中n表示有几个fd就绪了，同时就绪事件在就绪队列中连续存放，n可以用来遍历就绪事件。

注意！！

网上有些博客说, epoll中使用了内存映射机制

内存映射机制: 内核直接将就绪队列通过mmap的方式映射到用户态. 避免了拷贝内存这样的额外性能开销.

这种说法是不准确的. 我们定义的struct epoll_event是我们在用户空间中分配好的内存. 势必还是需要将内核的数据拷贝到这个用户空间的内存中的.

epoll_wait 调用在内核态维护一个就绪列表，并在有就绪事件时将文件描述符列表拷贝到用户态。这种实现方式虽然涉及一定的内存拷贝，但它在性能和复杂性之间取得了平衡。

标准的 epoll 实现中并没有使用 mmap 来将就绪队列直接映射到用户态。

要多对比总结select, poll, epoll之间的优点和缺点(重要, 面试中常见).

注意epoll是负责等待的，不负责socket的创建、绑定和监听

编写epoll服务器

Epoller.hpp

#pragma once#include "nocopy.hpp"#include <sys/epoll.h>#include <cerrno>#include <cstring>class Epoller:public nocopy{ static const int size = 128;public: Epoller() { _epfd = epoll_create(size); if(_epfd == -1) { lg(Error,"epoll_create error: %s",strerror(errno)); } else{ lg(Info,"epoll_create sucess: %d",_epfd); } } int EpollerWait(struct epoll_event revents[] , int num) { int n = epoll_wait(_epfd,revents,num,_timeout); return n; } int EpollerUpdate(int oper, int sock,uint32_t event) { int n =0; if(oper==EPOLL_CTL_DEL) { n = epoll_ctl(_epfd,oper,sock,nullptr); if(n!=0)//成功是0 { lg(Error,"epoll_ctl delete error!"); } }else{ // EPOLL_CTL_MOD || EPOLL_CTL_ADD struct epoll_event ev; ev.events = event; ev.data.fd = sock;//方便我们后面知道是哪一个fd就绪了 n = epoll_ctl(_epfd,oper,sock,&ev); if(n!=0)//成功是0 { lg(Error,"epoll_ctl error!"); } } return n; } ~Epoller() { if(_epfd>=0) { close(_epfd); } }private: int _epfd; int _timeout{ 3000};};

EpollServer.hpp

#pragma once#include <iostream>#include <sys/epoll.h>#include "Socket.hpp"#include "Epoller.hpp"#include <memory>#include "Log.hpp"#include "nocopy.hpp"uint32_t EVENT_IN = (EPOLLIN);uint32_t EVENT_OUT = (EPOLLOUT);class EpollServer: public nocopy{ static const int num = 64;public: EpollServer(uint16_t port):_port(port),_listsocket_ptr(new Sock()),_epoller_ptr(new Epoller()){ } void Init() { _listsocket_ptr->Socket(); _listsocket_ptr->Bind(_port); _listsocket_ptr->Listen(); lg(Info, "create listen socket success: %d\n",_listsocket_ptr->Fd()); } void Accepter() { //获取新链接 std::string clientip; uint16_t clientport; int sock = _listsocket_ptr->Accept(&clientip,&clientport); if(sock > 0) { //不能在这里读取，因为这只是链接建立好了，并不能在这里读取数据 _epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD,sock,EVENT_IN); lg(Info,"get a new link,client info@ %s:%d",clientip.c_str(),clientport); } } void Recver(int fd) { // demo char buffer[1024]; ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug? if (n > 0) { buffer[n] = 0; std::cout << "get a messge: " << buffer << std::endl; std::string echo_str = "server echo $ " +std::string(buffer); write(fd ,echo_str.c_str() ,echo_str.size()); } else if (n == 0) { lg(Info, "client quit, me too, close fd is : %d", fd); //细节3(先移除再关闭) _epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL,fd,0); close(fd); } else { lg(Warning, "recv error: fd is : %d", fd); _epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL,fd,0); close(fd); } } void Dispatcher(struct epoll_event revs[],int num) { for(int i=0;i<num;++i) { uint32_t events = revs[i].events; int fd = revs[i].data.fd; if(events & EVENT_IN) { if(fd == _listsocket_ptr->Fd()) { Accepter(); }else{ //其他fd上面的普通读事件就绪 Recver(fd); } }else if(events & EVENT_OUT) { }else{ } } } void Start() { //将listensock添加到epoll中 -> listensock和他关心的事件，添加到内核epoll模型中rb_tree. _epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD,_listsocket_ptr->Fd(),EVENT_IN); struct epoll_event revs[num];//这里规定了数组大小，但是没关系，一次捞不完等下一批捞， //一个文件描述符是否有读事件就绪的本质，看文件描述符及其关心的事件有没有放入到就绪队列中 for(;;) { int n = _epoller_ptr->EpollerWait(revs,num); if(n > 0) { //有事件就绪了 lg(Debug,"event happend,fd is : %d", revs[0].data.fd); Dispatcher(revs,n); }else if(n == 0) { lg(Info, "time out ..."); }else{ lg(Error,"epoll wait error"); } } } ~EpollServer(){ _listsocket_ptr->Close(); }private: std::shared_ptr<Sock> _listsocket_ptr; std::shared_ptr<Epoller> _epoller_ptr; uint16_t _port;};

Main.cc

#include <iostream>#include <memory>#include "EpollServer.hpp"#include "Epoller.hpp"int main(){ std::unique_ptr<EpollServer> epoll_svr(new EpollServer(8888)); epoll_svr->Init(); epoll_svr->Start(); Epoller ep; return 0; }

nocopy

#pragma onceclass nocopy{ public: nocopy(){ } nocopy(const nocopy &) = delete; const nocopy&operator=(const nocopy &) = delete;};

Socket.hpp

#pragma once#include <iostream>#include <string>#include <unistd.h>#include <cstring>#include <sys/types.h>#include <sys/stat.h>#include <sys/socket.h>#include <arpa/inet.h>#include <netinet/in.h>#include "Log.hpp"enum{ SocketErr = 2, BindErr, ListenErr,};// TODOconst int backlog = 10;class Sock{ public: Sock() { } ~Sock() { }public: void Socket() { sockfd_ = socket(AF_INET, SOCK_STREAM, 0); if (sockfd_ < 0) { lg(Fatal, "socker error, %s: %d", strerror(errno), errno); exit(SocketErr); } int opt = 1; setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)); } void Bind(uint16_t port) { 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) { lg(Fatal, "bind error, %s: %d", strerror(errno), errno); exit(BindErr); } } void Listen() { if (listen(sockfd_, backlog) < 0) { lg(Fatal, "listen error, %s: %d", strerror(errno), errno); exit(ListenErr); } } int Accept(std::string *clientip, uint16_t *clientport) { struct sockaddr_in peer; socklen_t len = sizeof(peer); int newfd = accept(sockfd_, (struct sockaddr*)&peer, &len); if(newfd < 0) { lg(Warning, "accept error, %s: %d", strerror(errno), errno); return -1; } char ipstr[64]; inet_ntop(AF_INET, &peer.sin_addr, ipstr, sizeof(ipstr)); *clientip = ipstr; *clientport = ntohs(peer.sin_port); return newfd; } bool Connect(const std::string &ip, const uint16_t &port) { struct sockaddr_in peer; memset(&peer, 0, sizeof(peer)); peer.sin_family = AF_INET; peer.sin_port = htons(port); inet_pton(AF_INET, ip.c_str(), &(peer.sin_addr)); int n = connect(sockfd_, (struct sockaddr*)&peer, sizeof(peer)); if(n == -1) { std::cerr << "connect to " << ip << ":" << port << " error" << std::endl; return false; } return true; } void Close() { close(sockfd_); } int Fd() { return sockfd_; }private: int sockfd_;};

Log.hpp

#pragma once#include <iostream>#include <time.h>#include <stdarg.h>#include <sys/types.h>#include <sys/stat.h>#include <fcntl.h>#include <unistd.h>#include <stdlib.h>#define SIZE 1024#define Info 0#define Debug 1#define Warning 2#define Error 3#define Fatal 4#define Screen 1#define Onefile 2#define Classfile 3#define LogFile "log.txt"class Log{ public: Log() { printMethod = Screen; path = "./log/"; } void Enable(int method) { printMethod = method; } std::string levelToString(int level) { switch (level) { case Info: return "Info"; case Debug: return "Debug"; case Warning: return "Warning"; case Error: return "Error"; case Fatal: return "Fatal"; default: return "None"; } } // void logmessage(int level, const char *format, ...) // { // time_t t = time(nullptr); // struct tm *ctime = localtime(&t); // char leftbuffer[SIZE]; // snprintf(leftbuffer, sizeof(leftbuffer), "[%s][%d-%d-%d %d:%d:%d]", levelToString(level).c_str(), // ctime->tm_year + 1900, ctime->tm_mon + 1, ctime->tm_mday, // ctime->tm_hour, ctime->tm_min, ctime->tm_sec); // // va_list s; // // va_start(s, format); // char rightbuffer[SIZE]; // vsnprintf(rightbuffer, sizeof(rightbuffer), format, s); // // va_end(s); // // 格式：默认部分+自定义部分 // char logtxt[SIZE * 2]; // snprintf(logtxt, sizeof(logtxt), "%s %s\n", leftbuffer, rightbuffer); // // printf("%s", logtxt); // 暂时打印 // printLog(level, logtxt); // } void printLog(int level, const std::string &logtxt) { switch (printMethod) { case Screen: std::cout << logtxt << std::endl; break; case Onefile: printOneFile(LogFile, logtxt); break; case Classfile: printClassFile(level, logtxt); break; default: break; } } void printOneFile(const std::string &logname, const std::string &logtxt) { std::string _logname = path + logname; int fd = open(_logname.c_str(), O_WRONLY | O_CREAT | O_APPEND, 0666); // "log.txt" if (fd < 0) return; write(fd, logtxt.c_str(), logtxt.size()); close(fd); } void printClassFile(int level, const std::string &logtxt) { std::string filename = LogFile; filename += "."; filename += levelToString(level); // "log.txt.Debug/Warning/Fatal" printOneFile(filename, logtxt); } ~Log() { } void operator()(int level, const char *format, ...) { time_t t = time(nullptr); struct tm *ctime = localtime(&t); char leftbuffer[SIZE]; snprintf(leftbuffer, sizeof(leftbuffer), "[%s][%d-%d-%d %d:%d:%d]", levelToString(level).c_str(), ctime->tm_year + 1900, ctime->tm_mon + 1, ctime->tm_mday, ctime->tm_hour, ctime->tm_min, ctime->tm_sec); va_list s; va_start(s, format); char rightbuffer[SIZE]; vsnprintf(rightbuffer, sizeof(rightbuffer), format, s); va_end(s); // 格式：默认部分+自定义部分 char logtxt[SIZE * 2]; snprintf(logtxt, sizeof(logtxt), "%s %s", leftbuffer, rightbuffer); // printf("%s", logtxt); // 暂时打印 printLog(level, logtxt); }private: int printMethod; std::string path;};Log lg;

epoll工作方式（水平触发LT和边缘触发ET）

水平触发Level Triggered工作模式

epoll默认状态下就是LT工作模式.

LT模式：事件到来，但是上层不处理，高电平，一直有效

当epoll检测到socket上事件就绪的时候, 可以不立刻进行处理. 或者只处理一部分.如上面的例子, 由于只读了1K数据, 缓冲区中还剩1K数据, 在第二次调用 epoll_wait 时, epoll_wait 仍然会立刻返回并通知socket读事件就绪.直到缓冲区上所有的数据都被处理完, epoll_wait 才不会立刻返回.支持阻塞读写和非阻塞读写

边缘触发Edge Triggered工作模式

ET模式：从无到有，从有到无，变化的时候，才会通知我们一次

如果我们在第1步将socket添加到epoll描述符的时候使用了EPOLLET标志, epoll进入ET工作模式.

当epoll检测到socket上事件就绪时, 必须立刻处理.如上面的例子, 虽然只读了1K的数据, 缓冲区还剩1K的数据, 在第二次调用 epoll_wait 的时候, epoll_wait 不会再返回了.也就是说, ET模式下, 文件描述符上的事件就绪后, 只有一次处理机会.ET的性能比LT性能更高( epoll_wait 返回的次数少了很多). Nginx默认采用ET模式使用epoll.只支持非阻塞的读写

select和poll其实也是工作在LT模式下. epoll既可以支持LT, 也可以支持ET.

对比LT和ET

LT是 epoll 的默认行为. 使用 ET 能够减少 epoll 触发的次数. 但是代价就是强逼着程序员一次响应就绪过程中就把 所有的数据都处理完（循环读取，直到读取出错，如果底层没有数据了就会阻塞住（fd默认是阻塞的），所以ET模式下所有的fd必须是non_block的）（重点）.所以ET的通知效率更高

相当于一个文件描述符就绪之后, 不会反复被提示就绪, 看起来就比 LT 更高效一些. 但是在 LT 情况下如果也能做到 每次就绪的文件描述符都立刻处理, 不让这个就绪被重复提示的话, 其实性能也是一样的.

如何理解ET的IO效率更高：ET要求程序员将本轮数据全部取走，从而让TCP向对方通告一个更大的窗口，从概率上让对方一次给本机发送更多的数据

另一方面, ET 的代码复杂程度更高了.

【问题】：但是ET一定比LT效率更高吗？

LT也可以将所有的fd设置成non_block,然后进行循环读取，通知第一次的时候，就全部取走就和ET一样了。

此外LT和ET的区别也是：一次向就绪队列里添加，还是次次添加

理解ET模式和非阻塞文件描述符

使用 ET 模式的 epoll, 需要将文件描述设置为非阻塞. 这个不是接口上的要求, 而是 “工程实践” 上的要求.

假设这样的场景: 服务器接受到一个10k的请求, 会向客户端返回一个应答数据. 如果客户端收不到应答, 不会发送第 二个10k请求.

epoll的使用场景

epoll的高性能, 是有一定的特定场景的. 如果场景选择的不适宜, epoll的性能可能适得其反.

对于多连接, 且多连接中只有一部分连接比较活跃时, 比较适合使用epoll.

例如, 典型的一个需要处理上万个客户端的服务器, 例如各种互联网APP的入口服务器, 这样的服务器就很适合epoll. 如果只是系统内部, 服务器和服务器之间进行通信, 只有少数的几个连接, 这种情况下用epoll就并不合适. 具体要根据需求和场景特点来决定使用哪种IO模型.

原因：

复杂性和开销：epoll 的实现相对复杂，它需要管理内核中的事件列表。对于少量连接，这种复杂性和管理开销并不能带来显著的性能提升，反而可能因为不必要的管理导致性能下降。轮询效率：对于少数连接，简单的轮询（例如 poll 或 select）足以胜任。因为连接数很少，轮询的开销也很小，使用这些机制更直接和高效。资源消耗：epoll 适用于大量连接的场景，因为它能够有效地处理大量并发连接而不显著增加资源消耗。但是，对于少数连接来说，这种优势并不明显，使用 epoll 反而可能造成资源浪费。实现简便性：select 和 poll 的实现更为简单，代码易于维护和理解。在连接数较少时，选择这种更简单的实现方式可以降低代码复杂度，提高开发效率。

epoll中的惊群问题

惊群问题

其他资料

epoll详解

apache/nginx网络模型