Linux高级IO

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

在平时 read/recv 时，如果内核缓冲区有数据，会拷贝到应用层缓冲区并返回，但如果没有数据，默认会阻塞，直到有数据时再读上来
write时也是一样，只不过内核缓冲区很少会被写满

因此，IO不仅涉及拷贝，更重要的是等待过程IO的本质为 等 + 拷贝，要想高效的IO，只要减少【等】的比重即可

五种IO模型

阻塞IO：默认的IO类型，也是最常见的IO模型

在内核将数据准备好之前，系统调用会一直等待，所有的套接字（文件描述符）默认都是阻塞方式

非阻塞IO：如果内核还未将数据准备好，系统调用仍然会直接返回，并且返回EWOULDBLOCK/EAGAIN错误码
当采用非阻塞IO，一般会用轮询的方式反复调用读取/写入接口

信号驱动IO：内核将数据准备好的时候，使用 SIGIO 信号通知应用程序进行IO操作

其余时间应用程序完全不管它

IO多路转接/复用：能够同时等待多个文件描述符的就绪状态，就像多进程/线程同时等待。但多进程/线程的成本太高，而多路转接成本很低

由于多路转接原理是可以同时等待多个文件描述符，就注定了之前的那些其他IO类接口不能直接使用，因此操作系统为了满足我们同时等待多个文件描述符的需求，必须单独设计其他的系统调用，例如select/poll/epoll

select接口只负责IO操作中【等】的部分，当等到数据后，会交给例如recvfrom进行数据拷贝工作

异步IO：由内核在数据拷贝完成时, 通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据)

进程发起读请求后立即返回，继续执行其他任务，内核自动完成数据拷贝，完成后通知进程，不参与进程参与数据拷贝过程

同步/异步通信？

虽然和线程/进程同步/异步一样都用的同步/异步这个词，但两者完全没关系。

线程/进程同步指的是协调任务执行顺序，例如加锁保护临界区，而线程/进程异步为多个任务并发执行

而判断是同步通信还是异步通信的核心依据是：数据从内核缓冲区复制到用户空间缓冲区时，应用程序的线程是否亲自参与了等待或执行。同步通信例如阻塞IO、非阻塞IO、多路转接IO、信号驱动IO，都参与了数据拷贝工作；异步IO当发送方发出请求/消息后，无需等待结果即可立即去处理其他事务，直到接收方数据拷贝工作完成后再通知发送方

非阻塞IO

要想在IO时为非阻塞，可以将例如 recv/recvfrom/send 接口的 flags 参数设为 MSG_DONTWAIT ，或在打开文件时就将open接口的flags参数加上 O_NONBLOCK 

ssize_t ret = recv(sockfd, buf, len, MSG_DONTWAIT);
int fd = open("/dev/mydevice", O_RDONLY | O_NONBLOCK);

这只能暂时设为非阻塞，要想将一个文件描述符永久设为非阻塞，可以用 fcntl 系统调用

int fcntl(int fd, int cmd, … /* arg */ );

fd为要操作的文件描述符，cmd为要执行的操作，后面为可变参数

当cmd为F_GETFL时，代表获取fd的访问模式与状态标志

当cmd为F_SETFL时，代表修改fd的状态标志位，若要设为非阻塞，就设为 O_NONBLOCK ，但这样会覆盖掉fd原来的状态标志，因此要先获取原状态标志，再按位或 O_NONBLOCK 代表加上该状态标志

bool SetNonBlock(int fd) //为fd文件描述符设置非阻塞
{
int fl = fcntl(fd, F_GETFL); // 获取原状态标志
if(fl < 0) // 获取失败
{
std::cerr << "fcntl(F_GETFL): " << strerror(errno) << std::endl;
return false;
}
// 给fd的状态标志加进非阻塞
if(fcntl(fd, F_SETFL, fl | O_NONBLOCK) < 0) // 若设置失败
{
std::cerr << "fcntl(F_SETFL): " << strerror(errno) << std::endl;
return false;
}
return true;
}

非阻塞IO一般和轮询一起使用，通过循环调用read/write等系统调用，不断询问内核数据是否就绪。

