UNIX网络编程卷一 学习笔记 第三十章 客户/服务器程序设计范式

开发一个Unix服务器程序时，我们本书做过的进程控制：
1.迭代服务器（iterative server），它的适用情形极为有限，因为这样的服务器在完成对当前客户的服务前无法处理已等待服务的新客户。

2.并发服务器（concurrent server），为每个客户调用fork派生一个子进程。传统上大多Unix服务器程序属于这种类型。

3.使用select函数处理多个客户的单个TCP服务器进程。

4.并发服务器的修改，为每个客户创建一个线程，取代一个进程。

我们本章探究并发服务器的另两类变体：
1.预先派生子进程，让服务器在启动阶段调用fork创建一个子进程池，每个客户请求由当前可用子进程池中的某个闲置子进程处理。

2.预先创建线程，让服务器在启动阶段创建一个线程池，每个客户由当前可用线程池中的某个闲置线程处理。

我们本章审视预先派生子进程和预先创建线程这两种类型的细节：如果池中进程和线程不够多怎么办？如果池中进程和线程过多怎么办？父进程与子进程之间以及各个线程之间怎样彼此同步？

客户程序的编写比服务器程序容易些，因为客户中进程控制要少得多。

本章会介绍9个不同的服务器程序设计范式，并用同一个客户程序访问这些服务器以便相互比较。我们的客户/服务器交互情形在web应用中是典型的，客户向服务器发送一个小请求，然后服务器响应以返回给客户的数据。

我们将针对每个服务器运行同一客户程序的多个实例，以测量服务某个固定数目的客户所需的CPU时间，以下是结果：

上图中测量的仅仅是用于进程控制所需的CPU时间，迭代服务器是基准，从其他服务器的实际CPU时间中减去迭代服务器的实际CPU时间就得到相应服务器用于进程控制所需的CPU时间，因为迭代服务器没有控制进程开销。本章我们使用进程控制CPU时间来称谓某个给定系统与基准的CPU时间之差。

以上测时数据都是与服务器运行在同一子网的两个不同主机上运行客户程序获得的，对于每个测试，这两个客户都派生5个子进程以建立到服务器的5个同时存在的连接，因此服务器在任意时刻最多有10个同时存在的连接。每个客户跨每个连接请求服务器返回4000字节的数据。对于涉及预先派生子进程或预先创建线程这两种服务器的测试，服务器在启动阶段派生15个子进程或15个线程。

有些服务器设计涉及创建一个子进程池或一个线程池，我们需要考虑闲置子进程或线程过多有什么影响，下图汇总了这些数据：

下图汇总了不同服务器情况下，客户请求在15个子进程或线程中的分布：

上图中第一列的含义是子进程或线程的编号。

以下客户程序用于测试我们的服务器的各个变体：

#include "unp.h"

#define MAXN 16384    /* max # bytes to request from server */

int main(int argc, char **argv) {
    int i, j, fd, nchildren, nloops, nbytes;
    pid_t pid;
    ssize_t n;
    char request[MAXLINE], reply[MAXN];

    if (argc != 6) {
        // 运行本客户需要指定服务器主机名或IP地址、服务器端口、
        // 由客户fork的子进程数（以允许客户并发地向同一服务器发起多个连接）、
        // 每个子进程发送给服务器的请求数、每个请求要求服务器返送的数据字节数
        err_quit("usage: client <hostname or IPaddr> <port> <#children>" 
                 "<#loops/child> <#bytes/request>");
    }

    nchildren  = atoi(argv[3]);
    nloops = atoi(argv[4]);
    nbytes = atoi(argv[5]);
    snprintf(request, sizeof(request), "%d\n", nbytes);    /* newline at end */

    for (i = 0; i < nchildren; ++i) {
        if ((pid = Fork()) == 0) {    /* child */
            for (j = 0; j < nloops; ++j) {
                // 每个子进程每次请求时，都要重新连接服务器
                fd = Tcp_connect(argv[1], argv[2]);
                
                // 向服务器发送一行文本，指出需由服务器返送的字节数
                Write(fd, request, strlen(request));

                // 在这个连接上读入所请求数据量（nbytes字节）的数据
                if ((n = Readn(fd, reply, nbytes)) != nbytes) {
                    err_quit("server returned %d bytes", n);
                }

                // 客户主动关闭连接，客户会进入TIME_WAIT状态，这是与通常的HTTP连接的差别之一
                Close(fd);    /* TIME_WAIT on client, not server */
            }
            printf("child %d done\n", i);
            exit(0);
        }
        /* parent loops around to fork() again */
    }

    // 父进程等待所有子进程终止
    while (wait(NULL) > 0);    /* now parent waits for all children */

    if (errno != ECHILD) {
        err_sys("wait error");
    }
    exit(0);
}

在测试本章中各个服务器程序时，用于执行本客户程序的命令如下：

这将建立2500个与服务器的TCP连接（5个子进程各自发起500次连接）。在每个连接上，客户向服务器发送5字节数据（4000\n），服务器于是向客户返送4000字节数据。我们在两个不同主机上针对同一个服务器执行本客户程序，于是总共有5000个TCP连接，且任意时刻服务器端最多同时存在10个连接。

迭代TCP服务器总是在完全处理某个客户的请求后才转向下一个客户，这样的服务器程序比较少见，一个例子是简单的时间获取服务器程序。

比较各个范式服务器程序时，迭代服务器程序很重要，如果我们针对迭代服务器程序执行如下客户程序：

同样会有5000个连接，跨每个连接传送的数据量也相同。由于服务器是迭代的，它没有执行任何进程控制，这就让我们测量出服务器处理这些数目的客户所需的CPU时间，改时间作为一个基准值，从其他服务器的实测CPU时间中减去该基准值就能得到它们的进程控制时间。从进程控制角度看迭代服务器是最快的，因为它不执行进程控制。

我们不给出本迭代服务器程序，它只是对下面介绍的并发服务器程序的少许修改。

传统上并发服务器调用fork派生一个子进程来处理每个客户，这使得服务器能同时为多个客户服务，每个进程一个客户。客户数目的唯一限制就是操作系统对服务器父进程所属用户ID能同时拥有的子进程数量的限制。

