【网络】（十）select的限制与解决方法poll

在前面的《第七篇》和《第八篇》文章中已经使用select函数对服务器和客户端进行了重写，但这有个问题存在，如果有非常多的客户端需要连接到服务器，那很可能连接到1020个客户端时，服务端就会报错了，原因在于用select实现的并发服务器能够达到的并发数，受两方面限制

一个进程能打开的最大文件描述符限制，这可通过调整内核参数修改！

select中的fd_set集合容量的限制（FD_SETSIZE）,它不易修改，否则需要重新编译内核！

2、解决方案

基于select函数的上述限制，所以当客户端数量比较大的时候，就得抛弃它使用poll函数！

poll函数原型

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

select函数原型

int select(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, struct timeval *timeout)

poll与select不同在于，它舍弃了fd_set集合，采用了struct pollfd结构体数组，这就避开了fd_set集合大小的限制！

完整源码

handle.h

#ifndef _HANDLE_H_
#define _HANDLE_H_

#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <stdlib.h>
#include <stdio.h>
#include <arpa/inet.h>
#include <stdbool.h>
#include <sys/resource.h>

#define handle_error(msg)   \
do{perror(msg);exit(EXIT_FAILURE);}while(0)

/*
* init -- 初始化环境设置，设置最大文件描述富
* 参数：
@maxfd:最大文件描述符
*/
void init(int maxfd);

/*
* readn -- 读取指定字节的数据后返回
* 参数：
@fd: 文件描述符
@buf: 数据接收缓冲区
@count: 要读取的字节数
* 返回值：
读取成功，返回值应该等于count；失败，返回-1
*/
ssize_t readn(int fd, void *buf, size_t count);

/*
* writen -- 写入指定字节的数据后返回
* 参数：
@fd: 文件描述符
@buf: 数据缓冲区
@count: 要写入的字节数
* 返回值：
写入成功，返回值应该等于count；失败，返回-1
*/
ssize_t writen(int fd, const void *buf, size_t count);

/*
* recv_peek -- 从套接字接收缓冲区中接收数据，但并不从缓冲去中清除该数据
* 参数：
@fd: 文件描述符
@buf: 数据接收缓冲区
@count: 要读取的字节数
* 返回值：
读取成功，返回读取到的字节数；失败，返回-1
*/
ssize_t recv_peek(int sockfd, void *buf, size_t len);

/*
* recvline -- 从套接字接收缓冲区中接收一行数据，该数据以'\n'结尾
* 参数：
@sockfd: 套接字描述符
@buf: 数据接收缓冲区
@maxlen: 缓冲区长度
* 返回值：
读取成功，返回读取到的字节数；失败，返回-1
*/
ssize_t recvline(int sockfd, void *buf, size_t maxlen);

#endif

handle.c

#include "handle.h"

/*
* init -- 初始化环境设置，设置最大文件描述富
* 参数：
@maxfd:最大文件描述符
*/
void init(int maxfd)
{
struct rlimit limit;
if(getrlimit(RLIMIT_NOFILE,&limit) == -1)
handle_error("getrlimit");

limit.rlim_cur = maxfd;
if(setrlimit(RLIMIT_NOFILE,&limit) == -1)
handle_error("setrlimit");

if(getrlimit(RLIMIT_NOFILE,&limit) == -1)
handle_error("getrlimit");
printf("getrlimit = %d\n",(int)limit.rlim_cur);
}

/*
* readn -- 读取指定字节的数据后返回
* 参数：
@fd: 文件描述符
@buf: 数据接收缓冲区
@count: 要读取的字节数
* 返回值：
读取成功，返回值应该等于count；失败，返回-1
*/
ssize_t readn(int fd, void *buf, size_t count)
{
if((fd < 0) || (buf == NULL) || (count < 0))
return -1;
size_t nleft = count;   //剩余字节数
ssize_t nread = 0;      //已读字节数
char *pbuf = (char*)buf;
while(nleft > 0)
{
if((nread = read(fd, pbuf, nleft)) < 0)
{
if(errno == EINTR)
continue;
return -1;
}
else if (nread == 0)
return count - nleft;
pbuf += nread;
nleft -= nread;
}
return count;
}

