进程间通信#

11.1 进程间通信的基本概念#

第 $10$ 章中通过 $fork$ 创建了多个进程，但父子进程拥有彼此独立的内存空间。即使子进程最初复制了父进程的数据，一方之后修改变量也不会影响另一方。因此，若两个进程需要交换数据，就必须借助操作系统提供的特殊通信机制，这类机制统称为进程间通信（ $Inter\ Process\ Communication$ ， $IPC$ ）。

通过管道实现进程间通信#

管道（ $Pipe$ ）是典型的 $IPC$ 方式。它并不属于某个进程，而是由操作系统管理的一段通信通道。两个进程只要持有该管道的文件描述符，就能通过这段由操作系统提供的空间交换数据。

$Linux$ 中通过 $pipe$ 创建管道：

1
#include <unistd.h>
2

3
int pipe(int filedes[2]);
4
//成功时返回0，失败时返回-1。

调用成功后，数组中会保存管道两端的文件描述符：

$filedes[0]$ 是管道出口，用于读取数据。
$filedes[1]$ 是管道入口，用于写入数据。

创建管道后再调用 $fork$ ，子进程会复制父进程持有的两个文件描述符。这里复制的只是文件描述符，父子进程仍通过它们访问操作系统中的同一条管道。

1
int fds[2];
2
char buf[30];
3
pid_t pid;
4

5
pipe(fds);
6
pid = fork();
7

8
if (pid == 0)
9
    write(fds[1], "Who are you?", 13);
10
else
11
{
12
    read(fds[0], buf, sizeof(buf));
13
    puts(buf);
14
}

该例中子进程只向 $fds[1]$ 写入数据，父进程只从 $fds[0]$ 读取数据，形成了一条由子进程流向父进程的单向通信路径。

通过一个管道双向通信的问题#

父子进程都拥有管道的入口和出口，所以从形式上看，同一条管道也能完成双向通信。子进程写入后由父进程读取，父进程再写入回复，由子进程读取即可。

但管道中的数据并不记录发送者与指定接收者。数据写入后，任何持有读端的进程都可以取走它，先调用 $read$ 的进程会得到数据。因此，子进程可能读回自己刚刚写入的内容，父进程则因读不到数据而一直阻塞。

通过 $sleep$ 控制执行顺序只能让示例暂时运行，并不能可靠地解决竞争问题。进程的调度顺序由操作系统决定，不应依赖延时去预测。

利用两个管道完成双向通信#

更清晰的办法是创建两条管道，让每条管道只负责一个方向：

$fds1$ 负责子进程向父进程传输数据。
$fds2$ 负责父进程向子进程传输数据。

1
int fds1[2], fds2[2];
2

3
pipe(fds1);
4
pipe(fds2);
5
pid = fork();
6

7
if (pid == 0)
8
{
9
    write(fds1[1], str1, sizeof(str1));
10
    read(fds2[0], buf, BUF_SIZE);
11
}
12
else
13
{
14
    read(fds1[0], buf, BUF_SIZE);
15
    write(fds2[1], str2, sizeof(str2));
16
}

两个方向互不争抢同一批数据，程序不再需要依赖父子进程的执行先后，通信关系也更容易理解。实际使用时还应关闭每个进程不需要的管道端点，避免资源长期占用，也能让 $EOF$ 等状态被正确识别。

11.2 运用进程间通信#

保存客户端消息的回声服务器#

多进程回声服务器中，每个子进程只负责一个客户端。如果希望把所有客户端发送的内容统一保存到文件，就需要让这些服务进程把数据汇总给同一个进程。

可以在服务器开始受理连接前创建管道，再创建一个专门保存消息的子进程：

保存进程从管道出口 $fds[0]$ 读取数据，并写入 $echomsg.txt$ 。
父进程继续通过 $accept$ 受理连接，并为每个客户端创建服务进程。
各服务进程在返回回声的同时，把收到的字符串写入管道入口 $fds[1]$ 。

1
pipe(fds);
2
pid = fork();
3

4
if (pid == 0)
5
{
6
    FILE *fp = fopen("echomsg.txt", "wt");
7

8
    for (i = 0; i < 10; i++)
9
    {
10
        len = read(fds[0], msgbuf, BUF_SIZE);
11
        fwrite(msgbuf, 1, len, fp);
12
    }
13

14
    fclose(fp);
15
    return 0;
16
}

