Linux进程通信的四种方式——共享内存、信号量、无名管道、消息队列｜实验、代码、分析、总结
每个进程各自有不同的用户地址空间，任何一个进程的全局变量在另一个进程中都看不到，所以进程之间要交换数据必须通过内核，在内核中开辟一块缓冲区，生产者进程把数据从用户空间拷到内核缓冲区，消费者进程再从内核缓冲区把数据读走，内核提供的这种机制称为进程间通信（IPC，InterProcess Communication）.
转载需注明出处：?? Sylvan Ding ??
实验目标
本实验主要实现单机上不同进程间的通信，故在网络环境中广泛应用的客户机-服务器系统通信的三种实现方法（套接字、远程过程调用和远程方法调用）不予实现。
问题描述
生产者一次生成一个元素放入缓冲池中，消费者一次可以从缓冲池中取出一个元素。生产者放入的元素个数要与消费者取出的元素个数一致。实验的输出要能跟踪生产者的每次“生产”行为，以及消费者的每次“消费”行为。
需求分析
利用OS提供的高级通信工具，在单机环境下，设计和编程实现进程间数据的高效传送，并对比分析各方法的优缺点。编写生产者和消费者的相关程序，在Linux系统下实现信号量、共享内存、管道（无名管道和有名管道）、消息队列等四种进程间通信方式。实验的输出要能跟踪生产者的每次“生产”行为，以及消费者的每次“消费”行为。
实验环境
- 操作系统：Ubuntu
- 编程语言：C
- 编译器：GCC 7.5.0
共享内存
共享存储区（Share Memory）是Linux系统中通信速度最高的通信机制，因为数据不需要在客户机和服务器端之间复制，数据直接写到内存，不用若干次数据拷贝，所以这是最快的一种IPC。该机制中共享内存空间和进程的虚地址空间满足多对多的关系。即一个共享内存空间可以映射多个进程的虚地址空间，一个进程的虚地址空间又可以连接多个共享存储区。当进程间预利用共享存储区通信时，先要在主存中建立一个共享存储区，然后将它附接到自己的虚地址空间。该机制只为进程提供了用于实现通信的共享存储区和对共享存储区进行操作的手段，然而并未提供对该区进行互斥访问及进程同步的措施，所以要使用信号量来实现对共享内存的存取的同步。
实验设计
共享内存原理
通过上图可知，共享内存是通过将不同进程的虚拟内存地址映射到相同的物理内存地址来实现的。
在Linux内核中，每个共享内存都由一个名为 struct shmid_kernel 的结构体来管理，而且Linux限制了系统最大能创建的共享内存为128个。通过类型为 struct shmid_kernel 结构的数组来管理，如下：
struct shmid_ds {
struct ipc_perm shm_perm; /* operation perms */
int shm_segsz; /* size of segment (bytes) */
__kernel_time_t shm_atime; /* last attach time */
__kernel_time_t shm_dtime; /* last detach time */
__kernel_time_t shm_ctime; /* last change time */
__kernel_ipc_pid_t shm_cpid; /* pid of creator */
__kernel_ipc_pid_t shm_lpid; /* pid of last operator */
unsigned short shm_nattch; /* no. of current attaches */
unsigned short shm_unused; /* compatibility */
void *shm_unused2; /* ditto - used by DIPC */
void *shm_unused3; /* unused */
};
struct shmid_kernel
{
struct shmid_ds u;
/* the following are private */
unsigned long shm_npages; /* size of segment (pages) */
pte_t *shm_pages; /* array of ptrs to frames -> SHMMAX */
struct vm_area_struct *attaches; /* descriptors for attaches */
};
static struct shmid_kernel *shm_segs[SHMMNI]; // SHMMNI等于128
struct shmid_kernel 结构体中，shm_npages 字段表示共享内存使用了多少个内存页，而 shm_pages 字段指向了共享内存映射的虚拟内存页表项数组。
使用函数介绍
sprintf()
C 库函数 int sprintf(char *str, const char *format, ...) 发送格式化输出到 str 所指向的字符串。下面是 sprintf() 函数的声明：
int sprintf(char *str, const char *format, ...)
- str – 这是指向一个字符数组的指针，该数组存储了 C 字符串。
- format – 这是字符串，包含了要被写入到字符串 str 的文本。它可以包含嵌入的 format 标签，format 标签可被随后的附加参数中指定的值替换，并按需求进行格式化。
ftoc()
进程间的通讯，必须有个公共的标识来确保使用同一个通讯通道。任何一个进程如果使用同一个通讯标识，则内核就可以通过该标识找到对应的那个信道，这个标识就是IPC键值。
函数 ftok 把一个已存在的路径名和一个整数标识符转换成一个key_t值，称为IPC键值。函数原型如下：
key_t ftok(const char *pathname, int proj_id)
- pathname：指定的文件，此文件必须存在且可存取
- proj_id：计划代号（project ID）
函数 ftok 把从pathname导出的信息与id的低序8位组合成一个整数IPC键，从而避免用户使用key值的冲突。所需头文件 #include <sys/types.h> 和 #include <sys/ipc.h>，成功：返回key_t值（即IPC 键值），出错返回-1，错误原因存于error中。
给semget、msgget、shmget传入key值，它们返回的都是相应的IPC对象标识符。IPC键值和IPC标识符是两个概念，后者是建立在前者之上。
上图画出了从 IPC 键值生成 IPC 标识符图，其中key为 IPC 键值，由ftok函数生成，ipc_id 为IPC标识符，由 semget、msgget、shmget 函数生成。ipc_id 在信号量函数中称为 semid，在消息队列函数中称为 msgid，在共享内存函数中称为 shmid，它们表示的是各自 IPC 对象标识符。
struct ipc_perm {
key_t key ; /* 此IPC对象的key键 */
uid_t uid ; /* 此IPC对象用户ID */
gid_t gid ; /* 此IPC对象组ID */
uid_t cuid ; /* IPC对象创建进程的有效用户ID */
gid_t cgid ; /* IPC对象创建进程的有效组ID */
mode_t mode ; /* 此IPC的读写权限 */
ulong_t seq ; /* IPC对象的序列号 */
};
系统为每一个IPC对象保存一个ipc_perm结构体，该结构说明了IPC对象的权限和所有者，并确定了一个IPC操作是否可以访问该IPC对象。
msgget、semget、shmget 函数最右边的形参flag（msgget中为msgflg、semget中为semflg、shmget中shmflg）为IPC对象创建权限。IPC对象创建权限（即flag）格式为0xxxxx，其中0表示8位制，低三位为用户、属组、其他的读、写、执行权限（执行位不使用）。IPC对象存取权限常与下面IPC_CREAT、IPC_EXCL两种标志进行或（|）运算完成对IPC对象创建的管理，下面是两种创建模式标志在<sys/ipc.h>头文件中的宏定义。
#define IPC_CREAT 01000 /* Create key if key does not exist. */
#define IPC_EXCL 02000 /* Fail if key exists. */
获取共享内存
使用 shmget() 函数获取共享内存，shmget() 函数的原型如下：
int shmget(key_t key, size_t size, int shmflg);
- 参数
key一般由ftok()函数生成，用于标识系统的唯一IPC资源。 - 参数
size指定创建的共享内存大小。 - 参数
shmflg指定shmget()函数的动作，比如传入IPC_CREAT表示要创建新的共享内存。
函数调用成功时返回一个新建或已经存在的的共享内存标识符，取决于shmflg的参数。失败返回-1，并设置错误码。需要头文件 #include <sys/shm.h>。
shmget() 函数的实现比较简单，首先调用 findkey() 函数查找值为key的共享内存是否已经被创建，findkey() 函数返回共享内存在 shm_segs数组 的索引。如果找到，那么直接返回共享内存的标识符即可。否则就调用 newseg() 函数创建新的共享内存。newseg() 函数的实现也比较简单，就是创建一个新的 struct shmid_kernel 结构体，然后设置其各个字段的值，并且保存到 shm_segs数组 中。
关联共享内存
shmget() 函数返回的是一个标识符，而不是可用的内存地址，所以还需要调用 shmat() 函数把共享内存关联到某个虚拟内存地址上。shmat() 函数的原型如下：
void *shmat(int shmid, const void *shmaddr, int shmflg);
- 参数
shmid是shmget()函数返回的标识符。 - 参数
shmaddr是要关联的虚拟内存地址，如果传入0，表示由系统自动选择合适的虚拟内存地址。 - 参数
shmflg若指定了SHM_RDONLY位，则以只读方式连接此段，否则以读写方式连接此段。
函数调用成功返回一个可用的指针（虚拟内存地址），出错返回-1。
shmget() 主要通过 shmid 标识符来找到共享内存描述符，系统中所有的共享内存到保存在 shm_segs 数组中。接着，找到一个可用的虚拟内存地址，如果在调用 shmat() 函数时没有指定了虚拟内存地址，那么就通过 get_unmapped_area() 函数来获取一个可用的虚拟内存地址。通过调用 kmem_cache_alloc() 函数创建一个 vm_area_struct 结构，vm_area_struct 结构用于管理进程的虚拟内存空间。shmat() 函数只是申请了进程的虚拟内存空间，而共享内存的物理空间并没有申请，当进程发生缺页异常的时候会调用 shm_nopage() 函数来恢复进程的虚拟内存地址到物理内存地址的映射。shm_nopage() 函数的主要功能是当发生内存缺页时，申请新的物理内存页，并映射到共享内存中。由于使用共享内存时会映射到相同的物理内存页上，从而不同进程可以共用此块内存。
取消关联共享内存
当一个进程不需要共享内存的时候，就需要取消共享内存与虚拟内存地址的关联。取消关联共享内存通过 shmdt() 函数实现，原型如下：
int shmdt(const void *shmaddr);
- 参数
shmaddr是要取消关联的虚拟内存地址，也就是shmat()函数返回的值。
函数调用成功返回0，出错返回-1。
实验设计与编程
设置两个进程，分别为生产者 进程A和消费者 进程B，进程A 创建一块共享内存，然后写入数据（字符串：Process A generated!），进程B 获取这块共享内存并且读取其字符串内容并输出。
生产者进程A
/* Process A - Producer */
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#define SHM_PATH "~/shm" // 指定文件，以此生成 IPC key
#define SHM_SIZE 128 // 设置共享内存区大小：128字节
int main(int argc, char *argv[])
{
int shmid; // 共享内存标识符
char *addr; // 虚存中的字符串地址
key_t key = ftok(SHM_PATH, 0x0066); // 生成 IPC key
// 在进程A的虚存中开辟对应key的共享存储空间
// 共享存储区大小为 128 字节
// 指定 flag 为 IPC_CREAT|IPC_EXCL|0666
// 因为有 IPC_EXCL 存在，所以第二次运行时，对应 key 已经存在，所以会失败
shmid = shmget(key, SHM_SIZE, IPC_CREAT|IPC_EXCL|0666);
if (shmid < 0) {
printf("failed to create share memory\n");
return -1;
}
// 将虚拟内存的共享空间和物理内存的共享空间相关联
// 返回对应的虚存地址
addr = shmat(shmid, NULL, 0);
if (addr <= 0) {
printf("failed to map share memory\n");
return -1;
}
// 向进程A的虚存地址 addr 中写入字符串
// 实际写入了在A和B的公共物理内存上
sprintf(addr, "%s", "Process A created!\n");
return 0;
}
消费者进程B
/* Process B - Consumer */
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#define SHM_PATH "~/shm" // 指定文件，以此生成 IPC key
#define SHM_SIZE 128 // 设置共享内存区大小：128字节
int main(int argc, char *argv[])
{
int shmid; // 共享内存标识符
char *addr; // 进程B虚存中的字符串地址
key_t key = ftok(SHM_PATH, 0x0066); // 生成 IPC key
char buf[128]; // 接受进程B虚存对应的共享物理内存中的内容
// 在进程B的虚存中开辟对应key的共享存储空间
// 对应 key 已经存在，所以 flag 参数不加 IPC_EXCL
shmid = shmget(key, SHM_SIZE, IPC_CREAT);
if (shmid < 0) {
printf("failed to get share memory\n");
return -1;
}
// 将虚拟内存的共享空间和物理内存的共享空间相关联
addr = shmat(shmid, NULL, 0);
if (addr <= 0) {
printf("failed to map share memory\n");
return -1;
}
// 拷贝出共享物理内存中对应的 128 字节数据
// 输出：Process A created!
strcpy(buf, addr);
printf("%s", buf);
return 0;
}
实验结果与分析
共享内存是进程间通信中最简单的方式之一。共享内存允许两个或更多进程访问同一块内存，当一个进程改变了这块地址中的内容的时候，其它进程都会察觉到这个更改。因为所有进程共享同一块内存，共享内存在各种进程间通信方式中具有最高的效率。访问共享内存区域和访问进程独有的内存区域一样快，并不需要通过系统调用或者其它需要切入内核的过程来完成。同时它也避免了对数据的各种不必要的复制。
但是，系统内核没有对访问共享内存进行同步，解决这些问题的常用方法是通过使用信号量进行同步。虽然每个使用者都可以读取写入数据，但是所有程序之间必须达成并遵守一定的协议，以防止诸如在读取信息之前覆写内存空间等竞争状态的出现。不幸的是，Linux无法严格保证提供对共享内存块的独占访问，甚至是在通过使用IPC_PRIVATE创建新的共享内存块的时候也不能保证访问的独占性。 同时，多个使用共享内存块的进程之间必须协调使用同一个键值。
信号量
信号量是一个计数器，可以用来控制多个线程对共享资源的访问。它不是用于交换大批数据，而用于多线程之间的同步。它常作为一种锁机制，防止某进程在访问资源时其它进程也访问该资源。因此，主要作为进程间以及同一个进程内不同线程之间的同步手段。
实验设计
信号量的工作原理
信号量本质上是一个计数器，它不以传送数据为主要目的，它主要是用来保护共享资源（信号量也属于临界资源），使得资源在一个时刻只有一个进程独享。
信号量有初值（>0），每当有进程申请使用信号量，通过一个P操作来对信号量进行-1操作，当计数器减到0的时候就说明没有资源了，其他进程要想访问就必须等待。当该进程执行完这段工作之后，就会执行V操作，对信号量进行+1操作。
当有进程要求使用共享资源时，需要执行以下操作：
- 系统首先要检测该资源的信号量
- 若该资源的信号量值大于0，则进程可以使用该资源，此时，进程将该资源的信号量值减-1
- 若该资源的信号量值为0，则进程进入休眠状态，直到信号量值大于0时进程被唤醒，访问该资源
- 当进程不再使用由一个信号量控制的共享资源时，该信号量值增+1，如果此时有进程处于休眠状态等待此信号量，则该进程会被唤醒
在信号量进行PV操作时都为原子操作（因为它需要保护临界资源）。
二元信号量（Binary Semaphore）是最简单的一种锁（互斥锁），它只用两种状态：占用与非占用，通常用来替代互斥锁实现线程同步。所以它的引用计数为1。
使用函数介绍
Linux提供了一组精心设计的信号量接口来对信号进行操作。
头文件
#include <sys/sem.h> // 有名信号量
信号量在进程是以有名信号量进行通信的，在线程中则是以无名信号进行通信的。本实验主要实现进程间信号量的通信，所以使用 <sys/sem.h> .
信号量创建
作用是创建一个新信号量或取得一个已有信号量，原型为：
int semget(key_t key, int num_sems, int sem_flags);
- num_sems指定需要的信号量数目，它的值几乎总是1
- sem_flags是一组标志，设置了IPC_CREAT标志后，即使给出的键是一个已有信号量的键，也不会产生错误。而IPC_CREAT | IPC_EXCL则可以创建一个新的，唯一的信号量，如果信号量已存在，返回一个错误。
信号量初始化和删除
semctl() 函数用来直接控制信号量信息，它的原型为：
int semctl(int sem_id, int sem_num, int command, ...);
- command 通常是下面两个值中的其中一个：
SETVAL用来把信号量初始化为一个已知的值，p 值通过union semun中的val成员设置，其作用是在信号量第一次使用前对它进行设。IPC_RMID用于删除一个已经无需继续使用的信号量标识符。
如果有第四个参数，它通常是一个union semum结构，定义如下：
union semun {
int val;
struct semid_ds *buf;
unsigned short *arry;
};
信号量值修改
作用是改变信号量的值，原型为：
int semop(int sem_id, struct sembuf *sem_opa, size_t num_sem_ops);
- sem_id是由semget()返回的信号量标识符
sembuf 结构的定义如下：
struct sembuf{
short sem_num; // 除非使用一组信号量，否则它为0
short sem_op; // 信号量在一次操作中需要改变的数据，通常是两个数，一个是-1，即P（等待）操作，
// 一个是+1，即V（发送信号）操作。
short sem_flg; // 通常为SEM_UNDO,使操作系统跟踪信号，
// 并在进程没有释放该信号量而终止时，操作系统释放信号量
};
实验设计与编程
使用二元信号量对“共享内存”方式进行改进。原本进程A和B能同时访问并修改共享内存区，进程A写入字符串，进程B读出字符串。现在，假定进程A写入时间需要5s，先启动进程A，A在执行写入前进行P操作，此时启动B，B执行读取操作，但因为有信号量对临界区的限制，所以进程B此时应当被挂起。5s后进程A写入成功，执行V操作，脱离临界区，进程B从休眠态被唤醒，从共享内存区中取进程A存入的字符串数据。最后，B释放共享内存区并删除信号量。
进程A
//
// Created by Sylvan Ding on 2022/3/5.
//
/* Process A - Producer */
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include "uni_clock.h"
#define SHM_PATH "~/shm"
#define SHM_SIZE 128
#define SEM_PATH "~/sem"
union semun
{
// 信息量的控制单元
int val;
struct semid_ds *buf;
unsigned short *arry;
};
static int set_semvalue(int semid);
static int semaphore_p(int semid);
static int semaphore_v(int semid);
int main(int argc, char *argv[])
{
int shmid;
char *addr;
key_t key = ftok(SHM_PATH, 0x0066);
// 生成信号量标识符
int semid;
key_t key_sem = ftok(SEM_PATH, 0x0066);
semid = semget(key_sem, 1, IPC_CREAT|IPC_EXCL|0666);
// 信号量初始化
if(set_semvalue(semid)<0) {
printf("failed to initialize semaphore\n");
return -1;
}
// 开辟共享存储空间
shmid = shmget(key, SHM_SIZE, IPC_CREAT|IPC_EXCL|0666);
if (shmid < 0) {
printf("failed to create share memory\n");
return -1;
}
// 将虚拟内存的共享空间和物理内存的共享空间相关联
addr = shmat(shmid, NULL, 0);
if (addr <= 0) {
printf("failed to map share memory\n");
return -1;
}
// 进入临界区
if(semaphore_p(semid)<0) {
printf("A semaphore_p failed\n");
return -1;
};
// 进程A开始向共享内存中写入字符串
printf("%s: Process A is on writing...\n", getTime());
sleep(5); // 模拟写入过程5s
sprintf(addr, "%s", "Hello, world!\n");
// 进程A提示写入完毕
printf("%s: Process A has finished writing!\n", getTime());
// 进程A写入完毕，退出临界区
if(semaphore_v(semid)<0) {
printf("A semaphore_v failed\n");
return -1;
}
// 断开进程A与共享内存区的连接
shmdt(addr);
return 0;
}
static int set_semvalue(int semid) {
// 信息量初始化
union semun sem_union;
sem_union.val = 1; // 二元信息量
if(semctl(semid, 0, SETVAL, sem_union) == -1) {
return -1;
}
return 0;
}
static int semaphore_p(int semid)
{
// 等待P（sv）
struct sembuf sem_b;
sem_b.sem_num = 0;
sem_b.sem_op = -1; // P()
sem_b.sem_flg = SEM_UNDO;
if(semop(semid, &sem_b, 1) == -1) {
return -1;
}
return 0;
}
static int semaphore_v(int semid)
{
// V（sv）
struct sembuf sem_b;
sem_b.sem_num = 0;
sem_b.sem_op = 1; // V()
sem_b.sem_flg = SEM_UNDO;
if(semop(semid, &sem_b, 1) == -1) {
return -1;
}
return 0;
}
进程B
//
// Created by Sylvan Ding on 2022/3/5.
//
/* Process B - Consumer */
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include "uni_clock.h"
#define SHM_PATH "~/shm"
#define SHM_SIZE 128
#define SEM_PATH "~/sem"
union semun
{
// 信息量的控制单元
int val;
struct semid_ds *buf;
unsigned short *arry;
};
static int semaphore_p(int semid);
static int semaphore_v(int semid);
static int del_semvalue(int semid);
int main(int argc, char *argv[])
{
int shmid;
char *addr;
key_t key = ftok(SHM_PATH, 0x0066);
// 生成信号量标识符
int semid;
key_t key_sem = ftok(SEM_PATH, 0x0066);
semid = semget(key_sem, 1, IPC_CREAT);
char buf[128];
// 进程B创建共享存储空间
shmid = shmget(key, SHM_SIZE, IPC_CREAT);
if (shmid < 0) {
printf("failed to get share memory\n");
return -1;
}
// 将虚拟内存的共享空间和物理内存的共享空间相关联
addr = shmat(shmid, NULL, 0);
if (addr <= 0) {
printf("failed to map share memory\n");
return -1;
}
// 进程B进入临界区
printf("%s: Process B started reading!\n", getTime());
if(semaphore_p(semid)<0) {
printf("B semaphore_p failed\n");
return -1;
};
// 进程B读共享内存中数据
sleep(2); // B读出数据需2s
strcpy(buf, addr);
printf("%s: %s", getTime(), buf);
// 退出临界区
if(semaphore_v(semid)<0) {
printf("B semaphore_v failed\n");
return -1;
}
// 销毁信号量
if(del_semvalue(semid)<0) {
printf("failed to delete semaphore\n");
return -1;
}
// 断开进程与共享内存区的连接
shmdt(addr);
// 删除共享内存区
shmctl(shmid, IPC_RMID, NULL);
return 0;
}
static int semaphore_p(int semid)
{
// 等待P（sv）
struct sembuf sem_b;
sem_b.sem_num = 0;
sem_b.sem_op = -1; // P()
sem_b.sem_flg = SEM_UNDO;
if(semop(semid, &sem_b, 1) == -1) {
return -1;
}
return 0;
}
static int semaphore_v(int semid)
{
// V（sv）
struct sembuf sem_b;
sem_b.sem_num = 0;
sem_b.sem_op = 1; // V()
sem_b.sem_flg = SEM_UNDO;
if(semop(semid, &sem_b, 1) == -1) {
return -1;
}
return 0;
}
static int del_semvalue(int semid)
{
// 删除信号量
union semun sem_union;
if(semctl(semid, 0, IPC_RMID, sem_union) == -1) {
return -1;
}
return 0;
}
uni_clock.h
//
// Created by Sylvan Ding on 2022/3/5.
//
#ifndef UNTITLED_UNI_CLOCK_H
#define UNTITLED_UNI_CLOCK_H
#include <time.h>
char *getTime() {
time_t t;
time(&t);
struct tm *tn;
tn = localtime(&t);
return asctime(tn);
}
#endif //UNTITLED_UNI_CLOCK_H
实验结果与分析
由于竞争信号量的时候，未能拿到信号的进程会进入睡眠，所以信号量可以适用于长时间持有。而且信号量不适合短时间的持有，因为会导致睡眠的原因，维护队列、唤醒等各种开销，在短时锁定，效率较低。由于睡眠的特性，只能在进程上下文进行调用，无法再中断上下文中使用信号量。一个进程可以在持有信号量的情况下去睡眠，另外的进程尝试获取该信号量时候，不会死锁。
管道
子进程从父进程继承文件描述符。来源于早起Unix命令行输入时的想法：能不能让上一个进程的输出重定向为下一个进程的输入。流水线方式，称为管道机制，即子进程共享父进程的一些资源。
管道的特点：
- 管道只允许具有血缘关系的进程间通信，如父子进程间的通信
- 管道只允许单向通信
- 管道内部保证同步机制，从而保证访问数据的一致性
- 面向字节流
- 管道随进程，进程在管道在，进程消失管道对应的端口也关闭，两个进程都消失管道也消失
实验设计
管道的实现机制
在Linux中，管道是一种使用非常频繁的通信机制。从本质上说，管道也是一种文件，但它又和一般的文件有所不同，管道可以克服使用文件进行通信的两个问题，具体表现为：
- 限制管道的大小。实际上，管道是一个固定大小的缓冲区。在Linux中，该缓冲区的大小为1页，即4K字节，使得它的大小不象文件那样不加检验地增长。使用单个固定缓冲区在写管道时可能变满，当这种情况发生时，随后对管道的write()调用将默认地被阻塞，等待某些数据被读取，以便腾出足够的空间供write()调用写；
- 读取进程也可能工作得比写进程快。当所有当前进程数据已被读取时，管道变空。当这种情况发生时，一个随后的read()调用将默认地被阻塞，等待某些数据被写入，这解决了read()调用返回文件结束的问题。
- 注意：从管道读数据是一次性操作，数据一旦被读，它就从管道中被抛弃，释放空间以便写更多的数据。
在 Linux 中，管道的实现并没有使用专门的数据结构，而是借助了文件系统的file结构和VFS的索引节点inode。通过将两个 file 结构指向同一个临时的 VFS 索引节点，而这个 VFS 索引节点又指向一个物理页面而实现的。
环形缓冲区
在内核中，管道 使用了环形缓冲区来存储数据。环形缓冲区的原理是：把一个缓冲区当成是首尾相连的环，其中通过读指针和写指针来记录读操作和写操作位置。
管道对象
在 Linux 内核中，管道使用 pipe_inode_info 对象来进行管理，pipe_inode_info 对象的定义，如下所示：
struct pipe_inode_info {
wait_queue_head_t wait;
unsigned int nrbufs,
unsigned int curbuf;
...
unsigned int readers;
unsigned int writers;
unsigned int waiting_writers;
...
struct inode *inode;
struct pipe_buffer bufs[16];
};
wait：等待队列，用于存储正在等待管道可读或者可写的进程。bufs：环形缓冲区，由 16 个pipe_buffer对象组成，每个pipe_buffer对象拥有一个内存页 。nrbufs：表示未读数据已经占用了环形缓冲区的多少个内存页。curbuf：表示当前正在读取环形缓冲区的哪个内存页中的数据。readers：表示正在读取管道的进程数。writers：表示正在写入管道的进程数。waiting_writers：表示等待管道可写的进程数。inode：与管道关联的inode对象。
环形缓冲区是由 16 个 pipe_buffer 对象组成，定义如下：
struct pipe_buffer {
struct page *page;
unsigned int offset;
unsigned int len;
...
};
page：指向pipe_buffer对象占用的内存页。offset：如果进程正在读取当前内存页的数据，那么offset指向正在读取当前内存页的偏移量。len：表示当前内存页拥有未读数据的长度。
无名管道的实验设计与编程
无名管道一般用于父子进程之间相互通信，具体流程如下：
- 父进程使用
pipe系统调用创建一个管道 - 父进程使用
fork系统调用创建一个子进程 - 由于子进程会继承父进程打开的文件句柄，所以父子进程可以通过新创建的管道进行通信
管道是单向的，所以要将管道分为读端和写端，需要两个文件描述符来管理管道：fd[0] 为读端，fd[1] 为写端。
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <stdlib.h>
#include <string.h>
int main()
{
int fd[2]; // 用于管理管道的文件描述符
pid_t pid; // 子进程pid
char buf[128] = {0};
char *msg = "Hello world!!!";
// 父进程创建管道
// pipe返回值：成功，返回0，否则返回-1.
// 参数数组包含pipe使用的两个文件的描述符.
// fd[0]:读管道，fd[1]:写管道.
if (-1 == pipe(fd)) {
printf("failed to create pipe\n");
return -1;
}
// 创建子进程
// fork函数将运行着的程序分成2个（几乎）完全一样的进程，
// 每个进程都启动一个从代码的同一位置开始执行的线程。
// 返回负值：创建子进程失败，零返回到新创建的子进程，
// 正值返回父进程或调用者。
pid = fork();
if (pid<0) {
printf("failed to fork\n");
return -1;
}
// 子进程-生产者
if (0 == pid) {
// 关闭管道的读端
close(fd[0]);
// 向管道写端写入数据
if(write(fd[1], msg, strlen(msg))<0) {
printf("failed to write data\n");
exit(1);
};
exit(0);
} else {
// 父进程-消费者
// 关闭管道的写端
close(fd[1]);
// 从管道的读端读取数据
if(read(fd[0], buf, sizeof(buf))<0) {
printf("failed to read data\n");
return -1;
};
printf("Parent read data: %s\n", buf);
}
return 0;
}
实验结果与分析
父子进程通过 pipe 系统调用打开的管道，在内核空间中指向同一个管道对象（pipe_inode_info）。所以父子进程共享着同一个管道对象，那么就可以通过这个共享的管道对象进行通信。
使用 pipe 系统调用打开管道时，并没有立刻申请内存页，而是当有进程向管道写入数据时，才会按需申请内存页。当内存页的数据被读取完后，内核会将此内存页回收，来减少管道对内存的使用。
消息队列
消息队列是消息的链表，存放在内核中并由消息队列标识符标识。消息队列克服了信号传递信息少，管道只能承载无格式字节流以及缓冲区大小受限等特点。消息队列是UNIX下不同进程之间可实现共享资源的一种机制，UNIX允许不同进程将格式化的数据流以消息队列形式发送给任意进程。对消息队列具有操作权限的进程都可以使用msget完成对消息队列的操作控制。通过使用消息类型，进程可以按任何顺序读信息，或为消息安排优先级顺序。
实验设计
消息队列的实现原理
- 消息队列的本质其实是一个内核提供的链表，内核基于这个链表，实现了一个数据结构;
- 向消息队列中写数据，实际上是向这个数据结构中插入一个新结点；从消息队列汇总读数据，实际上是从这个数据结构中删除一个结点;
- 消息队列提供了一个从一个进程向另外一个进程发送一块数据的方法;
- 消息队列也有管道一样的不足，就是每个数据块的最大长度是有上限的，系统上全体队列的最大总长度也有一个上限。Linux用宏MSGMAX和MSGMNB来限制一条消息的最大长度和一个队列的最大长度.
用户消息缓冲区
无论发送进程还是接收进程，都需要在进程空间中用消息缓冲区来暂存消息。该消息缓冲区的结构定义如下：
struct msgbuf {
long mtype; /* 消息的类型 */
char mtext[]; /* 消息正文 */
};
- 可通过
mtype区分数据类型，同过判断mtype，是否为需要接收的数据 mtext[]为存放消息正文的数组，可以根据消息的大小定义该数组的长度
使用函数介绍
消息队列的创建
通过msgget创建消息队列，函数原型如下：
#include <sys/msg.h>
int msgget(key_t key, int msgflg);
- 成功 msgget 将返回一个非负整数，即该消息队列的标识码；
- 失败 则返回“-1”
向消息队列中添加信息
向消息队列中添加数据，使用到的是msgsnd()函数，函数原型如下：
int msgsnd(int msgid, const void *msg_ptr, size_t msg_sz, int msgflg);
- msgid: 由msgget函数返回的消息队列标识码
- msg_ptr: 是一个指针，指针指向准备发送的消息，
- msg_sz: 是msg_ptr指向的消息长度，消息缓冲区结构体中mtext的大小,不包括数据的类型
- msgflg: 控制着当前消息队列满或到达系统上限时将要发生的事情。比如，
msgflg = IPC_NOWAIT表示队列满不等待，返回EAGAIN错误
注意，消息的数据结构却有一定的要求，指针msg_ptr所指向的消息结构一定要是以一个长整型成员变量开始的结构体，接收函数将用这个成员来确定消息的类型。所以消息结构要定义成这样：
struct my_message {
long int message_type;
/* The data you wish to transfer */
};
注意：
msg_sz是msg_ptr指向的消息的长度，注意是消息的长度，而不是整个结构体的长度，也就是说msg_sz是不包括长整型消息类型成员变量的长度。
从消息队列中读取信息
msgrcv() 用来从一个消息队列获取消息，它的原型为：
int msgrcv(int msgid, void *msg_ptr, size_t msg_st, long int msgtype, int msgflg);
- msgid: 由msgget函数返回的消息队列标识码
- msg_ptr: 是一个指针，指针指向准备接收的消息
- msgsz: 是msg_ptr指向的消息长度
- msgtype: 可以实现接收优先级的简单形式
- msgtype=0返回队列第一条信息
- msgtype>0返回队列第一条类型等于msgtype的消息
- msgtype<0返回队列第一条类型小于等于msgtype绝对值的消息
- msgflg: 控制着队列中没有相应类型的消息可供接收时将要发生的事。比如，
msgflg=IPC_NOWAIT，队列没有可读消息不等待，返回ENOMSG错误。msgflg=MSG_NOERROR，消息大小超过msgsz时被截断
调用成功时，该函数返回放到接收缓存区中的字节数，消息被复制到由msg_ptr指向的用户分配的缓存区中，然后删除消息队列中的对应消息。失败时返回-1。
消息队列的控制函数
控制函数原型如下：
int msgctl(int msqid, int command, strcut msqid_ds *buf);
- command: 是将要采取的动作，它可以取3个值
- IPC_STAT：把msgid_ds结构中的数据设置为消息队列的当前关联值，即用消息队列的当前关联值覆盖msgid_ds的值
- IPC_SET：如果进程有足够的权限，就把消息列队的当前关联值设置为msgid_ds结构中给出的值
- IPC_RMID：删除消息队列. 注意：若选择删除队列，第三个参数传NULL
- buf是指向msgid_ds结构的指针，它指向消息队列模式和访问权限的结构
- 如果操作成功，返回“0”；如果失败，则返回“－1”
msgid_ds结构至少包括以下成员：
struct msgid_ds
{
uid_t shm_perm.uid;
uid_t shm_perm.gid;
mode_t shm_perm.mode;
};
ipcs 命令
Linux系统下自带的ipcs命令，可以查看当前系统下共享内存、消息队列、信号量等等的使用情况，从而利于定位多进程通信中出现的通信问题。
上图分别查看了当前unix系统中信号量、共享内存和消息队列的使用情况。
命令格式如下：
ipcs [resource-option] [output-format]
resource选项：
ipcs -a显示系统内所有的IPC信息- …
输出格式控制：
ipcs -p查看IPC资源的创建者和使用的进程IDipcs -u查看IPC资源状态汇总信息
ipcrm通过指定ID删除IPC资源，同时将与IPC对象关联的数据一并删除，只有超级用户或IPC资源创建者能够删除。
实验设计与编程
- 编写消费者进程 A，A 生成 IPC key，并创建消息队列，使 A 保持对消息队列的监听，A 接受类型为 1 的消息；
- 编写生产者进程 B，B 和 A 使用相同的消息队列。在另一个终端上运行进程 B，B 向消息队列中发送自定义类型和内容的消息，观察 A 所在终端的输出。
消费者进程 A
//
// Created by Sylvan Ding on 2022/3/5.
//
/* Process A - Consumer */
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#define IPCKEY 666 // ipc key
#define MSGSIZE 128 // message size
struct MyMsg {
long msgtype; // msg类型
char msg_cont[MSGSIZE]; // msg内容
};
int msgid;
int ret_val;
long msgtype = 1; // msg type to receive
struct MyMsg mymsg = {0};
const char *quit_msg = "quit"; // message to quit
int main(int argc, char *argv[]) {
// 创建消息队列
msgid = msgget((key_t)IPCKEY, IPC_CREAT|IPC_EXCL|0666);
if (msgid < 0) {
printf("failed to create message queue\n");
return -1;
}
// 从消息队列中获取类型为1的消息
mymsg.msgtype = msgtype;
while (1) {
ret_val = msgrcv(msgid, &mymsg, sizeof(mymsg.msg_cont), msgtype, IPC_NOWAIT);
if (ret_val > 0) {
// 退出监听
if (!strcmp(mymsg.msg_cont, quit_msg)) {
printf("Process A received a command to QUIT!\n");
break;
}
// 打印接收信息
printf("A received a message from msq: %s\n", mymsg.msg_cont);
fflush(stdout);
}
}
// 消息队列释放
msgctl(msgid, IPC_RMID, NULL);
return 0;
}
生产者进程 B
//
// Created by Sylvan Ding on 2022/3/5.
//
/* Process B - Producer */
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#define IPCKEY 666 // ipc key
#define MSGSIZE 128 // message size
struct MyMsg {
long msgtype; // msg类型
char msg_cont[MSGSIZE]; // msg内容
};
int msgid;
int ret_val;
struct MyMsg mymsg = {0};
int main(int argc, char *argv[]) {
// 创建消息队列
msgid = msgget((key_t)IPCKEY, IPC_CREAT);
if (msgid < 0) {
printf("failed to create message queue\n");
return -1;
}
// 输入消息类型和内容
printf("Here is the procedure B: \n");
printf("mgs_type = ");
scanf("%ld", &(mymsg.msgtype));
printf("input your msg:\n");
scanf("%s", mymsg.msg_cont);
// 向消息队列发送信息
ret_val = msgsnd(msgid, &mymsg, sizeof(mymsg.msg_cont), IPC_NOWAIT);
if (ret_val < 0) {
printf("failed to send message\n");
return -1;
}
return 0;
}
实验结果与分析
总结和实验心得
本次实验基于Linux系统实现了进程的共享内存、管道和消息队列等三种IPC方式。其中，共享内存使各个进程共享一段物理内存空间，管道为父子进程间信息传递带来便捷，消息队列提供了消息类型选择机制。实验还通过信号量改进了共享内存，实现了进程对存储空间的互斥访问。通过实验，初步掌握了Linux系统的IPC操作，了解了他们各自的优缺点，现做总结如下：
如果用户传递的信息较少，或是需要通过信号来触发某些行为，软中断信号机制不失为一种简捷有效的进程间通信方式。但若是进程间要求传递的信息量比较大或者进程间存在交换数据的要求，那就需要考虑别的通信方式了。
无名管道简单方便。但局限于单向通信的工作方式。并且只能在创建它的进程及其子孙进程之间实现管道的共享：有名管道虽然可以提供给任意关系的进程使用。但是由于其长期存在于系统之中，使用不当容易出错，所以普通用户一般不建议使用。
消息缓冲可以不再局限于父子进程，而允许任意进程通过共享消息队列来实现进程间通信。并由系统调用函数来实现消息发送和接收之间的同步。从而使得用户在使用消息缓冲进行通信时不再需要考虑同步问题。使用方便，但是信息的复制需要额外消耗CPU的时间。不适宜于信息量大或操作频繁的场合。
共享内存针对消息缓冲的缺点改而利用内存缓冲区直接交换信息，无须复制，快捷、信息量大是其优点。但是共享内存的通信方式是通过将共享的内存缓冲区直接附加到进程的虚拟地址空间中来实现的。因此，这些进程之间的读写操作的同步问题操作系统无法实现。必须由各进程利用其他同步工具解决。另外，由于内存实体存在于计算机系统中。所以只能由处于同一个计算机系统中的诸进程共享。不方便网络通信。
不同的进程通信方式有不同的优点和缺点。因此。对于不同的应用问题，要根据问题本身的情况来选择进程间的通信方式。
参考文献
- Linux下进程间通信方式——共享内存
- 6种Linux进程间的通信方式
- 进程通信——百度百科
- Linux共享存储通信
- 一文搞定：Linux共享内存原理
- Linux信号量详解
- Linux下进程间通信方式——信号量（Semaphore）
- 进程间通信的方式（四）：信号量
- linux管道详解
- 图解 | Linux进程通信 - 管道实现
- Linux进程间通信——消息队列
- Linux进程间通信（七）：消息队列 msgget()、msgsend()、msgrcv()、msgctl()
- ipcs命令详解
转载注意事项
本文为原创文章，转载需要在文章开头或结尾注明出处！
?? ?? Sylvan Ding’s Blog ??