UNIX进程间通信

过去，UNIX系统IPC是各种进程通信方式的统称，但是这些通信方式极少有能在所有的UNIX系统实现中进行移植的。随着POSIX和The Open Group(以前的X/Open)标准化的推进和影响扩大，情况已得到改善，但差别仍然存在。

文章会简述一些经典的IPC：管道、FIFO、消息队列、信号量以及共享存储。

简述进程间通信（InterProcess Communication，IPC）的发展，Unix System V IPC 机制起源于 1983 年发布的 Unix System V 操作系统。在 System V 之前，不同的 Unix 供应商使用不同的 IPC 机制。为了标准化和提高 Unix 系统的互操作性，它引入了统一的 IPC 接口和实现：消息队列、共享存储、信号量。

而 System V 基于消息的通信，为了提高通信的吞吐量和降低延迟，提供更有效的顺序数据传输 XSI（X/Open System Interface）IPC 机制在传统的通信机制上做了许多扩展，例如有：

• 使用流通信，不在基于消息的通信
• 统一的API用于创建、管理和通信不同类型的IPC对象
• 能够通过名称访问IPC对象，而不仅仅局限于文件描述符访问
• 支持使用信号进行事件通知，而非轮询或阻塞的机制
• 提供了流控制、优先级和组播等功能

1 管道

管道是UNIX系统IPC的最古老形式，所有UNIX系统都提供此种通信机制。它有以下两种局限性：

1）管道只能在具有公共祖先的两个进程间使用。通常管道由一个进程创建，该进程调用fork后，该管道便能够在父子进程间使用

2）管道通常是半双工的（即数据只能在一个方向流动）。某些系统支持全双工，但考虑最佳的可移植性，开发者不应预先假定系统支持全双工管道

由于标准 I/O 的缓冲机制，数据可能不会立即被写入管道或者从管道中读取。例如，在一个管道连接的两个进程（进程 A 和进程 B）中，进程 A 使用标准 I/O 函数向管道写入数据。如果写入的数据没有填满缓冲区，这些数据可能会在缓冲区中停留一段时间，直到缓冲区被填满或者程序主动刷新缓冲区（如调用fflush函数）才会真正写入管道。这会导致进程 B 不能及时获取到进程 A 发送的数据。

1.1 函数pipe

管道通过调用pipe函数创建。经由参数 fd 返回两个文件描述符：fd[0]为读而打开，fd[1]为写而打开。

#include <unistd.h>

int pipe (int fd[2]);

示例：

int main(void) {
#define MAX_LINE 64
    size_t n;
    int fd[2];
    pid_t pid;
    char line[MAX_LINE];

    if (pipe(fd) < 0)
        printf("pipe error");
    if ((pid = fork()) < 0) {
        printf("fork error");
    } else if (pid > 0) {
        /* parent */
        close(fd[0]);
        write(fd[1], "hello world\n", 12);
    } else {
        /* child */
        close(fd[1]);
        n = read(fd[0], line, MAX_LINE);
        write(STDOUT_FILENO, line, n);
    }
    exit(0);
}

程序创建了一个从父进程到子进程的管道，并且由父进程经由该管道向子进程传递数据。

实例：

管道是用于进程间通信的，而进程的产生通常是为了执行特定的命令。

如，在命令ls -l | grep txt中，ls -l命令的输出被写入管道，而grep txt命令从管道中读取这些输出数据，用于查找包含 “txt” 的内容。

1.2 函数popen与pclose

标准I/O库提供了两个函数popen和pclose，两个函数实现的操作是：创建一个管道，fork一个子进程，关闭未使用的管道端，执行一个shell运行命令，然后等待命令终止。

#include <stdio.h>

// 出错返回 NULL；成功返回文件指针，type："r"返回指针可读、"w"则可写
FILE *popen (const char *cmdstring, const char *type);

// 成功返回终止状态；出错返回-1
int pclose(FILE *fp);

使用popen能有效减少需要编写的代码量。

1.3 FIFO

有时被称为命名管道，未命名的管道只能在两个相关进程间使用，而且需要有一个共同创建了它们的祖先进程。通过FIFO，不相关的进程也能交换数据。

#include <sys/stat.h>

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

int mkfifoat (int fd, const char *path, mode_t mode);

// 两个函数成功返回0；出错返回-1

FIFO是一种文件类型，通过stat结构的st_mode成员编码可得知文件是否是一种文件类型；可使用S_ISFIFO宏对此进行测试

FIFO有以下两种用途： 1）shell命令使用FIFO将数据从一条管道传送到另一条时，无需创建中间临时文件 2）客户进程-服务器进程应用程序中，FIFO用作汇聚点，在进程两者之间传递数据

实例：