之后创建的服务进程都会复制管道描述符，因此可以共同向同一入口传递消息：

1
while ((str_len = read(clnt_sock, buf, BUF_SIZE)) != 0)
2
{
3
    write(clnt_sock, buf, str_len);
4
    write(fds[1], buf, str_len);
5
}

这样，网络服务和文件保存被分配给不同进程。服务进程只负责接收、回送并转交数据，保存进程则统一完成文件 $I/O$ 。不过，随着功能继续增加，仅依靠多进程与管道组织服务器会使进程关系和通信路径越来越复杂，后续可以使用 $I/O$ 复用或线程模型实现更灵活的并发处理。

I/O复用#

12.1 基于I/O复用的服务器端#

多进程服务器会为每个客户端创建新的进程，逻辑直观，但创建进程需要复制运行环境、分配内存和进行上下文切换；进程间共享数据时，还要额外引入 $IPC$ 。当客户端数量增加时，这些成本也会随之增加。

复用（ $Multiplexing$ ）是使用较少的物理要素传递或处理更多数据的技术。把它应用到服务器端，就是让一个进程同时关注多个套接字，只处理当前已经准备好的 $I/O$ 对象。

在这种模型中，客户端连接数量不再直接决定服务进程数量。服务器进程先检查哪些套接字出现了连接请求或可读数据，再对这些套接字执行对应操作。因此， $I/O$ 复用的关键不是让一个进程同时执行多段代码，而是让它在多个 $I/O$ 对象之间高效切换。

这并不表示复用模型在所有场景中都优于多进程模型。若单次业务处理本身需要长时间运算，一个进程仍可能阻塞其他客户端；选择并发模型时，应同时考虑连接规模、任务耗时、隔离需求与实现复杂度。

12.2 理解select函数并实现服务器端#

$select$ 是实现 $I/O$ 复用的代表性函数。它可以把多个文件描述符集中起来，统一监视以下状态：

是否存在可以读取的数据。
是否可以无阻塞地写入数据。
是否发生异常情况。

这些被关注的状态通常称为事件。调用 $select$ 前需要准备监视集合和超时时间，返回后再从集合中确认具体是哪些文件描述符发生了事件。

设置文件描述符集合#

$select$ 使用 $fd\_set$ 保存监视对象。在 $Linux$ 中可将它理解为一个位数组：某个文件描述符对应位为 $1$ ，表示它属于当前监视集合。我们不直接操作其中的位，而是使用以下宏：

宏	作用
FD_ZERO(fdset)	将集合中的所有位初始化为 $0$
FD_SET(fd, fdset)	把文件描述符 $fd$ 加入集合
FD_CLR(fd, fdset)	从集合中移除文件描述符 $fd$
FD_ISSET(fd, fdset)	判断文件描述符 $fd$ 是否在集合中

例如，下面的代码先清空集合，再加入文件描述符 $1$ 和 $2$ ，最后移除 $2$ ：

1
fd_set set;
2

3
FD_ZERO(&set);
4
FD_SET(1, &set);
5
FD_SET(2, &set);
6
FD_CLR(2, &set);

设置监视范围及超时#

$Linux$ 中的 $select$ 函数形式如下：

1
#include <sys/select.h>
2
#include <sys/time.h>
3

4
int select(int maxfd, fd_set *readset, fd_set *writeset,
5
           fd_set *exceptset, struct timeval *timeout);
6
//发生事件时返回对应文件描述符数，
7
//超时时返回0，失败时返回-1。

$readset$ 、 $writeset$ 和 $exceptset$ 分别表示读取、写入和异常监视集合，不需要的项目可以传入 $NULL$ 。第一个参数应为所有监视对象中的最大文件描述符值加 $1$ ，因为 $Linux$ 文件描述符从 $0$ 开始。

$timeout$ 用于限制阻塞时间：

1
struct timeval
2
{
3
    long tv_sec;   //秒
4
    long tv_usec;  //微秒
5
};

如果监视对象一直没有变化， $select$ 会在指定时间后返回 $0$ ；传入 $NULL$ 则会一直等待，直到发生事件。部分系统会在函数返回时修改 $timeval$ 中的值，所以应在每次调用前重新设置超时时间。

查看select的调用结果#

$select$ 返回后会直接修改传入的 $fd\_set$ ：未发生事件的文件描述符对应位被清除，仍为 $1$ 的位才代表已经准备好的对象。

