进击的荆棘头像
Linux系统——进程间通信之管道封面图

Linux系统——进程间通信之管道

 💁‍♂️个人主页：进击的荆棘

👇作者其它专栏：

《数据结构与算法》《算法》《C++起始之路》《Linux》

目录

1.进程间通信介绍

2.管道

3.匿名管道

4.命名管道

5.system V共享内存

6.system V消息队列

7.system V信息量

8.内核是如何组织管理IPC资源的

1.进程间通信介绍

1.1进程间通信的目的

●数据传输：一个进程需要将它的数据发送给另一个进程

●资源共享：多个进程间共享同样的资源

●通知事件：一个进程需要向另一个或一组进程发送消息，通知它（它们）发生了某种事件（如进程终止时要通知父进程）

●进程控制：有些进程希望完全控制另一个进程的执行（如Debug进程），此时控制进程希望能够拦截另一个进程的所有陷入和异常，并能够及时知道它的状态改变

1.2进程间通信发展

●管道

●System V进程间通信

●POSIX进程间通信

1.3进程间通信分类

管道

●匿名管道pipe

●命名管道

System V IPC

●System V 消息队列

●System V 共享内存

●System V信号量

POSIX IPC

●消息队列

●共享内存

●信号量

●互斥量

●条件变量

●读写锁

2.管道

什么是管道

●管道是Unix中最古老的进程间通信的形式

●我们把从一个进程连接到另一个进程的一个数据流称为一个“管道”

3.匿名管道

#include <unistd.h>
功能:创建⼀⽆名管道
原型
int pipe(int fd[2]);
参数
fd：⽂件描述符数组,其中fd[0]表⽰读端, fd[1]表⽰写端
返回值:成功返回0，失败返回错误代码

3.1代码实例

例⼦：从键盘读取数据，写⼊管道，读取管道，写到屏幕
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

int main( void )
{
    int fds[2];
    char buf[100];
    int len;
    if ( pipe(fds) == -1 )
        perror("make pipe"),exit(1);
    // read from stdin
    while ( fgets(buf, 100, stdin) ) {
        len = strlen(buf);
        // write into pipe
        if ( write(fds[1], buf, len) != len ) {
            perror("write to pipe");
            break;
        }
        memset(buf, 0x00, sizeof(buf));
        // read from pipe
        if ( (len=read(fds[0], buf, 100)) == -1 ) {
            perror("read from pipe");
            break;
        }
        // write to stdout
        if ( write(1, buf, len) != len ) {
            perror("write to stdout");
            break;
        }
    }
}

3.2用fork来共享管道原理

3.3从文件描述符角度-理解管道

3.4从内核角度-管道本质

所以，可以把管道看成文件，管道的使用和文件一致，迎合了“Linux一切且文件的思想”

3.5管道实例

3.5.1测试管道读写

#include <iostream>
#include <unistd.h>
#include <sys/wait.h>
#include <cstdio>
#include <cstring>

void ChildWrite(int wfd){
    int cnt=0;
    while(true){
        char buf[1024];
        snprintf(buf,sizeof(buf),"I am child,pid=%d,%d",getpid(),cnt++);
        write(wfd,buf,strlen(buf));
        sleep(1);
    }
}
void FatherRead(int rfd){
    while(true){
        char buf[1024];
        int n=read(rfd,buf,sizeof(buf)-1);
        if(n>0){
            buf[n]=0;
            std::cout<<"child say:"<<buf<<std::endl;
        }
    }
}
int main(){
    int fds[2]={0};//fds[0] read ,fds[1] write
    pipe(fds);
    int fd=fork();
    if(fd==0){
        //child
        //关闭不需要的读写端
        //f -> r, c -> w
        close(fds[0]);
        ChildWrite(fds[1]);
        close(fds[1]);
    }
    //father
    //关闭不需要的读写端
    //f -> r, c -> w
    close(fds[1]);
    FatherRead(fds[0]);
    close(fds[0]);
    waitpid(fd,nullptr,0);
    return 0;
}

3.5.2创建进程池处理任务

ProcessPool.hpp

#ifndef __PROCESS_POOL_HPP__   //防止同一个头文件被重复包含
#define __PROCESS_POOL_HPP__

#include <iostream>
#include <vector>
#include <cstdlib>
#include <unistd.h>
#include <sys/wait.h>
#include "Task.hpp"