FIFO 可复制一系列输出流，并防止将数据写向中间磁盘文件（类似于使用管道来避免中间磁盘文件）。数据直接在内存中传递，避免了磁盘 I/O 和文件管理的复杂性。

如，将输入的数据 hello word 转换为大写后，读取转换后的文本。

echo "hello world" > temp_in.txt
tr '[:lower:]' '[:upper:]' < temp_in.txt > temp_out.txt
cat temp_out.txt
rm temp_in.txt temp_out.txt

若不使用管道，那么需要首先将文本写入到临时的 TXT 文件中，使用管道可避免中间文件产生：

mkfifo data_fifo
mkfifo tr_fifo

tr '[:lower:]' '[:upper:]' < data_fifo > tr_fifo &
cat < tr_fifo &

echo "Hello, World!" > data_fifo

mkfifo tr_fifo
rm data_fifo

案例中创建了两个管道，从 data_fifo 中读取数据，转换为大写后，写入 tr_fifo，第二个进程 cat 读取数据并显示。

这里并不需要两个临时文件，而使用临时的中间命名管道处理数据，避免了磁盘I/O。

2 XSI IPC

有三种称作 XSI IPC 的 IPC：消息队列、信号量、共享存储器。

每个内核中的 IPC 结构，都用一个非负整数的标识符（identifier）加以引用，只需要得到该标识符，则能够向一个 IPC 结构发送或获取消息。

#include <sys/ipc.h>
key_t ftok(const char *path ,int id);
// 成功返回键；失败返回(key_t)-1

可以使用函数 ftok 将路径名和项目ID 变换为一个键，其中 path 参数必须引用一个现有的文件，并函数只使用 id 参数的低8位。

1. 权限控制

   XSI IPC 制定了 ipc_perm 结构，来规定每个 IPC 结构的权限和所有者。

   cpp

   #include <sys/ipc.h>
   struct ipc_perm
   {
     __key_t __key;				/* Key.  */
     __uid_t uid;					/* Owner's user ID.  */
     __gid_t gid;					/* Owner's group ID.  */
     __uid_t cuid;					/* Creator's user ID.  */
     __gid_t cgid;					/* Creator's group ID.  */
     __mode_t mode;				/* Read/write permission.  */
     ...
   }

   使用 XSI IPC 结构时，根据不同的操作系统，它们都有各自的限制，这些限制能通过配置内核来改变。在 Linux 中，可以使用 ipcs -l 来显示 IPC 相关的限制：

   shell

   ubuntu@kwephispre11269:/$ ipcs -l
   
   ------ Messages Limits --------
   max queues system wide = 32000
   ...
   
   ------ Shared Memory Limits --------
   max number of segments = 4096
   ...
   
   ------ Semaphore Limits --------
   max number of arrays = 32000
   ...

2. 优缺点分析

   • XSI IPC 结构在系统范围内起作用，无引用计数。相比管道，最后一个引用管道的进程终止，管道就被完全删除了；相比 FIFO，当最后一个引用 FIFO 的进程终止时，虽然 FIFO 名仍保留，但能被显示删除。而这些 IPC 结构，需要调用 ipcrm 或 msgctl 进行删除。
   • 这些 IPC 结构在文件系统中没有名字，无法使用通用的函数来访问或修改它们的属性，需要额外学习新的系统调用（msgget、semop、shmat等）。
   • 又由于它们不使用文件描述符，因此这些 IPC 无法使用多路转接 I/O 函数（select和poll）。

2.1 消息队列

消息的链接表，存储在内核中，由队列标识符标识，以下简称队列ID。该队列不一定以先进先出次序取消息，可以按消息类型字段取消息。

其中 msgget 用于创建一个新队列或打开一个现有队列：

#include <sys/msg.h>
int msgget (key_t __key, int __msgflg);
// 成功返回队列ID；失败返回 -1

msgsnd 将新消息添加到队列尾端：

int msgsnd (int __msqid, const void *__msgp, size_t __msgsz, int __msgflg);
// 成功返回 0；失败返回 -1

每个消息由3部分组成：队列ID、消息类型、消息长度；msgflg 控制消息是否是阻塞I/O。其中 msgp 参数是指向 msgbuf 结构的指针，接收者可以使用消息类型以非先进先出的次序取消息。

msgrcv 用于从队列获取消息：

ssize_t msgrcv (int __msqid, void *__msgp, size_t __msgsz, long int __msgtyp, int __msgflg);
// 成功返回消息数据部分长度；失败返回 -1

其中，若返回消息长度大于 msgsz，且 msgflg 设置了 MSG_NOERROR 位，消息会被截断。若没设置这一标志，消息过长时会返回错误，消息仍被留在队列中。

参数 msgtype 可以指定获取消息的类型：

• =0：返回队列中第一个消息
• >0：返回队列中消息类型为 msgtype 的第一个消息
• <0：返回队列中消息类型值小于等于 msgtype 绝对值的消息，若该类消息有若干个，则取值最小的消息