/*
* writen -- 写入指定字节的数据后返回
* 参数：
@fd: 文件描述符
@buf: 数据缓冲区
@count: 要写入的字节数
* 返回值：
写入成功，返回值应该等于count；失败，返回-1
*/
ssize_t writen(int fd, const void *buf, size_t count)
{
if((fd < 0) || (buf == NULL) || (count < 0))
return -1;
size_t nleft = count;   //剩余字节数
ssize_t nwritten = 0;      //已发送字节数
char *pbuf = (char*)buf;
while (nleft > 0)
{
if((nwritten = write(fd, pbuf, nleft)) < 0)
{
if(errno == EINTR)
continue;
return -1;
}
else if(nwritten == 0)
continue;
pbuf += nwritten;
nleft -= nwritten;
}
return count;
}

/*
* recv_peek -- 从套接字接收缓冲区中接收数据，但并不从缓冲去中清除该数据
* 参数：
@fd: 文件描述符
@buf: 数据接收缓冲区
@count: 要读取的字节数
* 返回值：
读取成功，返回读取到的字节数；失败，返回-1
*/
ssize_t recv_peek(int sockfd, void *buf, size_t len)
{
while(true)
{
int iret = recv(sockfd, buf, len, MSG_PEEK);
if(iret == -1 && errno == EINTR)    //如果失败是因为信号中端，那么就重新再试
continue;
return iret;
}
}

/*
* recvline -- 从套接字接收缓冲区中接收一行数据，该数据以'\n'结尾
* 参数：
@sockfd: 套接字描述符
@buf: 数据接收缓冲区
@maxlen: 缓冲区长度
* 返回值：
读取成功，返回读取到的字节数；失败，返回-1
*/
ssize_t recvline(int sockfd, void *buf, size_t maxlen)
{
int iret = 0;
int nread = 0;  //已读数据
char *pbuf = (char*)buf;
int nleft = maxlen; //剩余字符
while(true)
{
iret = recv_peek(sockfd, pbuf, nleft);
if(iret < 0)
return iret;    //读取失败
else if(iret == 0)
return iret;    //对方关闭套接口

nread = iret;
if(nread > nleft)       //已经读取到的数据只可能小于或者等于剩余的数据
exit(EXIT_FAILURE);

for(int i = 0; i < nread; i++)
{
if(pbuf[i] == '\n')
{
iret = readn(sockfd, pbuf, i+1);    //从缓冲区中读走包括\n在内的数据
if(iret != i+1)
exit(EXIT_FAILURE);     //没有读取都i+1个数据，说明失败
return iret;                //读取到\n返回
}
}

//在当前读到的数据中没有发现\n,那么先将这部分数据从缓冲区中读走，然后接着偷窥后面的数据
nleft -= nread;
iret = readn(sockfd, pbuf, nread);
if(iret != nread)
exit(EXIT_FAILURE);
pbuf += nread;
}
return -1;
}

服务端源码

server.c

#include <string.h>
#include <signal.h>
#include <poll.h>

#include "handle.h"

#define FD_MAXSIZE 2048     //最大客户端数量

//为了防止SIGPIPE信号产生而终止了进程，所以捕获此信号
void handle_sigpipe(void)
{
struct sigaction act;
act.sa_handler = SIG_IGN;       //忽略SIGPIPE信号
act.sa_flags = 0;
sigemptyset(&act.sa_mask);
if(sigaction(SIGPIPE, &act, NULL) == -1)
handle_error("sigaction");
}