class Channel{
public:
    Channel(int fd,pid_t subid):_fd(fd),_subid(subid)
    {
        _name="channel-"+std::to_string(_fd)+"-"+std::to_string(_subid);
    }
    void Send(int taskcode){
        int n=write(_fd,&taskcode,sizeof(taskcode));
        (void)n;//消除编译器警告
    }
    void Close(){
        close(_fd);
    }
    void Wait(){
        pid_t rid=waitpid(_subid,nullptr,0);
        (void)rid;
    }
    ~Channel(){}
    int Fd(){return _fd;}
    int SubId(){return _subid;}
    std::string Name(){return _name;}
private:
    int _fd;
    int _subid;
    std::string _name;
};

class ChannelManager{
public:
    ChannelManager():_next(0)
    {}
    void Insert(int wfd,pid_t subid){
        _channels.emplace_back(wfd,subid);
    }
    Channel &Select(){
        auto &c=_channels[_next];
        _next++;
        _next%=_channels.size();
        return c;
    }
    void CloseAll(){
        for(auto &channel:_channels){
            channel.Close();
        }
    }
    void TestSlove(){
        for(int i=_channels.size()-1;i>=0;i--){
            _channels[i].Close();
            _channels[i].Wait();
            std::cout<<"等待子进程"<<_channels[i].Name()<<std::endl;
        }
    }
    void CloseAndWait(){
        for(auto &c:_channels){
            c.Close();
            c.Wait();
            std::cout<<"等待子进程"<<c.Name()<<std::endl;
        }
    }
    void StopChannel(){
        for(auto &c:_channels){
            c.Close();
            std::cout<<"关闭管道"<<c.Name()<<std::endl;
        }
    }
    void WaitSubProcess(){
        for(auto &c:_channels){
            c.Wait();
            std::cout<<"等待子进程"<<c.Name()<<std::endl;
        }
    }
    ~ChannelManager(){} 
private:
    std::vector<Channel> _channels;
    int _next;
};

int gdefaultnum = 5;

class ProcessPool{
public:
    ProcessPool(int num):_process_num(num)
    {
        _tm.Register(Task1);
        _tm.Register(Task2);
        _tm.Register(Task3);
    }
    void Work(int rfd){
        while(true){
            int code=0;
            ssize_t n=read(rfd,&code,sizeof(code));
            if(n>0){
                if(n!=sizeof(code)){
                    continue;
                }
                std::cout<<"子进程["<<getpid()<<"]收到一个任务码:"<<code<<std::endl;
                _tm.Excute(code);
            }
            else if(n==0){
                std::cout<<"子进程退出"<<std::endl;
                break;
            }
            else{
                std::cout<<"read error"<<std::endl;
                break;
            }
        }
    }
    bool Start(){
        for(int i=0;i<_process_num;i++){
            //创建管道
            int pfd[2]={0};
            int n=pipe(pfd);
            
            if(n<0){
                std::cout<<"pipe error"<<std::endl;
                return false;
            }

            pid_t subid=fork();
            if(subid<0){
                std::cout<<"fork error"<<std::endl;
                return false;
            }
            else if(subid==0){
                //child
                //关闭不需要的读/写端
                //child -> read
                close(pfd[1]);
                Work(pfd[0]);
                _cm.CloseAll();
                close(pfd[0]);
                exit(0);
            }
            else{
                //father
                //关闭不需要的读/写端
                //father -> write
                close(pfd[0]);
                _cm.Insert(pfd[1],subid);
            }
        }
        return true;
    }
    void Run(){
        //1.选择一个任务
        int taskcode=_tm.Code();
        //2.选择一个信道[子进程]，负载均衡的选择
        auto &c=_cm.Select();
        std::cout<<"选择了一个子进程"<<c.Name()<<std::endl;
        //3.发送任务
        c.Send(taskcode);
        std::cout<<"发送了一个任务码"<<taskcode<<std::endl;
    }

    void Stop(){
        // //关闭管道
        // _cm.StopChannel();
        // //结束子进程
        // _cm.WaitSubProcess();

        //_cm.CloseAndWait();
        //两个进程合在一起会报错,这是因为创建进程时产生的问题
        //原因：父进程创建子进程时，子进程会拷贝一份进程描述符表，当父进程的4号描述符指向管道文件的w端，
        //而3号指向r端，但是父进程是写入，所以3号被关闭，此时子进程1中还正常，只有父进程4号指向管道文件的w端
        //但子进程2中，父进程3号指向r端，5号指向w端，由于复制了父进程的描述符表所以子进程2的4也指向子进程1的w端，
        //以致于剩下的子进程都指向之前的子进程w端，导致关闭子进程时，因为有多个w端指向子进程所以无法正常关闭

        //解决方法一：倒着关闭子进程
        //_cm.TestSlove();

        //解决方法二：在创建这些子进程的时候就关闭掉它们的复制父进程的w端（Start()中添加）
        _cm.StopChannel();
        
        _cm.WaitSubProcess();
    }
    ~ProcessPool(){}
private:
    ChannelManager _cm;
    int _process_num;
    TaskManager _tm;
};

