二十八、高级IO与多路转接之select

一、五种IO模型

（一）阻塞IO:

在内核将数据准备好之前,，系统调用会一直等待，所有的套接字，默认都是阻塞方式。

（二）非阻塞IO:

非阻塞IO往往需要程序员循环的方式反复尝试读写文件描述符, 这个过程称为轮询. 这对CPU来说是较大的浪费, 一般只有特定场景下才使用.

（三）信号驱动IO:

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

（四）IO多路转接:

虽然从流程图上看起来和阻塞IO类似，实际上最核心在于IO多路转接能够同时等待多个文件描述符的就绪状态。

（五）异步IO:

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

任何IO过程中, 都包含两个步骤. 第一是等待, 第二是拷贝. 而且在实际的应用场景中, 等待消耗的时间往往都远远高于拷贝的时间. 让IO更高效, 最核心的办法就是让等待的时间尽量少。

二、高级IO重要概念

（一）同步通信 vs 异步通信(synchronous communication/ asynchronous communication)

同步和异步关注的是消息通信机制。

• 所谓同步，就是在发出一个调用时，在没有得到结果之前，该调用就不返回. 但是一旦调用返回，就得到返回值了; 换句话说，就是由调用者主动等待这个调用的结果;
• 异步则是相反，调用在发出之后，这个调用就直接返回了，所以没有返回结果; 换句话说，当一个异步过程调用发出后，调用者不会立刻得到结果; 而是在调用发出后，被调用者通过状态、通知来通知调用者，或通过回调函数处理这个调用.

另外, 我们回忆在讲多进程多线程的时候, 也提到同步和互斥. 这里的同步通信和进程之间的同步是完全不相干的概念。

• 进程/线程同步也是进程/线程之间直接的制约关系
• 是为完成某种任务而建立的两个或多个线程，这个线程需要在某些位置上协调他们的工作次序而等待、传递信息所产生的制约关系. 尤其是在访问临界资源的时候。

（二）阻塞 vs 非阻塞

阻塞和非阻塞关注的是程序在等待调用结果（消息，返回值）时的状态.

• 阻塞调用是指调用结果返回之前，当前线程会被挂起. 调用线程只有在得到结果之后才会返回。
• 非阻塞调用指在不能立刻得到结果之前，该调用不会阻塞当前线程。

（三）其他高级IO

非阻塞IO，纪录锁，系统V流机制，I/O多路转接（也叫I/O多路复用）,readv和writev函数以及存储映射IO（mmap），这些统称为高级IO。

（四）非阻塞IO

1.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).
我们此处只是用第三种功能, 获取/设置文件状态标记, 就可以将一个文件描述符设置为非阻塞
1
2
3
4
5
6
7
8
9
10
11

2.实现函数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); 
}
1
2
3
4
5
6
7
8

使用F_GETFL将当前的文件描述符的属性取出来(这是一个位图).
然后再使用F_SETFL将文件描述符设置回去. 设置回去的同时, 加上一个O_NONBLOCK参数。

• 轮询方式读取标准输入

#include 
#include 
#include 
void SetNoBlock(int fd) {
	 int fl = fcntl(fd, F_GETFL);
	 if (fl < 0) {
	 perror("fcntl");
	 return;
	 }
	 fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}
int main() {
	 SetNoBlock(0);
	 while (1) {
	 char buf[1024] = {0};
	 ssize_t read_size = read(0, buf, sizeof(buf) - 1);
	 if (read_size < 0) {
	 perror("read");
	 sleep(1);
	 continue;
	 }
	 printf("input:%s\n", buf);
	 }
	 return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

三、I/O多路转接之select

（一）初识select

系统提供select函数来实现多路复用输入/输出模型：

• select系统调用是用来让我们的程序监视多个文件描述符的状态变化的；
• 程序会停在select这里等待，直到被监视的文件描述符有一个或多个发生了状态改变；

（二）select函数原型

select的函数原型如下: #include <sys/select.h>
int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
1
2

1.参数解释:

• 参数nfds是需要监视的最大的文件描述符值+1；
• rdset,wrset,exset分别对应于需要检测的可读文件描述符的集合，可写文件描述符的集 合及异常文件描述符的集合;
• 参数timeout为结构timeval，用来设置select()的等待时间

2.参数timeout取值:

• NULL：则表示select（）没有timeout，select将一直被阻塞，直到某个文件描述符上发生了事件;
• 0：仅检测描述符集合的状态，然后立即返回，并不等待外部事件的发生。
• 特定的时间值：如果在指定的时间段里没有事件发生，select将超时返回。

3.关于fd_set结构

其实这个结构就是一个整数数组, 更严格的说, 是一个 “位图”. 使用位图中对应的位来表示要监视的文件描述符.
提供了一组操作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的全部位
1
2
3
4

4.关于timeval结构

timeval结构用于描述一段时间长度，如果在这个时间内，需要监视的描述符没有事件发生则函数返回，返回值为0。

函数返回值：

• 执行成功则返回文件描述词状态已改变的个数
• 如果返回0代表在描述词状态改变前已超过timeout时间，没有返回
• 当有错误发生时则返回-1，错误原因存于errno，此时参数readfds，writefds, exceptfds和timeout的值变成不可预测。

错误值可能为：

EBADF 文件描述词为无效的或该文件已关闭
EINTR 此调用被信号所中断
EINVAL 参数n 为负值。
ENOMEM 核心内存不足
1
2
3
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被清空。
1
2
3
4
5

（四）socket就绪条件

读就绪：

• socket内核中, 接收缓冲区中的字节数, 大于等于低水位标记SO_RCVLOWAT. 此时可以无阻塞的读该文件
• 描述符, 并且返回值大于0;
• socket TCP通信中, 对端关闭连接, 此时对该socket读, 则返回0;监听的socket上有新的连接请求;
• socket上有未处理的错误;
  写就绪：
• socket内核中, 发送缓冲区中的可用字节数(发送缓冲区的空闲位置大小), 大于等于低水位标记
  SO_SNDLOWAT, 此时可以无阻塞的写, 并且返回值大于0;
• socket的写操作被关闭(close或者shutdown). 对一个写操作被关闭的socket进行写操作, 会触发SIGPIPE信号;
• socket使用非阻塞connect连接成功或失败之后;
• socket上有未读取的错误;

（五）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支持的文件描述符数量太小.

（七）select使用示例

#pragma once
#include 
#include 
#include 
#include 
#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;
// 把 Select 封装成一个类. 这个类虽然保存很多 TcpSocket 对象指针, 但是不管理内存
class Selector {
public:
	 Selector() {
	 // [注意!] 初始化千万别忘了!!
	 max_fd_ = 0;
	 FD_ZERO(&read_fds_);
	 }
	 bool Add(const 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(const 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_;
	 // DEBUG
	 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_;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148