vxworks 并发服务器程序

基于vxworks 下的并发服务器访问 #include "ClientList.h" #define SA_LEN sizeof(struct sockaddr_in) #define MYPORT 5000 // 服务端口号 #define BACKLOG 50 //最大连接数 #define BUF_SIZE 200 //缓冲区大小 #define OUT_TIME_SEC 3 //连接超时(在该时间内无数据收发)时间设置（秒/单位） #define CHECK_DELAY 1 //检查超时连接的时间间隔（秒/单位） #define FILE_STATE_OUTTIME 10 //文件状态超时（select()超时 秒/单位） list_client *client_list; //客户端链表 pthread_attr_t pAttr; //线程属性 /********************************************************* * * 功能描述: 检查链表中超时的客户连接,并删除客户链表连接超时的节点 * * 参数: 无 * * 返回值: NULL * * 作者:胡士超 * * 完成日期: 2012.08.09 * * *******************************************************/ void * conn_Timeout(void * data) { time_t cur_time; //当前时间 client_node *temp,*next_node,*tmp_node; while (TRUE) { //获取用于刷新读描述符集的客户节点 tmp_node = client_list->client_head.prev; //获取当前时间 time(&cur;_time); //加锁 pthread_mutex_lock(&client;_list->mutex); //遍历客户链表检查超时的客户节点并将其从链表中移除 temp = client_list->client_head.next; while (temp != &client;_list ->client_head) { next_node = temp->next; //若该 socket fd是用于刷新的描述符则不移除 if(temp->client_sock_fd == tmp_node->client_sock_fd) { temp = next_node; continue; } //判断是否超时 if( (cur_time - temp->time) > OUT_TIME_SEC) { printf("delete a node fd is %d out time is %d\n",temp->client_sock_fd,(cur_time - temp->time)); //从链表中移除 deleteNode_by_sockfd(client_list, temp->client_sock_fd); //刷新读描述符集 if (write(client_list->refresh_fd, "refresh", 7) == ERROR) { perror("client : write failed."); } } temp = next_node; } //解除锁 pthread_mutex_unlock(&client;_list->mutex); //超时检查的间隔 sleep(CHECK_DELAY); } return NULL; } /************************************************** * * 功能描述: 线程函数,创建一个单独线程用来接收客户端发来的请求,并将请求插入客户端链表中 * * 参数: 用于监听的socket描述符指针 lsnfd * * 返回值: NULL * * 作者: 胡士超 * * 完成日期: 2012.08.09 * * ***********************************************/ void * ptr_accept(void * lsnfd) { int * sfd; int sock_fd,new_fd; int sin_size; struct sockaddr_in client_addr; // 客户端地址 int ret; pthread_t pid; sin_size = SA_LEN; sfd = (int *)lsnfd; sock_fd = *sfd; while (TRUE) { //接受新连接 new_fd = accept(sock_fd, (struct sockaddr *)&client;_addr, &sin;_size); if (new_fd <= 0) { perror("accept() failed\n"); continue; } //加锁 pthread_mutex_lock(&client;_list->mutex); //添加节点 ret = list_insert_node(client_list, 0, new_fd); if (ret == ERROR) { perror("insert node failed\n"); close(new_fd); //解除锁 pthread_mutex_unlock(&client;_list->mutex); continue; } //解除锁 pthread_mutex_unlock(&client;_list->mutex); //解除阻塞的select()刷新读描述符集 if (write(client_list->refresh_fd, "refresh", 7) == ERROR) { perror("client : write failed."); continue; } //通知等待条件的线程使之解除阻塞 //pthread_cond_signal(&client;_list->cond); } return NULL; } /********************************************** * * 功能描述:线程函数,创建一个单独线程用来接收所有客户端发来的数据 * * 参数: 无 * * 返回值: NULL * * 作者：胡士超 * * 完成日期：2012.08.09 * * ********************************************/ void * ptr_recv(void * data) { struct timeval tv; //文件描述符状态超时设置 client_node *temp; int ret; char buf[BUF_SIZE]; tv.tv_sec = FILE_STATE_OUTTIME; tv.tv_usec = 0; while (TRUE) { //清除文件描述符中所有的位 FD_ZERO(&client;_list->fdsr); //加锁 pthread_mutex_lock(&client;_list->mutex); //当链表中无客户连接时，等待客户连接 // while(client_list->node_count == 0) // { // pthread_cond_wait(&client;_list->cond, &client;_list->mutex); // } //遍历链表 temp = client_list->client_head.next; while (temp != &client;_list ->client_head) { /* 将socket描述符加入读描述符集 */ FD_SET (temp->client_sock_fd, &client;_list->fdsr); if (temp->client_sock_fd > client_list->maxsock) { client_list->maxsock = temp->client_sock_fd; } temp = temp->next; } //解除锁 pthread_mutex_unlock(&client;_list->mutex); //等待可读的文件描述 ret = select(client_list->maxsock + 1, &client;_list->fdsr, NULL, NULL, &tv;); if (ret < 0) { continue; } else if (ret == 0) { continue; } if (FD_ISSET(client_list->client_head.prev->client_sock_fd, &client;_list->fdsr)) { //清除读缓冲区内容 recv(client_list->client_head.prev->client_sock_fd, buf, sizeof(buf), 0); memset(buf,0,sizeof(buf)); continue; } //加锁 pthread_mutex_lock(&client;_list->mutex); //遍历链表,检查读描述符集中可读描述 temp = client_list->client_head.next; while (temp != &client;_list ->client_head) { //判断此描述符是否可读 if (FD_ISSET(temp->client_sock_fd, &client;_list->fdsr)) { //接收客户端数据 ret = recv(temp->client_sock_fd, buf, sizeof(buf), 0); //客户端关闭连接 if (ret <= 0) { printf("client[%d] close\n", temp->client_sock_fd); //关闭连接 close(temp->client_sock_fd); //删除节点 deleteNode_by_sockfd(client_list, temp->client_sock_fd); } else { //记录本次访问时间 time(&temp;->time); if (ret < BUF_SIZE) { buf[ret] = '\0'; printf("client[%d] send:%s\n", temp->client_sock_fd, buf); } } } temp = temp->next; } //解除锁 pthread_mutex_unlock(&client;_list->mutex); } return NULL; } /******************************************************* * * 功能描述:服务端主函数,开辟监听端口,创建接收客户端连接线程、接收客户端数据线程、客户连接超时线程 * * 参数:监听端口port * * 返回值: 成功返回0 失败返回-1 * * 作者:胡士超 * * 完成日期:2012.08.09 * * ****************************************************/ int Server(int port) { int sock_fd, new_fd; /* 监听描述符, 客户连接描述符 */ struct sockaddr_in serverAddr; /* 服务器端点地址 */ pthread_t pid_accept, pid_recv, pid_timeout; /* 线程id */ int ret; /* 创建socket */ if ((sock_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) { perror("socket"); exit(1); } /* 初始化服务器的本地端点地址 */ bzero((char *) &serverAddr;, SA_LEN); serverAddr.sin_family = AF_INET; serverAddr.sin_len = (u_char) SA_LEN; serverAddr.sin_port = htons (port); serverAddr.sin_addr.s_addr = htonl (INADDR_ANY); //与socket绑定 if (bind(sock_fd, (struct sockaddr *)&serverAddr;, sizeof(serverAddr)) == -1) { perror("bind"); exit(1); } //监听客户请求 if (listen(sock_fd, BACKLOG) == -1) { perror("listen"); exit(1); } printf("listen port= %d\n", MYPORT); /* 初始化客户链表 */ client_list = list_client_create(); if (client_list == NULL) { printf("client list create failed\n"); return ERROR; } client_list->refresh_fd = tcpClient("localhost",port); if( client_list->refresh_fd < 0) { perror("create tmpfd failed\n"); return ERROR; } client_list->maxsock = client_list->refresh_fd; /* 初始化线程属性 */ pthread_attr_init(&pAttr;); /* 设置线程离合状态 */ pthread_attr_setdetachstate(&pAttr;, PTHREAD_CREATE_DETACHED); //创建accept线程 ret = pthread_create(&pid;_accept, &pAttr;, ptr_accept, &sock;_fd); if(ret != 0) { perror("create ptr_accept() failed\n"); return ERROR; } //创建接收数据线程 ret =pthread_create(&pid;_recv, &pAttr;, ptr_recv, NULL); if(ret != 0) { perror("create ptr_recv() failed\n"); return ERROR; } //创建超时检查线程 ret =pthread_create(&pid;_recv, &pAttr;, conn_Timeout, NULL); if(ret != 0) { perror("create conn_Timeout() failed\n"); return ERROR; } return 0; }