并发服务器的问题在于为每个客户现场fork一个子进程比较耗费CPU时间，多年前（20世纪80年代后期），一个繁忙的服务器每天也就处理几百或几千个客户时，这点CPU时间是可以接受的，但随着web应用的爆发式增长，繁忙的web服务器每天的TCP连接数以百万计，这还是就单个主机而言，更繁忙的站点往往运行多个主机来分摊负荷。常用的负载散布方法是DNS轮询（DNS round robin）。

以下是我们的并发服务器的main函数：

#include "unp.h"

int main(int argc, char **argv) {
    int listenfd, connfd;
    pid_t childpid;
    void sig_chld(int), sig_int(int), web_child(int);
    socklen_t clilen, addrlen;
    struct sockaddr *cliaddr;

    // 调用tcp_listen使本函数变得协议无关
    if (argc == 2) {
        listenfd = Tcp_listen(NULL, argv[1], &addrlen);
    } else if (argc == 3) {
        listenfd = Tcp_listen(argv[1], argv[2], &addrlen);
    } else {
        err_quit("usage: serv01 [ <host> | <port#> ]");
    }
    cliaddr = Malloc(addrlen);

    // sig_chld信号处理函数简单地wait子进程，我们不再给出此函数
    Signal(SIGCHLD, sig_chld);
    Signal(SIGINT, sig_int);

    for (; ; ) {
        clilen = addrlen;
        if ((connfd = accept(listenfd, cliaddr, &clilen)) < 0) {
            if (errno == EINTR) {
                continue;    /* back to for() */
            } else {
                err_sys("accept error");
            }
        }

        if ((childpid = Fork()) == 0) {    /* child process */
            Close(listenfd);    /* close listening socket */
            web_child(connfd);    /* process request */
            exit(0);
        }
        Close(connfd);    /* parent closes connected socket */ 
    }
}

以上函数为每个客户连接fork一个子进程并处理来自垂死的子进程的SIGCHLD信号。

以上函数还捕获由终端中断键产生的SIGINT信号，在客户运行完毕后我们键入该键以显示服务器程序运行所需的CPU时间，以下是SIGINT的信号处理函数，它不返回而直接终止进程：

void sig_int(int signo) {
    void pr_cpu_time(void);

    pr_cpu_time();
    exit(0);
}

以下是pr_cpu_time函数：

#include "unp.h"
#include <sys/resource.h>

#ifndef HAVE_GETRUSAGE_PRPTO
int getrusage(int, struct rusage *);
#endif

void pr_cpu_time(void) {
    double user, sys;
    struct rusage myusage, childusage;

    // 获取调用进程的资源利用统计
    if (getrusage(RUSAGE_SELF, &myusage) < 0) {
        err_sys("getrusage error");
    }
    // 获取调用进程的所有已终止子进程的资源利用统计
    if (getrusage(RUSAGE_CHILDREN, &childusage) < 0) {
        err_sys("getrusage error");
    }

    // 获取耗费在执行用户进程上的CPU时间
    user = (double)myusage.ru_utime.tv_sec + myusage.ru_utime.tv_usec / 1000000.0;
    user += (double)childusage.ru_utime.tv_sec + childusage.ru_utime.tv_usec / 1000000.0;
    // 获取耗费在执行系统调用上的CPU时间
    sys = (double)myusage.ru_stime.tv_sec + myusage.ru_stime.tv_usec / 1000000.0;
    sys += (double)childusage.ru_stime.tv_sec + childusage.ru_stime.tv_usec / 1000000.0;

    printf("\nuser time = %g, sys time = %g\n", user, sys);
}

web_child函数处理每个客户请求：

#include "unp.h"

#define MAXN 16384    /* max # bytes client can request */

void web_child(int sockfd) {
    int ntowrite;
    ssize_t nread;
    char line[MAXLINE], result[MAXN];

    for (; ; ) {
        if ((nread = Readline(sockfd, line, MAXLINE)) == 0) {
            return;    /* connection closed by other end */
        }

        /* line from client specifies #bytes to write back */
        ntowrite = atol(line);
        if ((ntowrite <= 0) || (ntowrite > MAXN)) {
            err_quit("client request for %d bytes", ntowrite);
        }

        Writen(sockfd, result, ntowrite);
    }
}

以上函数中，客户会发来一行文本，指出需由服务器返送多少字节的数据给客户，这与HTTP有些类似：客户发送一个小请求，服务器响应以所期望的信息（如一个HTML文件或一幅GIF图像）。在HTTP应用系统中，服务器通常在发送回所请求的数据后就关闭连接，但较新版本允许使用持续连接（persistent connection），打开连接后会为后续客户的额外请求继续保持连接打开。以上函数中，服务器允许来自客户的额外请求，但客户每次建立连接后只发送一个请求，然后就自己关闭该连接。

以上这种传统的并发服务器所需CPU时间最多，与它为每个客户现场fork的做法相吻合。

本章中没有测量的服务器程序设计范式是由inetd激活的服务器，从进程控制角度看，由inetd激活的处理单个客户连接的每个服务器涉及一个fork和一个exec，因此所需CPU时间只会比以上传统并发服务器更多。

我们的第一个增强型服务器程序使用称为预先派生子进程的技术，使用该技术的服务器不像传统意义的并发服务器那样为每个客户现场派生一个子进程，而是在启动阶段预先派生一定数量的子进程，当各个客户连接到达时，这些子进程立即就能为它们服务。下图是服务器父进程预先派生出N个子进程且正有2个客户连接着的情形：

这种技术的优点在于无须父进程调用fork的开销就能处理新到的客户，缺点是父进程必须在服务器启动阶段猜测需要预先派生多少子进程。如果某个时刻客户数恰好等于子进程总数，那么新到的客户将被忽略，直到至少有一个子进程重新可用（内核会为每个新到的客户完成三路握手，直到达到相应套接字上listen函数的backlog参数为止，然后服务器调用accept时会把这些已完成的连接传递给服务器进程），这样客户就能察觉到服务器响应时间变慢，因为客户的connect函数可能立即返回，但它的第一个请求却在一段时间后才被服务器处理。

通过增加一些代码，服务器能应对客户负载的变动，即父进程要持续监视可用（即闲置）子进程数，一旦该值降到低于某个阈值就派生额外的子进程，同样，一旦该值超过另一个阈值就终止一些过剩的子进程，因为过多可用的子进程也会降低性能。

先查看预先派生子进程这类服务器程序的基本结构，以下是预先派生子进程服务器的第一个版本的main函数：

#include "unp.h"

