多路I/O转接之select模型

Posted 2020-09-21 gccbuaa

I/O复用使得程序可以同一时候监听多个文件描写叙述符。这对提高程序的性能至关重要。通常，网络程序同一时候处理或者监听多个socket文件描写叙述符的时候可以考虑使用I/O复用模型。

值得强调的是。I/O复用尽管可以同一时候监听多个文件描写叙述符。但它本身是堵塞的。当有多个文件描写叙述符就绪的时候，假设不採取额外的措施，程序就仅仅能按顺序依次处理当中的每个文件描写叙述符，这使得server程序看起来像串行工作的。

假设要实现并发。仅仅可以使用多进程或者多线程的手段。

select 系统调用的用途是：在一段指定的时间内，监听用户感兴趣的文件描写叙述符上的可读，可写和异常等事件。

值得说明的是：

1.select 能监听的文件描写叙述符个数受限于FD_SETSIZE。一般默觉得1024，单纯改变进程打开的文件符个数并不能改变select监听的文件描写叙述符的个数

2.解决1024一下client时候使用select非常合适（比方局域网中）。可是假设连接client过多，select採用是轮询模型（程序中会有所体现）。这样会大大减少server的响应效率。

相关系统调用函数介绍 

#include <sys/select.h> //所需头文件

int select(int nfds, fd_set *readfds, fd_set *writefds,
                 fd_set *exceptfds, struct timeval *timeout);
1.nfds 监控的文件描写叙述符集里最大文件描写叙述符加1，由于此參数会告诉内核检測前多少个文件描写叙述符的状态，之所以加1 由于文件描写叙述符是从0開始编号的
2.readfds 监控有读数据到达文件描写叙述符集合，传入传出參数 
3.writefds监控有写数据到达的文件描写叙述符集合，传入传出參数
4.exceptfds 监控异常发生达文件描写叙述符集合,如带外数据到达异常，传入传出參数
timeout：定时堵塞监控时间，3种情况
1.NULL，永远等下去，堵塞在这里
2.设置timeval，等待固定时间
3.设置timeval里时间均为0。检查描写叙述字后马上返回。轮询
struct timeval {
     long tv_sec; /* seconds */
     long tv_usec; /* microseconds */
};

void FD_CLR(int fd, fd_set *set);//将fd_set集中相应的文件描写叙述符清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

select 监听程序：

#include<stdio.h>
#include<string.h>
#include<sys/select.h>
#include <sys/un.h>
#include <sys/types.h>         
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include<errno.h>
int create_listen(int port)
{
	int listen_st,on;
	struct sockaddr_in s_addr;
	listen_st =socket(AF_INET,SOCK_STREAM,0);
	if(listen_st==-1)
	
	{
		perror("socket error ");
		return -1;
	}
	if(setsockopt(listen_st,SOL_SOCKET,SO_REUSEADDR,&on,sizeof(on))==-1)
	{
		perror("setsockopt error");
		return -1;
	}
	s_addr.sin_port=htons(port);
	s_addr.sin_family=AF_INET;
	s_addr.sin_addr.s_addr=htonl(INADDR_ANY);
	
	if(bind(listen_st,(struct sockaddr*)&s_addr,sizeof(struct sockaddr_in))==-1)
	{
		perror("bind error");
		return -1;
	}
	if (listen(listen_st, 5) == -1) // 设置文件描写叙述符具有监听的功能
    {  
        perror("listen error");
        return -1;  
    }  
    return listen_st;  
}

int run_server(int port)
{
	int i,maxi,maxfd,listen_st,conn_st,sockaddr_len;
	int nready,client[FD_SETSIZE];
	char buf[1024];
	struct sockaddr_in c_addr;
	fd_set rset,allset;
	listen_st=create_listen(port);
	if(listen_st==-1)
	{
		return -1;
	}
	for(i=0;i<FD_SETSIZE;i++)
	{
		client[i]=-1;
	}
	FD_ZERO(&allset);
	FD_SET(listen_st, &allset);
	maxfd = listen_st;
	maxi=-1;
	while(1)
	{
		rset=allset;
		nready = select(maxfd+1, &rset, NULL, NULL, NULL);//select 開始监听
		if(nready<0)
		{
			perror("select error");
			break;
		}
		if(FD_ISSET(listen_st,&rset))//检測listen_st 在fd_set有没有被事件到达
		{
			sockaddr_len=sizeof(c_addr);
			conn_st=accept(listen_st,(struct sockaddr *)&c_addr,&sockaddr_len);
			printf("received form %s at port:%d \n",inet_ntoa(c_addr.sin_addr),ntohs(c_addr.sin_port));
			for(i=0;i<FD_SETSIZE;i++)
			{
				if(client[i]<0)
				{
					client[i]=conn_st;
					break;
				}
			}
			if(i==FD_SETSIZE)
			{
				printf("too many client \n");
				close(conn_st);
			}else
			{
				FD_SET(conn_st,&allset);
				if(i>maxi) //记录最大下标
				{
					maxi=i;
				}
				if(conn_st>maxfd)//记录最大文件描写叙述符用于select用
				{
					maxfd=conn_st;
				}
			}
			if(--nready==0) continue;
		}
		
		for(i=0;i<=maxi;i++)
		{
			if((conn_st=client[i])<0)
			{
				continue;
			}
			
			if(FD_ISSET(conn_st,&rset))
			{
				memset(buf,0,sizeof(buf));
				if(read(conn_st,buf,sizeof(buf))==0)
				{
					printf("close client \n");
					close(conn_st);
					FD_CLR(conn_st,&allset);
					client[i]=-1;
				}
				else
				{
					printf("recv from client:%s \n",buf);
					write(conn_st,buf,strlen(buf));
				}
				if (--nready == 0) break;  //就绪个数减一
			}
			
		}
		sleep(1);
	}
	close(listen_st);
	return 0;
}