msgctl 能对队列执行多种操作，类似于 ioctl 函数（即垃圾桶函数）。

int msgctl (int __msqid, int __cmd, struct msqid_ds *__buf);

注意：

该类型 IPC 不再被推荐使用！

2.2 信号量

它是一个计数器，用于为多个进程提供对共享数据对象的访问。

/* Semaphore control operation.  */
extern int semctl (int __semid, int __semnum, int __cmd, ...) __THROW;

/* Get semaphore.  */
extern int semget (key_t __key, int __nsems, int __semflg) __THROW;

/* Operate on semaphore.  */
extern int semop (int __semid, struct sembuf *__sops, size_t __nsops) __THROW;

/* Operate on semaphore with timeout.  */
extern int semtimedop (int __semid, struct sembuf *__sops, size_t __nsops,
		       const struct timespec *__timeout) __THROW;

如果在多个进程间共享一个资源，记录锁、互斥量也是能够选择的方案。

在 Linux 上，在共享存储中使用互斥量是一个更快的选择，而记录锁更为便捷，因此开发过程中无特别考虑性能的要求，更推荐使用记录锁（XSI IPC 的信号量较慢）。

注意：

该类型 IPC 有更好的替代方法！

2.3 共享存储

允许两个或多个进程共享一个给定的存储区，因数据不需要在客户进程和服务器进程间复制，所以这是最快的一种 IPC。

shmctl 能对共享段执行多种操作：

int shmctl (int __shmid, int __cmd, struct shmid_ds *__buf);

参数 cmd能指定下列命令：

• IPC_STAT：取段的 shmid_ds 结构，将它存储在 buf指向的结构中
• IPC-SET：按 buf指向的结构中的值设置字段（shm_perm.uid|gid|mode），需要有效用户或超级用户权限
• IPC_RMID：从系统删除该共享存储段。需要有效用户或超级用户权限执行

其中 shmget 用于创建一个新共享存储段或引用一个现有共享存储段：

#include <sys/shm.h>
int shmget (key_t __key, size_t __size, int __shmflg);
// 成功返回共享存储ID；失败返回 -1

创建一个新段时，会初始化 shmid_ds 结构的成员，

参数 size 表示段长度，以字节为段位，通常填入系统页长的整数倍（非则最后一页不可用）；引用现存段时，size 指定为0。

父子进程、或一个进程内，可以通过共享存储ID来操作一块段内存；如果两个独立的进程，没有其他方式传递共享存储ID时，则它们需要使用相同的 key 来调用 shmget，系统会根据 key值索引到同一个内存段（但它们返回的段ID可能不同）。

shmat 将段连接到调用进程的哪个地址上：

void *shmat (int __shmid, const void *__shmaddr, int __shmflg);
// 成功返回指向共享存储段的指针；失败返回 -1

函数行为与 addr 参数以及 flag 是否指定 SHM_RND 位有关。

• addr 为 0：段连接到由内核选择的第一个可用地址上。这是推荐使用方式。
• addr 非 0，未指定 SHM_RND 位：段连接到 addr 指定的地址上。
• addr 非 0，指定了 SHM_RND 位：段连接到 (addr - (addr mod SHMLBA)) 所表示的地址上。算式表示将地址向下取最近 1个 SHMLBA 的整数倍。

若在 flag 中指定了 SHM_RDONLY 位，则以制度方式连接此段，否则以读写方式连接。

当 shmat 执行成功，内核将使用该段相关的 shmid_ds 结构中的 shm_nattch 计数器加1

共享段操作结束时，调用 shmdt 与该段分离：

int shmdt (const void *__shmaddr);
// 成功返回 0；失败返回 -1

参数 addr 是调用 shmat 函数时的返回值。成功调用后，将使 shm_nattch 计数器减1。

该函数不会让系统删除段的标识符以及相关数据结构，直到服务器进程带 IPC_RMID 命令的调用 shmctl 特地删除它为止

注意：

现代编程更推荐使用 mmap，它还可以用于文件I/O操作，使用更便捷

在需要 System V IPC 特定功能或细粒度控制时，shm 仍然是一个有效的选择。

3 POSIX 信号量

相比 XSI 信号量，POSIX 信号量在删除时表现更完美，并有更高性能的实现。该信号量有两种形式：命名和未命名，差异在于创建和销毁的形式上。

调用 sem_open 函数创建一个新的命令信号量或使用一个现有信号量：

#include <semaphore.h>
sem_t *sem_open (const char *__name, int __oflag, ...);
// 成功返回信号量指针；失败返回 SEM_FAILED

使用现有的命名信号量时，指定 name 参数 和 flag 参数的 0 值即可。当 flag 参数为 O_CREAT 标识时，信号量不存在则新建，否则仅被使用。