static int nchildren;
static pid_t *pids;

int main(int argc, char **argv) {
    int listenfd, i;
    socklen_t addrlen;
    void sig_int(int);
    pid_t child_make(int, int, int);

    if (argc == 3) {
        listenfd = Tcp_listen(NULL, argv[1], &addrlen);
    } else if (argc == 4) {
        listenfd = Tcp_listen(argv[1], argv[2], &addrlen);
    } else {
        err_quit("usage: serv02 [ <host> ] <port#> <#children>");
    }
    // 最后一个参数是用户指定的预先派生的子进程个数
    nchildren = atoi(argv[argc - 1]);
    // 分配一个存放各个子进程ID的数组，用于在父进程即将终止时终止所有子进程
    pids = Calloc(nchildren, sizeof(pid_t));

    for (i = 0; i < nchildren; ++i) {
        pids[i] = child_make(i, listenfd, addrlen);    /* parent returns */
    }

    Signal(SIGINT, sig_int);

    for (; ; ) {
        pause();    /* everything done by children */
    }
}

以下是SIGINT信号处理函数sig_int：

void sig_int(int signo) {
    int i;
    void pr_cpu_time(void);

    /* terminate all children */
    for (i = 0; i < nchildren; ++i) {
        kill(pids[i], SIGTERM);
    }
    while (wait(NULL) > 0);    /* wait for all children */
    if (errno != ECHILD) {
        err_sys("wait error");
    }

    // getrusage函数汇报的是已终止子进程的资源利用统计，因此调用pr_cpu_time前需要终止所有子进程
    pr_cpu_time();
    exit(0);
}

child_make函数由main调用以派生各个子进程：

#include "unp.h"

pid_t child_make(int i, int listenfd, int addrlen) {
    pid_t pid;
    void child_main(int, int, int);

    if ((pid = Fork()) > 0) {
        return pid;    /* parent */
    }

    child_main(i, listenfd, addrlen);    /* never returns */
}

以下是child_main函数：

void child_main(int i, int listenfd, int addrlen) {
    int connfd;
    void web_child(int);
    socklen_t clilen;
    struct sockaddr *cliaddr;

    cliaddr = Malloc(addrlen);

    printf("child %ld starting\n", (long)getpid());
    // 子进程一直循环，直到被父进程终止
    for (; ; ) {
        clilen = addrlen;
        // 每个子进程调用accept返回一个已连接套接字
        connfd = Accept(listenfd, cliaddr, &clilen);

        // 有客户连接到来后，调用web_child处理客户请求，最后关闭连接
        web_child(connfd);    /* process the request */
        Close(connfd);
    }
}

下面介绍一下在源自Berkeley的内核中，如何实现多个进程在同一监听描述符上调用accept

父进程在派生子进程前创建监听套接字，每次调用fork时，所有描述符也被复制。下图是proc结构（每个进程一个）、监听描述符的单个file结构、单个socket结构之间的关系：

上图中，父进程在fork完所有子进程后还不关闭listenfd，这是可以直接关闭的，但打开着是为了以后需要fork额外的子进程而做准备，这是一种对现有代码的改进。

描述符只是本进程的proc结构中的一个数组下标，其对应数组元素引用一个file结构。fork函数执行期间会复制描述符：子进程中一个给定描述符所引用的file结构与父进程中同一个描述符所引用的file结构是同一个。每个file结构都有一个引用计数，当打开一个文件或套接字时，内核将为之构造一个file结构，并由打开操作返回的描述符引用，它的引用计数初始为1，以后每当调用fork派生子进程，或调用dup复制描述符时，该file结构的引用计数就递增1。在我们的N个子进程的例子中，file结构的引用计数为N+1。

服务器进程在程序启动阶段派生N个子进程，它们各自调用accept并因此被内核投入睡眠。当第一个客户连接到达时，所有N个子进程均被唤醒，这是因为所有N个子进程的监听描述符指向同一个socket结构，因此它们都在同一个等待通道（wait channel），即在这个socket结构的so_timeo成员上进入睡眠。尽管N个子进程都被唤醒，但只有最先运行的子进程会获得那个客户连接，其余N-1个子进程继续睡眠，因为它们发现队列长度为0（最先运行的子进程取走了队列中的连接）。

这是有时称为惊群效应（thundering herd）的问题，尽管只有一个子进程将获得连接，但所有N个子进程都被唤醒了。尽管以上代码依然正常工作，但每当仅有一个连接准备好被接受时却唤醒太多进程的做法会导致性能受损。

图30-1中第二行里BSD/OS服务值为1.8的CPU时间的测试条件是：预先派生15个子进程且同时存在最多10个客户，为了测量惊群问题的影响，我们保持同时存在的最大客户数不变，只增加预先派生的子进程个数。

某些Unix内核有一个往往命名为wakeup_one的函数，它只是唤醒等待某个事件的多个进程中的一个，而非唤醒所有等待该事件的进程，BSD/OS内核没有这样的函数。

接着我们查看客户连接在进程池中的分配，为了采集这些信息，我们把main函数改为在共享内存区中分配一个长整数计数器数组，每个子进程一个计数器，main函数所增加代码如下：

long *cptr, *meter(int);    /* for counting #clients/child */

cptr = meter(nchildren);    /* before spawning children */

在分配共享内存区时，如果系统支持（如4.4 BSD），我们就使用匿名内存映射，否则使用/dev/zero映射（如SVR 4）。既然该数组是本进程在尚未派生各个子进程前调用mmap创建的，它将由本进程（父进程）和后来fork的所有子进程所共享。

meter函数在共享内存区中分配一个数组：

#include "unp.h"
#include <sys/mman.h>

/*
 * Allocate an array of "nchildren" longs in shared memory that can
 * be used as a counter by each child of how many clients it services.
 * See pp. 467-470 of "Advanced Programming in the Unix Environment."
 */

long *meter(int nchildren) {
    int fd;
    long *ptr;

#ifdef MAP_ANON
    ptr = Mmap(0, nchildren * sizeof(long), PROT_READ | PROT_WRITE, MAP_ANON | MAP_SHARED, -1, 0);
#else
    fd = Open("/dev/zero", O_RDWR, 0);

    ptr = Mmap(0, nchildren * sizeof(long), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);

    Close(fd);
#endif

    return ptr;
}

