I/O多路复用之——select

Posted 2021-07-28 清水寺扫地僧

先见：I/O多路复用之——背景知识

---

对于网络 I/O 使用socket套接字来通信，普通 I/O 模型只能监听一个 socket，假如能够预先传入一个 socket 列表（多个套接字），如果列表中的 socket都没有数据，挂起进程，直到有一个 socket 收到数据，唤醒进程。 这种方法很直接，也是 select 的设计思想。在下边的代码中，先准备一个数组 fds ，让 fds 存放着所有需要监视的 socket。然后调用 select，如果 fds 中的所有 socket 都没有数据，select 会阻塞，直到有一个 socket 接收到数据，select 返回，唤醒进程。用户可以遍历 fds，通过 FD_ISSET 判断具体哪个 socket 收到数据，然后做出处理。

实现思路/过程：

/*AF_INET（又称 PF_INET）是 IPv4 网络协议的套接字类型，AF_INET6 则是 IPv6 的；而
AF_UNIX 则是 Unix 系统本地通信。第二个参数指明连接类型，SOCK_STREAM是基于数据流的，
SOCK_DGRAM是基于数据报的。第三个参数用来指明所要接收的协议包，设置为0则用于接收任何的IP
数据，设置为0则用于接收任何的TCP数据等等。*/

int maxfd = 0；
lfd = socket(AF_INET, SOCK_STREAM, 0); // 第一个参数指明套接字类型,即创建套接字
maxfd = lfd;			//记录最大的fd值
bind();					//绑定地址结构
listen();				//设置监听上限
fd_set rset， allset;			//创建r监听集合
FD_ZERO(&allset);				//将r监听集合清空
FD_SET(lfd, &allset);			//将 lfd 添加至读集合中。

while(1) {
	rset = allset；			//保存/备份监听集合
	ret  = select(lfd+1， &rset， NULL， NULL， NULL);		//监听文件描述符集合对应事件
	if(ret > 0) {							//有监听的描述符满足对应事件
			if (FD_ISSET(lfd, &rset)) {		// 1 在, 0不在。
				cfd = accept();				//建立连接，返回用于通信的文件描述符
				maxfd = cfd;				//更新当前最大的fd值
				FD_SET(cfd, &allset);		//添加到监听通信描述符集合中。
			}

			for(i = lfd+1; i <= maxfd; i++){
				FD_ISSET(i, &rset)			//无有read、write事件
				//业务处理
				read();
				write();
			}	
	}
}
close(...);

select 的实现思路很直接，假如程序同时监视如下图的 sock1、sock2 和 sock3 三个 socket，那么在调用 select 之后，操作系统把进程 A 分别加入这三个 socket 的等待队列中（实际上就是把这三个套接字从运行空间拷贝到内核空间中，这里的拷贝涉及一次遍历）。

当任何一个 socket 收到数据后，中断程序将唤起进程。下图展示了 sock2 接收到了数据的处理流程（注： receive 和 select 的中断回调可以设置成不同的内容）：

sock2 接收到了数据，中断程序唤起进程 A，所谓唤起进程，就是将进程从所有的等待队列中移除，加入到工作队列里面，如下图所示：

将进程 A 从所有等待队列中移除，再加入到工作队列里面，经由这些步骤，当进程 A 被唤醒后，它知道至少有一个 socket 接收了数据。程序只需遍历一遍 socket 列表，就可以得到就绪的 socket。

---

Linux 系统中 select 函数的用法：

#include <sys/select.h>
/* According to earlier standards */
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
int select(int nfds, fd_set *readfds, fd_set *writefds,
			fd_set *exceptfds, struct timeval *timeout);

	nfds: 		监控的文件描述符集里最大文件描述符加1，因为此参数会告诉内核检测前多少个文件描述符的状态
	
	readfds：	监控有读数据到达文件描述符集合，传入传出参数
	
	writefds：	监控写数据到达文件描述符集合，传入传出参数
	
	exceptfds：	监控异常发生达文件描述符集合,如带外数据到达异常，传入传出参数
	
	timeout：	定时阻塞监控时间，3种情况
				1.NULL，永远等下去
				2.设置timeval，等待固定时间
				3.设置timeval里时间均为0，检查描述字后立即返回，轮询
				struct timeval {
					long tv_sec; /* seconds */
					long tv_usec; /* microseconds */
				};
	
	select I/O复用方式搭配辅助函数：
		void FD_CLR(int fd, fd_set *set); 	//把文件描述符集合里fd清0
		int FD_ISSET(int fd, fd_set *set); 	//测试文件描述符集合里fd是否置1
		void FD_SET(int fd, fd_set *set); 	//把文件描述符集合里fd位置1
		void FD_ZERO(fd_set *set); 			//把文件描述符集合里所有位清0

