【Linux】Linux 进程间通讯-管道

参考博客：https://blog.csdn.net/sjsjnsjnn/article/details/125864580

一、进程间通讯介绍

1.1 进程间通讯的概念

进程通信（Interprocess communication），简称：IPC

• 本来进程之间是相互独立的。但是由于不同的进程之间可能要共享某些信息，所以就必须要有通讯来实现进程间的互斥和同步。比如说共享同一块内存、管道、消息队列、信号量等等就是实现这一过程的手段，相当于移动公司在打电话的作用。

1.2 进程间通讯的目的

• 数据传输：一个进程需要将它的数据发送给另一个进程
• 资源共享：多个进程之间共享同样的资源。
• 通知事件：一个进程需要向另一个或一组进程发送消息，通知它（它们）发生了某种事件（如进程终止时要通知父进程）。
• 进程控制：有些进程希望完全控制另一个进程的执行（如Debug进程），此时控制进程希望能够拦截另一个进程的所有陷入和异常，并能够及时知道它的状态改变

1.3 进程间通信的前提

• 进程间通信的前提本质：由操作系统参与，提供一份所有通信进行都能看到的公共资源；
• 两个或多个进程相互通信，必须先看到一份公共的资源，这里的所谓的资源是属于操作系统的，就是一段内存（可能以文件的方式提供、可能以队列的方式提供，也有可能提供的就是原始内存块），这也就是通信方式有很多种的原因；

1.4 进程间通信的分类

管道

• 匿名管道pipe
• 命名管道

System V IPC

• System V 消息队列
• System V 共享内存
• System V 信号量

POSIX IPC

• 消息队列
• 共享内存
• 信号量
• 互斥量
• 条件变量
• 读写锁

二、管道通讯

2.1 管道的概念

• 管道是Unix中最古老的进程间通信的形式。
• 我们把从一个进程连接到另一个进程的一个数据流称为一个“管道”

比如下面的命令，我们通过管道连接了cat test.c和wc -l两个命令，本质是两个进程

cat test.c | wc -l

• 运行的结果如下，统计了test.c文件的行数(wc)，并且将对应的结果输出了出来(cat)
• 这里执行的顺序为从右到左，先执行wc -l

2.2 匿名管道

2.2.1 基本原理

• 匿名管道用于进程间通信，且仅限于父子进程之间的通信。

• 我们知道进程的PCB中包含了一个指针数组 struct file_struct，它是用来描述并组织文件的。父进程和子进程均有这个指针数组，因为子进程是父进程的模板，其代码和数据是一样的；

• 打开一个文件时，其实是将文件加载到内核中，内核将会以结构体(struct file)的形式将文件的相关属性、文件操作的指针集合（即对应的底层IO设备的调用方法）等；

• 当父进程进行数据写入时（例如：写入“hello Linux”），数据是先被写入到用户级缓冲区，经由系统调用函数，又写入到了内核缓冲区，在进程结束或其他的操作下才被写到了对应的设备中；

• 如果数据在写入设备之前，“hello Linux”是在内核缓冲区的，因为子进程和父进程是同时指向这个文件的，所以子进程是能够看到这个数据的，并且可以对其操作；

• 简单来说，父进程向文件写入数据时，不直接写入对应的设备中，而是将数据暂存在内核缓冲区中，交给子进程来处理；

所以这种基于文件的方式就叫做管道；

2.2.2 管道的创建步骤

• 在创建匿名管道实现父子进程间通信的过程中，需要pipe函数和fork函数搭配使用，具体步骤如下：

• 匿名管道属于单向通信，意味着父子进程只有一个端是打开的，实现父子通信的时候就需要根据自己的想要实现的情况，关闭对应的文件描述符；

pipe函数

#include <unistd.h>
int pipe(int pipefd[2]);

函数的参数是两个文件的描述符，是输出型参数：

• pipefd[0]：读管道 — 对应的文件描述符是3
• pipefd[1]：写管道 — 对应的文件描述符是4

返回值：成功返回0，失败返回-1；

2.2.3 匿名管道通讯

• 下面的代码通过使用fork和pipe函数实现父子进程之间的通讯
• 其中，父进程用于读取数据，子进程用于写入数据
• 由于管道是单向通讯的，因此需要关闭管道的另一端，即父进程关闭写端，子进程关闭读端

