epoll
- 接口认识
- epoll_create
- epoll_ctl
- epoll_wait
- epoll工作原理
- 在内核中创建的数据结构
- epoll模型的一个完整工作流程
- epoll工作模式
- LT-水平触发
- ET-边缘触发
- 两种方式的对比
- epoll的使用场景
- 对于poll的改进
- 惊群效应
- 什么是惊群效应
- 如何解决惊群效应
- 原子操作/mutex/spinlock如何选择
- 简单reactor模式的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);
//发送完全部数据直接break
if(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());
//关闭fd
close(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