#endif

Task.hpp

#pragma once

#include <iostream>
#include <vector>
#include <cstdlib>
#include <ctime>

//函数指针
typedef void (*func_t)();

void Task1(){
    std::cout<<"Task1 is write system log"<<std::endl;
}
void Task2(){
    std::cout<<"Task2 is upload"<<std::endl;
}
void Task3(){
    std::cout<<"Task3 is download"<<std::endl;
}

class TaskManager{
public:
    TaskManager(){
        srand(time(nullptr));
    }
    void Register(func_t f){
        _tasks.push_back(f);
    }
    int Code(){
        return rand()%_tasks.size();
    }
    void Excute(int code){
        if(code>=0&&code<_tasks.size()){
            _tasks[code]();
        }
    }
    ~TaskManager(){}
private:
    std::vector<func_t> _tasks;
};

main.cc

#include "ProcessPool.hpp"

int main(){
    //创建进程池对象
    ProcessPool pp(gdefaultnum);
    //启动进程池
    pp.Start();

    int cnt=10;
    while(cnt--){
        pp.Run();
        sleep(1);
    }

    //结束任务
    pp.Stop();
    return 0;
}

Makefile

process_pool:main.cc
	g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
	rm -f process_pool

3.6管道读写规则

●当没有数据可读时

        ▢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将保证写入的原子性（单次write()系统调用写入管道/FIFO的数据，作为一个不可分割的整体进入管道，不会和其它进程的write()数据相互交错）

●当要写入的数据量大于PIPE_BUF时，Linux将保证写入的原子性

3.7管道特性

●只能用于具有共同祖先的进程（具有亲缘关系的进程）之间进行通信。通常，一个管道由一个进程创建，然后该进程调用fork，此后父、子进程之间就可应用该管道

●管道提供流式服务

●一般而言，进程退出，管道释放，所以管道的生命周期随进程

●一般而言，内核会对管道操作进行同步和互斥

●管道是半双工的，数据只能向一个方向流动；需要双方通信时，需要建立起两个管道

3.8验证管道通信的4种情况

●读正常&&写满

●写正常&&读空

●写关闭&&读正常

●读关闭&&写正常

4.命名管道

●管道应用的一个限制就是只能在具有共同祖先（具有亲缘关系）的进程间通信

●若想不在不相关的进程间交换数据，可以使用FIFO文件来做这项工作，被称为命名管道

●命名管道是一种特殊类型的文件

4.1创建一个命名管道

●命名管道可以从命令行上创建，命令行方法是使用下面这个命令

$ mkfifo filename

●命名管道也可以从程序里创建，相关函数：

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

●创建命名管道

int main(int argc, char *argv[])
{
    mkfifo("p2", 0644);
    return 0;
}

4.2匿名管道与命名管道的区别

●匿名管道由pipe函数创建并打开

●命名管道由mkfifo函数创建，用open打开

●FIFO（命名管道）与pipe（匿名管道）之间唯一的区别在于它们创建和打开的方式不同，一旦这些工作完成之后，它们具有相同的语义（都是文件上的操作）

4.3命名管道的打开规则

●若当前打开操作是为读而打开FIFO时

        ▢O_NONBLOCK disable:阻塞直到有相应进程为写而打开该FIFO

        ▢O_NONBLOCK enable:立即返回成功

●若当前打开操作是为写而打开FIFO时

        ▢O_NONBLOCK disable:阻塞直到有相应进程为读而打开该FIFO

        ▢O_NONBLOCK enable:立即返回失败，错误码为ENXIO

实例1：用命名管道实现文件拷贝

读取文件，写入命名管道

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <string.h>

#define ERR_EXIT(m) \
do \
{ \
    perror(m); \
    exit(EXIT_FAILURE); \
} while(0)

int main(int argc, char *argv[])
{
    mkfifo("tp", 0644);
    int infd;    
    infd = open("abc", O_RDONLY);
    if (infd == -1) ERR_EXIT("open");
    int outfd;
    outfd = open("tp", O_WRONLY);
    if (outfd == -1) ERR_EXIT("open");
    char buf[1024];
    int n;
    while ((n=read(infd, buf, 1024))>0)
    {
        write(outfd, buf, n);
    }
    close(infd);
    close(outfd);
    return 0;
}

读取管道，写入目标文件

#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <string.h>