拿read举例，读取成功时返回读取的字节数（>0），若读取到文件结尾返回0，若读取失败返回-1，并且errno被设置
但是非阻塞情况下的read,如果没检测到数据，返回后，它的返回值也是-1，虽然不是错误，但以错误的形式被返回了，它的errno会被设置为11，代表资源暂时不可用，意思就是资源还没就绪

errno: 11,Resource temporarily unavailable

因此对于返回值为-1的情况需要先判断是否为非阻塞正常返回，虽然直接判断errno是否为11也可以，但Linux提供了对应的宏，可以不用硬编码：EAGAIN / EWOULDBLOCK，这两个宏的值在Linux中完全一致：

// <errno.h>
#define EAGAIN 11
#define EWOULDBLOCK EAGAIN // 直接别名

除此之外，EINTR错误代表被信号打断，并不清楚数据到底有没有，因此需要重来

完整代码：

#include "util.hpp"
#include <unistd.h>
using namespace std;

int main()
{
SetNonBlock(0);
char buffer[1024] = {0};
while (true)
{
printf(">>> ");
fflush(stdout);
ssize_t s = read(0, buffer, sizeof(buffer) – 1);
if (s > 0) // 读到数据
{
buffer[s – 1] = 0; // 把换行符去掉
printf("echo# %s\\n", buffer);
}
else if (s == 0) // 读到文件结尾,Linux下是Ctrl+D
{
printf("read end\\n");
break;
}
else if (s < 0) // 出错
{
if (errno == EWOULDBLOCK || errno == EINTR)//EWOULDBLOCK代表只是非阻塞的正常返回，EINTR代表被信号打断
{
sleep(1);
continue;
}
else // 真出错了
{
cerr << "errno: " << errno << "," << strerror(errno) << endl;
break;
}
}
sleep(1);
}
return 0;
}

也可以在轮询检测时干点别的事，例如使用回调函数

完整代码：

#include "util.hpp"
#include <unistd.h>
#include <vector>
#include <functional>
using namespace std;

//初始化
#define INIT(v) do{\\
v.push_back(printlog);\\
v.push_back(download);\\
v.push_back(executeSql);\\
}while(0)

//遍历任务
#define EXEC_OTHER(cbs) do{\\
for(auto cb : cbs)\\
cb();\\
}while(0)

int main()
{
vector<function<void ()>> cbs; // 任务集合
INIT(cbs);

SetNonBlock(0);
char buffer[1024] = {0};
while (true)
{
printf(">>> ");
fflush(stdout);
ssize_t s = read(0, buffer, sizeof(buffer) – 1);
if (s > 0) // 读到数据
{
buffer[s – 1] = 0; // 把换行符去掉
printf("echo# %s\\n", buffer);
}
else if (s == 0) // 读到文件结尾,Linux下是Ctrl+D
{
printf("read end\\n");
break;
}
else if (s < 0) // 出错
{
if (errno == EWOULDBLOCK)//EWOULDBLOCK/EAGAIN代表只是非阻塞的正常返回
{
EXEC_OTHER(cbs);
}
else if(errno == EINTR) // EINTR代表被信号打断，需要重来一次
{
continue;
}
else // 真出错了
{
cerr << "errno: " << errno << "," << strerror(errno) << endl;
break;
}
}
sleep(1);
}
return 0;
}

ps：printlog、download、executeSql为函数名

多路转接

select

select是多路转接的一种方式，它只负责等待，可以一次等待多个fd，select本身没有数据拷贝的能力，拷贝要例如read/write来完成

int select(int nfds, fd_set *_Nullable restrict readfds,
fd_set *_Nullable restrict writefds,
fd_set *_Nullable restrict exceptfds,
struct timeval *_Nullable restrict timeout);

nfds：select等待的多个文件描述符中最大的一个+1
告诉内核只需要在 0 到 nfds – 1 的范围内检查描述符的状态，避免内核进行不必要的全量检查，提升效率

timeout：输入输出型参数，输入时通过传入timeval结构体决定阻塞的时间，当超过该时间还没有就绪的文件描述符时就超时返回，返回值被置0，例如struct timeval timeout = {5, 0}代表5秒、timeout = {0, 0}代表非阻塞、timeout = nullptr代表阻塞式调用

struct timeval
{
time_t tv_sec; /* 秒 */
suseconds_t tv_usec; /* 微秒 */
};

