Linux-多路转接-epoll

接口认识

epoll_create

int epoll_create(int size);

🚀创建一个epoll模型，在Linux2.8.6之后size参数是被忽略的，通常传递一个大于0的整数。
🚀返回值是一个文件描述符，所以在使用完毕后要close掉，防止资源泄漏。

epoll_ctl

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

🚀第一个参数为创建的epoll模型对应的fd。
🚀第二个参数为在epoll模型中对让内核关心的文件描述符某些事件进行增删改。

EPOLL_CTL_ADD:在epoll模型中添加让内核关心对某个fd的某些事件。
EPOLL_CTL_MOD:修改epoll模型中某个fd所关心的事件。
EPOLL_CTL_MOD:删除epoll模型中某个fd，不用让内核关心此fd了。

🚀第三个参数为对哪个文件描述符fd进行操作。
🚀第四个参数是一个结构体类型，其内部结构如下：

epoll_event结构体中包含两个字段：uint32_t表示要让内核关心的哪些事件，epoll_data_t是一个联合体，通常设置fd字段即可，ptr指针可以用来存放用户自定义数据的地址，在epoll_wait返回时，会将我们当初设置的字段原样返回。
events字段：

epoll_wait

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

🚀第一个参数为创建的epoll模型对应的fd。
🚀第二个参数需要用户传递一个数组进去，返回时其内容会被填充，表示内核告诉用户哪些fd的哪些事件已经就绪了。
🚀第三个参数为用户传递数组的大小。
🚀第四个参数为等待时间：
大于0：表示经过timeout时间的等待没有事件就绪，就返回。
等于0：表示没有事件就绪就直接返回。返回值设置为0。
-1：表示阻塞等待模式

🚀返回值：大于0：表示用户传入的数组中多少个元素被设置。
等于0：表明timeout。
小于0：表明epoll_wait出错。

epoll工作原理

在内核中创建的数据结构

🚀在用户通过epoll_create创建一个epoll模型时，在内核中其实创建了以下的数据结构：

🚀一颗红黑树：

红黑树结点中存放的内容包括：fd，用户让内核关心这个fd的哪些事件，这个fd的哪些事件已经就绪了(revents)等。除此之外还会包括一些连接字段。红黑树是一颗近似平衡的搜索二叉树，所以用户通过epoll_ctl对某个fd进行插入，修改，删除，就是对红黑树结点做插入，修改，删除，效率是很高的。

🚀一个就绪队列：

就绪队列中的结点就表示，让内核关心的哪些fd上的哪些事件已经就绪了，用户调用epoll_wait时，内核就是将就绪队列中的信息拷贝给用户，另外epoll_wait可以以O（1）的事件复杂度来判断有没有事件就绪，因为如果就绪队列为空，表示没有事件就绪，如果事件不为空表示有事件就绪。

🚀struct file中的回调指针被设置

以读事件为例，内核怎么知道读事件就绪了呢？这我们就要弄清楚数据是从哪里来的，首先网络通信中接收到的数据一定是网卡先收到的，当网卡收到数据时可能会触发某种中断机制，通知操作系统，随后数据通过网络协议栈的层层解析最终被放到struct file中的缓冲区中，对于用户而言，数据被放到了struct file缓冲区中才代表实践就绪了，那么既然是操作系统将数据拷贝到的struct file缓冲区中的，在epoll的工作原理下，OS将数据拷贝到struct file缓冲区中时，会触发回调机制，将红黑树的结点链接到就绪队列中（内核中的数据结构的某个结点可以数据多个数据结构）。
写事件与读事件类似。

epoll模型的一个完整工作流程

epoll工作模式

LT-水平触发

🚀LT模式是epoll的默认工作方式，当内核缓冲区的数据越过设定的水位线后，epoll_wait就会返回，通知用户有事件已经就绪了，如果用户置之不理不去处理这个事件，那么epoll会一直通知用户这个事件已经就绪了。

ET-边缘触发

🚀ET工作模式下，epoll检测到某个套接字上的事件已经就绪了，但是用户并没有处理，那么下次调用epoll_wait的时候就不会返回，也就意味这只会通知一次，只有当套接字中有新的数据到来时才会再次通知用户一次。
🚀对于ET工作模式，假设某个客户端就给服务器发送了一次数据，那么对于服务器而言就必须要将数据全部读取上来，因为epoll不会通我第二次，并且如果采用阻塞式的读取，如果某次恰好将数据读取完，那么下次再读取时就会阻塞住，这时不合理的，所以ET模式一定要搭配非阻塞IO使用，ET这种工作模式倒逼用户一次就将数据读取完。

两种方式的对比

🚀由于ET模式倒逼用户一次将数据读取完，那么就能有效的减少epoll触发的次数，epoll可以利用重复通知某一进程的资源去通知其他进程，其IO效率是比LT要高的。但是，对于LT而言也可以采用非阻塞IO的方式一次将数据读取完，这种情况下LT和ET的IO效率是相等的。所以ET的IO效率是>=LT的iIO效率的。

