网站动图怎么做的,建立网站线上营销,wordpress开发cms,湖州专业做网站【Linux高级IO】select、poll、epoll
toc 作者#xff1a;爱写代码的刚子 时间#xff1a;2024.6.5 前言#xff1a;本篇博客将会介绍面试重点考察的select、poll、epoll IO: input Output
read write
应用层readwrite的时候#xff0c…【Linux高级IO】select、poll、epoll
toc 作者爱写代码的刚子 时间2024.6.5 前言本篇博客将会介绍面试重点考察的select、poll、epoll IO: input Output
read write
应用层readwrite的时候本质把数据从用户层写给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函数以及存储映射IOmmap这些统称为高级IO.
非阻塞IO
fcntl
一个文件描述符默认都是阻塞IO
#include unistd.h
#include fcntl.h
int fcntl(int fd, int cmd, ... /* arg */ );通过这个接口可以设置文件描述符属性为非阻塞
传入的cmd的值不同后面追加的参数也不相同。
fcntl函数有5种功能 复制一个现有的描述符(cmdF_DUPFD).获得/设置文件描述符标记(cmdF_GETFD或F_SETFD).获得/设置文件状态标记(cmdF_GETFL或F_SETFL).获得/设置异步I/O所有权(cmdF_GETOWN或F_SETOWN).获得/设置记录锁(cmdF_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 cstringusing 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就绪了 n0超时没有错误但是也没有fd就绪 n0等待出错了 获取时间的接口 其中struct timeval结构体 参数timeout取值 NULL:则表示select()没有timeoutselect将一直被阻塞直到某个文件描述符上发生了事件;0:仅检测描述符集合的状态然后立即返回并不等待外部事件的发生。特定的时间值:如果在指定的时间段里没有事件发生select将超时返回。 如果设置了timeout参数就是输入输出型参数
关于fd_set结构位图 比特位的位置表示文件描述符编号从右向左:0123…
比特位的内容0或者是1表示是否需要内核关心相应的事件。
参数作为返回值返回时比特位的内容0或者是1表示上面相应的事件是否就绪
fd_set是一张位图让用户-内核传递fd是否就绪的信息的
*fd_set readfdsfd_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此时参数readfdswritefds, exceptfds和timeout的 值变成不可预测。 错误值可能为: EBADF 文件描述词为无效的或该文件已关闭EINTR 此调用被信号所中断EINVAL 参数n 为负值。ENOMEM 核心内存不足 常见的程序片段如下:
fs_set readset;
FD_SET(fd,readset);
select(fd1,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。若fd5,执行FD_SET(fd,set); 后set变为0001,0000(第5位置为1)若再加入fd2fd1,则set变为0001,0011执行 select(6,set,0,0,0)阻塞等待若fd1,fd2上都发生可读事件则select返回此时set变为 0000,0011。注意:没有事件发生的fd5被清空。 编写select服务器
Main.cc
#include SelectServer.hpp
#include memoryint main()
{// std::cout fd_set bits num : sizeof(fd_set) * 8 std::endl;std::unique_ptrSelectServer 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.hppusing 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;elsebreak;}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){// demochar 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 0000int 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:// 有事件就绪了TODOcout get a new link!!!!! endl;Dispatcher(rfds); // 就绪的事件和fdbreak;}}}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.hppenum
{SocketErr 2,BindErr,ListenErr,
};// TODO
const 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*84096. 将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.hint 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.hppusing 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;elsebreak;}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){// demochar 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; // 3sfor (;;){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:// 有事件就绪了TODOcout 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 memoryint main()
{// std::cout fd_set bits num : sizeof(fd_set) * 8 std::endl;// std::unique_ptrSelectServer svr(new SelectServer());std::unique_ptrPollServer 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 cstringclass 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(operEPOLL_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_ADDstruct 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(_epfd0){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.hppuint32_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){// demochar 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 i0;inum;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_ptrSock _listsocket_ptr;std::shared_ptrEpoller _epoller_ptr;uint16_t _port;
};Main.cc
#include iostream
#include memory
#include EpollServer.hpp
#include Epoller.hpp
int main(){std::unique_ptrEpollServer 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 nocopyoperator(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.hppenum
{SocketErr 2,BindErr,ListenErr,
};// TODO
const 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.txtclass 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.txtif (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/FatalprintOneFile(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网络模型
