Linux 高級(jí)I/O之poll函數(shù)及簡(jiǎn)單服務(wù)器客戶端編程

當(dāng)需要同時(shí)監(jiān)聽(tīng)多個(gè)文件描述符時(shí)，就需要I/O復(fù)用函數(shù)，I/O復(fù)用函數(shù)有select、poll、epoll，今天主要使用poll函數(shù)。

poll()接受一個(gè)指向結(jié)構(gòu)'struct pollfd'列表的指針，其中包括了你想測(cè)試的文件描述符和事件。事件由一個(gè)在結(jié)構(gòu)中事件域的比特掩碼確定。當(dāng)前的結(jié)構(gòu)在調(diào)用后將被填寫并在事件發(fā)生后返回。

函數(shù)原型：

#include<poll.h>
int?poll(struct?pollfd?*fds,?nfds_t?nfds,?int?timeout);
struct?pollfd{
????int?fd;?????????/*file?descriptor*/
????short?events;???/*requested?events*/
????short?revents;??/*returned?events*/
}

函數(shù)參數(shù)：fds是要監(jiān)聽(tīng)的fd的數(shù)組，nfds是數(shù)組個(gè)數(shù)，timeout 超時(shí)時(shí)間 -1是阻塞；

函數(shù)說(shuō)用：通過(guò)傳入的events的類型去判斷返回的類型是否一致，如果一致就該干事了。

events：

接下來(lái)是是一個(gè)服務(wù)器監(jiān)聽(tīng)兩個(gè)socket的例子：

服務(wù)器代碼：

#include<stdio.h>
#include<stdlib.h>??
#include<string.h>??
#include<errno.h>??
#include<sys/types.h>??
#include<sys/socket.h>??
#include<netinet/in.h>?
#include<pthread.h>
#include<unistd.h>
#include<poll.h>

#define?MAX_POLLFD_NUM?2
//#define?SERVER_CONN_IP1?"1.1.1.1"
//#define?SERVER_CONN_IP2?"1.1.1.1"
#define?DEFAULT_PORT?8000
#define?BUFF_MAX?1024