🚀ET模式适合于高IO的场景。
🚀LT模式适合于高响应的模式，对于LT模式，可以每次读取一部分，对这一部分数据做处理尽快的响应客户端请求，所以LT适用于高响应的场景。

epoll的使用场景

🚀epoll适用于存在大量的客户端连接，但只有少部分的连接处于活跃状态的的场景下，例如各种互联网APP的入口服务器，就很适合使用epoll。但是，如果只有很少的连接，就不太适合使用epoll。

对于poll的改进

🚀在用户层面不再用维护pollfd这样类型的数组了，在epoll中由内核进行管理。
🚀不用将用户维护关于fd的某些事件的的信息，每次调用时都拷贝给内核。
🚀在epoll_wait返回时，不用对用户维护的数组做遍历来检测哪些fd上的哪些事件已经就绪，因为返回的都是已经就绪的事件。

惊群效应

什么是惊群效应

🚀在多线程或者多进程的场景下，多个线程或者多个进程共同监听同一个端口，当一个新的连接到来时，操作系统不知道通知哪个进程/线程去获取这个连接，所以就直接都通知，导致多个进程/线程只有一个能够成功的获取这个连接，而对于其他没获取连接成功的进程/线程，本有阻塞状态->运行状态，这种进程/线程的切换是浪费资源的，在一个多用户的服务上，如果存在这种惊群效应，那么对CPU是有一定消耗的。

如何解决惊群效应

🚀在Nginx中，是采用加锁的方式来处理惊群效应的，在连接到来时，多个进程谁先抢到锁，谁去获取这个连接。
🚀Nginx中锁的设计：在多个进程之间共享一块内存，在内存中设置一个变量-share_mtx,其值只有两种状态，0和1，当某个进程想去获取连接前，先去判断share_mtx，如果这个值为0，将其改为1，然后去获取连接。如果有的进程判断时，这个值本身就是1，那么去休眠一定时间，并将这个值改为0。这样设计肯定是存在漏洞的，势必会存在某几个进程同时判断share_mtx的值都是0，那么这几个进程都会去获取连接，所以对这个过程要做处理：1.可以采用CAS将这个判断+赋值的过程整合成一个原子操作2.使用mutex互斥锁3.使用spinlock自旋锁。

原子操作/mutex/spinlock如何选择

🚀临界区内代码量较小，且CPU体系结构的指令集中有指令能够支持，则选择原子操作。
🚀临界区内代码量很大，使用mutex互斥锁。
🚀介于两者之间的使用spinlock。

简单reactor模式的epoll服务器的编写

