[单刷APUE系列]第十五章——进程间通信

引言

在前面章节中讲解了进程的派生和常见调用，但是进程之间通信的唯一途径就是通过打开的文件，或者是使用进程之前信号传输，由于这些技术的局限性，Unix系统提供了一种进程间通信模型（IPC）。IPC是进程通信各种方式的统称，目前只有一些经典的IPC方式能做到移植使用：管道、FIFO、消息队列、信号量、共享存储。还有基于套接字技术的网络IPC。

管道

管道是很古老的进程间通信机制了，基本所有的Unix系统或者非Unix系统都支持这种方式，管道有以下特性：

1. 半双工，也就是数据只能做到单向流动

2. 管道只能在具有公共祖先的两个进城之间使用。

虽然具有局限性，但是由于它的可移植性，所以目前仍然是首选的进程间通信技术，管道在shell中非常常见，我们常常使用以下命令

> command1 | command2 ... commandn

shell使用管道将前一个进程的标准输出与后一条命令的标准输入相连接。开发者调用pipe函数创建管道

int pipe(int fildes[2]);

filedes参数包含了两个文件描述符，Unix手册上这么描述 The first descriptor connects to the read end of the pipe; the second connects to the write end. ，也就是说，filedes[0]是读的一端，filedes[1]是写入的一端，这就是一个半双工的管道，然后可以通过fork的方式将文件描述符分给父子进程，从而实现通信。在fork以后，双方都持有读写的端口，而父子进程可以关闭其中两个，每个进程只持有一个，从而做到真正的管道。管道有以下特点：

1. 读一个写口关闭的管道时，当缓冲区所有数据都读取后，会返回0，代表文件结束。

2. 写一个读口关闭的管道时，会产生SIGPIPE信号，如果忽略该信号或者捕获该信号并且从信号处理函数返回，则write返回-1，errno设置为EPIPE。

我们实际上可以将管道理解为一块内核维护的缓冲区，所以常量PIPE_BUF规定了管道的大小。

#include "include/apue.h"  int main(int argc, char *argv[]) {     int n;     int fd[2];     pid_t pid;     char line[MAXLINE];          if (pipe(fd) < 0)         err_sys("pipe error");     if ((pid = fork()) < 0)         err_sys("fork error");     else if (pid == 0) {         close(fd[0]);         write(fd[1], "hello world/n", 12);     } else {         close(fd[1]);         n = read(fd[0], line, MAXLINE);         write(STDOUT_FILENO, line, n);     }     exit(0); }

上面就是非常简单的父子进程使用同一个管道通信的小实例。

popen和close函数

除了上面的pipe函数以外，更常见的做法是创建一个连接到另一个进程的管道，然后读其输出或者向其输出端发送数据，为此标准IO库提供了popen和pclose函数

FILE *popen(const char *command, const char *mode); int pclose(FILE *stream);

协同进程

在日常Unix使用中，通常我们会使用管道连接多个命令，当一个程序标准输入输出都连接到管道的时候，这就是协同进程。下面是一个父子进程实现的协同进程

#include "include/apue.h"  int main(int argc, char *argv[]) {     int n, int1, int2;     char line[MAXLINE];          while ((n = read(STDIN_FILENO, line, MAXLINE)) > 0) {         line[n] = 0;         if (sscanf(line, "%d%d", &int1, &int2) == 2) {             sprintf(line, "%d/n", int1 + int2);             n = strlen(line);             if (write(STDOUT_FILENO, line, n) != n)                 err_sys("write error");         } else {             if (write(STDOUT_FILENO, "invalid args/n", 13) != 13)                 err_sys("write error");         }     }     exit(0); }

上面的代码非常简单，就是标准输入读取，计算后输出到标准输出，将其编译为add2程序。