当select返回时，timeout参数会返回剩余未等待的时间。如果输入时timeout = {5, 0}，等待了3秒后有文件描述符就绪返回，还剩余2秒，输出时timeout = {2, 0}

readfds：读位图，输入输出型参数，输入时通过传入fd_set结构体（位图）指针告知select需要关注可读事件的文件描述符集合，若不关心可传nullptr

当select返回时，readfds会返回读就绪的文件描述符集合（改变原位图）

writefds：写位图，输入输出型参数，输入时通过传入fd_set结构体指针告知select需要关注可写事件的文件描述符集合，若不关心可传nullptr
当select返回时，writefds会返回写就绪的文件描述符集合（改变原位图）

exceptfds：异常位图，输入输出参数，输入时通过传入fd_set结构体指针告知select需要关注带外数据（Out-of-band / OOB，即带外异常）的文件描述符集合，若不关心可传nullptr
当select返回时，exceptfds会返回异常就绪的文件描述符集合（改变原位图）

fd_set：本质上是位图，用bit位1代表被监视 

//内核版fd_set
typedef struct
{
/* XPG4.2 requires this member name. Otherwise avoid the name
from the global namespace. */
#ifdef __USE_XOPEN
__fd_mask fds_bits[__FD_SETSIZE / __NFDBITS];
# define __FDS_BITS(set) ((set)->fds_bits)
#else
__fd_mask __fds_bits[__FD_SETSIZE / __NFDBITS];
# define __FDS_BITS(set) ((set)->__fds_bits)
#endif
} fd_set;

// 简化版fd_set
typedef struct {
unsigned long fds_bits[1024 / (8 * sizeof(long))]; // 1024 位
} fd_set;

由于直接操作位图非常繁琐且不安全，系统提供了 4 个宏 来操作 fd_set 

FD_ZERO(fd_set *set) // 清空集合（将所有 bit 位置 0）
FD_SET(int fd, fd_set *set) // 将指定的 fd 加入集合（位置 1）
FD_CLR(int fd, fd_set *set) // 将指定的 fd 从集合中移除（位置 0）
FD_ISSET(int fd, fd_set *set) // 检查 fd 是否在集合中（常用于 select 返回后判断就绪状态）

当因为有就绪文件描述符而返回时，select的返回值为就绪文件描述符总数（readfds.size+writefds.size+exceptfds.size）；当调用失败时返回值被置-1，且errno被设置

select多路转接服务器

// socket.hpp

#pragma once
#include <iostream>
#include <string>

#include <cstring>
#include <cerrno>

#include <sys/socket.h>
#include <arpa/inet.h>

#include "log.hpp"

const static int backlog = 32;

class Socket
{
public:
static int create_socket() // 创建套接字
{

int sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == -1)
{
LogMessage(FATAL, (char *)"socket创建监听套接字失败, 错误码: %d, 错误描述：%s", errno, strerror(errno));
exit(SOCKET_ERR);
}
LogMessage(DEBUG, (char *)"socket创建监听套接字成功");

// 设置端口复用
int opt = 1;
setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));

return sockfd;
}

static void bind_socket(int sockfd, uint16_t port) // bind绑定自己的网络信息
{
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)
{
LogMessage(FATAL, (char *)"bind绑定失败, 错误码: %d, 错误描述：%s", errno, strerror(errno));
exit(BIND_ERR);
}
LogMessage(DEBUG, (char *)"bind绑定成功");
}

static void listen_socket(int sockfd) // 开启监听状态
{
if (listen(sockfd, backlog) != 0)
{
LogMessage(FATAL, (char *)"listen监听状态开启失败, 错误码: %d, 错误描述：%s", errno, strerror(errno));
exit(LISTEN_ERR);
}
LogMessage(DEBUG, (char *)"listen监听状态开启成功");
}

