# 12.2.1 基于 I/O 多路复用的并发事件驱动服务器

I/O 多路复用可以用做并发事件驱动(event-driven)程序的基础,在事件驱动程序中,某些事件会导致流向前推进。一般的思路是将逻辑流模型化为状态机。不严格地说,一个状态机(state machine)就是一组状态(state)、输入事件(input event)和转移(transition),其中转移是将状态和输入事件映射到状态。每个转移是将一个(输入状态,输入事件)对映射到一个输出状态。自循环(self-loop)是同一输入和输出状态之间的转移。通常把状态机画成有向图,其中节点表示状态,有向弧表示转移,而弧上的标号表示输入事件。一个状态机从某种初始状态开始执行。每个输入事件都会引发一个从当前状态到下一状态的转移。

对于每个新的客户端 k,基于 I/O 多路复用的并发服务器会创建一个新的状态机 sk,并将它和已连接描述符 dk 联系起来。如图 12-7 所示,每个状态机 sk 都有一个状态(“等待描述符 dk 准备好可读”)、一个输入事件(“描述符 dk 准备好可以读了”)和一个转移(“从描述符 dk 读一个文本行”)。

并发事件驱动 echo 服务器中逻辑流的状态机

图 12-7 并发事件驱动 echo 服务器中逻辑流的状态机

服务器使用 I/O 多路复用,借助 select 函数检测输入事件的发生。当每个已连接描述符准备好可读时,服务器就为相应的状态机执行转移,在这里就是从描述符读和写回一个文本行。

图 12-8 展示了一个基于 I/O 多路复用的并发事件驱动服务器的完整示例代码。一个 pool 结构里维护着活动客户端的集合(第 3~11 行)。在调用 init_pool 初始化池(第 27 行)之后,服务器进入一个无限循环。在循环的每次迭代中,服务器调用 select 函数来检测两种不同类型的输入事件:a)来自一个新客户端的连接请求到达,b)一个已存在的客户端的已连接描述符准备好可以读了。当一个连接请求到达时(第 35 行),服务器打开连接(第 37 行),并调用 add_client 函数,将该客户端添加到池里(第 38 行)。最后,服务器调用 check_clients 函数,把来自每个准备好的已连接描述符的一个文本行回送回去(第 42 行)。

code/conc/echoservers.c

#include "csapp.h"

typedef struct { /* Represents a pool of connected descriptors */
    int maxfd;        /* Largest descriptor in read_set */
    fd_set read_set;  /* Set of all active descriptors */
    fd_set ready_set; /* Subset of descriptors ready for reading */
    int nready;       /* Number of ready descriptors from select */
    int maxi;         /* High water index into client array */
    int clientfd[FD_SETSIZE];    /* Set of active descriptors */
    rio_t clientrio[FD_SETSIZE]; /* Set of active read buffers */
} pool;

int byte_cnt = 0; /* Counts total bytes received by server */

int main(int argc, char **argv)
{
    int listenfd, connfd;
    socklen_t clientlen;
    struct sockaddr_storage clientaddr;
    static pool pool;

    if (argc != 2) {
        fprintf(stderr, "usage: %s <port>\n", argv[0]);
        exit(0);
    }
    listenfd = Open_listenfd(argv[1]);
    init_pool(listenfd, &pool);

    while (1) {
        /* Wait for listening/connected descriptor(s) to become ready */
        pool.ready_set = pool.read_set;
        pool.nready = Select(pool.maxfd+1, &pool.ready_set, NULL, NULL, NULL);

        /* If listening descriptor ready, add new client to pool */
        if (FD_ISSET(listenfd, &pool.ready_set)) {
            clientlen = sizeof(struct sockaddr_storage);
            connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen);
            add_client(connfd, &pool);
        }

        /* Echo a text line from each ready connected descriptor */
        check_clients(&pool);
    }
}

图 12-8 基于 I/O 多路复用的并发 echo 服务器。每次服务器迭代都回送来自每个准备好的描述符的文本行

init_pool 函数(图 12-9)初始化客户端池。clientfd 数组表示已连接描述符的集合,其中整数 -1 表示一个可用的槽位。初始时,已连接描述符集合是空的(第 5~7 行),而且监听描述符是 select 读集合中唯一的描述符(第 10~12 行)。