#include "include/apue.h"  static void sig_pipe(int);  int main(int argc, char *argv[]) {     int n, fd1[2], fd2[2];     pid_t pid;     char line[MAXLINE];          if (signal(SIGPIPE, sig_pipe) == SIG_ERR)         err_sys("signal error");              if (pipe(fd1) < 0 || pipe(fd2) < 0)         err_sys("pipe error");          if ((pid = fork()) < 0)         err_sys("fork error");     else if (pid > 0) {         close(fd1[0]);         close(fd2[1]);                  while (fgets(line, MAXLINE, stdin) != NULL) {             n = strlen(line);             if (write(fd1[1], line, n) != n)                 err_sys("write error to pipe");             if ((n = read(fd2[0], line, MAXLINE)) < 0)                 err_sys("read error from pipe");             if (n == 0) {                 err_msg("child closed pipe");                 break;             }             line[n] = 0;             if (fputs(line, stdout) == EOF)                 err_sys("fputs error");         }         if (ferror(stdin))             err_sys("fgets error on stdin");         exit(0);     } else {         close(fd1[1]);         close(fd2[0]);         if (fd1[0] != STDIN_FILENO) {             if (dup2(fd1[0], STDIN_FILENO) != STDIN_FILENO)                 err_sys("dup2 error to stdin");             close(fd1[0]);         }         if (fd2[1] != STDOUT_FILENO) {             if (dup2(fd2[1], STDOUT_FILENO) != STDOUT_FILENO)                 err_sys("dup2 error to stdout");             close(fd2[1]);         }         if (execl("./add2", "add2", (char *)0) < 0)             err_sys("execl error");     }     exit(0); }

程序创建了两个管道，父子进程各自关闭不需要使用的管道。然后使用dup2函数将其移到标准输入输出，最后调用execl。

FIFO

int mkfifo(const char *path, mode_t mode);

1. 未指定标志的时候，只读的open会一直阻塞到有进程写打开，只写open会阻塞到有进程读打开。

2. 指定了标志，则只读open立刻返回。但是如果没有进程为只读打开，只写的open会出错。

和管道类似，如果write一个没有进程读打开的FIFO则会产生SIGPIPE信号。如果读完所有数据，read函数会返回文件结束。由于这是一个文件，所以多个进程读写是非常正常的事情，所以为了保证读写安全，原子写操作是必须要考虑的。FIFO有以下用途：

1. shell命令使用FIFO将数据从一条管道传输到另一条管道不需要创建中间文件

2. C/S架构中，FIFO用作中间点，在客户服务器之间传输数据

这里就不再讲实例了。需要的朋友自行寻找代码。

XSI IPC

在IPC中有三种被称为XSI IPC，他们有很多共同点。这里先讲解共同点。

标识符和键

1. 服务器进程指定IPC_PRIVATE键创建一个新的IPC结构。返回的标识符被存放在一个文件中，客户端进程读取这个文件来参与IPC结构。

2. 在一个公用头文件中定义一个统一的标识符，然后服务端根据这个标识符创建新的IPC结构，但是很有可能导致冲突

3. 客户端和服务端进程认同同一个路径名和项目ID。接着调用函数ftok将这两个值变为一个键，然后在创建一个IPC结构

key_t ftok(const char *path, int id);

path参数必须是一个现有的文件，当产生键的时候，只会使用id参数低八位。ftok创建键一般依据如下行为：首先根据给定的path参数获得对应文件的stat结构中st_dev和st_ino字段，然后将他们和项目ID组合。

权限结构

每个IPC结构都关联了一个ipc_perm结构，这个结构体关联了权限和所有者。

struct ipc_perm {         uid_t           uid;            /* [XSI] Owner's user ID */         gid_t           gid;            /* [XSI] Owner's group ID */         uid_t           cuid;           /* [XSI] Creator's user ID */         gid_t           cgid;           /* [XSI] Creator's group ID */         mode_t          mode;           /* [XSI] Read/write permission */         unsigned short  _seq;           /* Reserved for internal use */         key_t           _key;           /* Reserved for internal use */ };

上面是苹果平台的结构体内容，一般来说，都会有uid、gid、cuid、cgi、mode这些基本的内容，而其他则是各个实现自由发挥。在创建的时候，这些字段都会被赋值，而后，如果想要修改这些字段，则必须是保证具有root权限或者是创建者。

优点和缺点

XSI IPC一个问题就是：IPC结构是在系统范围内使用的，但是却没有引用计数。如果进程使用完但是没有对其进行删除就终止了，那就会导致IPC依然在系统中存在，而管道有引用计数，等最后一个引用管道的进程终止便会自动回收。FIFO就算没有删除，但是等最后一个引用FIFO的进程终止，里面的数据已经被删除了。XSI IPC还有一个问题就是它不是文件，我们不能使用ls和rm等文件操作函数或者命令处理它们，它们也没有文件描述符，这就限制了它们的使用，并且如果需要使用还得携带一大堆额外的API。