然后我们把child_main函数改为让每个子进程在accept函数返回后递增各自的计数器，把SIGINT信号处理函数改为在所有子进程终止后显示这个计数器数组。

图30-2给出这个分布，当可用子进程阻塞在accept函数上时，内核调度算法把各个连接均匀地散布到各个子进程。

TCPv2中提到过select函数的冲突（collision）现象，当多个进程在引用同一个套接字的描述符上调用select时会发生冲突，因为socket结构中为存放本套接字就绪时应唤醒哪些进程而分配的仅仅是一个进程id的空间，如果有多个进程在等待同一个套接字，那么内核必须唤醒阻塞在select函数上的所有进程，因为内核不知道哪些进程受到刚就绪的套接字的影响。

我们可以迫使本服务器程序发生select冲突，方法是在调用accept前加一个select调用，等待监听套接字变为可读，各个子进程将阻塞在select函数而非accept函数上，以下是对child_main函数的改动：

    printf("child %ld starting\n", (long)getpid());
+   FD_ZERO(&rset);
    for (; ; ) {
+       FD_SET(listenfd, &rset);
+       Select(listenfd + 1, &rset, NULL, NULL, NULL);
+       if (FD_ISSET(listenfd, &rset) == 0) {
+           err_quit("listenfd readable");   
+       }
+ 
        clilen = addrlen;
        connfd = Accept(listenfd, cliaddr, &clilen);
        
        web_child(connfd);    /* process the request */
        Close(connfd);
    }

修改好后，通过检查BSD/OS内核的nselcoll计数器在服务器运行前后的变化，我们发现某次运行本服务器出现1814个冲突，下一次运行出现2045个冲突，而两个客户为每次运行本服务器总共产生5000个连接，相当于有约35%~40%的select调用引起冲突。

比较BSD/OS服务器的CPU时间，加上select调用后其值由图30-1中的1.8增长到2.9，增长的原因一部分可能是新加了系统调用（由accept函数改为select+accept函数），另一部分可能是内核为处理select冲突而引入的额外开销。

以上是4.4 BSD实现，它允许多个进程在引用同一个监听套接字的描述符上调用accept，但这仅适用于在内核中实现accept函数的源自Berkeley的内核。作为一个库函数实现accept函数的System V内核可能不允许这么做，事实上如果我们在基于SVR 4的Solaris 2.5内核上运行以上服务器程序，那么客户开始连接到该服务器后不久，某个子进程的accept函数就会返回EPROTO错误（协议有错）。

造成本问题的原因在于SVR 4的流实现机制和库函数版本的accept函数并非一个原子操作，Solaris 2.6修复了这个问题，但大多其他SVR 4实现仍存在这个问题。

解决办法是让应用进程在调用accept前后安置某种形式的锁，这样任意时刻只有一个子进程阻塞在accept函数中，其他子进程则阻塞在试图获取用于保护accept函数的锁上。

我们使用fcntl函数的POSIX文件上锁功能保护accept函数。

main函数的唯一改动是在派生子进程的循环前增加一个对my_lock_init函数的调用：

+   my_lock_init("/tmp/lcok.XXXXXX");    /* one lock file for all children */
    for (i = 0; i < nchildren; ++i) {
        pids[i] = child_make(i, listenfd, addrlen);    /* parent returns */
    }

child_main函数的唯一改动是再调用accept前获取文件锁，在accept函数返回后释放文件锁：

    for (; ; ) {
        clilen = addrlen;
+       my_lock_wait();
        connfd = Accept(listenfd, cliaddr, &clilen);
+       my_lock_release();
        
        web_child(connfd);    /* process request */
        Close(connfd);
    }

以下是使用POSIX文件上锁的my_lock_init函数：

#include "unp.h"

static struct flock lock_it, unlock_it;
static int lock_fd = -1;    /* fcntl() will fail if my_lock_init() not called */

// 调用者将要上锁的文件的路径名模板传给my_lock_init函数
void my_lock_init(char *pathname) {
    char lock_file[1024];

    /* must copy caller's string, in case it's a constant */
    strncpy(lock_file, pathname, sizeof(lock_file));
    // mkstemp函数的参数是以6个字节的“XXXXXX”结尾的路径名，函数会将“XXXXXX“替换为一个串，该串使文件名唯一
    // 然后该函数创建此文件，并打开它，文件的权限为0600
    lock_fd = Mkstemp(lock_file);

    // 立即从文件系统目录中删除该路径名，之后如果程序崩溃，这个临时文件也将消失
    // 但还有进程打开着此文件（引用计数非0），因此文件本身不会被删除
    Unlink(lock_file);    /* but lock_fd remains open */

    // 初始化用于上锁的flock结构
    // 上锁范围为自字节偏移量0开始（l_whence为SEEK_SET，l_start为0），跨越整个文件（l_len为0）
    // 我们不往该文件中写任何东西，它的长度总为0，内核将正常处理该建议性锁
    lock_it.l_type = F_WRLCK;
    lock_it.l_whence = SEEK_SET;
    lock_it.l_start = 0;
    lock_it.l_len = 0;

    // 初始化用于解锁的flock结构
    unlock_it.l_type = F_UNLCK;
    unlock_it.l_whence = SEEK_SET;
    unlock_it.l_start = 0;
    unlock_it.l_len = 0;
}

以上函数中，作者（Stevens先生）声明两个flock结构时一开始使用以下方式：

static struct flock lock_it = { F_WRLCK, 0, 0, 0, 0 };
static struct flock unlock_it = { F_UNLCK, 0, 0, 0, 0 };

这样做有两个问题，首先，常值SEEK_SET为0并无保证；更重要的是，POSIX不保证flock结构中各成员的顺序，在Solaris和Digital Unix上l_type是第一个成员，在BSD/OS上它却不是。POSIX只是保证该结构中存在POSIX必需的成员，却不保证它们的先后顺序，更何况POSIX还允许flock结构中出现非POSIX的额外成员。因此，除非要把flock结构初始化为全0，否则总应该逐个成员赋值，而不应该在分配该结构时以初始化列表（initializer）初始化它。

这个规则的例外是初始化列表由实现具体提供的情形，如初始化Pthread互斥锁时：

pthread_mutex_t mlock = PTHREAD_MUTEX_INITIALIZER;

其中pthread_mutex_t通常是一个结构类型，而该初始化列表是由实现提供的，来自不同实现的该初始化列表可以不同。

