网络学习笔记：基于 I/O 复用 select 函数的并发服务器

2018-09-07

本文迁移并整理自个人之前在博客园的博客。

本文是并发服务器系列学习笔记的第 3 篇。

参考资料：《UNIX网络编程卷1》

I/O 复用

概念

在之前博文中，我们有简单介绍了 I/O 复用模型的概念，即可以复用同一个线程去完成多个描述符的 I/O 操作。

I/O 多路复用适用如下场合：
1）当客户处理多个描述符时（一般是交互式输入和网络套接口），必须使用 I/O 复用；
2）当一个客户同时处理多个套接字时，而这种情况是可能的，但很少出现；
3）如果一个 TCP 服务器既要处理监听套接口，又要处理已连接套接口，一般也要用到 I/O 复用；
4）如果一个服务器既要处理 TCP，又要处理 UDP，一般要使用 I/O 复用；
5）如果一个服务器要处理多个服务或多个协议，一般要使用 I/O 复用。

与多进程和多线程技术相比，I/O 多路复用技术的最大优势是系统开销小，系统不必创建进程/线程，也不必维护这些进程/线程，从而大大减小了系统的开销。

描述符就绪条件

可读

满足下列四个条件中的任何一个时，一个套接字准备好读：
1）该套接字接收缓冲区中的数据字节数大于等于套接字接收缓存区低水位标记的当前大小。对于 TCP 和 UDP 套接字而言，缓冲区低水位的值默认为 1。那就意味着，默认情况下，只要缓冲区中有数据，那就是可读的。我们可以通过使用 SO_RCVLOWAT 套接字选项（参见 setsockopt 函数）来设置该套接字的低水位大小。此种描述符就绪（可读）的情况下，当我们使用 read/recv 等对该套接字执行读操作的时候，套接字不会阻塞，而是成功返回一个大于 0 的值（即可读数据的大小）。

2）该连接的读半部关闭（也就是接收了 FIN 的 TCP 连接）。对这样的套接字的读操作，将不会阻塞，而是返回 0（也就是EOF）。

3）该套接字是一个 listen 的监听套接字，并且目前已经完成的连接数不为 0。对这样的套接字进行 accept 操作通常不会阻塞。

4）有一个错误套接字待处理。对这样的套接字的读操作将不阻塞并返回 -1（也就是返回了一个错误），同时把 errno 设置成确切的错误条件。这些待处理错误（pending error）也可以通过指定 SO_ERROR 套接字选项调用 getsockopt 获取并清除。

可写

满足下列四个条件中的任何一个时，一个套接字准备好写：
1）该套接字发送缓冲区中的可用空间字节数大于等于套接字发送缓存区低水位标记时，并且该套接字已经成功连接（UDP 套接字不需要连接）。对于 TCP 和 UDP 而言，这个低水位的值默认为 2048，而套接字默认的发送缓冲区大小是 8K，这就意味着一般一个套接字连接成功后，就是处于可写状态的。我们可以通过 SO_SNDLOWAT 套接字选项（参见 setsockopt 函数）来设置这个低水位。此种情况下，我们设置该套接字为非阻塞，对该套接字进行写操作（如 write、send 等），将不阻塞，并返回一个正值（例如由传输层接受的字节数，即发送的数据大小）。

2）该连接的写半部关闭。对这样的套接字的写操作将会产生 SIGPIPE 信号。所以我们的网络程序基本都要自定义处理 SIGPIPE 信号。因为 SIGPIPE 信号的默认处理方式是程序退出。

3）使用非阻塞的 connect 套接字已建立连接，或者 connect 已经以失败告终。

4）有一个错误的套接字待处理。对这样的套接字的写操作将不阻塞并返回 -1（也就是返回了一个错误），同时把 errno 设置成确切的错误条件。这些待处理的错误也可以通过指定 SO_ERROR 套接字选项调用 getsockopt 获取并清除。

异常处理

如果一个套接字存在带外数据（out-of-band data）或者仍处于带外标记，那么它有异常情况待处理。　

带外数据有时也称为经加速数据（expedited data）。其想法是一个连接的某端发生了重要的事情，而且该端希望迅速通告其对端。这里的“迅速”意味着这种通知应该在已经排队等待发送的任何“普通”（有时称为带内）数据之前发送。也就是说，带外数据被认为具有比普通数据更高的优先级。带外数据并不要求在客户和服务器之间再使用一个连接，而是被映射到已有的连接中。

UDP 套接字不存在带外数据。TCP 并没有真正的带外数据，不过提供了紧急模式（urgent mode）（TCP 数据包头部的紧急指针）。