• int __nfds是fd_set 中最大的描述符+1，当调用select时，内核态会判断fd_set中描述符是否就绪，__nfds告诉内核最多判断到哪一个描述符；
• readfds、writefds、exceptfds都是结构体 fd_set ，fd_set 可以看作是一个描述符的集合。 select 函数中存在三个fd_set 集合，分别代表三种事件， readfds 表示读描述符集合， writefds 表示读描述符集合， exceptfds 表示异常描述符集合。当对应的 fd_set = NULL 时，表示不监听该类描述符；
• timeval __timeout 用来指定select的工作方式，即当文件描述符尚未就绪时，select 是永远等下去，还是等待一定的时间，即使没有 socket 就绪，也直接返回；
• 函数返回值 int 表示： 就绪描述符的数量，如果为 -1 表示产生错误 ；
• fd 就是文件描述符 file descriptor；

---

运行机制：

Select会将全量fd_set 从用户空间拷贝到内核空间，并注册回调函数， 在内核态空间来判断每个请求是否准备好数据 。select在没有查询到有文件描述符（就是套接字）就绪的情况下，将一直阻塞（就是一直等在 receive 那里）。如果有一个或者多个描述符（就是套接字）就绪，那么select将就绪的文件描述符（就是套接字）置位，然后select返回。返回后，由程序遍历查看哪个请求有数据。

---

优点：

这种简单方式行之有效，在几乎所有操作系统都有对应的实现。

缺点：

• 每次调用 select 都需要将进程加入到所有 socket 的等待队列（套接字集合拷贝），每次唤醒都需要从每个队列中移除（检查到底是哪一个套接字就绪了），这里涉及了两次遍历：
    第一次遍历：每次都要将整个 fds 列表（就是套接字集合）传递给内核，这个传递，说白了就是将整个 fds 列表（就是套接字集合）从用户态拷贝到内核态，这波操作有一定的开销，而且 fd 越多开销则越大；
    第二次遍历：进程被唤醒后，程序并不知道哪些 socket 收到数据，select 返回的是整个 socket 数组的句柄。应用程序需要遍历整个socket 数组才知道谁发生了变化（谁就绪了），轮询代价大；
• 每次都会拷贝，每次都要轮询，最大开销和 fds列表（就是套接字集合）中的 fd（文件描述符/套接字）个数有关，为了防止fds太大，拷贝、轮询的开销太大，所以fds的长度是有限制的：
    select 的 fd_set 是基于一个位掩码（bit mask）实现的，因此 fd_set 有固定的长度（int[32]），一般是 32*32=1024，数组元素的每一位对应一个文件描述符。例如，一个 int 整数占 32 位，那么 32 个 int 整数组成的数组，第一个元素代表文件描述符 0 到 31，数组的第二个元素代表文件描述符 32 到 63，以此类推， fd_set 可以表示1024个文件描述符的就绪情况，所以默认就只能监视 1024 个 socket。内核在被编译的时候，可以不受这个长度的限制，因为 select() 允许应用程序自定义 FD_SETSIZE 的大小（在如今的系统头文件中可以看到这一点），但是这会增加额外的支出。

注意：这里只解释了 select 的一种情形。当程序调用 select 时，内核会先遍历一遍 socket，如果有一个以上的 socket 接收缓冲区有数据，那么 select 直接返回，不会阻塞。这也是为什么 select 的返回值有可能大于 1 的原因之一。如果没有 socket 有数据，说明该进程需要的网络数据还没到达，进程才会阻塞，才会处于等待状态。

---

使用场景：