因此，程序一般保留一份完整的监视集合，并在每次循环中复制出临时集合交给 $select$ ：

1
fd_set reads, temps;
2

3
FD_ZERO(&reads);
4
FD_SET(0, &reads);
5

6
while (1)
7
{
8
    temps = reads;
9
    timeout.tv_sec = 5;
10
    timeout.tv_usec = 0;
11

12
    result = select(1, &temps, NULL, NULL, &timeout);
13

14
    if (result > 0 && FD_ISSET(0, &temps))
15
        read(0, buf, BUF_SIZE);
16
}

若直接把长期使用的 $reads$ 交给 $select$ ，未就绪对象会在第一次调用后被清除，之后便无法继续监视。复制集合是使用 $select$ 时必须注意的基本模式。

实现I/O复用回声服务器#

复用服务器需要同时监视监听套接字和所有已连接客户端套接字。监听套接字变为可读，表示有新的连接请求；客户端套接字变为可读，则表示收到数据或对方已经断开连接。

1
FD_ZERO(&reads);
2
FD_SET(serv_sock, &reads);
3
fd_max = serv_sock;
4

5
while (1)
6
{
7
    cpy_reads = reads;
8
    timeout.tv_sec = 5;
9
    timeout.tv_usec = 0;
10

11
    fd_num = select(fd_max + 1, &cpy_reads,
12
                    NULL, NULL, &timeout);
13
    if (fd_num <= 0)
14
        continue;
15

16
    for (i = 0; i <= fd_max; i++)
17
    {
18
        if (!FD_ISSET(i, &cpy_reads))
19
            continue;
20

21
        if (i == serv_sock)
22
        {
23
            clnt_sock = accept(serv_sock,
24
                               (struct sockaddr *)&clnt_adr, &adr_sz);
25
            FD_SET(clnt_sock, &reads);
26
            if (fd_max < clnt_sock)
27
                fd_max = clnt_sock;
28
        }
29
        else
30
        {
31
            str_len = read(i, buf, BUF_SIZE);
32
            if (str_len == 0)
33
            {
34
                FD_CLR(i, &reads);
35
                close(i);
36
            }
37
            else
38
                write(i, buf, str_len);
39
        }
40
    }
41
}

整个循环可以概括为三步：先等待事件，再找出发生事件的套接字，最后根据套接字类型处理连接或数据。客户端断开时，不仅要调用 $close$ ，还要通过 $FD\_CLR$ 把该描述符移出长期监视集合。

该模型只需一个进程就能管理多个客户端，省去了频繁创建进程的成本。不过示例会从 $0$ 扫描到 $fd\_max$ ，描述符很多且分布稀疏时会产生额外检查开销，这也是之后出现 $poll$ 、 $epoll$ 等机制的原因之一。

12.3 基于Windows的实现#

$Windows$ 同样提供 $select$ ，参数顺序和整体用法与 $Linux$ 基本一致：

1
#include <winsock2.h>
2

3
int select(int nfds, fd_set *readfds, fd_set *writefds,
4
           fd_set *exceptfds, const struct timeval *timeout);

主要区别在于， $Windows$ 套接字句柄并不像 $Linux$ 文件描述符那样从 $0$ 开始连续增长，所以 $fd\_set$ 不是位数组，而是直接保存句柄数量和句柄数组：

1
typedef struct fd_set
2
{
3
    u_int  fd_count;
4
    SOCKET fd_array[FD_SETSIZE];
5
} fd_set;

因此， $Windows$ 中的第一个参数 $nfds$ 只是为了兼容 $UNIX$ 接口，没有实际作用，通常传入 $0$ 。检查集合时也不再遍历 $0$ 到最大描述符，而是遍历 $fd\_array$ 中的有效套接字：

1
for (i = 0; i < reads.fd_count; i++)
2
{
3
    SOCKET sock = reads.fd_array[i];
4

5
    if (FD_ISSET(sock, &cpy_reads))
6
    {
7
        /* 处理连接请求或客户端数据 */
8
    }
9
}

$FD\_ZERO$ 、 $FD\_SET$ 、 $FD\_CLR$ 和 $FD\_ISSET$ 的名称与用法保持不变。移植服务器时，还需使用 $SOCKET$ 类型、 $recv$ 、 $send$ 和 $closesocket$ ，并按照 $Winsock$ 程序的一般流程调用 $WSAStartup$ 与 $WSACleanup$ 。

Mkrari的个人博客