注意：当某个套接字上发生错误时，它将由 select 标记为既可写又可读。
关于带外数据更详细的请参考《UNIX网络编程卷1》第 24 章。

总结

接收低水位标记和发送低水位标记的目的在于：允许应用进程控制在 select 返回可读或可写条件之前有多少数据可读或有多大空间可用于写。举例来说，如果我们知道除非至少存在 64 个字节的数据，否则我们的应用进程没有任何有效工作可做，那么可以把接收低水位标记设置为 64，以防少于 64 个字节的数据准备好读时 select 唤醒我们。

任何 UDP 套接字只要发送低水位标记小于等于发送缓冲区大小（默认应该总是这种关系）就总是可写的，这是因为 UDP 套接字不需要连接。

下图汇总了导致select返回某个套接字就绪的条件：

I/O 复用 — 运用 Select 函数

函数说明

Select 函数可以在一段指定的时间内，监听用户感兴趣的文件描述符的可读、可写及异常事件。

Select 函数定义

#include <sys/select.h>
#include <sys/time.h>

int select(int maxfdp1, fd_set *readset, fd_set *writeset, fd_set *exceptset, const struct timeval *timeout)
// 返回值：就绪描述符的数目，超时返回0，出错返回-1

Select 函数说明
应用程序调用 select 函数时，通过 readset、writeset、exceptset 传入感兴趣的文件描述符，内核将修改它们来通知应用程序哪些文件描述符已经就绪。

Select 函数参数说明
1）maxfdp1 指定待测试的描述字个数，它的值是待测试的最大描述字加 1（因此把该参数命名为 maxfdp1），描述字 0、1、2…maxfdp1-1 均将被测试。
2）readset、writeset 和 exceptset 指定我们要让内核测试读、写和异常条件的描述字。如果对某一个的条件不感兴趣，就可以把它设为空指针。详细如下：

readset: 可读的文件描述符集合
writeset：可写的文件描述符集合
exceptset: 异常的文件描述符集合
3）timeout 告知内核等待所指定描述字中的任何一个就绪可花多少时间。

Select函数返回值
1）select 函数成功时，返回就绪（可读、可写、异常）文件描述符的总数；
2）如果在超时时间 timeout 内没有任何文件描述符就绪，则 select 函数返回 0；
3）select 函数失败时，返回 -1，并设置 errno；
4）如果在 select 函数等待期间，程序接收到信号，则 select 函数立即返回 -1，并设置 errno 为 EINTR。

fd_set 说明
struct fd_set 可以理解为一个集合，这个集合中存放的是文件描述符，可通过以下四个宏进行设置：

void FD_ZERO(fd_set *fdset);             //清除set的所有位
void FD_SET(int fd, fd_set *fdset);      //设置set的第fd位
void FD_CLR(int fd, fd_set *fdset);      //清除set的第fd位
void FD_ISSET(int fd, fd_set *fdset);    //测试set的第fd为是否被设置

timeval 说明
timeout 告知内核等待所指定描述字中的任何一个就绪可花多少时间。其 timeval 结构用于指定这段时间的秒数和微秒数。

struct timeval{
   long tv_sec;   //seconds
   long tv_usec;  //microseconds
};

这个参数有三种可能：
1）永远等待下去：仅在有一个描述字准备好 I/O 时才返回。为此，把该参数设置为空指针 NULL。
2）等待一段固定时间：在有一个描述字准备好 I/O 时返回，但是不超过由该参数所指向的 timeval 结构中指定的秒数和微秒数。
3）根本不等待：检查描述字后立即返回，这称为轮询（polling）。为此，该参数必须指向一个 timeval 结构，而且其中的定时器值（由该结构指定的秒数和微秒数）必须为 0。

修订的 str_cli 函数

我们这里修订的是以前博文基于多进程并发服务器的 str_cli 函数。在那篇博文，它的定义如下：

void
str_cli(FILE *fp, int sockfd)
{
    char sendline[MAXLINE], recvline[MAXLINE];

    while (Fgets(sendline, MAXLINE, fp) != NULL) {

        Writen(sockfd, sendline, strlen(sendline));

        if (Readline(sockfd, recvline, MAXLINE) == 0)
            err_quit("str_cli: server terminated prematurely");

        Fputs(recvline, stdout);
    }
}

早先 str_cli 版本阻塞于 fgets 调用，这个版本的将改为阻塞于 select 调用，或是等待标准输入可读，或是等待套接字可读。下图展示了调用 select 所处理的各种条件：