int main(void)
{
init(FD_MAXSIZE);
handle_sigpipe();       //捕获SIGPIPE信号

int sk_fd = socket(AF_INET, SOCK_STREAM , IPPROTO_TCP);
if(sk_fd < 0)
handle_error("socket");

//使用REUSEADDR，不必等待TIME_WAIT 状态消失，就可以重新使用端口
int on = 1;
if(setsockopt(sk_fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) < 0)
{
close(sk_fd);
handle_error("setsockopt");
}

struct sockaddr_in sr_addr;
memset(&sr_addr,0,sizeof(sr_addr));
sr_addr.sin_family = AF_INET;
sr_addr.sin_port = htons(5188);
sr_addr.sin_addr.s_addr = htonl(INADDR_ANY);

if(bind(sk_fd, (struct sockaddr*)&sr_addr, sizeof(sr_addr)) < 0)
{
close(sk_fd);
handle_error("bind");
}

//被动套接字
if(listen(sk_fd, SOMAXCONN) < 0)     //内核为此套接字排队的最大连接数由SOMAXCONN宏指定
{
close(sk_fd);
handle_error("listen");
}

//使用poll函数实现单进程并发服务器
int maxindex = 0;       //client_sk数组最大已用位置
int maxindex_last = maxindex;  //maxindex上一次的值

int nready = 0;
struct pollfd client_sk[FD_MAXSIZE];      //存储客户端连接信息
for(int i = 0; i < FD_MAXSIZE; i++)
{
client_sk[i].fd = -1;
}
client_sk[0].fd = sk_fd;
client_sk[0].events = POLLIN;   //监听的事件

while (true)
{
nready = poll(client_sk, maxindex+1, -1);        //超时10s
if(nready == 0)
{
printf("poll time out!\n");
continue;
}else if (nready == -1)
{
if(errno == EINTR)
continue;
handle_error("poll");
}

if(client_sk[0].revents & POLLIN)   //检测到监听套接字事件
{
//调用accept建立连接
struct sockaddr_in cl_addr;
socklen_t cl_length = sizeof(cl_addr);
memset(&cl_addr,0,sizeof(cl_addr));

int ac_sk = accept(sk_fd, (struct sockaddr *)&cl_addr, &cl_length);
if(ac_sk < 0)
{
if(errno == EINTR)
continue;
handle_error("accept");
}

int i;
for(i = 0; i < FD_MAXSIZE; i++)
{
if(client_sk[i].fd < 0)
{
client_sk[i].fd = ac_sk;  //加入到监听集合中
client_sk[i].events = POLLIN;
if(maxindex < i)
{
maxindex_last = maxindex;   //保存历史记录
maxindex = i;       //更新最大已用位置
}
break;
}
}
if(i == FD_MAXSIZE)     //client_sk数组空间已满，客户端太多
{
fprintf(stderr, "Client too many!");
exit(EXIT_FAILURE);
}
printf("Connect ip = %s\tport = %d\n",inet_ntoa(cl_addr.sin_addr),ntohs(cl_addr.sin_port));

if(--nready <= 0)   //如果除了监听套接字外没有其他套接字发生事件
continue;
}

//客户端连接套接字发生事件
for(int i = 0; i <= maxindex; i++)
{
int conn_sk = client_sk[i].fd;
if(client_sk[i].revents & POLLIN)
{
//有数据可以接收
char recvbuf[1024];
memset(recvbuf,0,sizeof(recvbuf));
int iret = recvline(conn_sk,recvbuf,sizeof(recvbuf));    //获取包数据长度
if(iret == -1)
handle_error("read");
else if(iret == 0)
{
//客户端关闭，删除对应的连接套接字
if(i == maxindex)
{
maxindex = maxindex_last;   //如果删除的描述符的位置是当前最大已用位置
}
client_sk[i].fd = -1;
printf("Client was closed!\n");
close(conn_sk);
}
fputs(recvbuf,stdout);
sleep(4);
writen(conn_sk, recvbuf, strlen(recvbuf));    //回传数据

if(--nready <= 0)       //判断是否处理完
break;
}
}

}
for(int i = 0; i <= maxindex; i++)
{
close(client_sk[i].fd);
}
close(sk_fd);
return 0;
}

客户端源码