static int accpet_socket(int listenfd, std::string &clientIp, uint16_t &clientPort) // 当接收到新连接时，通过输出型参数返回客户端ip+port
{
struct sockaddr_in ClientAddr;
memset(&ClientAddr, 0, sizeof(ClientAddr));
socklen_t socklen = sizeof(ClientAddr);

int sockfd = accept(listenfd, (struct sockaddr *)&ClientAddr, &socklen);
if (sockfd == -1)
{
LogMessage(WARNING, (char *)"accept建立新连接失败, 错误码: %d, 错误描述：%s", errno, strerror(errno));
// exit(ACCEPT_ERR);
}
else
{
LogMessage(DEBUG, (char *)"accept建立新连接成功,sockfd = %d", sockfd);
// 将IP传给输出参数clientIp
char ip_str[16] = {0};
if (inet_ntop(AF_INET, &(ClientAddr.sin_addr.s_addr), ip_str, sizeof(ip_str)) == nullptr)
{
LogMessage(FATAL, (char *)"inet_ntop()失败，错误码：%d，错误信息：%s", errno, strerror(errno));
exit(INETPN_ERR);
}
else
{
clientIp = ip_str;
}
// 将port传给输出参数clientPort
clientPort = ntohs(ClientAddr.sin_port);
}
return sockfd;
}
};

// SelectServer.hpp:

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include "socket.hpp"

namespace select_ns
{
static const uint16_t defaultport = 8080; // 默认端口
static const int fdnum = sizeof(fd_set) * 8; // fd_set可以存储的文件描述符数量
static const int defaultfd = -1; // 非法文件描述符的标志

using func_t = std::function<std::string(const std::string &)>;

class SelectServer
{
public:
SelectServer(func_t func, int port = defaultport) : _func(func), _port(port), _listensockfd(-1)
{
}

void init()
{
_listensockfd = Socket::create_socket();
Socket::bind_socket(_listensockfd, _port);
Socket::listen_socket(_listensockfd);

// 初始化_fdarray
_fdarray = new int[fdnum];
for (int i = 0; i < fdnum; i++)
_fdarray[i] = defaultfd;
_fdarray[0] = _listensockfd; // 先把listensock放进去
}

void Accepter()
{
// accept接收新连接
std::string ClientIp;
uint16_t ClientPort;
int sockfd = Socket::accpet_socket(_listensockfd, ClientIp, ClientPort);
if (sockfd < 0)
return; // accept失败

// 更新读位图
int i;
for (i = 0; i < fdnum; i++)
{
if (_fdarray[i] != defaultfd)
continue; // 跳过合法fd
break;
}
if (i == fdnum) // 读位图已满
{
LogMessage(WARNING, (char *)"读位图被占满");
close(sockfd);
}
else // 将新获取的sockfd添加到文件描述符集合_fdarray中
{
_fdarray[i] = sockfd;
}
}

void Recver(int sockfd, int pos /*在_fdarray中的位置*/)
{
// 读取
char request[1024];
ssize_t s = recv(sockfd, request, sizeof(request) – 1, 0); // 不考虑读不完的情况
if (s > 0) // 读到数据
{
request[s] = 0;
std::cout << "client# " << request << std::endl;
}
else if (s == 0) // 文件被关闭
{
close(sockfd);
_fdarray[pos] = defaultfd;
LogMessage(DEBUG, (char *)"客户端退出");
}
else // recv报错
{
close(sockfd);
_fdarray[pos] = defaultfd;
LogMessage(ERROR, (char *)"客户端退出，错误码：%d, 错误描述：%s", errno, strerror(errno));
}

// 请求转响应
std::string response = _func(request);
// 写入
write(sockfd, response.c_str(), response.size());
}

void HandleEvent(fd_set &fds)
{
for (int i = 0; i < fdnum; i++)
{
if (_fdarray[i] == defaultfd)
continue; // 过滤非法fd

if ((_fdarray[i] == _listensockfd) && (FD_ISSET(_fdarray[i], &fds))) // accept就绪
Accepter();
else if (FD_ISSET(_fdarray[i], &fds)) // 普通fd读就绪
Recver(_fdarray[i], i);
}

// Print
std::cout << "_fdarray: ";
for (int i = 0; i < fdnum; i++)
{
if (_fdarray[i] == defaultfd)
continue;
std::cout << _fdarray[i] << " ";
}
fflush(stdout);
}

void start()
{

while (true)
{
// std::string ClientIp;
// uint16_t ClientPort;
// int sock = Socket::accpet_socket(_listensockfd, ClientIp, ClientPort);
// if(sock < 0) continue;

// 初始化rfds与maxfd
fd_set rfds;
FD_ZERO(&rfds); // 初始化
int maxfd = _fdarray[0];
for (int i = 0; i < fdnum; i++)
{
if (_fdarray[i] == defaultfd)
continue; // 非法fd跳过

FD_SET(_fdarray[i], &rfds); // 将合法fd加进读位图
if (maxfd < _fdarray[i])
maxfd = _fdarray[i]; // 更新maxfd
}

// int n = select(_listensockfd + 1, &rfds, nullptr, nullptr, &timeout);
int n = select(maxfd + 1, &rfds, nullptr, nullptr, nullptr);
switch (n)
{
case 0: // 超时返回
std::cout << "timeout…\\n";
break;
case -1: // 错误返回
LogMessage(WARNING, (char *)"select失败，错误码：%d，错误描述：%s", errno, strerror(errno));
break;
default: // 就绪返回
LogMessage(DEBUG, (char *)"读就绪");
HandleEvent(rfds);
break;
}
}
}

~SelectServer()
{
if (_listensockfd == -1)
close(_listensockfd);

if (_fdarray)
delete[] _fdarray;
}

private:
uint16_t _port; // 服务端口号
int _listensockfd; // 监听套接字
int *_fdarray; // 文件描述符集合
func_t _func; // 用于将请求转响应的函数
};

}