客户的套接字上的三个条件处理如下：
1）如果对端 TCP 发送数据，那么该套接字变为可读，并且 read 返回一个大于 0 的值（即读入数据的字节数）；
2）如果对端 TCP 发送一个 FIN（对端进程终止），那么套接字变为可读，并且 read 返回 0（EOF）；
3）如果对端 TCP 发送一个 RST（对端主机崩溃并重新启动），那么该套接字变为可读，并且 read 返回 -1，而 errno 中含有确切的错误码。

新版本的 str_cli 函数如下：

void
str_cli_select(FILE *fp, int sockfd)
{
    int maxfdp, stdineof;
    fd_set rset;
    char buf[MAXLINE];
    int n;

    stdineof = 0;
    FD_ZERO(&rset);
    for(;;)
    {
        if(stdineof == 0)
            FD_SET(fileno(fp), &rset);
        FD_SET(sockfd, &rset);
        maxfdp = max(fileno(fp), sockfd) + 1;
        Select(maxfdp, &rset, NULL, NULL, NULL);

        if(FD_ISSET(sockfd, &rset))    /* socket is readable */
        {
            if((n == Read(sockfd, buf, MAXLINE)) == 0)
            {
                if(stdineof == 1)
                    return;    /* normal termination */
                else
                    err_quit("str_cli_select: server terminated prematurely");
            }
            Write(fileno(stdout), buf, n);
        }

        if(FD_ISSET(fileno(fp), &rset))    /* input is readable */
        {
            if((n = Read(fileno(fp), buf, MAXLINE)) == 0)
            {
                stdineof = 1;
                Shutdown(sockfd, SHUT_WR);    /* send FIN */
                FD_CLR(fileno(fp), &rset);
                continue;
            }
            Writen(sockfd, buf, n);
        }
    }
}

上述程序中的 stdineof 是一个初始化为 0 的新标志。只要该标识为 0，每次主循环中我们总是 select 标准输入的可读性。

1929：当我们在套接字上读到 EOF 时，如果我们已在标准输入上遇到 EOF，那就是正常的终止，于是函数返回；但是如果我们在标准输入上没有遇到 EOF，那么服务器进程已过早终止。
3141：当我们在标准输入上碰到 EOF 时，我们把新标志 stdineof 置为 1，并把第二个参数指定为 SHUT_WR 来调用 shutdown 以发送 FIN。

shutdown 函数

终止网络连接的通常方法是调用 close 函数。不过 close 有两个限制，却可以用shutdown来避免：
1）close 把描述符的引用计数减 1，仅在该计数变为 0 是才关闭套接字。使用 shutdown 可以不管引用计数就激发 TCP 的正常连接终止序列。
2）close 终止读和写两个方向的数据传送。既然 TCP 连接是全双工的，有时候我们需要告知对端我们已经完成了数据发送，即使对端仍有数据要发送给我们。

shutdown 函数定义如下：

1
2
3

#include <sys/socket.h>
int shutdown(int sockfd, int howto);
// 返回：若成功则返回0，若出错则返回-1

该函数的行为依赖于 howto 参数的值：
1）SHUT_RD：关闭连接的读这一半——套接字中不再有数据可接收，而且套接字接收缓冲区中的现有数据都被丢弃。进程不能再对这样的套接字调用任何读函数。对一个 TCP 套接字这样调用 shutdown 函数后，由该套接字接收来的来自对端的任何数据都被确认，然后悄然丢弃。
2）SHUT_WR：关闭连接的写这一半——对于 TCP 套接字，这称为半关闭（half close）。当前留在套接字发送缓冲区的数据将被发送掉，后跟 TCP 的正常连接终止序列。
3）SHUT_RDWR：连接的读半部和写半部都关闭——这与调用 shutdown 两次等效：第一次调用 SHUT_RD，第二次调用指定 SHUT_WR。

修订 TCP 回射程序

服务端

我们这里修订的是以前博文基于多进程并发服务器的服务端程序。在那篇博文，它的内容如下：

#include "unp.h"

void
sig_chld(int signo)
{
    pid_t pid;
    int stat;
    while((pid = waitpid(-1, &stat, WNOHANG)) > 0)
        printf("child %d terminated\n", pid);
    return;
}