typedef?void(*Server_Rrocess_Thread_Fun)(void?*arg);
/*thread?process?function?define?*/
typedef?struct?sever_thread_fun{
	Server_Rrocess_Thread_Fun?Server_Process_Client_Conn1;
	Server_Rrocess_Thread_Fun?Server_Process_Client_Conn2;
}Poll_Server_Process_Clinet_FUN_t;
/*connect?1?thread?function*/
void?*Poll_Conn1_Process(void?*arg)
{
	unsigned?char?ucBufArr[BUFF_MAX]?=?{0};
	unsigned?long?ulSize?=?0;
	int?connect_fd?;
????printf("come?pthread?conn1?fun?to?proess?conn1\n");
	if(?(connect_fd?=?accept(?*(int*)arg,?(struct?sockaddr*)NULL,?NULL))?==?-1){??
		printf("accept?socket?error:?%s(errno:?%d)",strerror(errno),errno);??
????}
	ulSize?=?recv(connect_fd,?ucBufArr,?BUFF_MAX,?0);??

	if(send(connect_fd,?"Hello,you?are?connected?1!\n",?26,0)?==?-1)??
	perror("send?error");??
	ucBufArr[ulSize]?=?'\0';??
????//sleep(20);
	printf("recv?msg?from?client:?%s\n",?ucBufArr);??
	close(connect_fd);??
	pthread_exit((void*)1);
}
/*connect?1?thread?function*/
void?*Poll_Conn2_Process(void?*arg)
{
	unsigned?char?ucBufArr[BUFF_MAX]?=?{0};
	unsigned?long?ulSize?=?0;	
	int?connect_fd?;
????printf("come?pthread?conn2?fun?to?proess?conn2\n");
	if(?(connect_fd?=?accept(?*(int*)arg,?(struct?sockaddr*)NULL,?NULL))?==?-1){??
		printf("accept?socket?error:?%s(errno:?%d)",strerror(errno),errno);??
????}
	ulSize?=?recv(connect_fd,?ucBufArr,?BUFF_MAX,?0);??

	if(send(connect_fd,?"Hello,you?are?connected?2!\n",?26,0)?==?-1)??
	perror("send?error");??
	ucBufArr[ulSize]?=?'\0';??
???//?sleep(20);
	printf("recv?msg?from?client:?%s\n",?ucBufArr);??
	close(connect_fd);??
	pthread_exit((void*)2);
}
int?main()
{	
????int?poll_ret?=?0;
	pthread_t?thread_conn1;
	pthread_t?thread_conn2;
????/*socket?var*/
	int?server_socket_fd_conn1;
	int?server_socket_fd_conn2;
	struct?sockaddr_in?servaddr_conn1;
	struct?sockaddr_in?serveraddr_conn2;
	/*poll?var*/
????struct?pollfd?pollfd_arr[MAX_POLLFD_NUM];
	/*init?thread?fun*/
	Poll_Server_Process_Clinet_FUN_t?server_conn_handel;
	server_conn_handel.Server_Process_Client_Conn1?=?Poll_Conn1_Process;
	server_conn_handel.Server_Process_Client_Conn2?=?Poll_Conn2_Process;
	/*create?two?server?socket*/
	if(?(server_socket_fd_conn1?=?socket(AF_INET,?SOCK_STREAM,?0))?==?-1?){??
????printf("create?socket?conn1?error:?%s(errno:?%d)\n",strerror(errno),errno);??
????exit(0);??
????}??
	if(?(server_socket_fd_conn2?=?socket(AF_INET,?SOCK_STREAM,?0))?==?-1?){??
????printf("create?socket?conn2?error:?%s(errno:?%d)\n",strerror(errno),errno);??
????exit(0);??
????}??
	
????/*init?socket?1?for?conn1*/??
????memset(&servaddr_conn1,?0,?sizeof(servaddr_conn1));??
????servaddr_conn1.sin_family?=?AF_INET;??	
????servaddr_conn1.sin_addr.s_addr?=?htonl(INADDR_ANY);?
????servaddr_conn1.sin_port?=?htons(DEFAULT_PORT);?
//	if(?inet_pton(AF_INET,?SERVER_CONN_IP1,?&servaddr_conn1.sin_addr.s_addr)?<=?0){??
?//???printf("inet_pton?error?for?%s\n",SERVER_CONN_IP1);??
??//??exit(0);??
???//?}??
	/*init?socket?2?for?conn2*/??
	memset(&serveraddr_conn2,?0,?sizeof(serveraddr_conn2));??
????serveraddr_conn2.sin_family?=?AF_INET;??
????serveraddr_conn2.sin_addr.s_addr?=?htonl(INADDR_ANY);?
????serveraddr_conn2.sin_port?=?htons(DEFAULT_PORT+1);?
//	if(?inet_pton(AF_INET,?SERVER_CONN_IP2,?&serveraddr_conn2.sin_addr.s_addr)?<=?0){??
?//???printf("inet_pton?error?for?%s\n",SERVER_CONN_IP2);??
??//??exit(0);??
???//?}??
??
????/*bind?connect?1?socket*/
????if(?bind(server_socket_fd_conn1,?(struct?sockaddr*)&servaddr_conn1,?sizeof(servaddr_conn1))?==?-1){??
????printf("bind?socket?error1:?%s(errno:?%d)\n",strerror(errno),errno);??
????exit(0);??
????}??
	/*bind?connect?2?socket*/
	if(?bind(server_socket_fd_conn2,?(struct?sockaddr*)&serveraddr_conn2,?sizeof(serveraddr_conn2))?==?-1){??
????printf("bind?socket?error2:?%s(errno:?%d)\n",strerror(errno),errno);??
????exit(0);??
????}??
????/*listen?connect?1*/??
????if(?listen(server_socket_fd_conn1,?10)?==?-1){??
????printf("listen?socket?error:?%s(errno:?%d)\n",strerror(errno),errno);??
????exit(0);??
????}??
	/*listen?connect?2*/??
????if(?listen(server_socket_fd_conn2,?10)?==?-1){??
????printf("listen?socket?error:?%s(errno:?%d)\n",strerror(errno),errno);??
????exit(0);??
????}??
	pollfd_arr[0].events?=?POLLRDNORM;/*only?read?event*/
	pollfd_arr[0].fd?=?server_socket_fd_conn1;
	
	pollfd_arr[1].events?=?POLLRDNORM;/*only?read?event*/
	pollfd_arr[1].fd?=?server_socket_fd_conn2;
????while(1)
	{
		poll_ret?=?poll(?pollfd_arr,?MAX_POLLFD_NUM,-1);
		if(?pollfd_arr[0].revents?&?POLLRDNORM?)
		{
			/*connect?1?process?task*/
			pthread_create(?&thread_conn1,?NULL,?server_conn_handel.Server_Process_Client_Conn1,?(void?*)(&pollfd_arr[0].fd));
		}
		if(?pollfd_arr[1].revents?&?POLLRDNORM?)
		{
			/*connect?2?process?task*/
			pthread_create(?&thread_conn2,?NULL,?server_conn_handel.Server_Process_Client_Conn2,?(void?*)(&pollfd_arr[1].fd));
		}
		sleep(1);
	}
	return?0;
}