// SelectServer.cpp:

#include <iostream>
#include <memory>
#include <string>

#include "SelectServer.hpp"
#include "log.hpp"

using namespace std;
using namespace select_ns;

static void usage(string proc) // 使用手册
{
cout << RED << "\\nUsage：\\n\\t" << proc << " " << "port\\n\\n " << ED;
}

string translation(const string &request) // [TODO] 请求转响应
{
return request;
}

int main(int argc, char *argv[])
{
if (argc != 2)
{
usage(argv[0]);
exit(USAGE_ERR);
}
uint16_t port = atoi(argv[1]);

unique_ptr<SelectServer> svr(new SelectServer(translation, port));
svr->init();
svr->start();

return 0;
}

select 缺点：

1. select能同时等待的文件fd是有上限的，除非重新改内核，否则无法解决
2. 必须借助第三方数组，来维护合法的fd集合
3. select的大部分参数是输入输出型的，调用select前，要重新设置所有的fd，调用之后，还需要检查更新所有的fd，遍历的成本很大
4. select 采用位图，输入：用户告诉内核，输出：内核告诉用户，来回的进行数据拷贝，拷贝成本大

poll

poll方案解决了select中fd有上限、每次调用都要重新设置所有关心的fd的问题

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

fds为 struct pollfd 类型的动态数组，用于存储需要关心的fd及事件

每个元素所包含的信息如下：

struct pollfd
{
int fd; /* 文件描述符 */
short events; /* 请求的事件 */
short revents; /* 返回的事件 */
};

fd为文件描述符

events为想要关心的事件，revents为返回时就绪的事件，可以填的值如下：

常用的为POLLIN（读）、POLLOUT（写）、POLLERR（错误）

每个宏都占用一个比特位，因此可以通过按位操作包含多个事件

fds通过将用户告诉内核与内核告诉用户的事件分开，避免了每次都要重新设置fd

nfds为fds的长度，也就是要关心的文件描述符的个数

timeout参数单位是ms（毫秒），>0代表阻塞timeout ms，超时则返回；=0代表非阻塞；<0代表阻塞；

有就绪事件就返回每个事件就绪的fd相加的值，若超时返回0，若失败返回-1

当调用poll接口时，用户通过fds[i].events告诉内核需要关心的fd及其事件，当poll返回时，内核通过fds[i].revents告诉用户每个关心的fd关心的事件中已就绪的事件

但poll为了知道哪些fd的哪些事件就绪，需要遍历struct pollfd数组，当fd多了后会影响效率

epoll

epoll是为了解决poll遍历问题的多路转接接口

epoll_create

int epoll_create(int size);
int epoll_create1(int flags);

创建一个epoll模型，size 参数为预期要管理的fd个数，现已被忽略，只需要保证>0即可
返回值为epoll模型的fd

现代写法推荐用 epoll_create1 ，flags填0和epoll_create(1)代表的意思一样
 epoll_create1(EPOLL_CLOEXEC) 代表防止 epoll 文件描述符泄露给子进程（例如 fork() 、 exec() ）

