【进程间通信】————匿名管道、模拟实现进程池

目录1. 进程间通信1.1 进程间通信的目的1.2 进程间通信分类2. 管道3. 匿名管道3.1 pipe函数3.2 用 fork 来共享管道原理3.3 从文件描述符角度理解3.4 从内核角度理解3.5 父子进程管道读写测试3.6 管道特性3.7 4种通信情况3.8 管道的原子性4. 进程池4.1 需要的几个类ChannelChannelManagerTaskManagerProcessPool4.2 主逻辑4.2.1 创建进程池对象4.2.2 启动进程池Start4.2.3 派发任务Run4.2.4 回收结束进程池Stop4.3 执行结果4.4 代码还存在的bugbug代码打印查看每个子进程继承的写端4.5 解决方法方法一倒着关闭管道方法二只让父进程指向所有管道的写端1. 进程间通信1.1 进程间通信的目的数据传输⼀个进程需要将它的数据发送给另⼀个进程资源共享多个进程之间共享同样的资源。通知事件⼀个进程需要向另⼀个或⼀组进程发送消息通知它它们发⽣了某种事件如进 程终⽌时要通知⽗进程。进程控制有些进程希望完全控制另⼀个进程的执⾏如Debug进程此时控制进程希望能够 拦截另⼀个进程的所有陷⼊和异常并能够及时知道它的状态改变。1.2 进程间通信分类匿名管道pipe命名管道2. 管道什么是管道管道是Unix中最古⽼的进程间通信的形式。我们把从⼀个进程连接到另⼀个进程的⼀个数据流称为⼀个“管道”3. 匿名管道匿名管道本质上是由内核管理的临时文件它不占用磁盘空间可即时创建和释放3.1 pipe函数#includeunisdt.h int pipe(int fd[2]);pipe函数作用创建一个匿名管道 fd[0] 表示读端fd[1] 表示写端 成功返回0失败返回错误代码。实例代码下面的代码的逻辑就是创建一个管道、从键盘读取内容、将内容写入管道、从管道中读取刚写入的内容、将读取到的内容输出至屏幕上。C语言字符串处理接口通常会自动处理\0终止符。系统调用位于更底层不涉及字符串操作因此需要手动为\0预留空间#include stdio.h #include stdlib.h #include string.h #include unistd.h int main() { int fds[2]; char buffer[100]; char msg[100]; // 创建管道从键盘读取内容写道管道中然后读取管道中的内容 // 创建管道 if (pipe(fds) 0) { perror(pipe); exit(1); } while (fgets(buffer, 100, stdin)) // 从键盘读取内容 { // 写道管道中 int len strlen(buffer); if (write(fds[1], buffer, len) 0) { perror(write); exit(1); } // 清空缓冲区 memset(buffer, 0, sizeof(buffer)); // 读取管道中的内容 if (read(fds[0], msg, sizeof(msg) - 1) 0) { perror(read); exit(1); } // 输出到屏幕上 if (write(1, msg, strlen(msg)) 0) { perror(write); exit(1); } // 清空缓冲区 memset(msg, 0, sizeof(msg)); } return 0; }3.2 用 fork 来共享管道原理文件描述符 fd[0] 和 fd[1] 分别对应管道的读端和写端。父进程通过 fork() 创建子进程时子进程会继承父进程的文件描述符表。为了实现单向通信如父进程读、子进程写父子进程需要各自关闭不需要的文件描述符父进程关闭读端 fd[1]子进程关闭写端 fd[0]。3.3 从文件描述符角度理解3.4 从内核角度理解3.5 父子进程管道读写测试创建管道、创建子进程、关闭父子各自不需要的文件描述符、父读子写、关闭剩下的文件描述符、回收子进程waitpid获取退出信息退出码、退出信号#include iostream #include unistd.h #include sys/types.h #include cstring #include sys/wait.h using namespace std; void ChildWrite(int wfd) { char c A; int cnt 0; while (true) { write(wfd, c, 1); // 一次写一个字符 printf(child:%d\n, cnt); sleep(1); } } void FatherRead(int rfd) { char buffer[1024]; int cnt4; while (cnt--) { // 系统调用不关系\0,需要我们自己预留空间 ssize_t n read(rfd, buffer, sizeof(buffer) - 1); // n0,表述写端退出管道里没数据时会阻塞等待 if (n 0) { buffer[n] 0; // 加上\0 cout child say: buffer endl; sleep(1); } else if (n 0) { cout n: n endl; cout 写端关闭,我也关闭 endl; break; } else break; } } int main() { // 创建管道 int pipefd[2]; int n pipe(pipefd); if (n 0) { cout pipe error endl; return 1; } cout pipefd[0]: pipefd[0] endl; cout pipefd[1]: pipefd[1] endl; // 创建子进程 int subid fork(); if (subid 0) { cout fork error endl; return 1; } if (subid 0) { // 子进程 // 关闭不需要的文件描述符 // f-r,z-w close(pipefd[0]); ChildWrite(pipefd[1]); // 关闭剩下的文件描述符 close(pipefd[1]); } close(pipefd[1]); FatherRead(pipefd[0]); close(pipefd[0]); // 回收子进程 int status 0; pid_t ret waitpid(subid, status, 0); if (ret 0) { printf(exit code:%d,exit signal:%d\n, (status 8) 0xff, status 0x7f); sleep(5); } return 0; }写端退出退出信号为13pipefd[0]:3 pipefd[1]:4 child:0 child say:A child:1 child say:A child:2 child say:A child:3 child say:A child:4 exit code:0,exit signal:133.6 管道特性匿名管道只能用来进行具有血缘关系的进程之间的进程通信常用于父子。管道文件自带同步机制写的慢读端阻塞写的快写端阻塞。管道是面向字节流的写入和读取的都是二进制。管道是单向通信的属于半双工的一种特殊情况。任何一个时刻一个发一个收——半双工任何一个时刻可以同时发收——全双工管道文件的生命周期是随进程的3.7 4种通信情况写慢读快——读端就要阻塞进程阻塞等写写快读慢——管道满了的时候写就要阻塞等待等读写端关闭继续读——read就会读到返回值为0表示文件结尾读端关闭继续写——写端再写入没有任何意义OS不会做没有意义的事OS会杀掉写端进程发送异常信号13号信号3.8 管道的原子性管道原子性保证较小的写操作是不可分割的。只要写入字节数 ≤ PIPE_BUFLinux 通常是 4096 字节多个进程同时 write 也不会数据交错。超过 PIPE_BUF 后原子性失效数据可能被分段、穿插。管道大小通常是64kb4. 进程池4.1 需要的几个类ChannelChannel 管道写端 子进程 PID一个子进程对应一个 Channel父进程通过 Channel 控制子进程发任务、关闭、回收Channel 就是父进程管理子进程的 “专属通信通道 控制句柄”class Channel { public: Channel(int wfd, pid_t subid) : _wfd(wfd), _subid(subid) { _name channel- to_string(_wfd) - to_string(_subid); } } ~Channel(){} private: int _wfd; pid_t _subid; string _name; };ChannelManager用来管理ChannelChannel是先描述ChannelManager是再组织。class ChannelManager { public: ChannelManager() {} ~ChannelManager() {} private: vectorChannel channels; int _next 0; // 下标用户选择进程执行任务轮询 };TaskManagerTaskManager 任务的 “仓库 调度器”专门管有哪些任务、随机选任务、执行任务。typedef void (*task_t)(); // 函数指针 class TaskManager { public: TaskManager() { srand(time(nullptr)); } void Register(task_t t) { _tasks.push_back(t); } int Code() { // 随机选取任务 return rand() % _tasks.size(); } void Execute(int code) { // 执行对应任务 _tasks[code](); } int Size() { return _tasks.size(); } ~TaskManager() {} private: vectortask_t _tasks; };ProcessPoolProcessPool就是我们的进程池了通过ChannelManager找到每个管道写端 子进程 PID从而控制子进程。class ProcessPool { public: ProcessPool(int num) : _process_num(num) { // 事先插入几个任务 _tm.Register(PrintLog); _tm.Register(DownLoad); _tm.Register(UpLoad); } ~ProcessPool() {} private: ChannelManager _cm; int _process_num;// 进程池中的进程个数 TaskManager _tm; };4.2 主逻辑创建进程池对象启动进程池给进程池中的子进程派发任务收回结束进程池。#include ProcessPool.hpp #include Task.hpp int main() { // 创建进程池对象 ProcessPool pp(gdefaultnum); // 启动进程池 pp.Start(); pp.Debug(); coutendl; // 给进程池中的子进程派发任务 int cnt 10; while (cnt--) { pp.Run(); sleep(1); } coutendl; // 回收结束进程池 pp.Stop(); return 0; }4.2.1 创建进程池对象进程池中的进程个数设为gdefaultnum然后调用 TaskManager 中的 Register 方法事先插入几个任务。hpp文件就是将头文件和实现方法写在一起ProcessPool.hpp:ProcessPool类 const int gdefaultnum 5; // 构造函数 ProcessPool(int num) : _process_num(num) { _tm.Register(PrintLog); _tm.Register(DownLoad); _tm.Register(UpLoad); } //////////////////////////////////////////// void PrintLog() { cout 我是一个打印日志的任务 endl; } void DownLoad() { cout 我是一个下载的任务 endl; } void UpLoad() { cout 我是一个上传的任务 endl; } //////////////////////////////////////////// Task.hpp:TaskManager类 void Register(task_t t) { _tasks.push_back(t); }4.2.2 启动进程池Start启动进程池就是创建num个管道个子进程。然后父子各自关闭不需要的文件描述符。子进程进入 Work 函数后会一直读取管道中的数据等待父进程发布任务。父进程会将 Channle组织起来以便后续控制子进程。bool Start() { // 创建num个管道和子进程 for (int i 0; i _process_num; i) { // 创建管道 int pipefd[2]; int n pipe(pipefd); if (n 0) { cout pipe error endl; return false; } // 创建子进程 pid_t subid fork(); if (subid 0) { cout fork error endl; return false; } else if (subid 0) { // 子进程 // 关闭不需要的文件描述符 close(pipefd[1]); Work(pipefd[0]); close(pipefd[0]); exit(0); } // 父进程 close(pipefd[0]); // 将wfd和subid管理起来 _cm.Insert(pipefd[1], subid); } return true; }workcode是任务码子进程不断从管道中读取任务码然后根据任务码找到相对应的任务并执行。管道是具有同步机制的就算管道中没有数据读端也会阻塞等待但如果写端关闭了读端调用read就会返回0表示写端关闭了。void Work(int rfd) { // 子进程一直从管道中读取内容 while (true) { int code; ssize_t n read(rfd, code, sizeof(code)); if (n 0) { if (n ! sizeof(code)) // 读取不完整 continue; cout 子进程: getpid() 接收到一个任务码 code endl; _tm.Execute(code); } else if (n 0) { // 写端退出 cout 子进程 getpid() 已退出 endl; break; } else { cout 读取失败 endl; break; } } }4.2.3 派发任务Run派发任务的逻辑就是选取任务即随机选取一个任务码选择进程轮询选择保证进程池的负载均衡发送任务父进程将任务码写入管道管道对应的子进程接收到任务码后找到相对应的任务并执行ProcessPool.hpp-ProcessPool类 void Run() { // 选择任务 int code _tm.Code(); // 选择进程 auto c _cm.Select(); // 发送任务 c.Send(code); }随机选取一个任务TaskManager类 TaskManager() { srand(time(nullptr)); } int Code() { // 随机选取任务 return rand() % _tasks.size(); }通过_next依次返回此次执行任务的进程ChannelManager类 Channel Select() { // 返回进程 int n _next; _next; _next % channels.size(); return channels[n]; } private: int _next 0; // 下标用户选择进程执行任务轮询送法任务就是将任务码写入选中进程的对应管道Channel类 void Send(int code) { // 向管道内写数据 write(_wfd, code, sizeof(code)); } private: int _wfd;4.2.4 回收结束进程池Stop结束进程池需要做两件事1.关闭管道2.回收子进程的PCB关闭管道其实就是关闭写端写端关闭后子进程调用 read 时不再阻塞直接返回0然后跳出Work 函数然后会回到 Start 函数执行剩下的代码。剩下的代码就是关闭子进程的读端。ProcessPool类 void Stop() { // 关闭管道 _cm.StopSubProcess(); cout endl; // 回收子进程 _cm.WaitSubProcess(); cout endl; }ChannelManager类 void StopSubProcess() { // 关闭管道 for (auto channel : channels) { channel.Stop(); } } void WaitSubProcess() { // 回收子进程 for (auto channel : channels) { channel.Wait(); } }Channel类 void Stop() { close(_wfd); cout 管道 _wfd 已关闭 endl; } void Wait() { int ret waitpid(_subid, nullptr, 0); (void)ret; // 绕过编译器对未使用变量的语法检查 cout 子进程 _subid 已回收 endl; }4.3 执行结果先打印管道号和对应的子进程pid。然后随机派发任务查看是否是轮询派发。回收进程池查看管道是否关闭子进程是否退出和回收[wsjiZgw05f0yp422tzyebhkoaZ 2.ProcessPool]$ ./process_pool channel-4-24948 channel-5-24949 channel-6-24950 channel-7-24951 channel-8-24952 任务个数:3 子进程:24948接收到一个任务码0 我是一个打印日志的任务 子进程:24949接收到一个任务码2 我是一个上传的任务 子进程:24950接收到一个任务码1 我是一个下载的任务 子进程:24951接收到一个任务码1 我是一个下载的任务 子进程:24952接收到一个任务码1 我是一个下载的任务 子进程:24948接收到一个任务码1 我是一个下载的任务 子进程:24949接收到一个任务码2 我是一个上传的任务 子进程:24950接收到一个任务码0 我是一个打印日志的任务 子进程:24951接收到一个任务码1 我是一个下载的任务 子进程:24952接收到一个任务码0 我是一个打印日志的任务 管道4已关闭 管道5已关闭 管道6已关闭 管道7已关闭 管道8已关闭 子进程24952已退出 子进程24951已退出 子进程24950已退出 子进程24949已退出 子进程24948已退出 子进程24948已回收 子进程24949已回收 子进程24950已回收 子进程24951已回收 子进程24952已回收 [wsjiZgw05f0yp422tzyebhkoaZ 2.ProcessPool]$4.4 代码还存在的bug父进程逐个关闭管道写端并同步回收子进程但由于其他子进程已继承了未关闭的管道写端导致目标子进程的管道读端无法读到 EOFread返回 0一直阻塞最终所有子进程无法退出、父进程无法回收形成进程阻塞卡死 BUG。bug代码如果我们不是先关闭父进程的所有写端再回收子进程而是父进程关闭一个写端就回收一个子进进程就会造成阻塞第一个进程在read阻塞了ChannelManager类 void StopSubProcess() { // 关闭管道 for (auto channel : channels) { channel.Stop(); } } void WaitSubProcess() { // 回收子进程 for (auto channel : channels) { channel.Wait(); } } void StopAndWait() { // 关闭管道回收子进程 for (auto channel : channels) { channel.Stop(); channel.Wait(); } } ProcessPool类 void Stop() { // // 关闭管道 // _cm.StopSubProcess(); // cout endl; // // 回收子进程 // _cm.WaitSubProcess(); // cout endl; _cm.StopAndWait(); }结果channel-4-27225 channel-5-27226 channel-6-27227 channel-7-27228 channel-8-27229 任务个数:3 子进程:27225接收到一个任务码0 我是一个打印日志的任务 子进程:27226接收到一个任务码2 我是一个上传的任务 子进程:27227接收到一个任务码0 我是一个打印日志的任务 子进程:27228接收到一个任务码2 我是一个上传的任务 子进程:27229接收到一个任务码0 我是一个打印日志的任务 子进程:27225接收到一个任务码2 我是一个上传的任务 子进程:27226接收到一个任务码2 我是一个上传的任务 子进程:27227接收到一个任务码0 我是一个打印日志的任务 子进程:27228接收到一个任务码0 我是一个打印日志的任务 子进程:27229接收到一个任务码2 我是一个上传的任务 管道4已关闭打印查看每个子进程继承的写端else if (subid 0) { // 子进程 // 打印每个子进程继承的文件文件描述符表 cout #################### 第 i 1 个子进程继承的写端 endl; _cm.PrintChannel(); cout ############################ endl endl; // 关闭不需要的文件描述符 close(pipefd[1]); Work(pipefd[0]); close(pipefd[0]); exit(0); }####################第1个子进程继承的写端 ############################ ####################第2个子进程继承的写端 channel-4-28604 ############################ ####################第3个子进程继承的写端 channel-4-28604 channel-5-28611 ############################ ####################第4个子进程继承的写端 channel-4-28604 channel-5-28611 channel-6-28621 ############################ ####################第5个子进程继承的写端 channel-4-28604 channel-5-28611 channel-6-28621 channel-7-28631 ############################4.5 解决方法方法一倒着关闭管道void StopAndWait() { // 方法一倒着关闭 for (int i channels.size() - 1; i 0; i--) { channels[i].Stop(); channels[i].Wait(); } }运行结果管道8已关闭 子进程28775已退出 子进程28775已回收 管道7已关闭 子进程28763已退出 子进程28763已回收 管道6已关闭 子进程28740已退出 子进程28740已回收 管道5已关闭 子进程28729已退出 子进程28729已回收 管道4已关闭 子进程28726已退出 子进程28726已回收方法二只让父进程指向所有管道的写端子进程也会继承父进程的 ChannelManager 而 Channel 中就有记录从父进程继承的写端。所以我们调用 Channel 中的 Close() 关闭继承的写端因为父子进程之间修改数据会发生写时拷贝所以子进程关闭写端不会影响父进程。ChannelManager类 void CloseAll() { // 关闭子进程的继承的写端 for (auto channel : channels) { channel.Close(); cout 子进程 channel.Name() 关闭继承的写端 endl endl; } } Channel类 void Close() { close(_wfd); cout 管道 _wfd 已关闭 endl; }####################第1个子进程继承的写端 ############################ ####################第2个子进程继承的写端 channel-4-29691 ############################ 管道4已关闭 子进程channel-4-29691 关闭继承的写端 ####################第3个子进程继承的写端 channel-4-29691 channel-5-29694 ############################ 管道4已关闭 子进程channel-4-29691 关闭继承的写端 管道5已关闭 子进程channel-5-29694 关闭继承的写端 ####################第4个子进程继承的写端 channel-4-29691 channel-5-29694 channel-6-29705 ############################ 管道4已关闭 子进程channel-4-29691 关闭继承的写端 管道5已关闭 子进程channel-5-29694 关闭继承的写端 管道6已关闭 子进程channel-6-29705 关闭继承的写端 ####################第5个子进程继承的写端 channel-4-29691 channel-5-29694 channel-6-29705 channel-7-29715 ############################ 管道4已关闭 子进程channel-4-29691 关闭继承的写端 管道5已关闭 子进程channel-5-29694 关闭继承的写端 管道6已关闭 子进程channel-6-29705 关闭继承的写端 管道7已关闭 子进程channel-7-29715 关闭继承的写端