消息队列

消息队列，正如其名称一样，是消息的链表形式，它由内核存储维护。并且和XSI IPC结构一样，由消息队列标识符标识。msgget函数创建一个队列或者打开一个现有队列，msgsnd将新数据添加到消息末尾，每个消息包含一个正的长整形字段、一个非负的长度以及实际数据字节数。msgrcv从队列中获得消息。

struct __msqid_ds {             struct __ipc_perm_new   msg_perm; /* [XSI] msg queue permissions */         __int32_t       msg_first;      /* RESERVED: kernel use only */         __int32_t       msg_last;       /* RESERVED: kernel use only */         msglen_t        msg_cbytes;     /* # of bytes on the queue */         msgqnum_t       msg_qnum;       /* [XSI] number of msgs on the queue */         msglen_t        msg_qbytes;     /* [XSI] max bytes on the queue */         pid_t           msg_lspid;      /* [XSI] pid of last msgsnd() */         pid_t           msg_lrpid;      /* [XSI] pid of last msgrcv() */         time_t          msg_stime;      /* [XSI] time of last msgsnd() */         __int32_t       msg_pad1;       /* RESERVED: DO NOT USE */         time_t          msg_rtime;      /* [XSI] time of last msgrcv() */         __int32_t       msg_pad2;       /* RESERVED: DO NOT USE */         time_t          msg_ctime;      /* [XSI] time of last msgctl() */         __int32_t       msg_pad3;       /* RESERVED: DO NOT USE */         __int32_t       msg_pad4[4];    /* RESERVED: DO NOT USE */ }

每个系统实现都会在SUS标准的基础上增加自己私有的字段，所以可能和上面的有所区别。这个结构体定义了队列的当前状态。

int msgget(key_t key, int flag);

msgget根据key获得已有队列或者新队列。

int msgctl(int msqid, int cmd, struct msqid_ds *buf);

msgctl用于对队列进行多种操作，其中，msqid则是消息队列ID，cmd参数是命令参数，可以取以下值：

1. IPC_STAT 取队列msqid_ds结构体，并且存放在buf参数指定的位置

2. IPC_SET 将buf参数指定的结构体复制到这个队列中的结构体，需要检查root权限或者创建者权限

3. IPC_RMID 从系统中删除消息队列

int msgsnd(int msqid, const void *ptr, size_t nbytes, int flag);

这个函数很好理解，就是把ptr指针对应的消息放入消息队列中，flag的值可以指定为IPC_NOWAIT，这点类似于文件IO非阻塞标志。

ssize_t msgrcv(int msqid, void *ptr, size_t nbytes, long type, int flag);

和msgsnd一样，这个ptr参数指向一个长整形数，随后跟随的是存储实际区域的缓冲区。nbytes指定数据缓冲区的长度。参数type则可以指定想要哪种消息。

信号量

信号量是一个计数器，针对多个进程提供对共享数据对象的访问。信号量的使用主要是用来保护共享资源的，使得资源在一个时刻只会被一个进程（线程）拥有。信号量的使用如下：

1. 测试控制该资源的信号量

2. 如果信号量为正，则进程可以使用该资源，这种情况下，信号量会减一，表示已经使用了一个资源。

3. 如果信号量为0，则进程进入休眠状态，知道信号量变为正，进程将会唤醒。

内核为每个信号量集合维护着一个semid_ds结构体，根据每个系统实现不同会有不同的字段，这里就不列出了。当我们想要使用信号量的时候，使用如下函数

int semget(key_t key, int nsems, int semflg);

我们知道，XSI IPC实际上具有其共性，所以如同消息队列一样，这里将key变换为标识符的规则也是一样的。

int semctl(int semid, int semnum, int cmd, ...);

