英国帮人做设计作业网站,请打开网站,帝国cms生成网站地图,建设银行银行号查询网站目录
一、五种IO模型
1、阻塞IO
2、非阻塞IO
3、信号驱动
4、IO多路转接
5、异步IO
6、总结
二、高级IO重要概念
1、同步通信与异步通信
2、阻塞 vs 非阻塞
三、非阻塞IO
1、fcntl
2、实现函数SetNoBlock
四、IO多路转接select
1、select
1.1、参数解释
1.2、…目录
一、五种IO模型
1、阻塞IO
2、非阻塞IO
3、信号驱动
4、IO多路转接
5、异步IO
6、总结
二、高级IO重要概念
1、同步通信与异步通信
2、阻塞 vs 非阻塞
三、非阻塞IO
1、fcntl
2、实现函数SetNoBlock
四、IO多路转接select
1、select
1.1、参数解释
1.2、参数timeout取值
1.3、select函数返回值
1.4、关于fd_set结构
1.5、常见的程序片段
2、selectserver
2.1、 select执行过程
2.2、socket就绪条件
2.2.1、读就绪
2.2.2、写就绪
2.3、实现代码
2.4、select的缺点
五、IO多路转接poll
1、poll函数接口
2、poll的特点
2.1、优点
2.2、缺点
3、pollserver
六、IO多路转接epoll
1、epoll
1.1、epoll_create
1.2、epoll_ctl
1.3、epoll_wait
2、epoll原理总结
3、epoll的优点
4、epollserver 基础版本
5、epoll的工作方式
5.1、水平触发Level Triggered 工作模式
5.2、边缘触发Edge Triggered工作模式
5.3、对比LT和ET
6、epollserver ET工作方式
7、Reactor IO 等待资源 拷贝资源。
等待资源等待要拷贝的数据、拷贝数据需要的空间。当资源就绪后该状态称为IO事件就绪。在 TCP协议中有一个状态标志位 PSH本质上就是把资源设置为就绪状态。
一、五种IO模型
1、阻塞IO
阻塞IO在内核将数据准备好之前系统调用会一直等待。所有的套接字默认都是阻塞方式。
阻塞IO是最常见的IO模型 2、非阻塞IO 非阻塞IO如果内核还未将数据准备好系统调用仍然会直接返回并且返回EWOULDBLOCK错误码。 非阻塞IO往往需要程序员循环的方式反复尝试读写文件描述符这个过程称为轮询。这对CPU来说是较大的浪费一般只有特定场景下才使用。 3、信号驱动
信号驱动IO内核将数据准备好的时候使用SIGIO信号通知应用程序进行IO操作。 4、IO多路转接 IO多路转接虽然从流程图上看起来和阻塞IO类似。实际上最核心在于IO多路转接能够同时等待多个文件描述符的就绪状态。 5、异步IO 异步IO由内核在数据拷贝完成时通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据)。 6、总结 任何IO过程中都包含两个步骤。第一是等待第二是拷贝。而且在实际的应用场景中等待消耗的时间往往都远远高于拷贝的时间。让IO更高效最核心的办法就是让等待的时间尽量少。
二、高级IO重要概念
1、同步通信与异步通信
同步和异步关注的是消息通信机制 所谓同步就是在发出一个调用时在没有得到结果之前该调用就不返回。但是一旦调用返回就得到返回值了。换句话说就是由调用者主动等待这个调用的结果。 异步则是相反 调用在发出之后这个调用就直接返回了所以没有返回结果。换句话说当一个异步过程调用发出后调用者不会立刻得到结果。而是在调用发出后 被调用者通过状态、通知来通知调用者或通过回调函数处理这个调用。
2、阻塞 vs 非阻塞
阻塞和非阻塞关注的是程序在等待调用结果消息返回值时的状态。 阻塞调用是指调用结果返回之前当前线程会被挂起。调用线程只有在得到结果之后才会返回。 非阻塞调用指在不能立刻得到结果之前该调用不会阻塞当前线程。
三、非阻塞IO
1、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)
我们此处只是用第三种功能获取/设置文件状态标记就可以将一个文件描述符设置为非阻塞。
2、实现函数SetNoBlock
基于fcntl我们实现一个SetNoBlock函数将文件描述符设置为非阻塞。
void SetNonBlock(int fd)
{int fl fcntl(fd, F_GETFD); //获取指定文件的状态标志位if(fl 0){std::cerr error string : strerror(error) error code: error std::endl;return;}fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}使用F_GETFL将当前的文件描述符的属性取出来(这是一个位图)。然后再使用F_SETFL将文件描述符设置回去。设置回去的同时加上一个O_NONBLOCK参数。 如果我们将文件设置为非阻塞状态那么一旦底层没有数据就绪就会以出错的形式返回但是不算真正的出错错误码会被设置为 11 代表临时资源没有就绪。
void SetNonBlock(int fd)
{int fl fcntl(fd, F_GETFD); //获取指定文件的状态标志位if(fl 0){std::cerr error string : strerror(errno) error code: errno std::endl;return;}fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}int main()
{char buffer[64];SetNonBlock(0);while(1){printf( );fflush(stdout);ssize_t n read(0, buffer, sizeof(buffer) - 1);if(n 0){buffer[n - 1] 0;std::cout echo# buffer std::endl;}else if(n 0){std::cout end file std::endl;break;}else{if(errno EAGAIN || errno EWOULDBLOCK){//因为底层数据没有准备好希望下次继续来检查std::cout data not really std::endl;sleep(1);continue;}else if(errno EINTR){//这次IO被信号中断了也需要重新读取continue;}else{//再下面才是真正的读取失败了std::cout read error error string : strerror(errno) error code: errno std::endl;break;}}}
} 四、IO多路转接select
1、select 系统提供select函数来实现多路复用输入/输出模型。
select系统调用是用来让我们的程序监视多个文件描述符的状态变化的;程序会停在select这里等待直到被监视的文件描述符有一个或多个发生了状态改变
int select(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, struct timeval *timeout); 1.1、参数解释
参数 nfds 是需要监视的最大的文件描述符值1。rdsetwrsetexset分别对应于需要检测的可读文件描述符的集合可写文件描述符的集合及异常文件描述符的集合。参数 timeout 为结构timeval用来设置select()的等待时间。
1.2、参数timeout取值 NULL阻塞等待。表示select没有timeout select将一直被阻塞直到某个文件描述符上发生了事件。{00}非阻塞等待。仅检测描述符集合的状态然后立即返回并不等待外部事件的发生。特定的时间值如果在指定的时间段里没有事件发生 select将超时返回。timeout是输入输出型参数被返回时值为剩余的时间。
1.3、select函数返回值
返回值n 0表示有 n 个文件描述符就绪。返回值n 0等待超时指定时间内没有文件描述符就绪。返回值n 0有等待失败的情况错误原因存于errno此时参数readfds writefdsexceptfds和timeout的值变成不可预测。 错误值可能为
EBADF 文件描述词为无效的或该文件已关闭EINTR 此调用被信号所中断EINVAL 参数n 为负值。ENOMEM 核心内存不足
1.4、关于fd_set结构 其实这个结构就是一个整数数组更严格的说是一个 位图。使用位图中对应的位来表示要监视的文件描述符。这个位图的上限在Linux系统中是 1024 。
提供了一组操作 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的全部位 select 函数中的 fd_set 类型的参数都是输入输出型参数。 用户首先通过 fd_set 类型参数输入告诉内核哪些 fd 需要内核关心。内核通过这些 fd_set 类型参数输出告诉用户哪些 fd 的对应事件已经就绪了。
1.5、常见的程序片段
fs_set readset
FD_SET(fd,readset);
select(fd1,readset,NULL,NULL,NULL);
if(FD_ISSET(fd,readset)){……}
2、selectserver
2.1、 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 fd1则set变为 0001,0011 执行select(6,set,nullptr,nullptr,nullptr)阻塞等待 若fd1fd2上都发生可读事件则select返回此时set变为 0000,0011。注意没有事件发生的fd5被清空。
2.2、socket就绪条件
2.2.1、读就绪 socket内核中接收缓冲区中的字节数大于等于低水位标记SO_RCVLOWAT。此时可以无阻塞的读该文件描述符并且返回值大于0。 socket TCP通信中对端关闭连接此时对该socket读则返回0。 监听的socket上有新的连接请求。 socket上有未处理的错误。
2.2.2、写就绪 socket内核中发送缓冲区中的可用字节数(发送缓冲区的空闲位置大小)大于等于低水位标记 SO_SNDLOWAT 此时可以无阻塞的写并且返回值大于0。 socket的写操作被关闭(close或者shutdown)。对一个写操作被关闭的socket进行写操作会触发SIGPIPE信号。 socket使用非阻塞connect连接成功或失败之后。 socket上有未读取的错误。
2.3、实现代码
select服务器在使用的时候需要程序员自己维护一个第三方数组对已经获取的sock做管理。 一是用于在select 返回后array作为源数据和fd_set进行FD_ISSET判断。 二是select返回后会把以前加入的但并无事件发生的fd清空则每次开始select前都要重新从array取得fd逐一加入(FD_ZERO最先)扫描array的同时取得fd最大值maxfd用于select的第一个参数。
typedef struct FdEvent
{int fd;uint8_t event;std::string clientip;uint16_t clientport;
} type_t;type_t _fdarray[N];
具体服务器代码如下
#pragma once#include iostream
#include string
#include sys/select.h
#include cstring#include Sock.hpp
#include log.hppconst static int gport 8888;#define READ_EVENT (0x1 0)
#define WRITE_EVENT (0x1 1)
#define EXCEPT_EVENT (0x1 2)typedef struct FdEvent
{int fd;uint8_t event;std::string clientip;uint16_t clientport;
} type_t;// typedef int type_t;
// static const int defaultfd -1;static const int defaultevent 0;class SelectServer
{static const int N (sizeof(fd_set) * 8);public:SelectServer(uint16_t port gport): _port(port){}void InitServer(){_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();for (int i 0; i N; i){_fdarray[i].fd defaultfd;_fdarray[i].event defaultevent;_fdarray[i].clientport 0;}}void Accepter(){std::cout 有一个新连接 std::endl;// 这里进行Accept就不会再被阻塞了std::string clientip;uint16_t clientport;int sock _listensock.Accept(clientip, clientport);if (sock 0)return;// 不能直接读因为不知道sock有没有数据就绪// 需要将sock交给select让select进行管理。logMessage(DEBUG, [%s:%d], sock: %d, clientip.c_str(), clientport, sock);// 只要把获得的sock添加到_fdarray数组里就可以了int pos 1;for (; pos N; pos){if (_fdarray[pos].fd defaultfd)break;}if (pos N){// 数组满了close(sock);logMessage(WARNING, sockfd array[] full);}else{_fdarray[pos].fd sock;_fdarray[pos].event READ_EVENT;_fdarray[pos].clientip clientip;_fdarray[pos].clientport clientport;}}void Recver(int index){// serverIO();int fd _fdarray[index].fd;char buffer[1024];ssize_t s recv(fd, buffer, sizeof(buffer) - 1, 0); // 这里读取不会被阻塞if (s 0){buffer[s - 1] 0;std::cout _fdarray[index].clientip : _fdarray[index].clientport : buffer std::endl;// 把数据发送回去也要被select管理std::string echo buffer;echo [select server echo];send(fd, echo.c_str(), echo.size(), 0);}else{if (s 0)logMessage(INFO, client quit, _fdarray[i] - defaultfd: %d-%d, fd, defaultfd);elselogMessage(WARNING, recv error, _fdarray[i] - defaultfd: %d-%d, fd, defaultfd);close(_fdarray[index].fd);_fdarray[index].fd defaultfd;_fdarray[index].event defaultevent;_fdarray[index].clientip.resize(0);_fdarray[index].clientport 0;}}void HandlerEvent(fd_set rfds, fd_set wfds){for (int i 0; i N; i){if (_fdarray[i].fd defaultfd)continue;if ((_fdarray[i].event READ_EVENT) (FD_ISSET(_fdarray[i].fd, rfds))){// 处理读取1.accept 2、recvif (_fdarray[i].fd _listensock.Fd()){Accepter();}else if (_fdarray[i].fd ! _listensock.Fd()) {Recver(i);}else{}}else if((_fdarray[i].event WRITE_EVENT) (FD_ISSET(_fdarray[i].fd, wfds))){}else{}}}void Start(){// 在网络中新连接到来被当作读事件就绪。//_listensock.Accept(); 不能直接进行accept因为如果没有链接到来程序会被阻塞住// 此时服务端只有一个文件描述符// 这种写法不正确因为直接把rfds写死了应该是动态变化的// struct timeval timeout {0, 0};// fd_set rfds;// FD_ZERO(rfds);// FD_SET(_listensock.Fd(), rfds);_fdarray[0].fd _listensock.Fd();_fdarray[0].event READ_EVENT;while (1){// 因为rfds是一个输入输出行参数注定了每次都要对rfds进行重置。// 重置就必须要知道历史上有哪些fd// 因为服务器在运行中sockfd的值一直在动态变化所以maxfd也在一直变化fd_set rfds;fd_set wfds;FD_ZERO(rfds);FD_ZERO(wfds);int maxfd _fdarray[0].fd;for (int i 0; i N; i){if (_fdarray[i].fd defaultfd)continue;// 合法fdif (_fdarray[i].event READ_EVENT)FD_SET(_fdarray[i].fd, rfds);if (_fdarray[i].event WRITE_EVENT)FD_SET(_fdarray[i].fd, wfds);if (maxfd _fdarray[i].fd)maxfd _fdarray[i].fd;}int n select(maxfd 1, rfds, wfds, nullptr, nullptr);switch (n){case 0:logMessage(DEBUG, time out, %d: %s, errno, strerror(errno));break;case -1:logMessage(WARNING, %d: %s, errno, strerror(errno));break;default:logMessage(DEBUG, 有一个就绪事件发生了 : %d, n);HandlerEvent(rfds, wfds);DebugPrint();break;}sleep(1);}}void DebugPrint(){std::cout fdarray[]: ;for (int i 0; i N; i){if (_fdarray[i].fd defaultfd)continue;std::cout _fdarray[i].fd ;}std::cout \n;}~SelectServer(){_listensock.Close();}private:uint16_t _port;Sock _listensock;type_t _fdarray[N];
};
2.4、select的缺点 每次调用select都需要手动设置fd集合从接口使用角度来说也非常不便。 每次调用select都需要把fd集合从用户态拷贝到内核态这个开销在fd很多时会很大。 同时每次调用select都需要在内核根据函数参数fd 1来遍历传递进来的所有fd这个开销在fd很多时也很大。 select支持的文件描述符数量太小。
五、IO多路转接poll
1、poll函数接口
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由于监听的文件描述符就绪而返回
2、poll的特点
2.1、优点
不同于 select 使用三个位图来表示三个 fdset 的方式 poll 使用一个 pollfd 的指针实现。 pollfd 结构包含了要监视的 event 和发生的 event 不再使用select “参数-值” 传递的方式。接口使用比select更方便。 poll并没有最大数量限制 (但是数量过大后性能也是会下降)。
2.2、缺点
poll中监听的文件描述符数目增多时 和select函数一样poll返回后需要轮询 pollfd 来获取就绪的描述符。 每次调用 poll 都需要把大量的 pollfd 结构从用户态拷贝到内核中。 同时连接的大量客户端在一时刻可能只有很少的处于就绪状态因此随着监视的描述符数量的增长其效率也会线性下降。
3、pollserver
实现代码
#pragma once#include iostream
#include string
#include sys/poll.h
#include cstring#include Sock.hpp
#include log.hppconst static int gport 8888;
static const int N 4096;
const static short defaultevent 0;typedef struct pollfd type_t;
// static const int defaultfd -1;class PollServer
{public:PollServer(uint16_t port gport): _port(port), _fdarray(nullptr){}void InitServer(){_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();_fdarray new type_t[N];for (int i 0; i N; i){_fdarray[i].fd defaultfd;_fdarray[i].events defaultevent;_fdarray[i].revents defaultevent;}}void Accepter(){std::cout 有一个新连接 std::endl;// 这里进行Accept就不会再被阻塞了std::string clientip;uint16_t clientport;int sock _listensock.Accept(clientip, clientport);if (sock 0)return;// 不能直接读因为不知道sock有没有数据就绪// 需要将sock交给select让select进行管理。logMessage(DEBUG, [%s:%d], sock: %d, clientip.c_str(), clientport, sock);// 只要把获得的sock添加到_fdarray数组里就可以了int pos 1;for (; pos N; pos){if (_fdarray[pos].fd defaultfd)break;}if (pos N){// 数组满了可以进行动态扩容close(sock); logMessage(WARNING, sockfd array[] full);}else{_fdarray[pos].fd sock;_fdarray[pos].events POLLIN; //POLLIN | POLLOUT_fdarray[pos].revents defaultevent;}}void HandlerEvent(){for (int i 0; i N; i){int fd _fdarray[i].fd;short revent _fdarray[i].revents;if (fd defaultfd)continue;if ((fd _listensock.Fd()) (revent POLLIN)){Accepter();}else if ((fd ! _listensock.Fd()) (revent POLLIN)){// serverIO();char buffer[1024];ssize_t s recv(fd, buffer, sizeof(buffer) - 1, 0); // 这里读取不会被阻塞if (s 0){buffer[s - 1] 0;std::cout client# buffer std::endl;// 把数据发送回去也要被select管理//向event里添加写事件_fdarray[i].events POLLIN | POLLOUT;std::string echo buffer;echo [select server echo];send(fd, echo.c_str(), echo.size(), 0);}else{if (s 0)logMessage(INFO, client quit, _fdarray[i] - defaultfd: %d-%d, fd, defaultfd);elselogMessage(WARNING, recv error, _fdarray[i] - defaultfd: %d-%d, fd, defaultfd);close(fd);_fdarray[i].fd defaultfd;_fdarray[i].events defaultevent;_fdarray[i].revents defaultevent;}}}}void Start(){// 在网络中新连接到来被当作读事件就绪。//_listensock.Accept(); 不能直接进行accept因为如果没有链接到来程序会被阻塞住// 此时服务端只有一个文件描述符// 这种写法不正确因为直接把rfds写死了应该是动态变化的// struct timeval timeout {0, 0};// fd_set rfds;// FD_ZERO(rfds);// FD_SET(_listensock.Fd(), rfds);_fdarray[0].fd _listensock.Fd();_fdarray[0].events POLLIN;while (1){// 因为rfds是一个输入输出行参数注定了每次都要对rfds进行重置。// 重置就必须要知道历史上有哪些fd// 因为服务器在运行中sockfd的值一直在动态变化所以maxfd也在一直变化int timeout -1; //设为-1表示阻塞式调用int n poll(_fdarray, N, timeout); //可以对_fdarrat内容进行管理合法的fd、event全部放入_fdarray的最左侧。switch (n){case 0:logMessage(DEBUG, time out, %d : %s, errno, strerror(errno));break;case -1:logMessage(WARNING, %d: %s, errno, strerror(errno));break;default:logMessage(DEBUG, 有一个就绪事件发生了 : %d, n);HandlerEvent();DebugPrint();break;}// sleep(1);}}void DebugPrint(){std::cout fdarray[]: ;for (int i 0; i N; i){if (_fdarray[i].fd defaultfd)continue;std::cout _fdarray[i].fd ;}std::cout \n;}~PollServer(){_listensock.Close();if(_fdarray)delete[] _fdarray;}private:uint16_t _port;Sock _listensock;type_t *_fdarray;
};
六、IO多路转接epoll
1、epoll epoll 是为处理大批量句柄而作了改进的 poll 。 它是在2.5.44内核中被引进的(epoll(4) is a new API introduced in Linux kernel 2.5.44)。它几乎具备了之前所说的一切优点被公认为Linux2.6下性能最好的多路I/O就绪通知方法。
epoll 有3个相关的系统调用
1.1、epoll_create
int epoll_create(int size); 创建一个epoll的句柄
自从linux2.6.8之后 size参数是被忽略的。用完之后必须调用close()关闭。 调用 epll_create 函数后会在OS内核中创建一个红黑树。红黑树节点管理的数据类型是 struct epoll_event 里面包含用户让OS管理的文件描述符以及对应的事件。 创建红黑树的同时也会创建一个就绪队列。 函数调用成功后返回文件描述符该文件中保存一个eventpoll结构体结构体中就存放红黑树的根节点与就绪队列的头节点。
1.2、epoll_ctl
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); epoll_ctl 函数的本质是通过epoll模型对红黑树做操作向红黑树中新增、删除或修改某一个节点。即对红黑树进行增删改。 epoll_ctl函数除了对红黑树进行增删改外还要给节点结构体里文件描述符所指向的file结构体注册回调机制。 file结构体中本来就有一个回调函数指针并在拷贝操作完成后自动调用。只不过一般这个指针都被设置为NULL现在给改结构体注册一个回调函数并把函数指针指向这个函数。 回调函数的功能在文件中数据拷贝完成之后把本文件描述符对应的红黑树节点添加到就绪队列中。 epoll的事件注册函数 它不同于select()是在监听事件时告诉内核要监听什么类型的事件而是在这里先注册要监听的事件类型。 第一个参数是 epoll_create() 的返回值(epoll的句柄)。 第二个参数表示动作用三个宏来表示。 第三个参数是需要监听的fd。 第四个参数是告诉内核需要监听什么事。
第二个参数的取值
EPOLL_CTL_ADD 注册新的fd到epfd中。EPOLL_CTL_MOD 修改已经注册的fd的监听事件。EPOLL_CTL_DEL 从epfd中删除一个fd。
struct epoll_event 结构如下 1.3、epoll_wait
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout); OS 会把已经就绪的文件描述符对应的红黑树节点中的epoll_event结构体也添加到到就绪队列里此时这个节点及隶属于红黑树也隶属于队列。 epoll_wait 负责以时间复杂度为O1的方式检测有没有事件就绪即检测队列是否为空。 epoll_wait收集在epoll监控的事件中已经发送的事件
参数events是分配好的epoll_event结构体数组。epoll将会把发生的事件从就绪队列里拷贝到events数组中 (events不可以是空指针内核只负责把数据就绪队列里拷贝到这个events数组中不会去帮助我们在用户态中分配内存)。maxevents告知内核这个events有多大这个 maxevents的值不能大于创建epoll_create()时的size。参数timeout是超时时间 (毫秒 0会立即返回 -1是永久阻塞)。如果函数调用成功返回对应I/O上已准备好的文件描述符数目如返回0表示已超时, 返回小于0表示函数失败。
2、epoll原理总结 当某一进程调用epoll_create方法时 Linux内核会创建一个eventpoll结构体这个结构体中有两个成员与epoll的使用方式密切相关。
truct 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_wait检查是否有事件发生时只需要检查eventpoll对象中的rdlist双链表中是否有epitem元素即可。 如果rdlist不为空则把发生的事件复制到用户态同时将事件数量返回给用户. 这个操作的时间复杂度是O(1)。
总结一下, epoll的使用过程就是三部曲
调用epoll_create创建一个epoll句柄;调用epoll_ctl, 将要监控的文件描述符进行注册;调用epoll_wait, 等待文件描述符就绪
3、epoll的优点 接口使用方便虽然拆分成了三个函数但是反而使用起来更方便高效。不需要每次循环都设置关注的文件描述符也做到了输入输出参数分离开。 数据拷贝轻量只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中这个操作并不频繁(而select/poll都是每次循环都要进行拷贝)。 事件回调机制避免使用遍历而是使用回调函数的方式将就绪的文件描述符结构加入到就绪队列中epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪。这个操作时间复杂度O(1)。即使文件描述符数目很多效率也不会受到影响。 没有数量限制文件描述符数目无上限。
4、epollserver 基础版本
实现代码
//Epoller.hpp
#pragma once#include iostream
#include string
#include sys/epoll.h
#include cstring
#include stdlib.h#include log.hpp
#include error.hppstatic const int defaultepfd -1;
static const int gsize 128;class Epoller
{
public:Epoller():_epfd(defaultepfd){}void Create(){_epfd epoll_create(gsize);if(_epfd 0){logMessage(FATAL, listen error, code: %d, errstring: %s, errno, strerror(errno));exit(EPOLL_CREATE_ERR);}}//向epoll中添加事件bool AddEvent(int fd, uint32_t events){struct epoll_event ev;ev.events events;ev.data.fd fd; //用户数据,epoll底层不对该数据做任何处理就是为了给未来返回准备的int n epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, ev);if(n 0){logMessage(FATAL, epoll_ctl error, code: %d, errstring: %s, errno, strerror(errno));return false;}return true;}bool DelEvent(int fd){//epoll在操作的时候fd必须得合法return epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr) 0;}int Wait(struct epoll_event* revs, int num, int timeout){return epoll_wait(_epfd, revs, num, timeout);}int Fd(){return _epfd;}void Close(){if(_epfd ! defaultepfd)close(_epfd);}~Epoller(){}
private:int _epfd;
};//Epollserver.hpp
#pragma once
#include iostream
#include string
#include assert.h
#include functional#include Sock.hpp
#include log.hpp
#include error.h
#include Epoller.hppconst static int gport 8888;using func_t std::functionstd::string(std::string);class EpollServer
{const static int gnum 64;public:EpollServer(func_t func, uint16_t port gport):_func(func) ,_port(port){}void InitServer(){_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();_epoller.Create();logMessage(DEBUG, initserver success);}void Start(){// 1. 将listensock添加到epoll中bool r _epoller.AddEvent(_listensock.Fd(), EPOLLIN);assert(r);(void)r;int timeout 1000;while (1){int n _epoller.Wait(_revs, gnum, timeout);switch (n){case 0:logMessage(DEBUG, timeout);break;case -1:logMessage(WARNING, epoll_wait filed);break;default:// n就是就绪事件的个数logMessage(DEBUG, 有%d个事件就绪了, n);HandlerEvents(n);break;}}}void HandlerEvents(int num){for (int i 0; i num; i){int fd _revs[i].data.fd;uint32_t events _revs[i].events;if (events EPOLLIN){// 读事件就绪if (fd _listensock.Fd()){// 1.新连接事件到来// logMessage(DEBUG, get a new line ... );std::string clientip;uint16_t clientport;int sock _listensock.Accept(clientip, clientport);if (sock 0)continue;logMessage(DEBUG, %s:%d 已经连接上了服务器, clientip.c_str(), clientport);// 1.1.此时不能直接recv/read因为需要使用多路转接bool r _epoller.AddEvent(sock, EPOLLIN);assert(r);(void)r;}else{// 2.读取事件char request[1024];ssize_t s recv(fd, request, sizeof(request) - 1, 0); // 这里读取不会被阻塞if (s 0){request[s - 1] 0; // \r\n的形式为结尾的request[s - 2] 0;std::string response _func(request);send(fd, response.c_str(), response.size(), 0);}else{if (s 0)logMessage(INFO, client quit... );elselogMessage(WARNING, recv error ...);//在处理异常的时候先移除再关闭_epoller.DelEvent(fd);close(fd);}}}}}~EpollServer(){_listensock.Close();_epoller.Close();}private:uint16_t _port;Sock _listensock;Epoller _epoller;struct epoll_event _revs[gnum];func_t _func;
}; 因为TCP协议中数据是以字节流的方式发送读取的完整报文由应用层协议规定。所以我们直接读取是没有办法保证读取到完整的报文的。 为了解决这个问题我们就需要自定义应用层协议并通过回调函数来处理数据得到完整报文。
5、epoll的工作方式
5.1、水平触发Level Triggered 工作模式 在我们使用 select、poll、epoll的时候在最基本的情况下一旦有事件就绪如果上层不取 底层就会一直通知用户事件已经就绪。这种工作方式为LT水平触发工作模式。
epoll默认状态下就是LT工作模式。 当epoll检测到socket上事件就绪的时候可以不立刻进行处理或者只处理一部分。 如果缓冲区中还有数据在第二次调用 epoll_wait 时epoll_wait仍然会立刻返回并通知socket读事件就绪。 直到缓冲区上所有的数据都被处理完epoll_wait 才不会立刻返回。 支持阻塞读写和非阻塞读写。
5.2、边缘触发Edge Triggered工作模式
如果我们在第1步将socket添加到epoll描述符的时候使用了EPOLLET标志epoll进入ET工作模式。 当epoll检测到socket上事件就绪时必须立刻处理。 如果缓冲区中还有数据在第二次调用 epoll_wait 的时候epoll_wait 不会再返回了。 也就是说ET模式下文件描述符上的事件就绪后只有一次处理机会。 ET的性能比LT性能更高( epoll_wait 返回的次数少了很多)。Nginx默认采用ET模式使用epoll。 只支持非阻塞的读写因为要强逼着程序员必须要将本轮数据全部读取完毕。因为缓冲区大小有限为了保证把数据读完采用的策略是进行循环读取直到某一次读取到的数据量少于预期值就说明数据已经读完没有剩余数据了。这时如果采用的是阻塞读取如果全部数据的数据量刚刚好是缓冲区大小的整数倍那么最后一次读完之后因为全部数据已经读完但是循环读取却没有遇到过读取数据量少于预期的情况还会继续读取又因为没有数据了从而陷入阻塞状态。
5.3、对比LT和ET LT是 epoll 的默认行为。使用 ET 能够减少 epoll 触发的次数。但是代价就是强逼着程序猿一次响应就绪过程中就把所有的数据都处理完。 相当于一个文件描述符就绪之后不会反复被提示就绪看起来就比 LT 更高效一些。但是在 LT 情况下如果也能做到每次就绪的文件描述符都立刻处理不让这个就绪被重复提示的话其实性能也是一样的。 另一方面ET 的代码复杂程度更高了。 ET强逼程序员尽快取走所有的数据本质上是让TCP底层更新出更大的接收窗口从而在较大概率上提供对方的滑块窗口的大小提高发送效率。
6、epollserver ET工作方式
//Epoll.hpp#pragma once#include iostream
#include string
#include sys/epoll.h
#include cstring
#include stdlib.h#include log.hpp
#include error.hppstatic const int defaultepfd -1;
static const int gsize 128;class Epoller
{
public:Epoller():_epfd(defaultepfd){}void Create(){_epfd epoll_create(gsize);if(_epfd 0){logMessage(FATAL, listen error, code: %d, errstring: %s, errno, strerror(errno));exit(EPOLL_CREATE_ERR);}}//合并添加和修改bool AddModEvent(int fd, uint32_t events, int op){struct epoll_event ev;ev.events events;ev.data.fd fd; //用户数据,epoll底层不对该数据做任何处理就是为了给未来返回准备的int n epoll_ctl(_epfd, op, fd, ev);if(n 0){logMessage(FATAL, epoll_ctl error, code: %d, errstring: %s, errno, strerror(errno));return false;}return true;}bool DelEvent(int fd){//epoll在操作的时候fd必须得合法return epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr) 0;}int Wait(struct epoll_event* revs, int num, int timeout){return epoll_wait(_epfd, revs, num, timeout);}int Fd(){return _epfd;}void Close(){if(_epfd ! defaultepfd)close(_epfd);}~Epoller(){}
private:int _epfd;
};//EpollServer.hpp
#pragma once
#include iostream
#include string
#include assert.h
#include functional
#include unordered_map#include Sock.hpp
#include log.hpp
#include error.h
#include Epoller.hpp
#include util.hpp
#include Protocol.hppusing namespace Protocol_ns;const static int gport 8888;
const static int bsize 1024;
class Connection;
class EpollServer;using func_t std::functionvoid(Connection *, const Request );
using callback_t std::functionvoid(Connection *);// 大号的结构体
class Connection
{
public:Connection(const int fd, const std::string clientip, const uint16_t clientport): _fd(fd), _clientip(clientip), _clientport(clientport){}void Register(callback_t recver, callback_t sender, callback_t excepter){_recver recver;_sender sender;_excepter excepter;}~Connection(){}public:// IO信息int _fd;std::string _inbuffer;std::string _outbuffer;// IO处理函数callback_t _recver;callback_t _sender;callback_t _excepter;// 用户信息, only debugstd::string _clientip;uint16_t _clientport;// 也可以给conn带上自己要关心的事件uint32_t events;// 回指指针EpollServer *R;
};class EpollServer
{const static int gnum 64;public:EpollServer(func_t func, uint16_t port gport) : _func(func), _port(port){}void InitServer(){_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();_epoller.Create();AddConnection(_listensock.Fd(), EPOLLIN | EPOLLET);logMessage(DEBUG, init server success);}// 事件派发器int Dispatcher() // 名字要改{int timeout -1;while (true){LoopOnce(timeout);}}void LoopOnce(int timeout){int n _epoller.Wait(_revs, gnum, timeout);for (int i 0; i n; i){int fd _revs[i].data.fd;uint32_t events _revs[i].events;if ((events EPOLLERR) || (events EPOLLHUP))//_connections[fd]-_excepter(_connections[fd]);events | (EPOLLIN | EPOLLOUT);//这一步是将所有的异常情况最后都转为recv send的异常。if ((events EPOLLIN) ConnIsExists(fd)) //判断fd还存不存在因为有可能在处理异常的时候把fd关掉了下面再使用就会报错_connections[fd]-_recver(_connections[fd]);if ((events EPOLLOUT) ConnIsExists(fd))_connections[fd]-_sender(_connections[fd]);}}void AddConnection(int fd, uint32_t events, const std::string ip 127.0.0.1, uint16_t port gport){// 设置fd是非阻塞if (events EPOLLET)Util::SetNonBlock(fd);// 为listensock创建对应的connect对象Connection *conn new Connection(fd, ip, port);if (fd _listensock.Fd()){conn-Register(std::bind(EpollServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);}else{conn-Register(std::bind(EpollServer::Recver, this, std::placeholders::_1),std::bind(EpollServer::Sender, this, std::placeholders::_1),std::bind(EpollServer::Excepter, this, std::placeholders::_1));}// 把events赋值给conn对象以便在下面进行处理数据的时候判断是ET的还是LT的conn-events events;conn-R this;// 将listensockconnection对象添加到connections中_connections.insert(std::pairint, Connection*(fd, conn));// 添加事件bool r _epoller.AddModEvent(fd, events, EPOLL_CTL_ADD);assert(r);(void)r;logMessage(DEBUG, addConnection success, fd: %d, clientinfo: [%s:%d], fd, ip.c_str(), port);}bool EnableReadWrite(Connection* conn, bool readable, bool writeable){uint32_t events EPOLLET;events | (readable ? EPOLLIN : 0);events | (writeable ? EPOLLOUT : 0);conn-events events;return _epoller.AddModEvent(conn-_fd, conn-events, EPOLL_CTL_MOD);}// 连接管理器void Accepter(Connection *conn){// 1.新连接事件到来// logMessage(DEBUG, get a new line ... );// 可能有多个连接同时到来因此需要循环读取保证每一个连接都被读取到了do{int err 0; // 用作Accept的输出型参数如果监听失败了用于返回accept函数的错误码std::string clientip;uint16_t clientport;int sock _listensock.Accept(clientip, clientport, err);if (sock 0){logMessage(DEBUG, %s:%d 已经连接上了服务器, clientip.c_str(), clientport);AddConnection(sock, EPOLLIN | EPOLLET, clientip, clientport);}else{if(err EAGAIN || err EWOULDBLOCK) //说明读完了新数据没来break;//下面的都没读完else if(err EINTR)continue;else{logMessage(WARNING, errstring: %s, errcode: %d, strerror(err), err);continue;}}} while (conn-events EPOLLET);}void Recver(Connection *conn){// 读取完毕本轮数据do{char buffer[bsize];ssize_t n recv(conn-_fd, buffer, sizeof(buffer) - 1, 0);if (n 0){buffer[n] 0;conn-_inbuffer buffer;// 根据基本协议进行数据分析 -- 自己定过一个std::string requestStr;int n Protocol_ns::ParsePackage(conn-_inbuffer, requestStr);if(n 0) {requestStr RemoveHeader(requestStr, n);Request req;req.Deserialize(requestStr);_func(conn, req); // request 保证是一个完整的请求报文}// logMessage(Debug, inbuffer: %s, [%d], conn-inbuffer_.c_str(), conn-fd_);}else if (n 0){conn-_excepter(conn);}else{if (errno EAGAIN || errno EWOULDBLOCK)break;else if (errno EINTR)continue;elseconn-_excepter(conn);}} while (conn-events EPOLLET);}void Sender(Connection *conn){do{ssize_t n send(conn-_fd, conn-_outbuffer.c_str(), conn-_outbuffer.size(), 0);if(n 0) //发送成功n是这次发送发送了多少数据{conn-_outbuffer.erase(0, n); //移除已经发送了的数据。std::cout you can see me std::endl;if(conn-_outbuffer.empty()) //如果数据已经发送完了就把写关心去掉{EnableReadWrite(conn, true, false);break;}else{EnableReadWrite(conn, true, true);}}else{if(errno EAGAIN || errno EWOULDBLOCK)break;else if(errno EINTR)continue;else{conn-_excepter(conn);break;}}}while(conn-events EPOLLET);}void Excepter(Connection *conn){logMessage(DEBUG, Excepter..., fd: %d, clientinfo: [%s:%d], conn-_fd, conn-_clientip.c_str(), conn-_clientport);}bool ConnIsExists(int fd){return _connections.find(fd) ! _connections.end();}~EpollServer(){_listensock.Close();_epoller.Close();}private:uint16_t _port;Sock _listensock;Epoller _epoller;struct epoll_event _revs[gnum];func_t _func;std::unordered_mapint, Connection * _connections;
};//main.cc
#include EpollServer.hpp
#include memoryResponse calculaterHelper(const Request req)
{Response resp(0, 0);switch(req._op){case :resp._result req._x req._y; break;case -: resp._result req._x - req._y; break;case *:resp._result req._x * req._y; break;case /:if(req._y 0)resp._code 1;elseresp._result req._x / req._y; break;case %:if(req._y 0)resp._code 2;elseresp._result req._x % req._y; break;default:resp._code 3;break;}return resp;}void Calculate(Connection* conn, const Request req)
{Response resp calculaterHelper(req);std::string sendStr;resp.Serialize(sendStr);sendStr Protocol_ns::AddHeader(sendStr);//发送//在epoll中关于fd的读取一般要常设置一直要让epoll关心//对于fd的写入一般是按需设置不能常设值只有需要发的时候才设置//V1conn-_outbuffer sendStr;//开启对写事件的关心conn-R-EnableReadWrite(conn, true, true); //一般初次设置对写事件的关心对应的fd会立刻触发一次就绪因为发送buffer一定是空的
}int main()
{ std::unique_ptrEpollServer svr(new EpollServer(Calculate));svr-InitServer();svr-Dispatcher();return 0;
}
7、Reactor
Reactor是基于多路转接包含事件派发器、连接管理器等的半同步半异步的IO服务器。 半同步半异步体现在Reactor可以只负责事件派发这是同步的而数据读写、业务处理交给上层的线程池来处理这是异步的。
由epoll进行驱动并进行事件派发这种服务器就是Reactor服务器。 Reactor模式翻译过来是反应堆模式他就像一个反应堆连接接了很多connection对象由epoll进行管理如果那个connction对象就绪了epoll就会通知上层进行处理。
几个重要的注意事项 在多路转接中对于任何文件描述符读要进行常设值写要进行按需设置。 在进行写入时把写使能一次就对应了写入一次。 在多路转接中读需要交给epoll等待而写可以直接写。这是因为读缓冲区默认为空即事件不满足因此需要等待。而写事件不满足是因为写缓冲区满了我们第一次发送的时候写缓冲区应该是空的所以可以直接写不用在epoll里等待如果一次没写完再等。 关于连接管理在客户端进行服务器连接时服务器要把所有的客户端连接都管理起来这无疑占用了服务器很多资源。由于可能由很多客户端仅仅与服务器建立了连接却没有与服务器进行信息交互因此服务器为了节省资源需要定期清理这些超时的客户端。 为了实现这个功能在连接类中增加了一个时间变量lasttime用于记录最后一次读取的时间每次进行读取时更新这个时间。在事件派发器中进行检测工作判断客户端最后一次访问的时间并进行与超时时间对比。 一个服务器中多数客户端剩余的连接时间都是不同的所以把epoll的等待时间timeout设置为所有客户端中剩余时间最短的那一个就好了。这个逻辑可使用最小堆来实现。
Reactor服务器
//Epoll.hpp
#pragma once#include iostream
#include string
#include sys/epoll.h
#include cstring
#include stdlib.h#include log.hpp
#include error.hppstatic const int defaultepfd -1;
static const int gsize 128;class Epoller
{
public:Epoller():_epfd(defaultepfd){}void Create(){_epfd epoll_create(gsize);if(_epfd 0){logMessage(FATAL, listen error, code: %d, errstring: %s, errno, strerror(errno));exit(EPOLL_CREATE_ERR);}}//合并添加和修改bool AddModEvent(int fd, uint32_t events, int op){struct epoll_event ev;ev.events events;ev.data.fd fd; //用户数据,epoll底层不对该数据做任何处理就是为了给未来返回准备的int n epoll_ctl(_epfd, op, fd, ev);if(n 0){logMessage(FATAL, epoll_ctl error, code: %d, errstring: %s, errno, strerror(errno));return false;}return true;}bool DelEvent(int fd){//epoll在操作的时候fd必须得合法return epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr) 0;}int Wait(struct epoll_event* revs, int num, int timeout){return epoll_wait(_epfd, revs, num, timeout);}int Fd(){return _epfd;}void Close(){if(_epfd ! defaultepfd)close(_epfd);}~Epoller(){}
private:int _epfd;
};//EpollServer.hpp
#pragma once
#include iostream
#include string
#include assert.h
#include functional
#include unordered_map
#include ctime#include Sock.hpp
#include log.hpp
#include error.h
#include Epoller.hpp
#include util.hpp
#include Protocol.hppusing namespace Protocol_ns;const static int gport 8888;
const static int bsize 1024;
const static int linkTimeOut 30; //连接保持时间class Connection;
class EpollServer;using func_t std::functionconst Response(const Request );
using callback_t std::functionvoid(Connection *);// 大号的结构体
class Connection
{
public:Connection(const int fd, const std::string clientip, const uint16_t clientport): _fd(fd), _clientip(clientip), _clientport(clientport){}void Register(callback_t recver, callback_t sender, callback_t excepter){_recver recver;_sender sender;_excepter excepter;}~Connection(){}public:// IO信息int _fd;std::string _inbuffer;std::string _outbuffer;// IO处理函数callback_t _recver;callback_t _sender;callback_t _excepter;// 用户信息, only debugstd::string _clientip;uint16_t _clientport;// 也可以给conn带上自己要关心的事件uint32_t events;// 回指指针EpollServer *R;//时间戳time_t lasttime; //该connection最近一次就绪的时间
};class EpollServer
{const static int gnum 64;public:EpollServer(func_t func, uint16_t port gport) : _func(func), _port(port){}void InitServer(){_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();_epoller.Create();AddConnection(_listensock.Fd(), EPOLLIN | EPOLLET);logMessage(DEBUG, init server success);}// 事件派发器int Dispatcher() {int timeout 1000;while (true){LoopOnce(timeout);checkLink();}}void LoopOnce(int timeout){int n _epoller.Wait(_revs, gnum, timeout);for (int i 0; i n; i){int fd _revs[i].data.fd;uint32_t events _revs[i].events;if ((events EPOLLERR) || (events EPOLLHUP))//_connections[fd]-_excepter(_connections[fd]);events | (EPOLLIN | EPOLLOUT);// 这一步是将所有的异常情况最后都转为recv send的异常。if ((events EPOLLIN) ConnIsExists(fd)) // 判断fd还存不存在因为有可能在处理异常的时候把fd关掉了下面再使用就会报错_connections[fd]-_recver(_connections[fd]);if ((events EPOLLOUT) ConnIsExists(fd))_connections[fd]-_sender(_connections[fd]);}}void AddConnection(int fd, uint32_t events, const std::string ip 127.0.0.1, uint16_t port gport){// 设置fd是非阻塞if (events EPOLLET)Util::SetNonBlock(fd);// 为listensock创建对应的connect对象Connection *conn new Connection(fd, ip, port);if (fd _listensock.Fd()){conn-Register(std::bind(EpollServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);}else{conn-Register(std::bind(EpollServer::Recver, this, std::placeholders::_1),std::bind(EpollServer::Sender, this, std::placeholders::_1),std::bind(EpollServer::Excepter, this, std::placeholders::_1));}// 把events赋值给conn对象以便在下面进行处理数据的时候判断是ET的还是LT的conn-events events;conn-R this;conn-lasttime time(nullptr);// 将listensockconnection对象添加到connections中_connections.insert(std::pairint, Connection *(fd, conn));// 添加事件bool r _epoller.AddModEvent(fd, events, EPOLL_CTL_ADD);assert(r);(void)r;logMessage(DEBUG, addConnection success, fd: %d, clientinfo: [%s:%d], fd, ip.c_str(), port);}// 在多路转接中对于任何描述符读要常设值写要按需设置bool EnableReadWrite(Connection *conn, bool readable, bool writeable){uint32_t events EPOLLET;events | (readable ? EPOLLIN : 0);events | (writeable ? EPOLLOUT : 0);conn-events events;return _epoller.AddModEvent(conn-_fd, conn-events, EPOLL_CTL_MOD);}// 连接管理器void Accepter(Connection *conn){// 1.新连接事件到来// logMessage(DEBUG, get a new line ... );// 可能有多个连接同时到来因此需要循环读取保证每一个连接都被读取到了do{int err 0; // 用作Accept的输出型参数如果监听失败了用于返回accept函数的错误码std::string clientip;uint16_t clientport;int sock _listensock.Accept(clientip, clientport, err);if (sock 0){logMessage(DEBUG, %s:%d 已经连接上了服务器, clientip.c_str(), clientport);AddConnection(sock, EPOLLIN | EPOLLET, clientip, clientport);}else{if (err EAGAIN || err EWOULDBLOCK) // 说明读完了新数据没来break;// 下面的都没读完else if (err EINTR)continue;else{logMessage(WARNING, errstring: %s, errcode: %d, strerror(err), err);continue;}}} while (conn-events EPOLLET);}void HandlerRequest(Connection *conn){int quit false;while (!quit){std::string requestStr;// 1.提取完整报文int n Protocol_ns::ParsePackage(conn-_inbuffer, requestStr);if (n 0){// 2.提取有效载荷requestStr RemoveHeader(requestStr, n);// 3.进行反序列化Request req;req.Deserialize(requestStr);// 4.进行业务处理Response resp _func(req); // request 保证是一个完整的请求报文// 5.序列化std::string RespStr;resp.Serialize(RespStr);// 6.添加报头RespStr AddHeader(RespStr);// 7.进行返回conn-_outbuffer RespStr;}elsequit true;}}bool RecverHelper(Connection *conn){int ret true;conn-lasttime time(nullptr); //更新conn最近访问的时间// 读取完毕本轮数据do{char buffer[bsize];ssize_t n recv(conn-_fd, buffer, sizeof(buffer) - 1, 0);if (n 0){buffer[n] 0;conn-_inbuffer buffer;// 根据基本协议进行数据分析 -- 自己定过一个// 可以边读取边分析也可以读完之后一起分析。// logMessage(Debug, inbuffer: %s, [%d], conn-inbuffer_.c_str(), conn-fd_);}else if (n 0){conn-_excepter(conn);ret false;break;}else{if (errno EAGAIN || errno EWOULDBLOCK)break;else if (errno EINTR)continue;else{conn-_excepter(conn);ret false;break;}}} while (conn-events EPOLLET);return ret;}void Recver(Connection *conn){// 读取数据if(!RecverHelper(conn))return;// 分析数据HandlerRequest(conn);// 一般在进行写入时直接写入没写完才交给epollif (!conn-_outbuffer.empty()){conn-_sender(conn);}}void Sender(Connection *conn){bool safe true;do{ssize_t n send(conn-_fd, conn-_outbuffer.c_str(), conn-_outbuffer.size(), 0);if (n 0) // 发送成功n是这次发送发送了多少数据{conn-_outbuffer.erase(0, n); // 移除已经发送了的数据。// 如果发送缓冲区已经为空直接break;if (conn-_outbuffer.empty())break;}else{// 把对方接收缓冲区写满了退出if (errno EAGAIN || errno EWOULDBLOCK){break;}else if (errno EINTR)continue;else{safe false;conn-_excepter(conn);break;}}} while (conn-events EPOLLET);if (!safe)return;// 如果对方接收缓冲区已经被写满了满了写不下了并且发送缓冲区中还有数据if (!conn-_outbuffer.empty()){// 设置写关心交给epoll管理EnableReadWrite(conn, true, true);}else{EnableReadWrite(conn, true, false);}}//这个函数要防止重复调用因此在上面很多函数中都添加了bool值进行判断void Excepter(Connection *conn){// 1.先从epoll中移除fd_epoller.DelEvent(conn-_fd);// 2.移除unordered_map中的KV关系_connections.erase(conn-_fd);// 3.关闭fdclose(conn-_fd);// 4.将connction对象释放delete conn;logMessage(DEBUG, Excepter...done, fd: %d, clientinfo: [%s:%d], conn-_fd, conn-_clientip.c_str(), conn-_clientport);}bool ConnIsExists(int fd){return _connections.find(fd) ! _connections.end();}//检查是否超时断开连接void checkLink(){time_t curr time(nullptr);for(auto connection : _connections){if(connection.second-lasttime linkTimeOut curr) //如果最后一次访问的时间 超时时间 当前时间就要断开连接了。Excepter(connection.second);else continue;}}~EpollServer(){_listensock.Close();_epoller.Close();}private:uint16_t _port;Sock _listensock;Epoller _epoller;struct epoll_event _revs[gnum];func_t _func;std::unordered_mapint, Connection * _connections;
};//main.cc
#include EpollServer.hpp
#include memoryResponse Calculater(const Request req)
{Response resp(0, 0);switch(req._op){case :resp._result req._x req._y; break;case -: resp._result req._x - req._y; break;case *:resp._result req._x * req._y; break;case /:if(req._y 0)resp._code 1;elseresp._result req._x / req._y; break;case %:if(req._y 0)resp._code 2;elseresp._result req._x % req._y; break;default:resp._code 3;break;}return resp;}int main()
{ std::unique_ptrEpollServer svr(new EpollServer(Calculater));svr-InitServer();svr-Dispatcher();return 0;
}
上面的Reactor服务器也可以实现成多线程版本。 在设计多线程服务器时最需要防备的是多个线程同时对同一个文件描述符进行读写这样会使数据混乱。 因此在使用多线程进行Reactor设计时有一个原则一个fd以及connection一定只能有一个线程来进行管理。