客戶端代碼：

#include<stdio.h>??
#include<stdlib.h>??
#include<string.h>??
#include<errno.h>??
#include<sys/types.h>??
#include<sys/socket.h>??
#include<netinet/in.h>??
??
#define?BUFF_MAX?1024
#define?DEFAULT_PORT?8000
int?main(int?argc,?char**?argv)??
{??
????int????sockfd,?n,rec_len;??
????char????recvline[BUFF_MAX],?sendline[BUFF_MAX];??
????char????buf[BUFF_MAX];??
????struct?sockaddr_in????servaddr;??
??
??
????if(?argc?!=?2){??
????printf("usage:?./client?<ipaddress>\n");??
????exit(0);??
????}??
??
????/*create?socket*/
????if(?(sockfd?=?socket(AF_INET,?SOCK_STREAM,?0))?<?0){??
????printf("create?socket?error:?%s(errno:?%d)\n",?strerror(errno),errno);??
????exit(0);??
????}??
	
????memset(&servaddr,?0,?sizeof(servaddr));??
????servaddr.sin_family?=?AF_INET;??
????servaddr.sin_port?=?htons(DEFAULT_PORT/*+1/);/*DEFAULT_PORT?is?connect?1?to?server,DEFAULT_PORT+1?is?connect?2?to?server*/??
	/*string?ip?to?int?ip*/
????if(?inet_pton(AF_INET,?argv[1],?&servaddr.sin_addr)?<=?0){??
????printf("inet_pton?error?for?%s\n",argv[1]);??
????exit(0);??
????}??
????/*connect?server*/
????if(?connect(sockfd,?(struct?sockaddr*)&servaddr,?sizeof(servaddr))?<?0){??
????printf("connect?error:?%s(errno:?%d)\n",strerror(errno),errno);??
????exit(0);??
????}??
	
????printf("send?msg?to?server:?\n");??
????fgets(sendline,?BUFF_MAX,?stdin);??
	/*write?data?to?sockfd*/
????if(?send(sockfd,?sendline,?strlen(sendline),?0)?<?0)??
????{??
????printf("send?msg?error:?%s(errno:?%d)\n",?strerror(errno),?errno);??
????exit(0);??
????}??
	/*read?socket?data*/
????if((rec_len?=?recv(sockfd,?buf,?BUFF_MAX,0))?==?-1)?{??
???????perror("recv?error");??
???????exit(1);??
????}??
????buf[rec_len]??=?'\0';??
????printf("Received?:?%s?\n",buf);??
	/*colse?socket*/
????close(sockfd);??
????exit(0);??
}

解釋：服務(wù)器端監(jiān)聽(tīng) ip 127.0.0.1 port 8000和ip 127.0.0.1 port 8001

結(jié)果演示：