epoll_ctl

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

控制指定epoll模型中需要关心的fd及其事件

• epfd：通过epoll_create创建的epoll模型
• op：有三种选项： EPOLL_CTL_ADD （增）、 EPOLL_CTL_DEL （删）、 EPOLL_CTL_MOD （改），代表要增/删/改指定fd及其事件（若为删，event参数被忽略，直接删除fd）
• fd：要操作的目标文件描述符
• event：描述感兴趣的事件和携带的用户数据

其中events参数为关心的事件类型，值如下：

• EPOLLIN : 表示对应的文件描述符可以读 (包括对端SOCKET正常关闭);
• EPOLLOUT : 表示对应的文件描述符可以写;
• EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来);
• EPOLLERR : 表示对应的文件描述符发生错误;
• EPOLLHUP : 表示对应的文件描述符被挂断;
• EPOLLET : 将EPOLL设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(Level Triggered)来说的.
• EPOLLONESHOT：只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个socket的话, 需要再次把这个socket加入到EPOLL队列里.

data为用户数据：

里面的fd成员用于表明该事件所属的文件描述符，epoll_wait拿取事件时只有struct epoll_event结构体，只能通过.data.fd确定文件描述符

成功返回0，失败返回-1

epoll_wait

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

相当于pool()或select()，用于等待文件描述符就绪

• epfd：通过epoll_create创建的epoll模型
• events：用于接收就绪事件，输出型参数
• maxevents：用于表示events的数组容量，即一次最多返回多少个事件，必须 > 0
• timeout：语义与poll()时一致

有就绪事件就返回就绪事件的个数，若超时返回0，若失败返回-1

epoll模型

epoll_create会创建一个epoll模型，它主要有两个核心数据结构：红黑树和就绪链表（双向）

当调用epoll_ctl接口时，会对红黑树进行增删改操作，每个节点对应一个需要关心的fd及其事件，并注册当该fd有数据时回调的接口 ep_poll_callback ，用于将事件挂载到就绪队列中
该红黑树的key为对应fd+struct file*指针组成

这其中 epitem.event 就是epoll_ctl时传入的event结构体，里面包含了events（事件）和data（用户数据，包含fd、指针等 ）

当底层的网卡驱动接收到数据后，传给协议栈解析并放在Socket的接收缓冲区中后，就会调用属于该fd的回调方法，而由于该fd的回调方法是 ep_poll_callback ， 就会将epoll模型中红黑树的对应节点挂载到就绪队列（将epitem.rdlink插入到就绪双向链表的末尾）

当调用epoll_wait()时，就会从就绪队列中拿取事件

epitem结构：

struct epitem {
/* 红黑树节点：用于将当前 epitem 挂载到 epoll 实例的红黑树中 */
struct rb_node rbn;

/* 双向链表节点：用于将当前 epitem 挂载到就绪队列 (rdllist) 或 overflow 队列中 */
struct list_head rdlink;

/* 指向当前 epitem 所属的 epoll 实例 (struct eventpoll) */
struct eventpoll *ep;

/* 包含用户态传入的 fd 和 event（用户感兴趣的事件掩码及用户自定义数据 epoll_data） */
struct epoll_filefd ffd;

/* 用户注册的关注事件掩码（如 EPOLLIN | EPOLLOUT | EPOLLET） */
struct epoll_event event;

/* 等待队列头：用于管理当前 epitem 挂载在底层 Socket (sk_sleep) 上的等待项 */
struct eppoll_entry *pwqlist;

/* 指向该 fd 在 VFS 层的内核文件结构体 struct file */
struct file *ffd_file;

// … 其他内核同步/统计相关的内部字段
};

因此epoll并不需要遍历数据结构，直接用O(1)的复杂度就可以知道哪些fd就绪

epoll_create返回的是epol模型抽象成文件的fd

水平触发（LT）/边缘触发（ET）

 epoll_wait 将所有就绪的事件，按照顺序放到用户传入的数组的中，如果就绪队列很多数据节点，一次拿不完，下次 epoll_wait 时默认会继续通知，也就是水平触发（LT）