int
main()
{
    int listenfd, connfd;
    pid_t childpid;
    socklen_t clilen;
    struct sockaddr_in cliaddr, servaddr;

    void sig_chld(int);

    listenfd = Socket(AF_INET, SOCK_STREAM, 0);
    
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(SERV_PORT);

    Bind(listenfd, (SA *) &servaddr, sizeof(servaddr));
    Listen(listenfd, LISTENQ);
    Signal(SIGCHLD, sig_chld);    /* must call waitpid */
    for( ; ; )
    {
        clilen = sizeof(cliaddr);
        if((connfd = accept(listenfd, (SA *)&cliaddr, &clilen)) < 0)
        {
            if(errno == EINTR)
                continue;
            else
                err_sys("accept error");
        }
        if((childpid = Fork()) == 0)
        {
            Close(listenfd);
            str_echo(connfd);
            Close(connfd);
            exit(0);
        }
        Close(connfd);
    }

    exit(0);
}

在这里，我们将把上边程序改写成使用 select 来处理任意个客户的单进程程序。

完整的服务器程序如下：

#include "unp.h"

int
main()
{
    int i, maxi, maxfd, listenfd, connfd, sockfd;
    int nready, client[FD_SETSIZE];
    ssize_t n;
    fd_set rset, allset;
    char buf[MAXLINE];
    socklen_t clilen;
    struct sockaddr_in cliaddr, servaddr;

    listenfd = Socket(AF_INET, SOCK_STREAM, 0);
    
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(SERV_PORT);

    Bind(listenfd, (SA *) &servaddr, sizeof(servaddr));
    Listen(listenfd, LISTENQ);
    
    maxfd = listenfd;    /* initialize */
    maxi = -1;            /* index into client[] array */
    for(i = 0; i < FD_SETSIZE; i++)
        client[i] = -1;    /* -1 indicates available entry */
    FD_ZERO(&allset);
    FD_SET(listenfd, &allset);
    
    for( ; ; )
    {
        rset = allset;        /* structure assignment */
        nready = Select(maxfd + 1, &rset, NULL, NULL, NULL);

        if(FD_ISSET(listenfd, &rset))    /* new client connection */
        {
            clilen = sizeof(cliaddr);
            connfd = Accept(listenfd, (SA *) &cliaddr, &clilen);

            for(i = 0; i < FD_SETSIZE; i++)
            {
                if(client[i] < 0)
                {
                    client[i] = connfd;    /* save descriptor */
                    break;
                }
            }
            if(i == FD_SETSIZE)
                err_quit("too many clients");
        
            FD_SET(connfd, &allset);    /* add new descriptor to set */
            if(connfd > maxfd)
                maxfd = connfd;    /* for select */
            if(i > maxi)
                maxi = i;        /* max index in client[] array */

            if(--nready <= 0)
                continue;        /* no more readable descriptor */
        }

        for(i = 0; i <= maxi; i++)    /* check all clients for data */
        {
            if((sockfd = client[i]) < 0)
                continue;
            if(FD_ISSET(sockfd, &rset))
            {
                if((n = Read(sockfd, buf, MAXLINE)) == 0)
                {
                    /* connection closed by client */
                    Close(sockfd);
                    FD_CLR(sockfd, &allset);
                    client[i] = -1;
                }
                else
                    Writen(sockfd, buf, n);

                if(--nready <= 0)
                    break;    /* no more available descriptors */
            }
        }
    }

    exit(0);
}

上边程序主要逻辑可用流程图表示如下：
图片来源

客户端程序

相应的客户端程序如下：

#include "unp.h"

void
str_cli_select(FILE *fp, int sockfd);

int
main(int argc, char **argv)
{
    int sockfd;
    struct sockaddr_in servaddr;

    if(argc != 2)
        err_quit("usage: tcpcli <IPaddress>");

    sockfd = Socket(AF_INET, SOCK_STREAM, 0);
    
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(SERV_PORT);
    Inet_pton(AF_INET, argv[1], &servaddr.sin_addr.s_addr);

    Connect(sockfd, (SA *) &servaddr, sizeof(servaddr));

    str_cli_select(stdin, sockfd);

    exit(0);
}

void
str_cli_select(FILE *fp, int sockfd)
{
    int maxfdp, stdineof;
    fd_set rset;
    char buf[MAXLINE];
    int n;

    stdineof = 0;
    FD_ZERO(&rset);
    for(;;)
    {
        if(stdineof == 0)
            FD_SET(fileno(fp), &rset);
        FD_SET(sockfd, &rset);
        maxfdp = max(fileno(fp), sockfd) + 1;
        Select(maxfdp, &rset, NULL, NULL, NULL);

        if(FD_ISSET(sockfd, &rset))    /* socket is readable */
        {
            if((n == Read(sockfd, buf, MAXLINE)) == 0)
            {
                if(stdineof == 1)
                    return;    /* normal termination */
                else
                    err_quit("str_cli_select: server terminated prematurely");
            }
            Write(fileno(stdout), buf, n);
        }

        if(FD_ISSET(fileno(fp), &rset))    /* input is readable */
        {
            if((n = Read(fileno(fp), buf, MAXLINE)) == 0)
            {
                stdineof = 1;
                Shutdown(sockfd, SHUT_WR);    /* send FIN */
                FD_CLR(fileno(fp), &rset);
                continue;
            }
            Writen(sockfd, buf, n);
        }
    }
}

