IO多路复用之select

2023-05-16

一、IO多路复用

IO多路复用是IO模式中的一种，它建立在内核提供的多路分离函数select基础之上，我们使用select函数可以避免同步非阻塞IO模型中轮询等待的问题，此外poll、epoll都是这种模型。

轮询等待问题伪代码：

while(1)
{
read(gps_fd, buf, sizeof(buf));
read(socket_fd, buf, sizeof(buf));
read(serialport_fd, buf, sizeof(buf));
}

可见当我们在阻塞模式下同时有多个read系统调用时，如果第一个read读取不到数据，那么它就一直不会继续执行下一个read，从而导致轮询等待的问题。

而在使用select进行多路复用时，用户首先将需要进行IO操作的socket添加到select中，然后阻塞等待select系统调用返回。当数据到达时，socket被激活，select函数返回。用户线程正式发起read请求，读取数据并继续执行。这样虽然多了监视socket的过程，但是这个过程避免了轮询等待问题，达到了可以同时在一个线程中处理多个IO请求的目的。而如果选择同步阻塞模式并不可以在单个线程中达到此目的。

二、select函数

select()函数有着很好的跨平台性，几乎所以平台都支持，它允许进程指示内核等待多个事件(文件描述符)中的任何一个发生，并只在有一个或多个事件发生或经历一段指定时间后才唤醒它，然后接下来判断究竟是哪个文件描述符发生了事件并进行相应的处理。但是select也有着系统开销很大和单个进程能够监视的文件描述符的数量存在最大限制的缺点。

man手册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);

       void FD_CLR(int fd, fd_set *set);//清空集合
       int  FD_ISSET(int fd, fd_set *set);//将给定的描述符加入集合
       void FD_SET(int fd, fd_set *set);//判断指定描述符是否在集合中
       void FD_ZERO(fd_set *set);//将给定的描述符从文件中删除

参数解释：max_fd指待测试的fd的总个数，它的值是待测试的最大文件描述符加1；readset、writeset和exceptset分别为指定要让内核测试读、写和异常条件的文件描述符集合，如果不需要参数为NULL；timeout为超时时间，如果设置为NULL则永不超时。

timeval结构体如下：

struct timeval
{
 long tv_sec; 
 long tv_usec; 
};

返回值：返回0超时；返回1出错。

注意：是待测试的描述集总是从0开始，如果你要检测的描述符为9，实际上真正测试描述符我10个。因为Linux内核有个参数__FD_SETSIZE定义了每个FD_SET的句柄个数中，所以select只能同时处理1024个客户端的连接请求。

三、select实现socket服务器

（1）流程图：

在这里插入图片描述

（2）代码讲解：

fd_set 			rdset

fd集合通过fd_set类型定义。

for(i=0; i<ARRAY_SIZE(fds_array) ; i++)
{
	fds_array[i]=-1;
}
fds_array[0] = listenfd;

这里通过一个循环来将更新fd集合的数组全部清零，因为对应的是文件描述符，所以给他们所有赋值为-1来清空集合，然后将listenfd放入数组。

if ( FD_ISSET(listenfd, &rdset) )
		{
			if( (connfd=accept(listenfd, (struct sockaddr *)NULL, NULL)) < 0)
			{
				printf("accept new client failure: %s\n", strerror(errno));
				continue;
			}
			found = 0;
			for(i=0; i<ARRAY_SIZE(fds_array) ; i++)
			{
				if( fds_array[i] < 0 )
				{
					printf("accept new client[%d] and add it into array\n", connfd );
					fds_array[i] = connfd;
					found = 1;
					break;
				}
			}
			if( !found )
			{
				printf("accept new client[%d] but full, so refuse it\n", connfd);
				close(connfd);
			}
		}
		else 
		{
			for(i=0; i<ARRAY_SIZE(fds_array); i++)//遍历
			{
				if( fds_array[i]<0 || !FD_ISSET(fds_array[i], &rdset) )
					continue;
				if( (rv=read(fds_array[i], buf, sizeof(buf))) <= 0)
				{
					printf("socket[%d] read failure or get disconncet.\n", fds_array[i]);
					close(fds_array[i]);
					fds_array[i] = -1;
				}
				else
				{
					printf("socket[%d] read get %d bytes data\n", fds_array[i], rv);
					for(j=0; j<rv; j++)
						buf[j]=toupper(buf[j]);
					if( write(fds_array[i], buf, rv) < 0 )
					{
						printf("socket[%d] write failure: %s\n", fds_array[i], strerror(errno));
						close(fds_array[i]);
						fds_array[i] = -1;
					}
				}
			}
		}