client.c

#include <string.h>
#include <poll.h>

#include "handle.h"

#define FD_MAXSIZE 2

void client_handler(int sk_fd)
{
//使用poll函数实现单进程并发服务器
int nready = 0;
struct pollfd client_sk[FD_MAXSIZE];      //存储客户端连接信息
for(int i = 0; i < FD_MAXSIZE; i++)
{
client_sk[i].fd = -1;
}

int fd_stdin = fileno(stdin);   //获得标准输入的描述符，防止输入重定向
bool stdinflag = true;      //表示是否需要select监听stdin

client_sk[0].fd = fd_stdin;
client_sk[0].events = POLLIN;
client_sk[1].fd = sk_fd;
client_sk[1].events = POLLIN;

while(true)
{
if(!stdinflag)
{
client_sk[0].fd = -1;
}

nready = poll(client_sk, 2, -1);
if(nready == -1)
{
if(errno == EINTR)
continue;
handle_error("poll");
}else if(nready == 0)
{
printf("poll time out!\n");
continue;
}
if(client_sk[0].revents & POLLIN)    //标准输入产生读事件
{
char sendbuf[1024] = {0};
if(fgets(sendbuf, sizeof(sendbuf), stdin) == NULL)
{
//如果从标准输入接收到EOF，表示我们想关闭发送端口
stdinflag = false;      //让select不再对stdin进行监听
printf("Ctrl + D\n");
shutdown(sk_fd, SHUT_WR);   //只关闭写端，对方会收到FIN段
}
else
{
writen(sk_fd, sendbuf, strlen(sendbuf));   //发送数据
memset(&sendbuf, 0, sizeof(sendbuf));
}
}

if(client_sk[1].events & POLLIN)   //套接口产生了读事件
{
char recvbuf[1024] = {0};
int iret = recvline(sk_fd, recvbuf, sizeof(recvbuf)); //接收包数据长度
if(iret == -1)
handle_error("read");
else if(iret == 0)
{
printf("Server was closed!\n");
break;
}
fputs(recvbuf, stdout);
memset(&recvbuf, 0, sizeof(recvbuf));
}

}
}

int main(void)
{
int sk_fd;
sk_fd = socket(AF_INET, SOCK_STREAM , IPPROTO_TCP);
if(sk_fd < 0)
handle_error("socket");
struct sockaddr_in sr_addr;
memset(&sr_addr,0,sizeof(sr_addr));
sr_addr.sin_family = AF_INET;
sr_addr.sin_port = htons(5188);
sr_addr.sin_addr.s_addr = inet_addr("127.0.0.1");

if(connect(sk_fd, (struct sockaddr*)&sr_addr, sizeof(sr_addr)) < 0)
{
close(sk_fd);
handle_error("connect");
}

client_handler(sk_fd);
close(sk_fd);
return 0;
}

压力测试客户端

conncilent.c

发起大量的连接请求，测试服务器的容量！

#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <string.h>
#include <arpa/inet.h>
#include <unistd.h>

#include "handle.h"

#define FD_MAXSIZE 4096

//发起4096次连接请求
int main(void)
{
init(FD_MAXSIZE);
int sk_fd;
for(int i = 0; i < FD_MAXSIZE; i++)
{
sk_fd = socket(AF_INET, SOCK_STREAM , IPPROTO_TCP);
if(sk_fd < 0)
handle_error("socket");
struct sockaddr_in sr_addr;
memset(&sr_addr,0,sizeof(sr_addr));
sr_addr.sin_family = AF_INET;
sr_addr.sin_port = htons(5188);
sr_addr.sin_addr.s_addr = inet_addr("127.0.0.1");

if(connect(sk_fd, (struct sockaddr*)&sr_addr, sizeof(sr_addr)) < 0)
{
handle_error("connect");
}
printf("Count = %d: Connect ip = %s\tport = %d\n",i,inet_ntoa(sr_addr.sin_addr),ntohs(sr_addr.sin_port));
}

return 0;
}