以下是用于上锁和解锁的两个函数，它们使用我们在my_lock_init函数中初始化过的flock结构调用fcntl：

void my_lock_wait() {
    int rc;

    while ((rc = fcntl(lock_fd, F_SETLKW, &lock_it)) < 0) {
        if (errno == EINTR) {
            continue;
        } else {
            err_sys("fcntl error for my_lock_wait");
        }
    }
}

void my_lock_release() {
    if (fcntl(lock_fd, F_SETLKW, &unlock_it) < 0) {
        err_sys("fcntl error for my_lock_release");
    }
}

现在使用文件锁包围accept函数的预先派生子进程服务器程序在SVR 4系统上照样可以工作，因为它保证每次只有一个子进程阻塞在accept调用中。对比图30-1中Digital Unix和BSD/OS服务器的行2和行3，我们看到这种围绕accept函数的上锁增加了服务器的控制进程CPU时间。

Apache Web服务器程序的1.1版本在预先派生子进程后，如果实现允许所有子进程都阻塞在accept调用中，就不再对accept调用加锁，否则就使用以上包围accept函数的文件上锁技术。

我们可以查看围绕accept函数加文件锁的版本是否存在不加锁版本的惊群现象，图30-1A给出了增加子进程数时的CPU时间，在使用文件上锁保护accept函数的Solaris一栏中，我们看到子进程数超过75时，会引起CPU时间剧增，一个可能的原因是系统因进程过多而耗尽内存，导致开始对换。

我们可用上面的meter函数查看全体客户连接在可用子进程池上的分布，图30-2给出结果显示所有3个系统都均匀地把文件锁散布到等待进程中。

有多种方法可用于实现进程之间的上锁，以上使用的POSIX文件锁可移植到所有兼容POSIX的系统，但它涉及文件系统操作，比较耗时，下面我们使用线程上锁保护accept耗时，这种方法不仅用于同一进程内各线程之间的上锁，也适用于不同进程之间的上锁。

为了使用线程上锁，main、child_make、child_main函数都保持不变，唯一需要改动的是那3个上锁函数。在不同进程之间使用线程上锁要求：
1.互斥锁变量必须存放在由所有进程共享的内存区中。

2.告知线程库这是在不同进程间共享的互斥锁。

将线程上锁用于多个进程要求线程库支持PTHREAD_PROCESS_SHARED属性。

有多种方法在不同进程间共享内存空间，下例使用mmap函数和/dev/zero设备，它在Solaris和其他SVR 4内核上均可运行。以下是新版my_lock_init函数：

#include "unpthread.h"
#include <sys/mman.h>

static pthread_mutex_t *mptr;    /* actual muutex will be in shared memory */

