【Linux高级IO】select、poll、epoll

【Linux高级IO】select、poll、epoll

toc

作者：爱写代码的刚子

时间：2024.6.5

前言：本篇博客将会介绍面试重点考察的select、poll、epoll

IO: input && Output

read && write

1. 应用层read&&write的时候，本质把数据从用户层写给OS —— 本质就是拷贝函数
2. 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 
#include 
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 
#include 
#include 
#include 
#include 
#include 
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

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 
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 
#include 
#include 
#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 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#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 
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 
#include 
#include 
#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 
int main()
{
// std::cout <<"fd_set bits num : " << sizeof(fd_set) * 8 << std::endl;
// std::unique_ptr 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 
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 
#include 
#include 
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 
#include 
#include "Socket.hpp"
#include "Epoller.hpp"
#include 
#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
",_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 
#include 
#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 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#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 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#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
", 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模型.

原因：

1. 复杂性和开销：epoll 的实现相对复杂，它需要管理内核中的事件列表。对于少量连接，这种复杂性和管理开销并不能带来显著的性能提升，反而可能因为不必要的管理导致性能下降。
2. 轮询效率：对于少数连接，简单的轮询（例如 poll 或 select）足以胜任。因为连接数很少，轮询的开销也很小，使用这些机制更直接和高效。
3. 资源消耗：epoll 适用于大量连接的场景，因为它能够有效地处理大量并发连接而不显著增加资源消耗。但是，对于少数连接来说，这种优势并不明显，使用 epoll 反而可能造成资源浪费。
4. 实现简便性：select 和 poll 的实现更为简单，代码易于维护和理解。在连接数较少时，选择这种更简单的实现方式可以降低代码复杂度，提高开发效率。

epoll中的惊群问题

惊群问题

其他资料

epoll详解

apache/nginx网络模型