• 可移植性好，几乎适用于所有平台。select 已经存在很长时间了，你可以确定每个支持网络和非阻塞套接字的平台都会支持select，而它可能还不支持 poll。另一种选择是你仍然使用 poll 然后在那些没有 poll 的平台上使用 select 来模拟它；
• select的超时时间理论(timeout)上可以精确到纳秒级别。而poll和epoll的精度只有毫秒级。这对于桌面或者服务器系统来说没有任何区别，因为它们不会运行在纳秒精度的时钟上，但是在某些与硬件交互的实时嵌入式平台，降低控制棒关闭核反应堆，可能是需要的。(这就可以作为一个更加精确的sleep()来用);
• select 的缺点，使得其无法支持高并发连接。假设我们的服务器需要支持100万的并发连接，则在 _FD_SETSIZE 为1024的情况下，则我们至少需要开辟1k个进程才能实现100万的并发连接。除了进程间上下文切换的时间消耗外，从内核/用户空间大量的无脑内存拷贝、数组轮询等，是系统难以承受的。因此，基于 select 模型的服务器程序，要达到10万级别的并发访问，是一个很难完成的任务。

---

触发方式：

select 的触发方式是水平触发，应用程序如果没有完成对一个已经就绪的文件描述符的IO，那么之后再次 select 调用还是会将这些文件描述符通知进程。

---

select 完整调用过程图：

---

使用 select 轮询方式进行I/O复用的 server 简单实现：

/* server.c */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "wrap.h"

#define MAXLINE 80
#define SERV_PORT 6666

int main(int argc, char *argv[])
{
	int i, maxi, maxfd, listenfd, connfd, sockfd;
	int nready, client[FD_SETSIZE]; 	/* FD_SETSIZE 默认为 1024 */
	ssize_t n;
	fd_set rset, allset;
	char buf[MAXLINE];
	char str[INET_ADDRSTRLEN]; 			/* #define INET_ADDRSTRLEN 16 */
	socklen_t cliaddr_len;
	struct sockaddr_in cliaddr, servaddr;

	listenfd = Socket(AF_INET, SOCK_STREAM, 0);

bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(SERV_PORT);

Bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr));

Listen(listenfd, 20); 		/* 默认最大128 */

maxfd = listenfd; 			/* 初始化 */
maxi = -1;					/* client[]的下标 */

for (i = 0; i < FD_SETSIZE; i++)
	client[i] = -1; 		/* 用-1初始化client[] */

FD_ZERO(&allset);
FD_SET(listenfd, &allset); /* 构造select监控文件描述符集 */

for ( ; ; ) {
	rset = allset; 			/* 每次循环时都从新设置select监控信号集 */
	nready = select(maxfd+1, &rset, NULL, NULL, NULL);

	if (nready < 0)
		perr_exit("select error");
	if (FD_ISSET(listenfd, &rset)) { /* new client connection */
		cliaddr_len = sizeof(cliaddr);
		connfd = Accept(listenfd, (struct sockaddr *)&cliaddr, &cliaddr_len);
		printf("received from %s at PORT %d\\n",
				inet_ntop(AF_INET, &cliaddr.sin_addr, str, sizeof(str)),
				ntohs(cliaddr.sin_port));
		for (i = 0; i < FD_SETSIZE; i++) {
			if (client[i] < 0) {
				client[i] = connfd; /* 保存accept返回的文件描述符到client[]里 */
				break;
			}
		}
		/* 达到select能监控的文件个数上限 1024 */
		if (i == FD_SETSIZE) {
			fputs("too many clients\\n", stderr);
			exit(1);
		}

		FD_SET(connfd, &allset); 	/* 添加一个新的文件描述符到监控信号集里 */
		if (connfd > maxfd)
			maxfd = connfd; 		/* select第一个参数需要 */
		if (i > maxi)
			maxi = i; 				/* 更新client[]最大下标值 */

		if (--nready == 0)
			continue; 				/* 如果没有更多的就绪文件描述符继续回到上面select阻塞监听,
										负责处理未处理完的就绪文件描述符 */
		}
		for (i = 0; i <= maxi; i++) { 	/* 检测哪个clients 有数据就绪 */
			if ( (sockfd = client[i]) < 0)
				continue;
			if (FD_ISSET(sockfd, &rset)) {
				if ( (n = Read(sockfd, buf, MAXLINE)) == 0) {
					Close(sockfd);		/* 当client关闭链接时，服务器端也关闭对应链接 */
					FD_CLR(sockfd, &allset); /* 解除select监控此文件描述符 */
					client[i] = -1;
				} else {
					int j;
					for (j = 0; j < n; j++)
						buf[j] = toupper(buf[j]);
					Write(sockfd, buf, n);
				}
				if (--nready == 0)
					break;
			}
		}
	}
	close(listenfd);
	return 0;
}