指定了 O_CREAT 标识时，需要额外提供 mode 参数，用于指定访问信号量的权限：用户[组|其他]读[写|执行]；需要指定 value 参数设置信号量初始值，取值范围是 0 ~ SEM_VALUE_MAX。

为了增加可移植性，建议名字第一个字符为斜杠（/），这样信号量的实现使用了文件系统，可以消除命名的二义性

可调用 sem_close 释放信号量相关资源，调用 sem_unlink 销毁一个命名信号量：

int sem_close (sem_t *__sem);
int sem_unlink (const char *__name);
// 成功返回 0；失败返回 -1

函数 sem_close 关闭一个信号量。它不会销毁信号量本身，只是断开当前进程与信号量对象的连接。

函数 sem_unlink 删除信号量的名字，如没有其他额外信号量的引用，该信号量会直接被销毁，否则销毁将延迟到最后一个打开的引用关闭。

使用 sem_wait 或 sem_trywait 函数实现信号量减 1 操作：

int sem_wait (sem_t *__sem);
int sem_trywait (sem_t *__sem);
int sem_timedwait (sem_t *__restrict __sem, const struct timespec *__restrict __abstime);
// 成功返回 0；失败返回 -1

使用 sem_wait 函数时，若信号量计数是 0会发生阻塞，直到成功使信号量减 1或被信号中断时才返回。

能使用 sem_trywait 函数来避免阻塞，如果信号量为 0时，该函数不阻塞而是返回 -1并将 errno 置为 EAGAIN。

如果选择阻塞一段确定的时间，使用 sem_timewait 函数并利用 tsptr 参数指定绝对时间（基于 CLOCK_REALTIME 时钟），如果时间到期且信号量没能减 1则函数返回 -1并将 errno 置为 ETIMEDOUT。

使用 sem_post 函数实现信号量增 1 操作：

int sem_post (sem_t *__sem);
// 成功返回 0；失败返回 -1

调用 sem_post 时，如果进程有调用 wait 函数的相关阻塞，那么进程将会被唤醒，并将 sem_post 增1 的信号量计数减 1。

若希望在单个进程中使用 POSIX信号量时，可以使用 sem_init 函数创建一个未命名信号量：

int sem_init (sem_t *__sem, int __pshared, unsigned int __value);
// 成功返回 0；失败返回 -1

需要声明一个 sem_t 类型的变量并传递地址给 sem_init 函数 实现初始化，参数 value 指定了信号量的初始值。

如果需要在两个进程间使用信号量，需要确保 sem 参数 指向两个进程之间共享的内存范围。此时，参数 pshared 需要设置为一个非0 值；

对未命名信号量使用已经完成时，使用 sem_destroy 函数丢弃它：

int sem_destroy (sem_t *__sem);
// 成功返回 0；失败返回 -1

当调用了该函数后，不能再使用任何带有 sem 的信号量 函数，除非 init 重新初始化它。

函数 sem_getvalue 函数检索信号量值：

int sem_getvalue (sem_t *__restrict __sem, int *__restrict __sval);
// 成功返回 0；失败返回 -1

成功后，valp 指向的整数值将包含信号量值。但注意：试图使用刚读取出来的值时，该值可能已经变化！因此，除非使用额外的同步机制来避免该竞争，``sem_getvalue 函数` 只能用于调试。

实例

#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <malloc.h>
#include <unistd.h>
#include <semaphore.h>

struct slock {
    sem_t *semp;
    char name[_POSIX_NAME_MAX];
};

struct slock *s_alloc()
{
    struct slock *sp;
    static int cnt;
    if ((sp = malloc(sizeof(struct slock))) == NULL)
        return (NULL);
    do {
        snprintf(sp->name, sizeof(sp->name), "/%ld.%d", (long) getpid(), cnt++);
        sp->semp = sem_open(sp->name, O_CREAT | O_EXCL, S_IRWXU, 1);
    } while ((sp->semp = SEM_FAILED) && (errno == EEXIST));

    if (sp->semp == SEM_FAILED) {
        free(sp);
        return (NULL);
    }

    sem_unlink(sp->name);
    return (sp);
}

void s_free(struct slock *sp) {
    sem_destroy(sp->semp);
    free(sp);
}

int s_lock(struct  slock *sp) {
    return (sem_wait(sp->semp));
}

int s_unlock(struct  slock *sp) {
    return (sem_post(sp->semp));
}

可以通过信号量来创建自己的锁原语从而提供互斥。

4 小结

管道、FIFO 两种基本技术仍可有效地应用于大量的应用程序。

在新的应用程序中，要尽可能避免使用消息队列以及信号量，而应当考患全双工管道和记录锁，它们使用起来会简单得多。

共享存储仍然有它的用途，虽然通过mmap函数也能提供同样的功能。