这个地方首先用FD_ISSET()来判断是否是listenfd发生的事件如果是表示有新的客户端连就继续accept，如果不是则执行else以后的操作。

有新客户端则把进行accept操作，并用for循环把文件描述符加入更新数组，if( !found )来判断客户端是否因为fd集合满了而被拒绝。

如果是已经连上了的client发的数据，就直接用if( fds_array[i]<0 || !FD_ISSET(fds_array[i], &rdset) )找到对应文件描述符，然后再进行read等操作。

static inline void print_usage(char *progname)

此自定义函数用于输出相关帮助信息。

int socket_server_init(char *listen_ip, int listen_port)

此自定义函数用于封装了socket、bind、listen等操作，详细见流程图。

四、总结

以上就是今天要讲的内容，本文讲解了IO多路复用，详细的讲解了select函数，并通过select实现了select服务器程序实例让大家深入了解了select多路复用，希望这篇博客可以让大家有所收获。

代码示例：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <ctype.h>
#include <time.h>
#include <pthread.h>
#include <getopt.h>
#include <libgen.h>
#include <sys/types.h> 
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>

#define ARRAY_SIZE(x) (sizeof(x)/sizeof(x[0]))

static inline void msleep(unsigned long ms);
static inline void print_usage(char *progname);
int socket_server_init(char *listen_ip, int listen_port);