void my_lock_init(char *pathname) {
    int fd;
    pthread_mutexattr_t mattr;

    fd = Open("/dev/zero", O_RDWR, 0);

    mptr = Mmap(0, sizeof(pthread_mutex_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
    // 内存映射后，就可以关闭描述符了
    Close(fd);

    // 以默认属性初始化互斥量属性结构pthread_mutexattr_t
    Pthread_mutexattr_init(&mattr);
    // 赋予互斥量属性结构PTHREAD_PROCESS_SHARED属性，该属性默认值为PTHREAD_PROCESS_PRIVATE，只能用于单个进程内
    Pthread_mutex_attr_setpshared(&mattr, PTHREAD_PROCESS_SHARED);
    // 以刚设置的属性初始化共享内存中的互斥锁
    Pthread_mutex_init(mptr, &mattr);
}

以下是线程上锁版本的my_lock_wait和my_lock_release函数：

void my_lock_wait() {
    Pthread_mutex_lock(mptr);
}

void my_lock_release() {
    Pthread_mutex_unlock(mptr);
}

比较图30-1中Solaris服务器的行3和行4，可见线程互斥锁上锁快于文件上锁。

对预先派生子进程服务器程序的另一个修改版本是只让父进程调用accept，然后把所接受的已连接套接字传递给某个子进程，这么做绕过了为所有子进程的accept调用上锁保护的可能需求，但需要从父进程到子进程的某种形式的描述符传递。这种技术会使代码变得复杂，因为父进程必须跟踪子进程的忙闲状态，以便给空闲子进程传递新套接字。

之前的预先派生子进程的例子中，父进程无需关心由哪个子进程接收客户连接，操作系统处理这个细节（给予某个子进程首先调用accept的机会，或给予某个子进程所需的文件锁或互斥锁）。图30-2的前5列表明我们测量的这3个系统以公平的轮询方式将客户连接分配给子进程。

对于传递描述符的预先派生子进程例子，我们需要为每个子进程维护一个信息结构以便管理，以下是child.h头文件中定义的Child结构：

typedef struct {
    pid_t child_pid;    /* process ID */
    int child_pipefd;    /* parent's stream pipe to/from child */
    int child_status;    /* 0 = ready */
    long child_count;    /* # connections handled */
} Child;

Child *cptr;    /* array of Child structures; calloc'ed */

我们在该结构中存放相应子进程的进程ID、父进程中连接到子进程的Unix域字节流管道描述符、子进程状态、该子进程已处理客户的计数。我们的SIGINT信号处理函数将在终止程序前显示各个子进程已处理的客户计数，以便观察全体客户请求在各个子进程之间的分布。

以下是传递描述符的child_make函数，它在fork前先创建一个字节流管道，它是一对Unix域字节流套接字，派生出子进程后，父进程关闭其中一个描述符（sockfd[1]），子进程关闭另一个描述符（sockfd[0]）：

#include "unp.h"
#include "child.h"

pid_t child_make(int i, int listenfd, int addrlen) {
    int sockfd[2];
    pid_t pid;
    void child_main(int, int, int);

    Socketpair(AF_LOCAL, SOCK_STREAM, 0, sockfd);

    if ((pid = Fork()) > 0) {
        Close(sockfd[1]);
        cptr[i].child_pid = pid;
        cptr[i].child_pipefd = sockfd[0];
        cptr[i].child_status = 0;
        return pid;    /* parent */
    }

    // 子进程把流管道的自身拥有端复制到标准错误输出，这样每个子进程就能通过读写标准错误与父进程通信
    Dup2(sockfd[1], STDERR_FILENO);    /* child's stream pipe to parent */
    Close(sockfd[0]);
    Close(sockfd[1]);
    Close(listenfd);    /* child does not need this open */
    child_main(i, listenfd, addrlen);    /* never returns */
}

所有子进程均派生之后的进程关系：

我们关闭每个子进程中的监听套接字，因为只有父进程才调用accept。父进程必须处理监听套接字和所有字节流套接字，父进程使用select函数多路选择它的所有描述符。

以下是传递描述符版本的main函数，变动在于，分配描述符集、打开监听套接字和到各个子进程的字节流管道对应的位、计算最大描述符值、分配Child结构数组的内存空间、主循环由一个select函数驱动：

#include "unp.h"
#include "child.h"

static int nchildren;

int main(int argc, char **argv) {
    // 计数器navail用于跟踪当前可用的子进程数
    int listenfd, i, navail, maxfd, nsel, connfd, rc;
    void sig_int(int);
    pid_t child_make(int, int, int);
    ssize_t n;
    fd_set rset, masterset;
    socklen_t addrlen, clilen;
    struct sockaddr *cliaddr;

    if (argc == 3) {
        listenfd = Tcp_listen(NULL, argv[1], &addrlen);
    } else if (argc == 4) {
        listenfd = Tcp_listen(argv[1], argv[2], &addrlen);
    } else {
        err_quit("usage: serv05 [ <host> ] <port#> <#children>");
    }

    FD_ZERO(&masterset);
    FD_SET(listenfd, &masterset);
    maxfd = listenfd;
    cliaddr = Malloc(addrlen);

    nchildren = atoi(argv[argc - 1]);
    navail = nchildren;
    cptr = Calloc(nchildren, sizeof(Child));

    /* prefork all the children */
    for (i = 0; i < nchildren; ++i) {
        child_make(i, listenfd, addrlen);    /* parent returns */
        FD_SET(cptr[i].child_pipefd, &masterset);
        maxfd = max(maxfd, cptr[i].child_pipefd);
    }

    Signal(SIGINT, sig_int);

    for (; ; ) {
        rset = masterset;
        // 如果没有可用子进程，就从读描述符集中关掉监听套接字对应的位
        // 这样可防止父进程在无子进程可用的情况下accept新连接
        // 内核会将这些多余连接排入队列，直到达到listen函数的backlog参数指示的值为止
        if (navail <= 0) {
            FD_CLR(listenfd, &rset);    /* turn off if no available children */
        }
        nsel = Select(maxfd + 1, &rset, NULL, NULL, NULL);

        /* check for new connections */
        // 监听套接字变为可读，一个新连接准备好accept
        if (FD_ISSET(listenfd, &rset)) {
            clilen = addrlen;
            connfd = Accept(listenfd, cliaddr, &clilen);
            
            // 找出第一个可用子进程
            // 我们总是从Child结构数组的第一个元素开始搜索可用子进程
            // 这意味着该数组中靠前的子进程比靠后的子进程更优先接收新连接
            // 如果不希望偏向于较早的子进程，我们可以记住最近一次接收新连接的子进程在Child结构数组中的位置
            // 下一次搜索就从该位置后面开始，如果到达数组末端就绕回第一个元素
            // 但这么做没什么优势，如果有多个子进程可用，由哪个子进程处理一个客户请求无关紧要
            // 除非操作系统进程调度算法惩罚（降低其优先级）总CPU时间较长的进程
            // 如果在各个子进程间更均匀地分摊负荷，那么每个子进程会在各自的总CPU时间上更为一致
            for (i = 0; i < nchildren; ++i) {
                if (cptr[i].child_status == 0) {
                    break;    /* available */
                }
            }

            if (i == nchildren) {
                err_quit("no available children");
            }
            cptr[i].child_status = 1;    /* mark child as busy */
            ++cptr[i].child_count;
            --navail;

            // 将新的已连接套接字传递给找到的空闲子进程
            // 我们随作为辅助数据传递的描述符写一个单字节普通数据，但接收进程不查看该字节内容
            n = Write_fd(cptr[i].child_pipefd, "", 1, connfd);
            // 父进程关闭新的已连接套接字
            Close(connfd);

            if (--nsel == 0) {
                continue;    /* all done with select() results */
            }
        }

        /* find any newly-available children */
        for (i = 0; i < nchildren; ++i) {
            if (FD_ISSET(cptr[i].child_pipefd, &rset)) {
                // 如果子进程处理完一个客户，会通过该子进程的字节流管道拥有端向父进程写回单个字节
                // 这使得该字节流管道的父进程拥有端变为可读
                // 父进程会读入这个单字节，忽略其值，然后把该子进程标记为可用，并递增navail计数器
                // 如果子进程意外终止，它的字节流管道拥有端将被关闭，read函数会返回0
                // 之后父进程会终止运行，但更好的做法是登记这个错误，并重新派生一个子进程取代意外终止的那个子进程
                // 本程序使用的Unix域套接字是字节流套接字，当子进程终止时，父进程会从Unix域套接字中读到EOF
                // 这个Unix域套接字也可换为数据报套接字，但这样就需要通过SIGCHLD信号来判断子进程是否终止
                // 当Unix域套接字两端进程非父子关系时，如想知道对端进程是否终止，只能用字节流套接字
                if ((n = Read(cptr[i].child_pipefd, &rc, 1)) == 0) {
                    err_quit("child %d terminated unexpectedly", i);
                }
                cptr[i].child_status = 0;
                ++navail;
                if (--nsel == 0) {
                    break;    /* all done with select() results */
                }
            }
        }
    }
}

以下是传递描述符式预先派生子进程服务器程序的child_main函数：

void child_main(int i, int listenfd, int addrlen) {
    char c;
    int connfd;
    ssize_t n;
    void web_child(int);

    printf("child %ld starting\n", (long)getpid());
    for (; ; ) {
        if ((n = Read_fd(STDERR_FILENO, &c, 1, &connfd)) == 0) {
            err_quit("read_fd returned 0");
        }
        if (connfd < 0) {
            err_quit("no descriptor from read_fd");
        }

        web_child(connfd);    /* process request */
        Close(connfd);

        // 完成客户处理后，子进程通过它的字节流管道拥有端写出1个字节到父进程，告知父进程本子进程已可用
        Write(STDERR_FILENO, "", 1);    /* tell parent we're ready again */
    }
}

以上child_main函数不再调用accept，而是阻塞在read_fd函数中，等待父进程传递来一个已连接套接字描述符。

比较图30-1中Solaris服务器的行4和行5，可见传递描述符的预先派生子进程的版本慢于使用线程上锁的预先派生子进程版本；再比较Digital Unix和BSD/OS服务器的行3和行5，我们能得出类似结论：父进程通过字节流管道把描述符传递到各个子进程，且各个子进程通过字节流管道写回单个字节，无论是与使用共享内存区中的互斥锁相比，还是与使用文件锁相比，都更费时。

图30-2给出了Child结构中child_count计数器值的分布，它是在终止服务器时由SIGINT信号处理函数显示的，正如以上讨论，越早派生（在Child结构数组中排名越前）的子进程所处理的客户数越多。

如果服务器主机支持线程，我们就可以将以上讨论中的子进程换成线程。

以下是第一个创建线程的服务器版本：

#include "unpthread.h"

int main(int argc, char **argv) {
    int listenfd, connfd;
    void sig_int(int);
    void *doit(void *);
    pthread_t tid;
    socklen_t clilen, addrlen;
    struct sockaddr *cliaddr;

    if (argc == 2) {
        listenfd = Tcp_listen(NULL, argv[1], &addrlen);
    } else if (argc == 3) {
        listenfd = Tcp_listen(argv[1], argv[2], &addrlen);
    } else {
        err_quit("usage: serv06 [ <host> ] <port#>");
    }
    cliaddr = Malloc(addrlen);

    Signal(SIGINT, sig_int);

    for (; ; ) {
        clilen = addrlen;
        // 主线程大多时间阻塞在accept函数中，每当它返回一个客户连接，就调用pthread_create创建一个新线程
        connfd = Accept(listenfd, cliaddr, &clilen);

        // 新线程执行的函数是doit，参数是新的已连接套接字
        Pthread_create(&tid, NULL, &doit, (void *)connfd);
    }
}

void *doit(void *arg) {
    void web_child(int);

    // 先让自己脱离，这样主线程就不用等待它
    Pthread_detach(pthread_self());
    web_child((int)arg);
    Close((int)(arg));
    return NULL;
}

图30-1表明这个简单的创建线程版本在Solaris和Digital Unix上都快于所有预先派生子进程的版本。这个为每个客户现场创建一个线程的版本比为每个客户现场派生一个子进程版本快很多倍。

我们的web_child函数调用readline，而readline函数是非线程安全的，本例中，readline函数仅仅用于读入来自客户的5字符的带换行符的数字，为了简单起见，我们使用第三章中给出的效率稍低的readline函数版本。

在支持线程的系统上，我们预期在服务器启动阶段预先创建一个线程池以取代为每个客户现场创建一个线程的做法会有性能提升。以下服务器的基本设计是预先创建一个线程池，并让每个线程各自调用accept，并且用互斥锁保证任何时刻只有一个线程在调用accept。这里我们没有理由使用文件锁保护各个线程中的accept函数，因为进程共享这个互斥锁。

以下给出pthread07.h头文件，其中定义了用于维护关于每个线程信息的Tthread结构：

typedef struct {
    pthread_t thread_tid;    /* thread ID */
    long thread_count;    /* # connections handled */
} Thread;
Thread *tptr;    /* array of Thread structures; calloc'ed */

int listenfd, nthreads;
socklen_t addrlen;
pthread_mutex_t mlock;

以下是预先创建线程版本的main函数：

#include "unpthread.h"
#include "pthread07.h"

pthread_mutex_t mlock = PTHREAD_MUTEX_INITIALIZER;

int main(int argc, char **argv) {
    int i;
    void sig_int(int), thread_make(int);

    if (argc == 3) {
        listenfd = Tcp_listen(NULL, argv[1], &addrlen);
    } else if (argc == 4) {
        listenfd = Tcp_listen(argv[1], argv[2], &addrlen);
    } else {
        err_quit("usage: serv07 [ <host> ] <port #> <#threads>");
    }
    nthreads = atoi(argv[argc - 1]);
    tptr = Calloc(nthreads, sizeof(Thread));

    for (i = 0; i < nthreads; ++i) {
        thread_make(i);    /* only main thread returns */
    }
    
    Signal(SIGINT, sig_int);

    for (; ; ) {
        pause();    /* everything done by threads */
    }
}

以下是thread_make和thread_main函数：

#include "unpthread.h"
#include "pthread07.h"

void thread_make(int i) {
    void *thread_main(void *);

    // 创建线程并使其执行thread_main函数，该函数的唯一参数是本线程在Thread结构数组中的下标
    Pthread_create(&tptr[i].thread_tid, NULL, &thread_main, (void *)i);
    return;    /* main thread returns */
}

void *thread_main(void *arg) {
    int connfd;
    void web_child(int);
    socklen_t clilen;
    struct sockaddr *cliaddr;

    cliaddr = Malloc(addrlen);

    printf("thread %d starting\n", (int)arg);
    for (; ; ) {
        clilen = addrlen;
        // 调用accept前后用互斥量保护
        Pthread_mutex_lock(&mlock);
        connfd = Accept(listenfd, cliaddr, &clilen);
        Pthread_mutex_unlock(&mlock);
        ++tptr[(int)arg].thread_count;

        web_child(connfd);    /* process request */
        Close(connfd);
    }
}

比较图30-1中Solaris和Digital Unix服务器的行6和行7，我们看到互斥量保护的预先创建子进程版本快于为每个客户现场创建一个线程的版本，且在这两个主机上，是所有版本中最快的。

图30-2给出了Thread结构中thread_count计数器值的分布，它们由SIGINT信号处理函数在服务器终止前显示输出。这个分布的均匀性是由线程调度算法带来的，该算法在选择由哪个线程接收互斥锁上表现为按顺序轮询所有线程。

在诸如Digital Unix等源自Berkeley的内核上，我们不必为调用accept上锁，因此可将使用互斥量保护accept函数的预先创建线程版本改为没有互斥锁的版本，但这么做会导致图30-1中行7的3.5秒增长到3.9秒，如果继续查看CPU时间的两个构成部分（用户时间和系统时间），我们发现没有上锁的用户时间有所减少（因为上锁是由在用户空间中执行的线程函数库完成的），系统时间却增长较多（因为一个连接到达时所有阻塞在accept函数中的线程都被唤醒，引发内核的惊群问题）。不管是否加锁都需要某种形式的互斥来把每个连接分配到线程池中某个线程，因此让内核执行分配不如让线程自行通过线程函数库执行分配来得快。

最后一个使用线程的服务器程序设计范式是在程序启动阶段创建一个线程池后，只让主线程调用accept，并把每个客户连接传递给池中某个可用线程。这里有多个实现手段，我们可以使用前面的描述符传递，但既然所有线程和所有描述符都在同一进程内，我们就没有必要使用描述符传递。接收线程只需知道这个已连接套接字描述符的值，而上面的描述符传递的实际并非这个值，而是对这个套接字的一个引用，因此可能返回一个不同于原值的描述符（该套接字的引用计数也会被递增）。以下是主线程统一accept的预先创建线程版本的头文件pthread08.h，其中也定义了Thread结构：

typedef struct {
    pthread_t thread_tid;    /* thread ID */
    long thread_count;    /* # connections handled */
} Thread;
Thread *tptr;

#define MAXNCLI 32
// clifd数组中是由主线程存入已accept的已连接套接字描述符，会由线程池中的可用线程从中取出一个以服务相应客户
// iput是主线程将往该数组中存入元素的下标，iget是线程池中某个线程将从该数组中取元素的下标
int clifd[MAXNCLI], iget, iput;
// 以上3个由所有线程共享的数据结构使用以下互斥锁和条件变量来保护
pthread_mutex_t clifd_mutex;
pthread_cond_t clifd_cond;

以下是主线程accept的预先创建线程版本的main函数：

#include "unpthread.h"
#include "pthread08.h"

static int nthreads;
pthread_mutex_t clifd_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t clifd_cond = PTHREAD_COND_INITIALIZER;

int main(int argc, char **argv) {
    int i, listenfd, connfd;
    void sig_int(int), thread_make(int);
    socklen_t addrlen, clilen;
    struct sockaddr *cliaddr;

    if (argc == 3) {
        listenfd = Tcp_listen(NULL, argv[1], &addrlen);
    } else if (argc == 4) {
        listenfd = Tcp_listen(argv[1], argv[2], &addrlen);
    } else {
        err_quit("usage: serv08 [ <host> ] <port#> <#threads>");
    }
    cliaddr = Malloc(addrlen);

    nthreads = atoi(argv[argc - 1]);
    tptr = Calloc(nthreads, sizeof(Thread));
    iget = iput = 0;

    /* create all the threads */
    for (i = 0; i < nthreads; ++i) {
        thread_make(i);    /* only main thread returns */
    }

    Signal(SIGINT, sig_int);

    for (; ; ) {
        clilen = addrlen;
        connfd = Accept(listenfd, cliaddr, &clilen);

        // 一旦某个客户连接到达，把新的已连接套接字描述符加入clifd数组
        Pthread_mutex_lock(&clifd_mutex);
        clifd[iput] = connfd;
        if (++iput == MAXNCLI) {
            iput = 0;
        }
        // 如果iput赶上了iget，说明该数组不够大
        if (iput == iget) {
            err_quit("iput = iget = %d", iput);
        }
        // 发送信号到条件变量，然后释放互斥锁，以允许线程池中某个线程为这个客户服务
        Pthread_cond_signal(&clifd_cond);
        Pthread_mutex_unlock(&clifd_mutex);
    }
}

以下是主线程accept的预先创建线程版本的thread_make和thread_main函数：

#include "unpthread.h"
#include "pthread08.h"

void thread_make(int i) {
    void *thread_main(void *);

    Pthread_create(&tptr[i].thread_tid, NULL, &thread_main, (void *)i);
    return;    /* main thread returns */
}

void *thread_main(void *arg) {
    int connfd;
    void web_child(int);

    printf("thread %d starting\n", (int)arg);
    for (; ; ) {
        // 线程池中每个线程都试图获取保护clifd数组的互斥锁，获得后就测试iput与iget是否相等
        // 如果相等，说明无事可做，于是通过调用pthread_cond_wait睡眠在条件变量上
        // 主线程接受一个连接后将调用pthread_cond_signal向条件变量发送信号，以唤醒睡眠在其上的线程
        // 若测得iput与iget不等则从clifd数组中取出一个连接，然后调用web_child
        Pthread_mutex_lock(&clifd_mutex);
        while (iget == iput) {
            Pthread_cond_wait(&clifd_cond, &clifd_mutex);
        }
        connfd = clifd[iget];    /* connected socket to service */
        if (++iget == MAXNCLI) {
            iget = 0;
        }
        Pthread_mutex_unlock(&clifd_mutex);
        ++tptr[(int)arg].thread_count;

        web_child(connfd);    /* process request */
        Close(connfd);
    }
}

图30-1中的测时数据表明这个版本的服务器慢于互斥锁保护accept的预先创建线程的版本，原因在于此版本同时需要互斥锁和条件变量，而互斥锁保护accept的预先创建线程的版本只需要互斥锁。

检查线程池中各个线程所服务客户数的分布直方图，我们发现它类似图30-2中的最后一列，这意味着当主线程调用pthread_cond_signal引起线程函数库基于条件变量执行唤醒工作时，该函数库在所有可用线程中轮询唤醒其中一个。

经过比较各个版本的服务器程序，我们得出以下总结性意见：
1.当系统负载较轻时，每来一个客户连接现场派生一个子进程为之服务的传统并发服务器程序模型就足够了，这个模型甚至可以和inetd结合使用，即inetd处理每个连接的接受。下面的意见是就重负荷运行的服务器而言的，如web服务器。

2.相比传统的每个客户fork一次设计范式，预先创建一个子进程池或一个线程池的设计范式能把进程控制CPU时间降低10倍以上，编写这些范式的程序并不复杂，但超越本章所给例子的是要监视子进程个数，随着所服务客户数的动态变化而增加或减少。

3.某些实现允许多个子进程或线程阻塞在同一个accept调用中，另一些实现却要求包绕accept调用安置某种类型的锁加以保护，可用文件锁或Pthread互斥锁。

4.让所有子进程或线程自行调用accept通常比让父进程或主线程独自调用accept并把描述符传递给子进程或线程来得简单而快速。

5.由于潜在select冲突，让所有子进程或线程阻塞在同一个accept调用中比让它们阻塞在同一个select调用中更可取。

6.使用线程通常远快于使用进程，但选择进程还是线程要取决于操作系统提供什么支持，还可能取决于是否要为客户执行exec。