#ifndef __M_SERVER_H__
#define __M_SERVER_H__ 
#include "Sock.hpp"
#include "err.hpp"
#include "log.hpp"
#include <memory>
#include <iostream>
#include "epoll.hpp"
#include "Protocol.hpp"
#include "util.hpp"
#include <unordered_map>
#include <functional>
#include <cerrno>
const int N = 1024;
const int PORT = 8888;
class Connection;
using func_t = std::function<void (Connection*)>;
using Func_t = std::function<void (const Request&,Response&)>;
class Connection {private:int _fd;std::string _inbuffer;std::string _outbuffer;std::string _ip;uint16_t _port;uint32_t _events = 0;func_t _recver;func_t _sender;func_t _excepter;public:Connection(int fd,const std::string& ip,uint16_t port) :_fd(fd),_ip(ip),_port(port) {}int GetFd() {return _fd;}std::string Ip() {return _ip;}uint16_t Port() {return _port;}void SetEvents(uint32_t events) {_events = events;}void SetRecver(const func_t& recver) {_recver = recver;}void SetSender(const func_t& sender) {_sender = sender;}void SetExcepter(const func_t& excepter) {_excepter = excepter;}uint32_t GetEvents() {return _events;}func_t& GetRecver() {return _recver;}func_t& GetSender() {return _sender;}func_t& GetExcepter() {return _excepter;}string& GetInBuffer() {return _inbuffer;}string& GetOutBuffer() {return _outbuffer;}
};
class EpollServer {private:std::string _server_ip;uint16_t _server_port;sock::Sock _listensock;Epoll _epoll;epollfd _events[N];std::unordered_map<int,Connection*> _connections;Func_t _func;public:using ptr = std::unique_ptr<EpollServer>;EpollServer(const Func_t& func,uint16_t port = PORT,const std::string& ip = "0.0.0.0"):_server_ip(ip),_server_port(port),_func(func) {}~EpollServer() {_listensock.Close();_epoll.Close();}void AddEvents(int fd,uint32_t events,std::string ip = "127.0.0.1",uint16_t port = 0) {//1.将套接字添加到epoll模型中if(false == _epoll.AddEvent(fd,events)) {std::cerr << "将套接字设置进epoll模型失败\n";}//2.将套接字添加到_connections中Connection* conn = new Connection(fd,ip,port);conn->SetEvents(events);//注册recv方法if(fd == _listensock.Fd()) {conn->SetRecver(std::bind(&EpollServer::Acceptor,this,std::placeholders::_1));} else {conn->SetRecver(std::bind(&EpollServer::Reader,this,std::placeholders::_1));}//注册send方法conn->SetSender(std::bind(&EpollServer::Writer,this,std::placeholders::_1));//注册异常方法conn->SetExcepter(std::bind(&EpollServer::Excepter,this,std::placeholders::_1));_connections.insert({fd,conn});}void Init() {_listensock.Socket();_listensock.Bind(_server_port);_listensock.Listen();//创建epoll模型if(_epoll.Create() < 0) {std::cerr << "创建epoll模型失败\n";}//将listen套接字设置为非阻塞模式Util::SetNonBlock(_listensock.Fd());AddEvents(_listensock.Fd(),READ | ET | EXCEPT);}void Acceptor(Connection* conn) {do {//处理新获取的连接std::string ip;uint16_t port;int err = 0;int n = _listensock.Accept(&ip,&port,&err);if(n > 0) {//获取新的连接成功//将套接字设置设置为非阻塞Util::SetNonBlock(n);std::cout << "获取了一个新的连接: " << ip << ":" << port << std::endl; AddEvents(n,READ | ET | EXCEPT,ip,port);} else if(n < 0) {if((err & EAGAIN) || (err & EWOULDBLOCK)) {break;} else if(err & EINTR){continue;} else {std::cerr << "获取连接失败\n";continue;}} }while(conn->GetEvents() & ET);}void HandlerBusiness(Connection* conn) {bool stop = false;while(!stop) {//1.获取inbuffer中完整的报文std::string read_str;int n = ReadPackge(conn->GetInBuffer(),&read_str);if(n <= 0) {break;} //这一轮没有读取到完整的一个报文//2.去掉报头read_str = RemoveHeaders(read_str,n);//3.反序列化Request req;req.Deserialize(read_str);//4.业务处理Response res;_func(req,res);//5.序列化std::string out_str;res.Serialize(&out_str);//6.添加报头out_str = AddHeaders(out_str);conn->GetOutBuffer() += out_str;//7.给客户端响应conn->GetSender()(conn);}}bool ReaderHelp(Connection* conn) {bool safe = true;do {std::cout << "读取开始,fd: " << conn->GetFd() << std::endl;std::string ip = conn->Ip();uint16_t port = conn->Port();char buffer[4096] {0};ssize_t n = recv(conn->GetFd(),buffer,sizeof(buffer) - 1,0);if(n > 0) {buffer[n] = 0;conn->GetInBuffer() += buffer;std::cout << ip << ":" << port << "inbuffer>>>" << conn->GetInBuffer(); } else {//读取失败if(n < 0) {if((errno & EAGAIN) || (errno & EWOULDBLOCK)) {break;} else if(errno & EINTR) {continue;} else {//读出错，将inbuffer置空然后断开连接conn->GetInBuffer() = "";conn->GetExcepter()(conn);safe = false;break;}}else if(n == 0) { //客户端关闭std::cout << "客户端关闭连接: " << ip << ":" << port << std::endl;conn->GetExcepter()(conn);safe = false;break;}}}while(conn->GetEvents() & ET);return safe;}void Reader(Connection* conn) {bool res = ReaderHelp(conn);if(false == res) return;//完成读取后进行业务处理HandlerBusiness(conn);}void Writer(Connection* conn) {bool safe = true;do {ssize_t n = send(conn->GetFd(),conn->GetOutBuffer().c_str(),conn->GetOutBuffer().size(),0);if(n > 0) {conn->GetOutBuffer().erase(0,n);//发送完全部数据直接breakif(conn->GetOutBuffer().empty()) {break;}} else {if((errno & EAGAIN) || (errno & EWOULDBLOCK)) {break;} else if(errno & EINTR) {continue;} else {conn->GetExcepter()(conn);safe = false;}}}while(conn->GetEvents() & ET);if(false == safe) {return;}if(conn->GetOutBuffer().empty()) {//数据都发送完了,让epoll不再关心此次写事件_epoll.ModifyEvent(conn->GetFd(),READ | ET | EXCEPT); } else {_epoll.ModifyEvent(conn->GetFd(),WRITE | READ | ET | EXCEPT);//没有写完让将写事件添加到epoll中}}void Excepter(Connection* conn) {//从epoll模型中去除此fd_epoll.DelEvent(conn->GetFd());//从connections中去除此fd_connections.erase(conn->GetFd());//关闭fdclose(conn->GetFd());//delete connection对象delete conn;}bool IsConnExist(int fd) {return _connections.find(fd) != _connections.end();}void LoopOnce(int timeout) {epollfd fds[N];int n = _epoll.Wait(fds,N,timeout);for(int i = 0; i < n; i++) {if(fds[i].events & EXCEPT) {fds[i].events |= (READ | WRITE);}if((fds[i].events & READ) && IsConnExist(fds[i].fd)) {std::cout << "有一个读事件就绪\n";std::cout << "fd: " << fds[i].fd << std::endl;_connections[fds[i].fd]->GetRecver()(_connections[fds[i].fd]);} if((fds[i].events & WRITE) && IsConnExist(fds[i].fd))  {//写事件就绪_connections[fds[i].fd]->GetSender()(_connections[fds[i].fd]);    }}}void Start() {int timeout = -1;while(true) {LoopOnce(timeout);}}
};#endif