I/O多路转接之select
程序员文章站
2022-06-14 14:39:50
...
I/O多路转接之select
一、五种IO模型
-
阻塞IO: 在内核将数据准备好之前, 系统调用会一直等待. 所有的套接字, 默认都是阻塞方式
-
非阻塞IO: 如果内核还未将数据准备好, 系统调用仍然会直接返回, 并且返回EWOULDBLOCK错误码.非阻塞IO往往需要程序员循环的方式反复尝试读写文件描述符, 这个过程称为轮询. 这对CPU来说是较大的浪费
3. 信号驱动IO: 内核将数据准备好的时候, 使用SIGIO信号通知应用程序进行IO操作
4. IO多路转接: 虽然从流程图上看起来和阻塞IO类似. 实际上最核心在于IO多路转接能够同时等待多个文件描述符的就绪状态
- 异步IO: 由内核在数据拷贝完成时, 通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据
高级IO相关概念：
同步通信 vs 异步通信
同步和异步关注的是消息通信机制（注意和线程同步区分）
- 所谓同步，就是在发出一个调用时，在没有得到结果之前，该调用就不返回. 但是一旦调用返回，就得 到返回值了;换句话说，就是由调用者主动等待这个调用的结果;
- 异步则是相反，调用在发出之后，这个调用就直接返回了，所以没有返回结果; 换句话说，当一个异步过程调用发出后，调用者不会立刻得到结果; 而是在调用发出后，被调用者通过状态、通知来通知调用者，或通过回调函数处理这个调用
阻塞 vs 非阻塞
阻塞和非阻塞关注的是程序在等待调用结果（消息，返回值）时的状态.
- 阻塞调用是指调用结果返回之前，当前线程会被挂起. 调用线程只有在得到结果之后才会返回.
- 非阻塞调用指在不能立刻得到结果之前，该调用不会阻塞当前线程
二、I/O多路转接之select原理
1.函数原型
int select(int nfds, fd_set *readfds, fd_set *writefds,
fd_set *exceptfds, struct timeval *timeout);
| 参数 | 作用 |
|---|---|
| nfds | 是需要监视的最大文件描述值+1 |
| fd_set *readfds | 需要检测的可读文件描述符的集合， |
| fd_set *writefds | 需要检测的可写文件描述符的集合 |
| fd_set *exceptfds | 需要检测的异常文件描述符的集合； |
| struct timeval *timeout | 为结构timeval，用来设置select（）的等待时间 |
函数返回值
1.执行成功则返回文件描述符状态已经改变的个数；
2.如果返回0代表在描述词状态改变之前已经超过 timeout时间，没有返回；
3.当有错误发生时则返回-1，错误原因存于error，此时参数readfd、writefd、exceptfd和timeout的值变成不可预测。
2.理解fd_set结构
其实这个结构就是一个整型数组，更严格的说，是一个“位图”，使用位图中对应的位表示要监视的文件描述符
- 用来消除描述词组 set中相关fd的位
void FD_CLR(int fd,fd_set *set);
- 用来测试描述词组set中的相关fd的位是否为真；
void FD_ISSET(int fd,fd_set *set);
- 用来设置描述词组set中相关的位；
void FD_SET(int fd,fd_set *set);
- 用来消除描述词组set的全部位；
void FD_ZERO(int fd,fd_set *set);
3.socket就绪条件
读就绪：
1.socket内核中，接收缓冲区中的字节数，大于等于低水位标记SO_RCVLOWAT。此时就可以无阻塞的读该文件描述符，并且返回值大于0；
2.socket TCP通信中，对端关闭连接，此时对该socket读，则返回0（四次挥手）；
3.监听的socket上有新的连接请求；
4.socket上有未处理的错误。
写就绪：
1.socket内核中，发送缓冲区中的可用字节数（发送缓冲区的空闲位置大小），大于等于低水位标记SO_RCVLOWAT。此时可以无阻塞的写，并且返回值大于0；
2.socket的写操作被关闭。对一个写操作被关闭的socket进行写操作，会触发SIGPIPE信号；
3.socket使用非阻塞connect连接成功或失败之后；
4.socket上有未读取的错误；
异常就绪：
socket上带外数据。关于带外数据，和TCP紧急模式相关（TCP协议头中的紧急指针字段）
4.理解select执行过程
理解select模型的关键在于理解fd_set,为说明方便，取fd_set长度为1字节，fd_set中的每一bit可以对应一个文件描述符fd。则1字节长的fd_set最大可以对应8个fd.
（1）执行fd_set set; FD_ZERO(&set);则set用位表示0000,0000。
（2）若fd＝5,执行FD_SET(fd,&set);后set变为0001,0000(第5位置为1)
（3）若再加入fd＝2，fd=1,则set变为0001,0011
（4）执行select(6,&set,0,0,0)阻塞等待
（5）若fd=1,fd=2上都发生可读事件，则select返回，此时set变为
0000,0011。注意：没有事件发生的fd=5被清空
5.select优缺点
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的第一个参数。
select缺点
- 每次调用select, 都需要手动设置fd集合, 从接口使用角度来说也非常不便.
每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大
同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大
select支持的文件描述符数量太小
6.select实现tcp服务器（支持多用户）
#pragma once
#include <vector>
#include <unordered_map>
#include <functional>
#include <sys/select.h>
#include "tcp_socket.hpp"
// 必要的调试函数
inline void PrintFdSet(fd_set* fds, int max_fd)
{
printf("select fds: ");
for (int i = 0; i < max_fd + 1; ++i)
{
if (!FD_ISSET(i, fds))
{
continue;
}
printf("%d ", i);
}
printf("\n");
}
typedef std::function<void (const std::string& req, std::string* resp)> Handler;
class Selector
{
public:
Selector()
{
// [注意!] 初始化千万别忘了!!
max_fd_ = 0;
FD_ZERO(&read_fds_);
}
bool Add( TcpSocket& sock)
{
int fd = sock.GetFd();
printf("[Selector::Add] %d\n", fd);
if (fd_map_.find(fd) != fd_map_.end())
{
printf("Add failed! fd has in Selector!\n");
return false;
}
fd_map_[fd] = sock;
FD_SET(fd, &read_fds_);
if (fd > max_fd_)
{
max_fd_ = fd;
}
return true;
}
bool Del( TcpSocket& sock)
{
int fd = sock.GetFd();
printf("[Selector::Del] %d\n", fd);
if (fd_map_.find(fd) == fd_map_.end())
{
printf("Del failed! fd has not in Selector!\n");
return false;
}
fd_map_.erase(fd);
FD_CLR(fd, &read_fds_);
// 重新找最大的文件描述符, 从右往左找比较快
for (int i = max_fd_; i >= 0; --i)
{
if (!FD_ISSET(i, &read_fds_))
{
continue;
}
max_fd_ = i;
break;
}
return true;
}
// 返回读就绪的文件描述符集
bool Wait(std::vector<TcpSocket>* output)
{
output->clear();
// [注意] 此处必须要创建一个临时变量, 否则原来的结果会被覆盖掉
fd_set tmp = read_fds_;
PrintFdSet(&tmp, max_fd_);
int nfds = select(max_fd_ + 1, &tmp, NULL, NULL, NULL);
if (nfds < 0)
{
perror("select");
return false;
}
// [注意!] 此处的循环条件必须是 i < max_fd_ + 1
for (int i = 0; i < max_fd_ + 1; ++i)
{
if (!FD_ISSET(i, &tmp))
{
continue;
}
output->push_back(fd_map_[i]);
}
return true;
}
private:
fd_set read_fds_;
int max_fd_;
// 文件描述符和 socket 对象的映射关系
std::unordered_map<int, TcpSocket> fd_map_;
};
class TcpSelectServer
{
public:
TcpSelectServer(const std::string& ip, uint16_t port) : ip_(ip), port_(port)
{}
bool Start(Handler handler) const
{
//1. 创建 socket
TcpSocket listen_sock;
bool ret = listen_sock.Socket();
if (!ret)
{
return false;
}
// 2. 绑定端口号
ret = listen_sock.Bind(ip_, port_);
if (!ret)
{
return false;
}
// 3. 进行监听
ret = listen_sock.Listen(5);
if (!ret)
{
return false;
}
// 4. 创建 Selector 对象
Selector selector;
selector.Add(listen_sock);
// 5. 进入事件循环
for (;;)
{
std::vector<TcpSocket> output;
bool ret = selector.Wait(&output);
if (!ret)
{
continue;
}
// 6. 根据就绪的文件描述符的差别, 决定后续的处理逻辑
for (size_t i = 0; i < output.size(); ++i)
{
if (output[i].GetFd() == listen_sock.GetFd())
{
// 如果就绪的文件描��符是 listen_sock, 就执行 accept, 并加入到 select 中
TcpSocket new_sock;
listen_sock.Accept(&new_sock, NULL, NULL);
selector.Add(new_sock);
}
else
{
// 如果就绪的文件描述符是 new_sock, 就进行一次请求的处理
std::string req, resp;
bool ret = output[i].Recv(&req);
if (!ret)
{
selector.Del(output[i]);
// [注意!] 需要关闭 socket
output[i].Close();
continue;
}
// 调用业务函数计算响应
handler(req, &resp);
// 将结果写回到客户端
output[i].Send(resp);
}
} // end for
} // end for (;;)
return true;
}
private:
std::string ip_;
uint16_t port_;
};
下一篇： 三国美男周瑜的四种死法！哪种最靠谱？