int main(int argc, char **argv)
{
	int				listenfd, connfd;
	int				serv_port = 0;
	int 			daemon_run = 0;
	char			*progname = NULL;
	int				opt;
	fd_set 			rdset;//definition fd set
	int 			rv;
	int 			i, j;
	int				found;
	int				maxfd=0;
	char 			buf[1024];
	int 			fds_array[1024];
	struct 			option long_options[] =
	{
		{"daemon", no_argument, NULL, 'b'},
		{"port", required_argument, NULL, 'p'},
		{"help", no_argument, NULL, 'h'},
		{NULL, 0, NULL, 0}
	};
	progname = basename(argv[0]);//eg:/local/user/src/ basename it will get src
	
	while ((opt = getopt_long(argc, argv, "bp:h", long_options, NULL)) != -1)
	{
		switch (opt)
		{
			case 'b':
				daemon_run=1;
				break;
			case 'p':
				serv_port = atoi(optarg);
				break;
			case 'h':
				print_usage(progname);
				return EXIT_SUCCESS;
			default:
				break;
		}
	}

	if( !serv_port )
	{ 
		print_usage(progname);
		return -1;
	}

	if( (listenfd=socket_server_init(NULL, serv_port)) < 0 )
	{
		printf("ERROR: %s server listen on port %d failure\n", argv[0],serv_port);
		return -2;
	}
	printf("%s server start to listen on port %d\n", argv[0],serv_port);
	
	if( daemon_run )
	{
		daemon(0, 0);
	}
	for(i=0; i<ARRAY_SIZE(fds_array) ; i++)
	{
		fds_array[i]=-1;
	}
	fds_array[0] = listenfd;

	for ( ; ; )
	{
		FD_ZERO(&rdset);
		for(i=0; i<ARRAY_SIZE(fds_array) ; i++)
		{
			if( fds_array[i] < 0 )
				continue;
				
			maxfd = fds_array[i]>maxfd ? fds_array[i] : maxfd;
			FD_SET(fds_array[i], &rdset);
		}
		
		rv = select(maxfd+1, &rdset, NULL, NULL, NULL);//int select(int max_fd, fd_set *readset, fd_set *writeset, fd_set *exceptset, struct timeval *timeout);usually just use max_fd and *readset.
		if(rv < 0) 
		{
			printf("select failure: %s\n", strerror(errno));
			break;
		}
		else if(rv == 0)
		{
			printf("select get timeout\n");
			continue;
		}
		
		if ( FD_ISSET(listenfd, &rdset) )
		{
			if( (connfd=accept(listenfd, (struct sockaddr *)NULL, NULL)) < 0)
			{
				printf("accept new client failure: %s\n", strerror(errno));
				continue;
			}
			found = 0;
			for(i=0; i<ARRAY_SIZE(fds_array) ; i++)
			{
				if( fds_array[i] < 0 )
				{
					printf("accept new client[%d] and add it into array\n", connfd );
					fds_array[i] = connfd;
					found = 1;
					break;
				}
			}
			if( !found )
			{
				printf("accept new client[%d] but full, so refuse it\n", connfd);
				close(connfd);
			}
		}
		else 	
		{
			for(i=0; i<ARRAY_SIZE(fds_array); i++)
			{
				if( fds_array[i]<0 || !FD_ISSET(fds_array[i], &rdset) )
					continue;
				if( (rv=read(fds_array[i], buf, sizeof(buf))) <= 0)
				{
					printf("socket[%d] read failure or get disconncet.\n", fds_array[i]);
					close(fds_array[i]);
					fds_array[i] = -1;
				}
				else
				{
					printf("socket[%d] read get %d bytes data\n", fds_array[i], rv);
					
					for(j=0; j<rv; j++)
						buf[j]=toupper(buf[j]);
					if( write(fds_array[i], buf, rv) < 0 )
					{
						printf("socket[%d] write failure: %s\n", fds_array[i], strerror(errno));
						close(fds_array[i]);
						fds_array[i] = -1;
					}
				}
			}
		}
	}

CleanUp:
	close(listenfd);
	return 0;
}

static inline void print_usage(char *progname)
{
	printf("Usage: %s [OPTION]...\n", progname);

	printf(" %s is a socket server program, which used to verify client and echo back string from it\n",
		progname);
	printf("\nMandatory arguments to long options are mandatory for short options too:\n");

	printf(" -b[daemon ] set program running on background\n");
	printf(" -p[port ] Socket server port address\n");
	printf(" -h[help ] Display this help information\n");

	printf("\nExample: %s -b -p 8900\n", progname);
	return ;
}

int socket_server_init(char *listen_ip, int listen_port)
{
	struct sockaddr_in servaddr;
	int rv = 0;
	int on = 1;
	int listenfd;
	if ( (listenfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
	{
		printf("Use socket() to create a TCP socket failure: %s\n", strerror(errno));
		return -1;
	}
	
	setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
	
	memset(&servaddr, 0, sizeof(servaddr));
	servaddr.sin_family = AF_INET; 
	servaddr.sin_port = htons(listen_port);
	
	if( !listen_ip )
	{
		servaddr.sin_addr.s_addr = htonl(INADDR_ANY); 
	}
	else 
	{
		if (inet_pton(AF_INET, listen_ip, &servaddr.sin_addr) <= 0)
		{
			printf("inet_pton() set listen IP address failure.\n");
			rv = -2;
			goto CleanUp;
		}
	}
	
	if(bind(listenfd, (struct sockaddr *) &servaddr, sizeof(servaddr)) < 0)
	{
		printf("Use bind() to bind the TCP socket failure: %s\n", strerror(errno));
		rv = -3;
		goto CleanUp;
	}
	
	if(listen(listenfd, 13) < 0)
	{
		printf("Use bind() to bind the TCP socket failure: %s\n", strerror(errno));
		rv = -4;
		goto CleanUp;
	}
CleanUp:
	if(rv<0)
	close(listenfd);
	else
	rv = listenfd;
	return rv;
}

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)