void test1(){
    int pipe_fd[2];
    memset(pipe_fd,0,sizeof(pipe_fd));

    int ret = pipe(pipe_fd);
    if(ret < 0 ){
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if(ret == 0){ //子进程
        close(pipe_fd[0]); //关闭子进程读端

        for(int i =1;i<=10;++i){
            std::string msg = "hello from child process :" + std::to_string(i) +" times";
            write(pipe_fd[1],msg.c_str(),msg.size());
            sleep(1);
        }

        exit(0);
    }

    close(pipe_fd[1]); //关闭父进程写端
    
    char buffer[1024];
    memset(buffer,0,sizeof(buffer));
    while(1){
        ssize_t s = read(pipe_fd[0],buffer,sizeof(buffer));
        if(s <= 0){
            std::cout << "read finished !" << std::endl;
            break; 
        }
        else{
            buffer[s] = '\0';
            std::cout << "read from child process : " << buffer << std::endl;
        }

    }
}

• 可以发现，管道的读写端的文件描述符为3，4，其中0,1，2通常是输入流、输出流和错误流
• 通过打印结果，可以发现父子进程成功通讯了

2.2.4 匿名管道通讯的特点

五个特点

1. 管道仅限父子通讯，只能单向通讯
2. 管道提供流式服务
3. 管道自带同步与互斥机制
4. 进程退出，管道随之释放，因此管道的生命周期随进程
5. 如果需要双向通讯，则需要建立两个管道

四个情况

1. 读端不读或者读得慢，写端要等待读端
2. 读端关闭，写端收到SIGPIPE信号后终止
3. 写端不写或者写得慢，读端要等待写端
4. 写端关闭，读端读到EOF后退出

2.2.5 字节流通讯

• 字节流的特征就是没有边界，每次读取指定的字节
• 我们发送数据的时候是先把数据写到内核缓冲区中，读取的时候也是从内核缓冲区读取指定的字节
• 因此，如果写端慢了，那么读取的数据会重合在一起，如下面的程序所示

void test1(){
    int pipe_fd[2];
    memset(pipe_fd,0,sizeof(pipe_fd));

    int ret = pipe(pipe_fd);
    if(ret < 0 ){
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if(ret == 0){ //子进程
        close(pipe_fd[0]); //关闭子进程读端

        for(int i =1;i<=10;++i){
            std::string msg = "hello from child process :" + std::to_string(i) +" times";
            write(pipe_fd[1],msg.c_str(),msg.size());
            sleep(1);
        }

        exit(0);
    }

    close(pipe_fd[1]); //关闭父进程写端
    
    char buffer[1024];
    memset(buffer,0,sizeof(buffer));
    while(1){
	    sleep(10);
        ssize_t s = read(pipe_fd[0],buffer,sizeof(buffer));
        if(s <= 0){
            std::cout << "read finished !" << std::endl;
            break; 
        }
        else{
            buffer[s] = '\0';
            std::cout << "read from child process : " << buffer << std::endl;
        }
    }
}

• 可以发现，读取的数据全都合并在一起了，因为我们指定读取的字节数较大

2.2.6 同步机制

• 内核的缓冲区是有大小限制的，下面我们不断发送数据到内核缓冲区，到了65536字节后，就发送不了了，此时阻塞了进程

void test2(){
    int pipe_fd[2];
    memset(pipe_fd,0,sizeof(pipe_fd));

    int ret = pipe(pipe_fd);
    if(ret < 0 ){
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if(ret == 0){ //子进程
        close(pipe_fd[0]); //关闭子进程读端
        int writedBytes = 0;
        for(int i =1;i<=10000000;++i){
            
            write(pipe_fd[1],"a",1);
            
            writedBytes ++;
            std::cout << "child process send msg: " << "a" <<",writed Bytes = " << writedBytes<< std::endl;
            
        }

        exit(0);
    }

    
    close(pipe_fd[1]); //关闭父进程写端
    char buffer[1024];
    memset(buffer,0,sizeof(buffer));
    while(1){
        sleep(1);
    }
}

• 管道通讯自带同步机制和互斥机制，也就是发送端和接收端看到的数据是一致的，并且同时只有一段可以读或者写
• 下面的程序在内核缓冲区写满了以后，尝试读取数据，发现只有读取了一些数据之后，才能继续往内核写入数据，而不是读取一个字节可以写入一个字节

void test3()
{
    int pipe_fd[2];
    memset(pipe_fd,0,sizeof(pipe_fd));

    int ret = pipe(pipe_fd);
    if(ret < 0 ){
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if(ret == 0){ //子进程
        close(pipe_fd[0]); //关闭子进程读端
        int writedBytes = 0;
        for(int i =1;i<=10000000;++i){
            
            write(pipe_fd[1],"a",1);
            
            writedBytes ++;
            std::cout << "child process send msg: " << "a" <<",writed Bytes = " << writedBytes<< std::endl;
        }

        exit(0);
    }

    
    close(pipe_fd[1]); //关闭父进程写端
    sleep(5);
    while(1){
        char c = 0;
        read(pipe_fd[0],&c,1);

        std::cout << "read :" << c << std::endl;
        sleep(1);
    }
}

• 读取部分数据后，才会继续写入
• 读端太慢，会导致写端等待读端

void test3()
{
    int pipe_fd[2];
    memset(pipe_fd,0,sizeof(pipe_fd));

    int ret = pipe(pipe_fd);
    if(ret < 0 ){
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if(ret == 0){ //子进程
        close(pipe_fd[0]); //关闭子进程读端
        int writedBytes = 0;
        for(int i =1;i<=10000000;++i){
            
            write(pipe_fd[1],"a",1);
            
            writedBytes ++;
            std::cout << "child process send msg: " << "a" <<",writed Bytes = " << writedBytes<< std::endl;
        }

        exit(0);
    }

    
    close(pipe_fd[1]); //关闭父进程写端
    sleep(5);
    char buffer[1024];
    memset(buffer,0,sizeof(buffer));
    int readBytes = 0;
    while(1){
        char c = 0;
        ssize_t s = read(pipe_fd[0],buffer,sizeof(buffer));
        buffer[s] = '\0';

        std::cout << "read :" << buffer << std::endl;
        std::cout << "read bytes = " << readBytes << std::endl;
        sleep(1);
        readBytes += s;
    }
}

2.2.7 写端关闭

• 写端关闭，那么读端会读到EOF后自动退出
• 比如下面的程序，我们让读进程先休眠一会，然后写进程写了一些数据后退出，那么读进程读到EOF后也就退出了

void test4()
{
    int pipe_fd[2];
    memset(pipe_fd, 0, sizeof(pipe_fd));

    int ret = pipe(pipe_fd);
    if (ret < 0)
    {
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if (ret == 0)
    {                      // 子进程
        close(pipe_fd[0]); // 关闭子进程读端
        for (int i = 1; i <= 10; ++i)
        {
            write(pipe_fd[1], "abcdefg", 7);
        }

        exit(0);
    }

    sleep(5);
    close(pipe_fd[1]); // 关闭父进程写端
    char buffer[1024];
    memset(buffer, 0, sizeof(buffer));

    int readBytes = 0;
    while (1)
    {
        char c = 0;
        ssize_t s = read(pipe_fd[0], buffer, sizeof(buffer));
        if (s <= 0)
        {
            std::cout << "read finished !" << std::endl;
            break;
        }
        buffer[s] = '\0';
        readBytes += s;
        std::cout << "read :" << buffer << std::endl;
        std::cout << "read bytes = " << readBytes << std::endl;
        sleep(1);
    }
}

2.2.8 读端关闭

• 读端关闭，写段会收到SIGPIPE信号，然后中断进程
• 当我们的读端关闭，写端还在写入，在操作系统的层面上，严重不合理；这本质上就是在浪费操作系统的资源，所以操作系统在遇到这样的情况下，会将子进程杀掉（发送13号信号—SIGPIPE）

下面的shell脚本用于持续跟踪测试进程

while :; do 
    ps axj | grep pipe_process | grep -v grep; 
    sleep 1; 
    echo "####################";
done;

• 可以发现，子进程退出后，父进程随之也退出了

• 这里我们添加上父进程等待子进程，也就是waitpid，然后输出对应的信号值
• 可以发现，退出后的信号值为13，对应的是SIGPIPE

void test5()
{
    int pipe_fd[2];
    memset(pipe_fd, 0, sizeof(pipe_fd));

    int ret = pipe(pipe_fd);
    if (ret < 0)
    {
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if (ret == 0)
    {                      // 子进程
        close(pipe_fd[0]); // 关闭子进程读端
        for (int i = 1; i <= 10; ++i)
        {
            write(pipe_fd[1], "abcdefg", 7);
            sleep(1);
        }

        exit(0);
    }

    close(pipe_fd[1]); // 关闭父进程写端
    char buffer[7];
    memset(buffer, 0, sizeof(buffer));

    int readBytes = 0;
    while (1)
    {
        char c = 0;
        ssize_t s = read(pipe_fd[0], buffer, sizeof(buffer));
        if (s <= 0)
        {
            std::cout << "read finished !" << std::endl;
            break;
        }
        buffer[s] = '\0';

        std::cout << "read :" << buffer << std::endl;
        readBytes += s;
        std::cout << "read bytes = " << readBytes << std::endl;
        sleep(2);
        close(pipe_fd[0]);
		
		int status = 0;
        waitpid(-1, &status, 0);
        printf("exit code: %d\n",(status >> 8)& 0xFF);
        printf("exit signal: %d\n",status& 0x7F);
    }
}

• 查询对应的信号，符合预期

kill -l

2.2.9 非阻塞管道

int pipe2(int pipefd[2], int flags);

• 可以通过pip2函数，设置管道通讯的阻塞与非阻塞
• 可以通过设置O_NONBLOCK标志为非阻塞，默认为阻塞，或者传入0

当没有数据可读时

• O_NONBLOCK disable：read调用阻塞，即进程暂停执行，一直等到有数据来到为止。
• O_NONBLOCK enable：read调用返回-1，errno值为EAGAIN。

当管道满的时候

• O_NONBLOCK disable： write调用阻塞，直到有进程读走数据
• O_NONBLOCK enable：调用返回-1，errno值为EAGAIN
• 如果所有管道写端对应的文件描述符被关闭，则read返回0
• 如果所有管道读端对应的文件描述符被关闭，则write操作会产生信号SIGPIPE,进而可能导致write进程退出
• 当要写入的数据量不大于PIPE_BUF时，linux将保证写入的原子性。
• 当要写入的数据量大于PIPE_BUF时，linux将不再保证写入的原子性。

2.2.9.1 非阻塞写入满了

• 下面的程序演示非阻塞管道写端，在内核写入满了以后的返回值

void test6(){
    int pipe_fd[2];
    memset(pipe_fd, 0, sizeof(pipe_fd));

    int ret = pipe2(pipe_fd,O_NONBLOCK);
    if (ret < 0)
    {
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if (ret == 0)
    {                      // 子进程
        close(pipe_fd[0]); // 关闭子进程读端
        int writedBytes = 0;
        for (int i = 1; i <= 10000000; ++i)
        {

            int ret = write(pipe_fd[1], "a", 1);
            if(ret == -1 && errno == EAGAIN){
                std::cout << "errno: EAGAIN !" << std::endl;
                sleep(1);
                continue;
            }

            writedBytes++;
            std::cout << "child process send msg: " << "a" << ",writed Bytes = " << writedBytes << std::endl;
        }

        exit(0);
    }

    close(pipe_fd[1]); // 关闭父进程写端

    int readBytes = 0;
    while (1)
    {
        sleep(1);
    }
}

2.2.9.2 非阻塞无数据可读

• 下面的程序演示非阻塞管道读取，无数据可读的返回值

void test7(){
    int pipe_fd[2];
    memset(pipe_fd, 0, sizeof(pipe_fd));

    int ret = pipe2(pipe_fd,O_NONBLOCK);
    if (ret < 0)
    {
        std::cout << "error:" << strerror(ret) << std::endl;
        return;
    }

    std::cout << "pipe_fd[0] = " << pipe_fd[0] << std::endl;
    std::cout << "pipe_fd[1] = " << pipe_fd[1] << std::endl;

    ret = fork();
    if (ret == 0)
    {                      // 子进程
        close(pipe_fd[0]); // 关闭子进程读端
        sleep(60);
    }

    close(pipe_fd[1]); // 关闭父进程写端

    int readBytes = 0;
    char buffer[1024] = {0};
    while (1)
    {
        ssize_t s = read(pipe_fd[0],buffer, sizeof(buffer));
        if(s == -1 && errno == EAGAIN){
            std::cout << "errno = " << "EAGAIN !" << std::endl;
            sleep(1);
            continue;
        }
    }
}

2.3 命名管道

2.3.1 命名管道的概念

• 匿名管道应用的一个限制就是只能在具有共同祖先（具有亲缘关系）的进程间通信。
• 如果我们想在不相关的进程之间交换数据，可以使用FIFO文件来做这项工作，它经常被称为命名管道。 命名管道是一种特殊类型的文件

2.3.2 命名管道的创建

2.3.2.1 命令行创建

可以通过命令行创建命名管道，使用mkfifo指令

创建管道之后，可以使用cat指令读取数据，使用echo指令写入数据

2.3.2.2 代码创建

• 使用mkfifo函数可以创建一个命名管道

函数原型

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

pathname：表示你要创建的命名管道文件

• 如果pathname是以文件的方式给出，默认在当前的路径下创建；
• 如果pathname是以某个路径的方式给出，将会在这个路径下创建；

mode：表示给创建的命名管道设置权限

• 我们在设置权限时，例如0666权限，它会受到系统的umask（文件默认掩码）的影响，实际创建出来是(mode & ~umask)0664;
• 所以想要正确的得到自己设置的权限（0666），我们需要将文件默认掩码设置为0；

返回值：命名管道创建成功返回0，失败返回-1

#define MY_FIFO "myfifo"

int main(int argc,char* argv[])
{
    umask(0);
    int ret = mkfifo(MY_FIFO,0666);
    if(ret < 0){
        perror("mkfifo");
        return 1;
    }

    return 0;
}

2.3.3 使用命名管道通讯

• 我们使用CS模型，在服务端创建命名管道，同时服务端不断读取数据，客户端发送数据
• 使用阻塞管道，当没有数据可读，服务端持续阻塞等待客户端的数据

server

#include<iostream>
#include<sys/stat.h>
#include<unistd.h>
#include<fcntl.h>
#include<cstring>

#define MY_FIFO "fifo"
int main(){
    umask(0);
    int ret = mkfifo(MY_FIFO,0666);

    if(ret < 0){
        perror("mkfifo");
        return 1;
    }

    int fd = open(MY_FIFO,O_RDONLY);//只读模式
    if(fd < 0){
        perror("open");
        return 1;
    }

    while(1){
        char buffer[1024];
        memset(buffer,0,sizeof(buffer));

        ssize_t len = read(fd,buffer,sizeof(buffer) - 1);
        if(len == 0){
            std::cout << "read fifo finished !" << std::endl;
            break;
        }
        else if(len > 0){
            buffer[len] = '\0';
            std::cout << "read from client : " << buffer << std::endl;
        }
        else{
            perror("open");
            break;
        }
    }

    close(fd);

    return 0;
}

client

#include<iostream>
#include<sys/stat.h>
#include<unistd.h>
#include<fcntl.h>
#include<cstring>

#define MY_FIFO "fifo"
int main(){
    int fd = open(MY_FIFO,O_WRONLY);//写入模式
    if(fd < 0){
        perror("open");
        return 1;
    }

    while(1){
        std::string str;
        std::cout << "Please enter message :";
        std::cin >> str;
        
        ssize_t len = write(fd,str.c_str(),str.size());
        if(len <= 0){
            perror("write");
            break;
        }
    }

    close(fd);

    return 0;
}

2.4 管道的总结

管道：

• 管道分为匿名管道和命名管道；

• 管道通信方式的中间介质是文件，通常称这种文件为管道文件；

• 匿名管道:管道是半双工的，数据只能单向通信；需要双方通信时，需要建立起两个管道；只能用于父子进程或者兄弟进程之间（具有亲缘关系的进程）。

• 命名管道：不同于匿名管道之处在于它提供一个路径名与之关联，以FIFO的文件形式存在于文件系统中。这样，即使与FIFO的创建进程不存在亲缘关系的进程，只要可以访问该路径，就能够彼此通过FIFO相互通信

• 利用系统调用pipe()创建一个无名管道文件，通常称为无名管道或PIPE；利用系统调用mkfifo()创建一个命名管道文件，通常称为有名管道或FIFO。

• PIPE是一种非永久性的管道通信机构，当它访问的进程全部终止时，它也将随之被撤消。

• FIFO是一种永久的管道通信机构，它可以弥补PIPE的不足。管道文件被创建后，使用open()将文件进行打开，然后便可对它进行读写操作，通过系统调用write()和read()来实现。通信完毕后，可使用close()将管道文件关闭。

• 匿名管道的文件是内存中的特殊文件，而且是不可见的，命名管道的文件是硬盘上的设备文件，是可见的。

更多资料：https://github.com/0voice