就如同是前面的ioctl等函数一样，这个函数也是用于控制信号量，其中第四个参数是可选的，取决于cmd参数。

     IPC_STAT     Fetch the semaphore set's struct semid_ds, storing it in the memory pointed to by arg.buf.       IPC_SET      Changes the sem_perm.uid, sem_perm.gid, and sem_perm.mode members of the semaphore set's struct semid_ds to match those of the struct pointed to                   by arg.buf.  The calling process's effective uid must match either sem_perm.uid or sem_perm.cuid, or it must have superuser privileges.       IPC_RMID     Immediately removes the semaphore set from the system.  The calling process's effective uid must equal the semaphore set's sem_perm.uid or                   sem_perm.cuid, or the process must have superuser privileges.       GETVAL       Return the value of semaphore number semnum.       SETVAL       Set the value of semaphore number semnum to arg.val.  Outstanding adjust on exit values for this semaphore in any process are cleared.       GETPID       Return the pid of the last process to perform an operation on semaphore number semnum.       GETNCNT      Return the number of processes waiting for semaphore number semnum's value to become greater than its current value.       GETZCNT      Return the number of processes waiting for semaphore number semnum's value to become 0.       GETALL       Fetch the value of all of the semaphores in the set into the array pointed to by arg.array.       SETALL       Set the values of all of the semaphores in the set to the values in the array pointed to by arg.array.  Outstanding adjust on exit values for                   all semaphores in this set, in any process are cleared.

上面就是cmd可选值，除了GETALL以外的所有命令，semctl都返回相应值，除此以外，还有个semop函数自动执行信号量集合上的操作数组

int semop(int semid, struct sembuf *sops, size_t nsops);

sembuf参数是一个指针，指向了sembuf结构体，实际上这是一个数组，

共享存储

共享存储是一项很有用的技术，它允许两个或者更多的进程共享同一个给定存储区，由于数据不需要复制，所以这是一种最快的IPC方式，共享存储最重要的就是资源的竞争，所以信号量一般用于共享存储访问。在前面的章节中，我们看到了一种共享存储的方式，就是内存映射技术，但是相比存储映射，共享存储不需要创建中间文件。

int shmget(key_t key, size_t size, int shmflg); int shmctl(int shmid, int cmd, struct shmid_ds *buf);

就如同是其他的XSI IPC一样，这里也是一个创建函数，通过这个函数获得存储标识符。然后就是shmctl函数，具体的使用直接查手册，基本上都是差不多的。当创建完成共享存储段后，使用shmat将其链接到自己的地址空间中。

void *shmat(int shmid, const void *shmaddr, int shmflg); int shmdt(const void *shmaddr);

实际上，由于架构的不同和平台不同，为了保持可移植性，我们不应当去指定共享存储段的地址，而是由系统自行分配，最终返回的就是共享存储段的地址，当操作完成后，我们需要使用shmdt函数将其分离。

POSIX信号量

POSIX信号量是三种IPC机制之一，相比XSI标准规定的IPC方式，POSIX的方式更加简洁好用。POSIX信号量有两种类型：命名的和未命名的，他们两者的区别就像是命名管道和未命名管道一样，有了标识符的信号量就能全局使用，而没有标识符的信号量只能在同一内存区域内使用。

sem_t *sem_open(const char *name, int oflag, ...); The parameters "mode_t mode" and "unsigned int value" are optional.  The value of oflag is formed by or'ing the following values:      O_CREAT         create the semaphore if it does not exist     O_EXCL          error if create and semaphore exists

实际上这个函数技能创建也能使用现有信号量，上面的是Unix手册节选的内容，应该算是相当清楚了，当函数返回的时候，sem_open会返回一个指针，让我们传递到其他的函数上，等一切结束，使用 sem_close 关闭信号量指针。

int sem_close(sem_t *sem);

当然，也能使用sem_unlink函数销毁一个命名信号量。

int sem_unlink(const char *name);

在这里我们能看出来，POSIX信号量的命名信号量和文件很像。不像XSI信号量，POSIX信号量的值只能通过一个函数调用来调节，也就是sem_wait函数

int sem_wait(sem_t *sem); int sem_trywait(sem_t *sem);

后一个是sem_wait函数的尝试版本，还有一个是超时版本，但是苹果下好像不存在，建议少使用。还可以调用sem_post函数使信号量值+1，这个解锁一个二进制信号量或者释放一个技术信号量资源的过程是很像的。

int sem_post(sem_t *sem);

对于未命名信号量，只能使用在单进程或者单线程中，是非常容易的，我们只需要使用sem_init函数创建，然后使用sem_destroy函数销毁。这里就不在赘述。