#define ERR_EXIT(m) \
do \
{ \
    perror(m); \
    exit(EXIT_FAILURE); \
} while(0)

int main(int argc, char *argv[])
{
    int outfd;
    outfd = open("abc.bak", O_WRONLY | O_CREAT | O_TRUNC, 0644);
    if (outfd == -1) ERR_EXIT("open");
    int infd;
    infd = open("tp", O_RDONLY);
    if (outfd == -1)
        ERR_EXIT("open");
    char buf[1024];
    int n;
    while ((n=read(infd, buf, 1024))>0)
    {
        write(outfd, buf, n);
    }
    close(infd);
    close(outfd);
    unlink("tp");
    return 0;
}

实例2：用命名管道实现server&client通信

Makefile

.PHONY:all
all:server client
server:server.cc
	g++ -o $@ $^
client:client.cc
	g++ -o $@ $^

.PHONY:clean
clean:
	rm -f server client

comm.hpp

#pragma once

#include <iostream>
#include <cstdio>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <string>

#define PATH "."
#define FIFONAME "fifo"

#define ERR_EXIT(n) \
do{ \
    perror(n); \
    exit(EXIT_FAILURE); \
}while(0) \

class FileFifo{
public:
    FileFifo(const std::string &path,const std::string &name)
        :_path(path),_name(name)
    {
        _pathname=_path+"/"+_name;
        umask(0);
        //创建管道文件
        int n=mkfifo(_pathname.c_str(),0666);
        if(n<0){
            std::cerr<<"mkfifo error"<<std::endl;
            ERR_EXIT("mkfifo");
        }
    }
    ~FileFifo(){
        //删除管道文件
        int n=unlink(_pathname.c_str());
        if(n==0){
            std::cout<<"unlink success"<<std::endl;
        }
        else{
            std::cerr<<"remove fifo failed"<<std::endl;
            ERR_EXIT("unlink");
        }
    }
private:
    std::string _path;
    std::string _name;
    std::string _pathname;
};

class FifoOper{
public:
    FifoOper(const std::string &path,const std::string &name)
        :_path(path),_name(name),_fd(-1)
    {
        _pathname=_path+"/"+_name;
    }
    void OpenForRead(){
        //打开文件
        //当write没有执行open时，read会在open内部进行阻塞
        //直到有人把管道文件打开了，open才会返回
        _fd=open(_pathname.c_str(),O_RDONLY);
        if(_fd<0){
            std::cerr<<"open mkfifo failed"<<std::endl;
            ERR_EXIT("open");
        }
        std::cout<<"open fifo success"<<std::endl;
    }
    void OpenForWrite(){
        //打开文件
        _fd=open(_pathname.c_str(),O_WRONLY);
        if(_fd<0){
            std::cerr<<"open mkfifo failed"<<std::endl;
            ERR_EXIT("open");
        }
        std::cout<<"open fifo success"<<std::endl;
    }
    void Reader(){
        //读取文件内容
        char buffer[1024];
        while(true){
            int n=read(_fd,buffer,sizeof(buffer)-1);
            if(n>0){
                buffer[n]=0;
                std::cout<<"Client say# "<<buffer<<std::endl;
            }
            else if(n==0){
                std::cout<<"Client quit,me too"<<std::endl;
                break;
            }
            else {
                std::cout<<"read error"<<std::endl;
                ERR_EXIT("read");
            }
        }
    }
    void Writer(){
        //向文件写入内容
        std::string message;
        int cnt=1;
        pid_t id=getpid();
        while(true){
            std::cout<<"Please say# ";
            std::getline(std::cin,message);
            message+=(",message number:"+std::to_string(cnt++)+",["+std::to_string(id)+"]");
            write(_fd,message.c_str(),message.size());
        }
    }
    void Close(){
        if(_fd>0)
            close(_fd);
    }
    ~FifoOper(){}
private:
    std::string _path;
    std::string _name;
    std::string _pathname;
    int _fd;
};

server.cc

#include "comm.hpp"

int main(){
    FileFifo fifo(PATH,FIFONAME);

    FifoOper readerfile(PATH,FIFONAME);
    readerfile.OpenForRead();
    readerfile.Reader();
    readerfile.Close();
    return 0;
}

client.cc

#include "comm.hpp"

int main(){
    FifoOper writerfile(PATH,FIFONAME);
    writerfile.OpenForWrite();
    writerfile.Writer();
    writerfile.Close();
    return 0;
}

结果：

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2501_93697341/article/details/167586574

文章来源转载

评论

评论列表

微信小程序
QQ小程序

关于作者

点赞数：0
关注数：0
粉丝：0
文章：0
关注标签：0
加入于：--