状态跟踪

现在我们来跟踪一下服务器端的状态。

下图给出了第一个客户建立连接前服务器的状态：

服务器有单个监听描述符，我们用一个圆点来表示。

服务器只维护一个读描述符集，如图 6-15 所示。假设服务器是在前台启动的，那么描述符 0、1 和 2 将分别被设置为标准输入、标准输出和标准错误输出。可见监听套接字的第一个可用描述符是 3。图 6-15 还展示了一个名为 client 的整形数组，它含有每个客户的已连接套接字描述符。该数组的所有元素都被初始化为 -1。

描述符中的唯一的非 0 项是表示监听套接字的项，因此 select 的第一个参数将为 4。
当一个客户与服务器建立连接时，监听描述符变为可读，我们的服务器于是调用 accept。在本例的假设下，有 accept 返回的新的已连接描述符将是 4。图 6-16 展示了从用户到服务器的连接：

从现在起，我们的服务器必须在其 client 数组中记住每个新的已连接描述符，并把它加到描述符集中去。图 6-17 展示了这样更新后的数据结构：

稍后，第二个客户与服务器建立连接，图 6-18 展示了这种情形：

新的已连接描述符（假设是 5）必须被记住，从而给出如图 6-19 所示的数据结构：

我们接着假设第一个客户终止它的连接。该客户的 TCP 发送一个 FIN，使得服务器中的描述符 4 变为可读。当服务器读这个已连接套接字时，read 将返回 0。我们于是关闭该套接字并相应地更新数据结构：把 client[0] 的值置为 -1，把描述符集中描述符 4 的为设置为 0，如图 6-20 所示。注意，maxfd 的值没有改变。

总之，当有客户到达时，我们在 client 数组中的第一个可用项（即值为 -1 的第一个项）中记录其已连接套接字的描述符。我们还必须把这个已连接描述符加到读描述符集中。变量 maxi 是 client 数组当前使用项的最大下标，而变量 maxfd（加1之后）是 select 函数第一个参数的当前值。对于本服务器所能处理的最大客户数目的限制是以下两个之中的较小者：FD_SETSIZE 和内核允许本进程打开的最大描述符数。

Select 函数存在的问题

这段摘自博文 select、poll、epoll 之间的区别总结[整理]。

select 的调用过程如下所示：
图片来源

1）使用 copy_from_user 从用户空间拷贝 fd_set 到内核空间
2）注册回调函数 __pollwait
3）遍历所有 fd，调用其对应的 poll 方法（对于 socket，这个 poll 方法是 sock_poll，sock_poll 根据情况会调用到 tcp_poll,udp_poll 或者 datagram_poll）
4）以 tcp_poll 为例，其核心实现就是 __pollwait，也就是上面注册的回调函数。
5）__pollwait 的主要工作就是把 current（当前进程）挂到设备的等待队列中，不同的设备有不同的等待队列，对于 tcp_poll 来说，其等待队列是 sk->sk_sleep（注意把进程挂到等待队列中并不代表进程已经睡眠了）。在设备收到一条消息（网络设备）或填写完文件数据（磁盘设备）后，会唤醒设备等待队列上睡眠的进程，这时 current 便被唤醒了。
6）poll 方法返回时会返回一个描述读写操作是否就绪的 mask 掩码，根据这个 mask 掩码给 fd_set 赋值。
7）如果遍历完所有的 fd，还没有返回一个可读写的 mask 掩码，则会调用 schedule_timeout 是调用 select 的进程（也就是 current）进入睡眠。当设备驱动发生自身资源可读写后，会唤醒其等待队列上睡眠的进程。如果超过一定的超时时间（schedule_timeout指定），还是没人唤醒，则调用 select 的进程会重新被唤醒获得 CPU，进而重新遍历 fd，判断有没有就绪的 fd。
8）把 fd_set 从内核空间拷贝到用户空间。

select的几大缺点：
1）每次调用 select，都需要把 fd 集合从用户态拷贝到内核态，这个开销在 fd 很多时会很大
2）同时每次调用 select 都需要在内核遍历传递进来的所有 fd，这个开销在 fd 很多时也很大
3）select 支持的文件描述符数量太小了，默认是 1024