int main(int argc,char *argv[])
{
	if(argc<2)
	{
		printf("usage:%s port \n",argv[0]);
		return 0;
	}
	int port=atoi(argv[1]);
	if(port==0)
	{
		printf("port error \n");
		return 0;
	}
	printf("start server \n");
	run_server(port);
	return 0;
}

client程序

#include<stdio.h>
#include<stdlib.h>
#include<string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include<unistd.h>
#include<errno.h>
#include<pthread.h>
#define BUFFSIZE 1024
#define ERRORCODE -1

static void *thread_send(void *arg)
{
	char buf[BUFFSIZE];
	int sd = *(int *) arg;
	while (1)
	{
		memset(buf, 0, sizeof(buf));
		read(STDIN_FILENO, buf, sizeof(buf));
		if (send(sd, buf, strlen(buf), 0) == -1)
		{
			printf("send error:%s \n", strerror(errno));
			break;
		}
	}
	return NULL;
}
static void* thread_recv(void *arg)
{
	char buf[BUFFSIZE];
	int sd = *(int *) arg;
	while (1)
	{
		memset(buf, 0, sizeof(buf));
		int rv = recv(sd, buf, sizeof(buf), 0);
		if (rv <= 0)
		{
			if(rv == 0) //server socket关闭情况
			{
				printf("server have already full !\n");
				exit(0);//退出整个客服端
			}
			printf("recv error:%s \n", strerror(errno));
			break;
		}
		printf("%s", buf);//输出接收到的内容
	}
	return NULL;
}
int run_client(char *ip_str, int port)
{
	int client_sd;
	int con_rv;
	pthread_t thrd1, thrd2;
	struct sockaddr_in client_sockaddr; //定义IP地址结构

	client_sd = socket(AF_INET, SOCK_STREAM, 0);
	if (client_sd == -1)
	{
		printf("socket create error:%s \n", strerror(errno));
		return ERRORCODE;
	}

	memset(&client_sockaddr, 0, sizeof(client_sockaddr));
	client_sockaddr.sin_port = htons(port); //指定一个端口号并将hosts字节型传化成Inet型字节型（大端或或者小端问题）
	client_sockaddr.sin_family = AF_INET;	//设置结构类型为TCP/IP
	client_sockaddr.sin_addr.s_addr = inet_addr(ip_str);
	//将字符串的ip地址转换成int型,客服端要连接的ip地址
	con_rv = connect(client_sd, (struct sockaddr*) &client_sockaddr,
			sizeof(client_sockaddr));
	//struct sockaddr 是非常早曾经定义的 struct sockaddr_in 是后定义的,眼下用的比較多
	//调用connect连接到指定的ip地址和端口号,建立连接后通过socket描写叙述符通信
	if (con_rv == -1)
	{
		printf("connect error:%s \n", strerror(errno));
		return ERRORCODE;
	}
	if (pthread_create(&thrd1, NULL, thread_send, &client_sd) != 0)
	{
		printf("thread error:%s \n", strerror(errno));
		return ERRORCODE;

	}
	if (pthread_create(&thrd2, NULL, thread_recv, &client_sd) != 0)
	{
		printf("thread error:%s \n", strerror(errno));
		return ERRORCODE;
	}
	pthread_join(thrd2, NULL);
	pthread_join(thrd1, NULL);
	close(client_sd);
	return 0;
}
int main(int argc, char *argv[])
{
	if (argc < 3)
	{
		printf("Usage:ip port,example:127.0.0.1 8080 \n");
		return ERRORCODE;
	}
	int port = atoi(argv[2]);
	char *ip_str = argv[1];
	run_client(ip_str,port);
	return 0;
}