【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,
};
// 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*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 once
class 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,
};
// 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.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网络模型