【Linux进程间通信】用管道实现简单的进程池、命名管道

【Linux进程间通信】用管道实现简单的进程池、命名管道

作者：爱写代码的刚子

时间：2024.2.10

前言：本篇博客将会介绍并实现简单的线程池

为什么要实现进程池？

• 系统调用是有成本的，池化技术是为了我们的访问速度和效率
• 在需要频繁的创建删除较多进程的情况下,导致计算机资源消耗过多
• 进程池则是创建指定进程数量等待执行事件，避免了不必要的创建和销毁过程

代码实现

• ProcessPool_Task.hpp

#pragma once#include <iostream>
#include <functional>
#include <vector>using task_t=std::function<void()>;
//typedef void(*task_t)();
void task1()
{std::cout<<"task1"<<std::endl;
}void task2()
{std::cout<<"task2"<<std::endl;
}void task3()
{std::cout<<"task3"<<std::endl;
}void task4()
{std::cout<<"task4"<<std::endl;
}void LoadTask(std::vector<task_t> *tasks)
{tasks->push_back(task1);tasks->push_back(task2);tasks->push_back(task3);tasks->push_back(task4);}

• ProcessPool.cc

#include "ProcessPool_Task.hpp"
#include <string>
#include <vector>
#include <ctime>
#include <unistd.h>
#include <cstdlib>
#include <cassert>
#include <iostream>#include <sys/wait.h>
#include <sys/stat.h>const int processnum = 5;
//描述std::vector<task_t> tasks;
class channel
{ 
public:channel(int cmdfd,int slaverid,const std::string &processname):_cmdfd(cmdfd),_slaverid(slaverid),_processname(processname){}public:int _cmdfd;pid_t _slaverid;std::string _processname;
};void slaver()
{while(true){int cmdcode = 0;int n = read(0,&cmdcode, sizeof(int));//如果父进程不给子进程发送数据，则会阻塞等待if(n == sizeof(int)){//执行cmdcode对应的任务列表std::cout <<"slaver say@ get a command: "<<getpid() << ": cmdcode: "<< cmdcode <<std::endl;if(cmdcode >=0 && cmdcode<tasks.size()) tasks[cmdcode]();}if(n == 0)break;}
}
//参数规范
//输入：const &
//输出：*
//输入输出：&void InitProcessPool(std::vector<channel> *channels)
{//确保每一个子进程都只有一个写端std::vector<int> oldfds;for(int i=0;i<processnum;++i){int pipefd[2];//临时空间int n = pipe(pipefd);assert(!n);(void)n;pid_t id = fork();if(id==0)//子进程,子进程拿到的pipefd都是3{std::cout<< "child" << getpid() << "close history fd: ";for(auto fd : oldfds) {std::cout<<fd<<" ";close(fd);}std::cout<<"\n";close(pipefd[1]);dup2(pipefd[0],0);//将pipefd[0]重定向到0，将来直接往键盘文件（fd为0）文件里面读即可。close(pipefd[0]);slaver();std::cout<< "process : "<< getpid() << "quit" <<std::endl;//方法一：//slaver(pipefd[0]);exit(0);}//父进程,父进程拿到的pipefd是4，5，6...close(pipefd[0]);//添加channel字段std::string name = "process-"+ std::to_string(i);channels->push_back(channel(pipefd[1],id,name ));//pipefd[1]表示父进程要往pipefd[1]里面写oldfds.push_back(pipefd[1]);}
}void Debug(const std::vector<channel> &channels)
{for(const auto &c : channels){std::cout<<c._cmdfd<<" "<<c._slaverid<<" "<<c._processname << std::endl;}
}void ctrlSlaver(const std::vector<channel> &channels)
{int which = 0;//轮转的方式while(true){//1.选择任务int cmdcode = rand()%tasks.size();//2.选择进程//[负载均衡(1.随机数 2.轮转)]int processpos = rand()%channels.size();//3.发送任务write(channels[which]._cmdfd,&cmdcode,sizeof(cmdcode));++which;which %= channels.size();sleep(1);}
}void QuitProcess(const std::vector<channel> &channels)
{//for(const auto &c : channels) close(c._cmdfd);//for(const auto &c : channels) waitpid(c._slaverid,nullptr,0);//这里存在子进程有多个写端的问题，解决办法：//方法一.从后往前关闭子进程int last = channels.size()-1;for(int i= last;i>= 0;i--){close(channels[i]._cmdfd);waitpid(channels[i]._slaverid,nullptr,0);}//方法二.确保每一个子进程都只有一个写端}
int main()
{LoadTask(&tasks);//随机数srand(time(nullptr)^getpid()^1023);//组织std::vector<channel> channels;//将特定的结构转化为数据的增删查改//初始化InitProcessPool(&channels);Debug(channels);//控制子进程ctrlSlaver(channels);//清理收尾QuitProcess(channels);return 0;
}

图解：

命名管道

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

• 如果我们想在不相关的进程之间交换数据，可以使用FIFO文件来做这项工作，它经常被称为命名管道。

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

创建一个命名管道

• $ mkfifo filename在命令行上创建命名管道

p开头表示这是命名管道（但是并不在磁盘上）,同时管道文件的大小为0

• *int mkfifo(const char filename, mode_t mode); 程序中创建命名管道的函数

理解命名管道

不同的两个进程打开同一个文件的时候，在内核中操作系统文件描述符只会指向同一个文件，进程间通信的前提：先让不同的进程看到同一份资源，管道文件则不需要进行刷盘（内存级文件），所以大小为0字节。

【问题】：如何保证打开的是同一个文件？看到同一个路径下的同一个文件名。（inode），即= 路径 + 文件名（唯一性）

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

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

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

FIFO（命名管道）与pipe（匿名管道）之间唯一的区别在它们创建与打开的方式不同，一但这些工作完成之后，它们具有相同的语义。

命名管道的打开规则

• 如果当前打开操作是为读而打开FIFO时

  • O_NONBLOCK disable：阻塞直到有相应进程为写而打开该FIFO
  • O_NONBLOCK enable：立刻返回成功

• 如果当前打开操作是为写而打开FIFO时

  • O_NONBLOCK disable：阻塞直到有相应进程为读而打开该FIFO
  • O_NONBLOCK enable：立刻返回失败，错误码为ENXIO