水平触发：只要状态“处于就绪状态”（例如缓冲区还有数据未读完、或者就绪队列里还有未处理完的事件）， epoll_wait 就会不断通知。

但如果将对应epitem节点改为边缘触发（events），对于 ET 模式的节点，内核在将事件拷贝给用户态数组后，直接从就绪链表（rdllist）中移除该节点，不再放回。下一次调用 epoll_wait 时，即使底层缓冲区还有未读完的数据/就绪队列之前没处理完，epoll_wait 也不会通知你，它会一直阻塞等待，直到这个 fd 上再次有新的数据到达（发生新的状态边沿变化）。
边缘触发：只有当状态“发生变化/出现新边缘”（例如数据从无到有、从不可写变为可写）时，epoll_wait 才会通知一次

要对某个fd设置边缘触发，就将 events |= EPOLLET 

struct epoll_event ev;

// 设置为 读事件(EPOLLIN) + 边缘触发(EPOLLET)
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = sockfd;

// 将该 fd 注册到 epoll 实例中
epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);

ET模式的优点

虽然在LT模式下也可以通过尽快将数据读上来而达到和ET一样的效果，但ET模式下强制约束了代码要高效的IO数据，否则就跑不对，适用于高并发场景

在ET模式下，为防止服务器因fd阻塞而挂起，必须要将fd设为非阻塞。由于直到read/recv返回EWOULDBLOCK/EAGAIN时才可以确认数据读完，但在阻塞模式下会导致整个事件循环卡死（因为只会通知一次）。但只要是非阻塞，可以一直读取，直到返回EWOULDBLOCK/EAGAIN

并且由于ET模式下就算有数据没有读完，也不会一直通知，减少了重复通知，可以将这个名额用到其他新事件上（因为epoll_wait()传入的事件数组是有大小限制的！）

由于ET模式的机制，应用层会及时读取，那么就会给TCP接收缓冲区腾出更大的空间，从而让TCP可以通告更大的窗口大小，发送方就可以一次发出更多数据

应用层及时 read()
→ 内核接收缓冲区 (sk_rmem) 腾出空间
→ TCP 通告更大的 Window Size
→ 发送方可以一次发出更多数据
→ 吞吐量提升

Reactor

Reactor模式（反应堆模式），半同步（IO同步）半异步（事件分发异步），通过让主线程监听，分线程负责执行IO和业务。与之对应的还有Proactor模式，这里不做介绍

Reactor实现代码已放出，下面主要讲设计中的一些问题（只实现了单线程Reactor）

为什么要用Connection来管理一个fd，每个fd都有自己的IO缓冲区？

ET模式下的epoll，由于阻塞模式下可能会出现一次读取读不完的情况（例如被信号中断、应用层缓冲区过小等原因），因此需要设置非阻塞。
但如果其中一个fd的报文只来了一半，多个连接的数据是交错到达的，共用缓冲区会互相覆盖。每个 fd 一个 Connection ，本质上是给每个连接一个独立的"记忆"，让它记住自己读到哪了、发到哪了、状态是什么。 

写事件也是一样，若一个报文只发送了一半，而此时返回了 EWOULDBLOCK ，就要将数据存到写就绪时再发送。但等到写就绪时，不一定是这个fd先执行，如果共用一个缓冲区，就会被覆盖掉

一个 fd 对应一个 Connection，是为了给每个网络连接建立独立的上下文环境。由于 TCP 数据可能分批到达、非阻塞 IO 可能部分读写、多个连接事件交错发生，因此每个 fd 必须独立保存自己的输入输出缓冲区和状态，否则数据会互相覆盖，连接状态会混乱。

epoll中对于IO就绪事件的不同设计

对于读取事件，会将EPOLLIN常驻在epoll中，只要有数据到来就通知
对于写入事件，会先尝试直接写入，如果没有写完或返回了EWOULDBLOCK（因为是非阻塞），再让epoll关注写就绪事件（将EPOLLOUT添加到epoll），当底层的输出缓冲区空出来，就会通知。并且在确定写完数据后再将EPOLLOUT从epoll中移除！

这种读取与写入的不同处理，是因为大部分情况下，写总是就绪的，如果epoll一直关注写就绪，可能我们还没有想写入的数据，但epoll一直在通知写就绪…