code/conc/echoservers.c

void init_pool(int listenfd, pool *p)
{
    /* Initially, there are no connected descriptors */
    int i;
    p->maxi = -1;
    for (i = 0; i < FD_SETSIZE; i++)
        p->clientfd[i] = -1;

    /* Initially, listenfd is only member of select read set */
    p->maxfd = listenfd;
    FD_ZERO(&p->read_set);
    FD_SET(listenfd, &p->read_set);
}

图 12-9 init_pool 初始化活动客户端池

add_client 函数(图 12-10)添加一个新的客户端到活动客户端池中。在 clientfd 数组中找到一个空槽位后,服务器将这个已连接描述符添加到数组中,并初始化相应的 RIO 读缓冲区,这样一来我们就能够对这个描述符调用 rio_readlineb(第 8~9 行)。然后,我们将这个已连接描述符添加到 select 读集合(第 12 行),并更新该池的一些全局属性。maxfd 变量(第 15~16 行)记录了 select 的最大文件描述符。maxi 变量(第 17~18 行)记录的是到 clientfd 数组的最大索引,这样 check_clients 函数就无需搜索整个数组了。

code/conc/echoservers.c

void add_client(int connfd, pool *p)
{
    int i;
    p->nready--;
    for (i = 0; i < FD_SETSIZE; i++) /* Find an available slot */
        if (p->clientfd[i] < 0) {
            /* Add connected descriptor to the pool */
            p->clientfd[i] = connfd;
            Rio_readinitb(&p->clientrio[i], connfd);

            /* Add the descriptor to descriptor set */
            FD_SET(connfd, &p->read_set);

            /* Update max descriptor and pool high water mark */
            if (connfd > p->maxfd)
                p->maxfd = connfd;
            if (i > p->maxi)
                p->maxi = i;
            break;
        }
    if (i == FD_SETSIZE) /* Couldn't find an empty slot */
        app_error("add_client error: Too many clients");
}

图 12-10 add_client 向池中添加一个新的客户端连接

图 12-11 中的 check_clients 函数回送来自每个准备好的已连接描述符的一个文本行。如果成功地从描述符读取了一个文本行,那么就将该文本行回送到客户端(第 15~18 行)。注意,在第 15 行我们维护着一个从所有客户端接收到的全部字节的累计值。如果因为客户端关闭这个连接中它的那一端,检测到 EOF,那么将关闭这边的连接端(第 23 行),并从池中清除掉这个描述符(第 24~25 行)。

根据图 12-7 中的有限状态模型,select 函数检测到输入事件,而 add_client 函数创建一个新的逻辑流(状态机)。check_clients 函数回送输入行,从而执行状态转移,而且当客户端完成文本行发送时,它还要删除这个状态机。

练习题 12.4 图 12-8 所示的服务器中,我们在每次调用 select 之前都立即小心地重新初始化 pool.ready_set 变量。为什么?

旁注 事件驱动的 Web 服务器

尽管有 12.2.2 节中说明的缺点,现代高性能服务器(例如 Node.js、nginx 和 Tornado)使用的都是基于 I/O 多路复用的事件驱动的编程方式,主要是因为相比于进程和线程的方式,它有明显的性能优势。

code/conc/echoservers.c

void check_clients(pool *p)
{
    int i, connfd, n;
    char buf[MAXLINE];
    rio_t rio;

    for (i = 0; (i <= p->maxi) && (p->nready > 0); i++) {
        connfd = p->clientfd[i];
        rio = p->clientrio[i];

        /* If the descriptor is ready, echo a text line from it */
        if ((connfd > 0) && (FD_ISSET(connfd, &p->ready_set))) {
            p->nready--;
            if ((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0) {
                byte_cnt += n;
                printf("Server received %d (%d total) bytes on fd %d\n",
                       n, byte_cnt, connfd);
                Rio_writen(connfd, buf, n);
            }

            /* EOF detected, remove descriptor from pool */
            else {
                Close(connfd);
                FD_CLR(connfd, &p->read_set);
                p->clientfd[i] = -1;
            }
        }
    }
}

图 12-11 check_clients 服务准备好的客户端连接