文章目录
- I/O 多路转接 之 Select
- 1. 了解select
- 2. select 函数原型
- ① fd_set 结构
- ② 详细理解参数（readfds为例）
- 3. 理解select的执行过程
- 4. select代码实例：监视多个文件描述符
- 5. Socket就绪条件
- 6. select代码实例：多路复用服务器
- ① Sock.hpp
- ② Log.hpp
- ③ SelectServer.hpp
- 7. select的优缺点
I/O 多路转接 之 Select
1. 了解select
select 是一个用于多路复用 I/O 的系统调用，它 允许一个进程监视多个文件描述符等待其中任何一个变为可读或可写状态，然后进行相应的 I/O 操作。
- select 函数会阻塞程序，直到被监视的文件描述符之一准备好执行相应的 I/O 操作，或者超时。一旦有文件描述符准备就绪，select 函数会返回，并告诉你哪些文件描述符已经就绪。
2. select 函数原型
select 函数的原型在头文件 <sys/select.h> 中声明，原型如下：
#include <sys/select.h>
int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
参数含义：
-
nfds：需要检查的文件描述符的数量，通常设置为需要监视的文件描述符中最大的描述符值加1。 -
readfds：指向一个fd_set结构的指针，包含了需要监视读事件的文件描述符集合。 -
writefds：指向一个fd_set结构的指针，包含了需要监视写事件的文件描述符集合。 -
exceptfds：指向一个fd_set结构的指针，包含了需要监视异常事件的文件描述符集合。 -
timeout：设置超时时间，即select最多阻塞等待的时间。如果设置为NULL，select将一直阻塞，直到有文件描述符就绪或者被信号中断。如果设置为不为NULL的值，则指定最大等待时间。
对于struct timeval：
struct timeval {
time_t tv_sec; // 秒数
suseconds_t tv_usec; // 微秒数
};
tv_sec表示秒数部分，是一个 time_t 类型的整数，通常用于表示秒数。tv_usec表示微秒数部分，是一个 suseconds_t 类型的整数，通常用于表示不足一秒的时间部分。
而对于中间三个参数（readfds、writefds、execptfds），均为输入输出型参数：
-
它们用于指定需要监视的文件描述符集合，并在
select函数返回后用于表示哪些文件描述符已经就绪。readfds用于监视读事件的文件描述符集合，即当指定的文件描述符可以从中读取数据时，select将会通知。writefds用于监视写事件的文件描述符集合，即当指定的文件描述符可以向其中写入数据时，select将会通知。exceptfds用于监视异常事件的文件描述符集合，例如带外数据到达时，select将会通知。
在调用 select 函数之前，通常会将需要监视的文件描述符添加到这些集合中，然后 select 函数会在返回时更新这些集合，以指示哪些文件描述符已经就绪。
① fd_set 结构
fd_set是一个位图，用于表示一组文件描述符。
需要注意的是，fd_set 的大小是由宏 FD_SETSIZE 定义的，这决定了 fd_set 能表示的最大文件描述符数目。在使用 fd_set 时，要确保不会超出这个限制。
② 详细理解参数（readfds为例）
对于readfds：
- 输入时：用户->内核，当前的比特位的值表示文件描述符值，比如为：0000 1010 意味着关心1号和3号文件描述符的读事件
- 输出时：内核->用户，作为操作系统，用户让自身关心的多个fd有结果了，0000 1000 表示3号文件描述符已就绪可以读取数据（后续用户可以直接读取三号而不会被阻塞）
3. 理解select的执行过程
-
初始化
fd_set：- 在调用
select之前，你通常会用FD_ZERO清空fd_set，然后用FD_SET将你感兴趣的文件描述符添加到集合中。FD_SET用于将一个文件描述符添加到fd_set中，FD_ZERO用于清空一个fd_set。
- 在调用
-
调用
select：select调用会将你传入的fd_set结构体中的文件描述符集合拷贝到内核中。此时，内核会监视这些文件描述符，并等待它们的状态变化。
-
阻塞等待：
select会阻塞（或者在设定的超时时间到达后返回），直到其中至少一个文件描述符的状态发生变化。状态变化包括文件描述符变得可读、可写，或者发生了异常情况。
-
内核检查事件：
- 内核会检查每个文件描述符的状态，并检测是否满足你的要求（例如是否有数据可读，是否可以写入数据，是否有异常条件）。如果状态发生变化，内核会更新传入的
fd_set结构体，使其反映文件描述符的实际状态。
- 内核会检查每个文件描述符的状态，并检测是否满足你的要求（例如是否有数据可读，是否可以写入数据，是否有异常条件）。如果状态发生变化，内核会更新传入的
-
返回并处理结果：
- 当
select返回时，你可以检查fd_set结构体，查看哪些文件描述符的状态发生了变化。你会看到之前传入的fd_set结构体中相应的位已经被设置，表示这些文件描述符有事件发生。
- 当
-
清理和准备下一次调用：
- 在下一次调用
select之前，你需要再次初始化fd_set结构体，并设置你感兴趣的文件描述符。请注意，在每次调用select后，传入的fd_set会被修改，所以在每次调用之前，你需要重新设置它们。
- 在下一次调用
4. select代码实例：监视多个文件描述符
下面使用select()监视多个文件描述符：
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/select.h>
int main() {
fd_set readfds;
int maxfd;
struct timeval timeout;
// 设置文件描述符
int fd1 = 0; // 标准输入
int fd2 = 1; // 标准输出
// 设定 `select` 需要监视的最大文件描述符
maxfd = (fd1 > fd2 ? fd1 : fd2) + 1;
while (1) {
// 清空文件描述符集合
FD_ZERO(&readfds);
// 添加文件描述符到集合
FD_SET(fd1, &readfds);
FD_SET(fd2, &readfds);
// 设置超时时间
timeout.tv_sec = 5; // 5秒
timeout.tv_usec = 0;
// 调用 `select`
int retval = select(maxfd, &readfds, NULL, NULL, &timeout);
if (retval == -1) {
perror("select");
exit(EXIT_FAILURE);
} else if (retval == 0) {
printf("Timeout occurred! No data after 5 seconds.\n");
} else {
if (FD_ISSET(fd1, &readfds)) {
printf("File descriptor %d is ready for reading.\n", fd1);
}
if (FD_ISSET(fd2, &readfds)) {
printf("File descriptor %d is ready for writing.\n", fd2);
}
}
}
return 0;
}
5. Socket就绪条件
在使用套接字（Socket）进行网络编程时，套接字的就绪条件表示可以进行某种操作的条件，主要用于异步 I/O 操作，例如通过 select、poll、epoll 等函数实现的多路复用。
套接字的就绪条件通常包括以下几种：
-
读就绪（Read Ready）：套接字缓冲区中有数据可供读取，即接收缓冲区中有数据到达，可以调用
recv函数读取数据。 -
写就绪（Write Ready）：套接字缓冲区有足够的空间可以写入数据，即发送缓冲区有足够的空间可以发送数据，可以调用
send函数写入数据。 -
异常就绪（Exception Ready）：套接字发生异常情况，如带外数据到达或者连接错误。这通常通过
select或者poll函数的异常集合来检查。 -
连接就绪（Connection Ready）：套接字连接已经建立，可以进行数据交换。对于服务器套接字来说，连接就绪表示已经有客户端连接请求到达，可以调用
accept函数接受连接。对于客户端套接字来说，连接就绪表示连接成功建立，可以进行数据交换。
这些就绪条件可以通过多路复用函数（如
select、poll、epoll
等）来监视，当套接字处于就绪状态时，这些函数会通知应用程序执行相应的操作。
6. select代码实例：多路复用服务器
对于下面的代码，重点放在关于select方面的代码↓
对于套接字代码的IO通信，首先自然需要一个套接字，下面是一个封装了套接字的类Sock的头文件：
① Sock.hpp
#pragma once
#include <iostream>
#include <string>
#include <cstring>
#include <cerrno>
#include <cassert>
#include <unistd.h>
#include <memory>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <ctype.h>
#include "log.hpp"
using std::string;
class Sock
{
private:
// 将listen的第二个参数设为1
const static int gbacklog = 1; // 监听队列的最大数量
public:
Sock() {}
~Sock() {}
static int Socket() // 创建监听socket
{
int listenSock = socket(AF_INET, SOCK_STREAM, 0);
if(listenSock < 0) // 失败
{
logMessage(FATAL, "socket error");
exit(2);
}
logMessage(NORMAL, "create socket successfully");
return listenSock;
}
static void Bind(int sock, uint16_t port, string ip = "0.0.0.0")
{
struct sockaddr_in local;
memset(&local, 0, sizeof(local));
local.sin_family = AF_INET;
local.sin_port = htons(port);
// local.sin_addr.s_addr = inet_addr(ip.c_str());
inet_pton(AF_INET, ip.c_str(), &local.sin_addr);
if(bind(sock, (struct sockaddr*)&local, sizeof(local)) < 0)
{
logMessage(FATAL, "bind error | %d : %s", errno, strerror(errno));
exit(3);
}
}
static void Listen(int sock)
{
if(listen(sock, gbacklog) < 0)
{
logMessage(FATAL, "listen error | %d : %s", errno, strerror(errno));
exit(4);
}
logMessage(NORMAL, "init server successfully");
}
static int Accept(int sock, string* ip, uint16_t* port)
{
// 接收连接请求
struct sockaddr_in src;
socklen_t len = sizeof(src);
int serviceSock = accept(sock, (struct sockaddr*)&src, &len);
if(serviceSock < 0)
{
logMessage(FATAL, "accept error | %d : %s", errno, strerror(errno));
exit(5);
}
if(port) *port = ntohs(src.sin_port);
if(ip) *ip = inet_ntoa(src.sin_addr);
return serviceSock;
}
// 连接
static bool Connect(int sock, const string& server_ip, const uint16_t& server_port)
{
struct sockaddr_in server;
memset(&server, 0, sizeof(server));
server.sin_family = AF_INET;
server.sin_port = htons(server_port);
server.sin_addr.s_addr = inet_addr(server_ip.c_str());
if(connect(sock, (struct sockaddr*)&server, sizeof(server)) == 0)
{
logMessage(NORMAL, "connect to %s successfully", server_ip.c_str());
return true;
}
else
return false;
}
// 主动关闭连接
void Close(int sock)
{ }
};
② Log.hpp
该文件用于打印日志：
#pragma once
#include <iostream>
#include <cstdio>
#include <cstdarg>
#include "log.hpp"
// 宏定义 日志级别
#define DEBUG 0
#define NORMAL 1
#define WARNING 2
#define ERROR 3
#define FATAL 4
// 全局字符串数组 : 将日志级别映射为对应的字符串
const char *gLevelMap[] = {
"DEBUG",
"NORMAL",
"WARNING",
"ERROR",
"FATAL"
};
#define LOGFILE "./threadpool.log" // LOGFILE: 表示日志文件的路径
void logMessage(int level, const char* format, ...)
{
// 判断DEBUG_SHOW 是否定义，分别执行操作
#ifndef DEBUG_SHOW // 将日志级别映射为对应的字符串
if(level == DEBUG) return; // DEBUG_SHOW不存在 且 日志级别为 DEBUG时，返回
#endif
// DEBUG_SHOW存在 则执行下面的日志信息
char stdBuffer[1024];
time_t timestamp = time(nullptr);
// 将日志级别和时间戳格式化后的字符串将会被写入到 stdBuffer 缓冲区中
snprintf(stdBuffer, sizeof(stdBuffer), "[%s] [%ld] ", gLevelMap[level], timestamp);
char logBuffer[1024];
va_list args;
va_start(args, format);
vsnprintf(logBuffer, sizeof(logBuffer), format, args);
va_end(args);
printf("%s%s\n", stdBuffer, logBuffer);
}
③ SelectServer.hpp
下面进行select的部分编写：
我们主要创建一个SelectServer类：
成员变量包括：
uint16_t _port; // 端口号
int _listensock; // 监听套接字
int _fd_array[NUM]; // 文件描述符数组
随后编写构造函数与析构函数：
// 构造函数：初始化监听套接字，并绑定到指定端口。
SelectServer(const uint16_t& port = 8080): _port(port)
{
_listensock = Sock::Socket();
Sock::Bind(_listensock, _port);
Sock::Listen(_listensock);
logMessage(DEBUG, "%s", "create base socket success.");
// 初始化_fd_array
for(int i = 0; i < NUM; ++i) _fd_array[i] = FD_NONE;
_fd_array[0] = _listensock;
}
// 析构函数：关闭监听套接字。
~SelectServer()
{
if(_listensock != FD_NONE) // FD_NONE定义为-1
{
close(_listensock);
logMessage(DEBUG, "%s", "close base sock success.");
}
}
接下来编写Start函数：
- 服务器的主循环，使用 select() 函数监听所有的文件描述符，一旦有事件发生，就调用相应的处理函数
// Start() 函数：服务器的主循环，使用 select() 函数监听所有的文件描述符，一旦有事件发生，就调用相应的处理函数。
void Start()
{
DebugPrint();
while(true)
{
fd_set rfds; //一组文件描述符的集合
FD_ZERO(&rfds); // 清空
int maxfd = _listensock; // 最大文件描述符值
for(int i = 0; i < NUM; ++i)
{
if(_fd_array[i] == FD_NONE) continue; // 无效文件描述符，继续
FD_SET(_fd_array[i], &rfds); // 添加到rfds
if(maxfd < _fd_array[i]) maxfd = _fd_array[i]; // 更新maxfd
}
int n = select(maxfd + 1, &rfds, NULL, NULL, NULL); // 监视文件描述符可读性
switch(n)
{
case 0:
logMessage(DEBUG, "%s", "select timeout.");
break;
case -1:
logMessage(WARNING, "select error %d : %s", errno, strerror(errno));
break;
default: // 有事件发生，执行操作：
logMessage(DEBUG, "select success: get a new link event.");
HandlerEvent(rfds);
break;
}
}
}
对于DebugPrint()：
void DebugPrint() // 打印当前的fd信息
{
cout << "_fd_array[]: ";
for(int i = 0; i < NUM; ++i)
{
if(_fd_array[i] == FD_NONE) continue;
cout << _fd_array[i] << " ";
}
cout << endl;
}
跟随Start中select监听到了事件发生时调用的HandlerEvent函数：
- 下面Handler处理给定的文件描述符集合的所以fd，应该执行什么操作事件：
void HandlerEvent(const fd_set& rfds)
{
for(int i = 0; i < NUM; ++i) // rfds中可能存在重复的sock
{
// 去掉无效sock
if(_fd_array[i] == FD_NONE) continue;
//
if(FD_ISSET(_fd_array[i], &rfds))
{
// 此时当前的_fd_array[i]的读事件就绪
if(_fd_array[i] == _listensock) Accepter(); // 该fd是监听套接字，有新的连接请求需要处理
else Recver(i); // 该fd已就绪，有数据可读，执行读操作
}
}
}
随后编写建立连接请求的Accepter() 与 Recver()：
这两个函数分别用于处理新连接的接受和已建立连接的数据接收。
下面是思路：
Accepter() 函数：
- 首先调用
Sock::Accept()函数接受新的连接，并获取客户端的 IP 地址、端口号和新的客户端套接字。 - 如果接受连接失败，记录错误信息并返回。
- 如果成功接受连接，将新的客户端套接字添加到
_fd_array数组中，以便后续可以监视其读取事件。 - 如果
_fd_array已满，则记录警告信息并关闭新的连接。 - 否则，将新的客户端套接字添加到
_fd_array中。
void Accepter()
{
string clientIp; // 客户端ip
uint16_t clientPort; // 客户端端口
int clientSock = Sock::Accept(_listensock, &clientIp, &clientPort); // 建立连接
if(clientSock == -1)
{
logMessage(WARNING, "accept error %d : %s", errno, strerror(errno));
return;
}
logMessage(DEBUG, "get a new link: [%s:%d], sock: %d.", clientIp.c_str(), clientPort, clientSock);
// 直接read / recv可能会被阻塞（sock数据到来的时间未知）
// 得到新连接时，将新的sock托管给select，用select检查sock上是否有新数据
// 如果sock上有数据，select读事件就绪，select会进行通知，随后我们再读取，不会被阻塞
int pos = 1;
for(; pos < NUM; ++pos)
{
if(_fd_array[pos] == FD_NONE) break;
}
if(pos == NUM) {
logMessage(WARNING, "%s:%d", "select already full, can't accept new link.\n close: %d", clientIp.c_str(), clientPort, clientSock);
close(clientSock);
} else {
_fd_array[pos] = clientSock;
}
}
Recver() 函数：
- 首先从
_fd_array中获取指定位置的套接字。 - 然后尝试从该套接字中接收数据。
- 如果成功接收到数据，则打印接收到的消息。
- 如果对端关闭了连接，则关闭对应的套接字并在
_fd_array中将其标记为无效。 - 如果接收数据失败，则记录错误信息，关闭套接字并将其标记为无效。
void Recver(int pos)
{
int& sock = _fd_array[pos];
// 读事件就绪
logMessage(DEBUG, "message in, get IO event: %d", sock);
// 此时fd的数据一定是就绪的
char buffer[1024];
int n = recv(sock, buffer, sizeof(buffer) - 1, 0);
if(n > 0) { // 成功接收数据
buffer[n] = 0;
logMessage(NORMAL, "client[%d]# %s", sock, buffer);
} else if (n == 0) { // 对端关闭连接
logMessage(DEBUG, "client[%d] quit, me, too.", sock);
close(sock);
sock = FD_NONE;
} else { // 读取失败
logMessage(WARNING, "%d sock recv error, %d : %s", sock, errno, strerror(errno));
close(sock);
sock = FD_NONE;
}
}
7. select的优缺点
优点：
- 多文件描述符监视：可以同时监视多个文件描述符的状态。
- 超时控制：支持设置超时时间来控制等待时间。
- 适合低并发场景：在处理文件描述符数量不多的情况下表现良好。
缺点：
- 能够同时管理的fd数量有上限
- 几乎每个参数都是输入输出型的，select会频繁的进行 “用户到内核，内核到用户”的参数数据拷贝，开销大
- 编码较麻烦复杂，每次调用需要手动设置fd_set
-
- 为维护第三方数组，select服务器不得不进行大量的遍历操作，底层操作系统去关心fd时，一样需要遍历（时间开销大大大）
- 每一次都需要对select参数进行重新设定
![[Python可视化]空气污染物浓度地图可视化](https://i-blog.csdnimg.cn/direct/9ae234d7ce834bd0b5f2086270e225cb.png)
