Socket

09 select并发服务器

day09

本章目标

  • select
  • 读、写、异常事件发生条件
  • 用select改进回射服务器程序

select

再复习和补充一点select的参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#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);

#include <sys/select.h>

int pselect(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, const struct timespec *timeout, const sigset_t *sigmask);

  • 参数一：int nfds

    因为select的对感兴趣的fd取值范围为[0, nfds)，前闭后开区间，所以需要最大的文件描述符+1，才能覆盖整个感兴趣的文件描述符号的范围。

  • 参数二：fd_set *readfds

    可读事件集合，添加感兴趣可读fd

  • 参数三：fd_set *writefds

    可写事件集合，添加感兴趣可写fd

  • 参数四：fd_set *exceptfds

    异常事件集合

  • 参数五：struct timeval *timeout

    超时时间，在响应的超时时间内没有发生任何一个I/O事件(读/写事件)，则select返回0，指示超时。出现错误select返回-1.

监测到多个事件，需要遍历，readfds、writefds、exceptfds这些集合来一一处理这些事情，则将select实现的服务器称为并发服务器。因为没有能力并行处理这些I/O事件，只能一一遍历处理这些事件。所以称为并发服务器，而不是并行服务器。并发服务器不能长时间处理这些事件，因为长时间处理则不能去执行其他的动作。且事件间的处理间隔时间也不能太长。无法充分利用多核处理器。使用多进程select或者多线程处理这些事件。

select可以同时检测网络I/O和标准I/O事件。那个事件产生则处理那个事件。不会因为某一个阻塞导致另一个无法处理。

读、写、异常事件发生条件

  • 可读
    • 套接口缓冲区有数据可读，底层本质：如果缓冲区的数据超过一定的范围才会通知有数据可读
    • 连接的读一般关闭，即接收到FIN段，读操作返回0
    • 如果是监听套接口，已完成连接队列不为空时。
    • 套接口上发生了一个错误待处理，错误可以通过getsockopt指定SO_ERROR选项来获取
  • 可写
    • 套接口发送缓冲区有空间容纳数据
    • 连接的写一半关闭，即收到RST段之后，再次调用write操作
    • 套接口上发生一个错误待处理，错误可以通过getsockopt指定SO_ERROR选项来获取
  • 异常
    • 套接口存在带外数据。

并发服务器

对上一次的服务器端通过select使用一个进程实现。并发服务器的连接客户端的限制，FD_SETSZIE大小和一个进程当中能够打开的I/O的最大数目。FD_SETSIZE不能被更改(非要更改，需要更改内核代码，并进行重新编译)，而一个进程的可以打开的最大的I/O数目可以更改，通过ulimit -n 1000命令更改一个进程可以打开的最大I/O数目为1000。FD_SETSIZE默认情况下和进程能够打开的最大I/O数目为1024个。

核心代码(部分代码)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
int i = 0;
int num = 0;
int maxi = 0; //最大不空闲的套接口下标
int nready = 0;
int maxfd = listenfd;
int conn[FD_SETSIZE] = {0};
fd_set rset;
fd_set allset;
FD_ZERO(&rset);
FD_ZERO(&allset);
FD_SET(listenfd, &allset);
struct sockaddr_in sockname;
socklen_t addrlen = sizeof(sockname);//一定要有初始值，为结构体大小，输入输出型参数。
for(i = 0; i < FD_SETSIZE; i++)//初始化conn数组
{
        conn[i] = -1;
}
while(1)//一个进程处理并发
{
        rset = allset;
        nready = select(maxfd + 1, &rset, NULL, NULL, NULL);
        if(nready == -1)
        {
                //errno 是记录系统的最后一次错误代码。
                //EINTR错误的产生：当阻塞于某个慢系统调用的一个进程捕获某个信号且相应信号处理函数返回时，该系统调用可能返回一个EINTR错误。
                if(errno == EINTR)//如果被信号中断则继续
                        continue;
                ERR_EXIT("select");//其他错误退出
        }
        else if(nready == 0)
        {
                //时间超时,因为这里没有设置超时时间，理论不会出现nready == 0的情况
                continue;
        }
        if(FD_ISSET(listenfd, &rset))//判断监听套接口是否在集合当中,处理监听套接口的数据
        {
                int ret = -1;
                if((ret = accept(listenfd, (struct sockaddr *)&peeraddr, (socklen_t *)&peerlen)) < 0)
                {
                        ERR_EXIT("accept");
                }
                for(i = 0; i < FD_SETSIZE; i++)
                {
                        if(conn[i] < 0)
                        {
                                conn[i] = ret;
                                if(i > maxi)
                                        maxi = i;//最大不空闲套接口下标，优化遍历时间，提高效率
                                break;
                        }
                }
                if(i == FD_SETSIZE)
                {
                        //stderr标准错误
                        fprintf(stderr, "链接的客户端太多了... ...");
                        exit(EXIT_FAILURE);
                }
                //获取远程的地址
                if(0 > getpeername(ret , (struct sockaddr *)&sockname, (socklen_t*)&addrlen))
                {
                        perror("getsockname");
                        exit(EXIT_FAILURE);//退出程序
                }
                printf("远程的IP: %s, Port: %d\n", inet_ntoa(sockname.sin_addr), ntohs(sockname.sin_port));
                FD_SET(ret, &allset);//添加新的套接口到allset中下一次select会检测这个套接口
                if(maxfd < ret)//需要更新最大描述符
                        maxfd = ret;
                if(--nready <= 0)//已经处理完了监听套接口的事件，把nready减一操作，如果等于零，说明没有要处理的事件了，为了提高效率直接continue,如果不小于等于零，说明还有其他事件，即为连接客户端的事件，>然后对客户端的事件进行处理，
                        continue;
        }
        for(i = 0; i <= maxi; i++)
        {
                if(conn[i] == -1)
                        continue;
                if(FD_ISSET(conn[i], &rset))//判断是否在读的集合当中
                {
                        //读取数据
                        char recvbuf[1024] = {0};
                        int ret = readline(conn[i], recvbuf, 1024);
                        if(ret == -1)
                        {
                                ERR_EXIT("readline");
                        }
                        if(0 == ret)
                        {
                                printf("对方关闭了Socket readline\n");
                                FD_CLR(conn[i], &allset);//从allset中清除
                                conn[i] = -1;//标志为-1
                                continue;
                        }
                        printf("服务器接收到的数据:");
                        fputs(recvbuf, stdout);
                        writen(conn[i], recvbuf, strlen(recvbuf));//数据回射回去:
                        if(--nready <= 0)
                                break;
                }
        }
        FD_ZERO(&rset);//清空rset
}