Linux 进程池实战：匿名管道、轮询调度与隐藏的阻塞陷阱

文章目录

• 前言
• 代码展示
• 代码功能
• • Task.hpp
  • ProcessPool.hpp
• 整体运行思路
• 运行演示
• 隐藏的问题

前言

本次介绍的是使用匿名管道创建进程池的操作，其中会存在一个隐藏的堵塞问题，这个在后面会提及。本人所认为的重要部分会以注释的形式展出，有一些也会在后面强调

代码展示

ProcessPool.hpp

#ifndef__PROCESS_POOL_HPP__#define__PROCESS_POOL_HPP__#include"Task.hpp"#include<unistd.h>#include<stdlib.h>#include<errno.h>#include<string.h>#include<sys/types.h>#include<sys/wait.h>usingnamespacestd;classChannel{public:Channel(intfd,pid_t id):_wfd(fd),_subid(id){// 切记要to_string，转换一下格式_name="channel- "+to_string(_wfd)+"-"+to_string(_subid);}intFd(){return_wfd;}pid_tSubid(){return_subid;}stringName(){return_name;}voidSend(intcode){// 发送本质上就是把任务编码通过写端写到管道文件write(_wfd,&code,sizeof(code));}voidClose(){// 关闭写端close(_wfd);cout<<"关闭："<<_name<<endl;}voidWait(){// 等待本管道对应的子进程退出waitpid(_subid,nullptr,0);cout<<"等待子进程["<<_name<<"]回收"<<endl;}private:// 每个信道里都包含文件描述符 _wfd// 子进程id _subid// 信道名字 _nameint_wfd;pid_t _subid;string _name;};classChannelManager{public:ChannelManager():_next(0){}voidInsert(intfd,pid_t id){// 向_channels里插入Channel_channels.emplace_back(fd,id);}Channel&Select(){// 轮询选择信道并返回auto&c=_channels[_next];_next++;_next%=_channels.size();returnc;}voidStopSubprocess(){// 通过挨个关闭信道的方式关闭进程池for(auto&channel:_channels){channel.Close();}}voidWaitSubprocess(){// 挨个等待每个管道的子进程退出for(auto&channel:_channels){channel.Wait();}}private:vector<Channel>_channels;//_next用于轮询选择_channels里的Channelint_next;};constintgg=10;classProcessPool{public:ProcessPool(intnum):_process_num(num){// 注册任务_tm.Register(Printlog);_tm.Register(Download);_tm.Register(Upload);}boolStart(){// 主要作用是创建管道和父子进程for(inti=0;i<_process_num;i++){// 创建匿名管道intpipefd[2]={0};if(pipe(pipefd)<0){perror("pipe");returnfalse;}// 创建子进程intid=fork();if(id<0){perror("fork");returnfalse;}// 子进程，只读不写elseif(id==0){close(pipefd[1]);Work(pipefd[0]);close(pipefd[0]);exit(0);}// 父进程，只写不读else{close(pipefd[0]);_cm.Insert(pipefd[1],id);}}returntrue;}voidRun(){// 该函数的作用就是选择一个信道，生成任务码，并将任务码发送至管道里inttaskcode=_tm.code();auto&c=_cm.Select();cout<<"选择一个信道："<<c.Name()<<endl;c.Send(taskcode);cout<<"发送了一个任务码："<<taskcode<<endl;}voidStop(){// 关闭所有父进程的写端即可，子进程读端读取不到内容自己就结束了_cm.StopSubprocess();// 回收所有子进程_cm.WaitSubprocess();}private:// Work的主要工作就是在管道里读取任务码，并执行相应任务voidWork(intrfd){while(true){intcode;intnum=read(rfd,&code,sizeof(code));if(num<0){perror("read");break;}elseif(num>0){if(num!=sizeof(code)){continue;}cout<<"子进程["<<getpid()<<"]收到了一个任务码："<<code<<endl;_tm.Execute(code);}else{cout<<"子进程结束"<<endl;break;}}}private://_process_num是要创建的信道数量ChannelManager _cm;TaskManager _tm;int_process_num;};#endif

Task.hpp

#pragmaonce#include<iostream>#include<ctime>#include<vector>usingnamespacestd;typedefvoid(*_task)();voidPrintlog(){cout<<"这是一个打印日志的任务"<<endl;}voidUpload(){cout<<"这是一个上传的任务"<<endl;}voidDownload(){cout<<"这是一个下载的任务"<<endl;}classTaskManager{public:TaskManager(){srand(time(nullptr));}voidRegister(_task task){// 在_channels中插入任务_tasks.push_back(task);}intcode(){// 生成随机任务号码returnrand()%_tasks.size();}voidExecute(intcode){// 根据编号执行任务// 先判断编号是否合格，再执行if(code>=0&&code<_tasks.size()){_tasks[code]();}}private:vector<_task>_tasks;};

Main.cc

#include"ProcessPool.hpp"#include<cstdio>intmain(){ProcessPoolpp(gg);pp.Start();intcnt=10;while(cnt--){pp.Run();sleep(1);cout<<endl;}pp.Stop();return0;}

Makefile

Process_Pool:Main.cc ProcessPool.hpp Task.hpp g++-o $@ $<-std=c++11.PHONY:clean clean:rm-f Process_Pool

---

代码功能

Task.hpp

利用函数指针_task调用各种任务函数：Printlog()Upload()Download()

• TaskManager
  创建名为_tasks存放_task函数指针的vector容器；
  生成随机的、不超过vector容器长度的任务码（后面会提到任务码的用处）；
  通过任务码执行任务

---

ProcessPool.hpp

• Channel
  负责记载管道的写端和子进程的id，以及信道的名字;
  把任务码通过写端传给管道;
  关闭写端;
  等待子进程退出

• ChannelManager
  将一个个的信道通过Insert函数放入名为_channels的verctor容器里;
  利用成员变量_next进行轮询选择信道;
  实现每个信道的关闭和等待操作,其本质就是对信道的增删操作

• ProcessPool
  因本类相较于前面几个来说，较为复杂，故此按函数名讲解
  构造函数，运用成员变量_tm里的Register注册任务
  Start()函数，创建匿名管道，创建子进程，每创建一个管道就要创建一个子进程
  Run()函数,前面提到的随机任务码，在此处被信道通过写端写入了管道内。该函数先选择信道，再把任务码发到信道里的管道内
  Work()函数，它其实是子进程的工作函数，使子进程能够在读端读取任务码，并将其传送给成员变量_tm，让它去根据任务码执行相应的任务。有一点要强调一下，当read返回值为0的时候，意味着父进程的写端关闭，也就意味着子进程的任务结束了
  Stop()函数，调用ChannelManager关闭所有父进程的写端，并等待子进程退出

---

整体运行思路

我们按照main函数的思路来梳理

1. ProcessPool构造，_process_num为想要创建的信道数目，同时也是子进程的数目，用TaskManager里的Register()插入任务
2. 在每一次for循环中，pipe创建匿名管道，创建父子进程，父写子读，父进程关闭读端，然后创建并插入信道；子进程关闭写端，用读端去读取父进程写在管道里的随机任务码，读取完成后关闭子进程读端，之后exit，子进程退出
3. 循环运行Run()，cnt的大小就是派发任务的数量，此数量可以大于信道的数量。先获取随机任务码，再选择一个信道，再将其发送给信道
4. 调用Stop()，关闭所有父进程的写端，接着回收刚才exit的子进程

运行演示

隐藏的问题

下一篇再讲