15.进程间通信（一）

一、进程间通信介绍

进程间通信目的：

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

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

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

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

怎么通信？

进程间通信本质：先让不同的进程，看到同一份资源（内存）（然后才有通信的条件）

不能由任何一个进程提供，进程间数据隔离->OS提供系统调用->设计统一的通信接口。

什么是通信？

二、具体通信方式

1）基于文件的，管道通信

2）System V 本机通信

1.背景

基于已有的技术，直接进行通信。

2.原理

单独设计了一个内存级的文件，管道（复用了文件管理的代码）。

独特的系统调用：

/* On all other architectures */

int pipe(int pipefd[2]);  //数组第一个参数是读的fd，第二个参数的写的fd。

返回值：成功返回0，失败返回-1，错误码被设置。

管道：通过创建子进程，子进程拷贝一份和父进程一样的文件描述符表，指向同一个“文件”（管道），关闭相应的读写端，使得单向通信。

3.demo代码，测试接口。

测试代码如下：

#include <iostream>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>void ChildWrite(int wfd)
{std::cout << "子进程wfd:" << wfd << std::endl;char buff[1024];int cnt = 0;while (true){sleep(1);snprintf(buff, sizeof(buff), "第%d次:子进程写入\n", cnt++);ssize_t ret = write(wfd, buff, strlen(buff));(void)ret;}
}void FatherRead(int rfd)
{std::cout << "父进程rfd:" << rfd << std::endl;char buff[1024];int cnt = 1;while (cnt--){sleep(5);ssize_t ret = read(rfd, buff, sizeof(buff) - 1);buff[ret] = 0;std::cout << buff << std::endl;if (ret == 0){std::cout << "写关闭了，读关闭" << std::endl;break;}}
}int main()
{// 1.创建管道int pipefd[2] = {0};int ret = pipe(pipefd);if (ret == -1)exit(1);// 2.创建子进程，子进程写，父进程读pid_t pid = fork();if (pid < 0)exit(1);else if (pid == 0){// 子进程close(pipefd[0]);ChildWrite(pipefd[1]);close(pipefd[1]);exit(0);}else{// 父进程close(pipefd[1]);FatherRead(pipefd[0]);close(pipefd[0]);int status;waitpid(pid, &status, 0);std::cout << "子进程 exit code:" << ((status >> 8) & 0xFF) << " exit signal:" << (status & 0x7F) << std::endl; }return 0;
}

5种特性（重点）：

1）匿名管道，只能用来进行具有血缘关系的进程进行进程间通信（常用于父子）

2）管道文件，自带同步机制

3）管道是面向字节流的。（怎么读和怎么写没有必然关系）

4）管道是单向通信的。

属于半双工的一种特殊情况。

半双工：任何时刻，一个发，一个收。

全双工：任何时刻，可以同时收发（吵架）。

5）（管道）文件的生命周期是随进程的（引用计数）。

4中通信情况：

1）写慢，读快 ------ 读端就要阻塞（进程）

2）写快，读慢 ------ 满了的时候，写端就要阻塞等待

3）写关，读继续 ------ read读到返回值为0，表示文件结尾。

4）写继续，读关 ------ 写端写入无意义，OS不会做无意义的事情->OS会杀掉写端进程->发送异常信号 13 SIGPIPE 

在小于pipe_buf时，管道的写入被要求是原子性的（一次要写全写完）。

测试管道容量：一次写一个字节，写入测试信息。

4.基于匿名管道 --- 进程池 

原理：

父进程通过向指定的管道写入的方式来向子进程发送对应的任务。

父进程的wfd关闭，子进程会读到0，可以退出。

总体结构：

        要对匿名管道进行管理，先描述在组织，要有对应的类Channel，记录父进程的wfd和子进程的pid，在有对应管理多个类Channel的数据结构vector，对于匿名管道的管理就转换成了对于vector的增删查改。对于任务也做管理，与匿名管道管理类似。

逻辑：

        父进程创建若干个子进程，让子进程阻塞在read中，并死循环执行，子进程一直阻塞，等待父进程发送对应的任务码，根据任务码执行相应的任务。父进程采用轮询的方式，确保能够负载均衡，选择子进程发送随机的任务码，即向对应的匿名管道内写入任务码。

易错点：

回收时需注意，每次创建子进程时，子进程会继承上一次父进程的wfd，导致第1个管道有n个引用，第2个管道有n-1个引用 ... 第n个管道有一个引用。

解决方法1：倒着关闭

解决方法2：随便关闭。创建子进程后，子进程把之前从父进程继承下来的wfd全部关闭。

完整代码如下：

Processpool.hpp

#pragma once#include <vector>
#include <cstdlib>
#include <cstdio>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include "Task.hpp"class Channel
{
public:Channel(int wfd, int pid): _wfd(wfd), _pid(pid){}~Channel(){}int Wfd() const { return _wfd; }int Pid() const { return _pid; }void Close() const { close(_wfd); }void Wait() const { waitpid(_pid, nullptr, 0); }private:int _wfd; // 控制子进程int _pid; // 拿到子进程pid，方便回收
};class ChannelManager
{
public:ChannelManager() : _next(0){}// 为了保证负载均衡，采用轮询的方式const Channel& SelectChannel(){const Channel& c = _channels[_next];// 选出下标++_next;_next %= _channels.size();return c;}void InsertChannel(int wfd, int pid){_channels.emplace_back(wfd, pid);}void CloseFd() const{for(const auto &e : _channels){e.Close();}}void CloseChannels() const{// 2.正着关，创建子进程时就关闭从父进程继承下来的wfdfor (size_t i = 0; i < _channels.size(); i++){// 关闭父进程的写fd，让子进程读到0个字节退出。_channels[i].Close();_channels[i].Wait();printf("等待成功，回收了子进程：%d\n", _channels[i].Pid());}// // 1.倒着关闭// for (int i = _channels.size() - 1; i >= 0; i--)// {//     // 关闭父进程的写fd，让子进程读到0个字节退出。//     _channels[i].Close();//     _channels[i].Wait();//     printf("等待成功，回收了子进程：%d\n", _channels[i].Pid());// }}~ChannelManager(){}private:std::vector<Channel> _channels;int _next;
};class ProcessPool
{
public:ProcessPool(){}~ProcessPool(){}void ChildRead(int rfd) const{int taskcode = 0;while (true){ssize_t ret = read(rfd, &taskcode, sizeof(taskcode));// 父进程写端关闭了，子进程要结束if (ret == 0){std::cout << "父进程写端关闭，子进程:" << getpid() << "退出" << std::endl;break;}// 读到的不是4字节，丢弃，重新读if (ret != sizeof(taskcode)){printf("丢弃\n");return;}// 执行相应任务printf("进程：%d ExcuteTask开始，ret:%d，taskcode:%d\n", getpid(), (int)ret, taskcode);_tm.ExecuteTask(taskcode);}}void Create(int num){for (int i = 0; i < num; i++){// 1.创建管道int pipefd[2] = {0};int ret = pipe(pipefd);if (ret != 0)exit(1);// 2.创建子进程pid_t pid = fork();// 3.关闭父读，子写if (pid < 0)exit(1);else if (pid == 0){// 关掉从父进程继承下来的wfd_cm.CloseFd();// 子进程关闭写close(pipefd[1]);// 子进程工作printf("ChildRead开始，进程为%d\n", getpid());ChildRead(pipefd[0]);close(pipefd[0]);// 子进程完成工作，退出exit(0);}// 父进程关闭读close(pipefd[0]);// emplace_back直接构造，插入到_channels_cm.InsertChannel(pipefd[1], pid);// 循环num次}}// 选择一个子进程，随机发送任务void Run(){// 1.选择一个子进程const Channel &c = _cm.SelectChannel();printf("挑选的子进程为：%d\n", c.Pid());// 2.获取任务码int taskcode = _tm.TaskCode();// 3.发送任务码给子进程，子进程执行(写给子进程)printf("父进程：%d 写入taskcode：%d\n", getpid(), taskcode);ssize_t ret = write(c.Wfd(), &taskcode, sizeof(taskcode));}void Close() const{_cm.CloseChannels();}private:TaskManager _tm;ChannelManager _cm;
};

task.hpp

#pragma once#include <iostream>
#include <cstdio>
#include <vector>
#include <cstdlib>
#include <ctime>void Open()
{std::cout << "这是一个打开的任务" << std::endl;
}
void Download()
{std::cout << "这是一个下载的任务" << std::endl;
}
void Upload()
{std::cout << "这是一个上传的任务" << std::endl;
}typedef void (*task_t)();
class Task
{
public:Task(task_t task) : _task(task){}~Task(){}void operator()() const{_task();}task_t getTask() const {return _task;}
private:task_t _task;
};class TaskManager
{
public:TaskManager(){// 设置种子数，采用随机任务形式srand((unsigned)time(nullptr));Register(Open);Register(Download);Register(Upload);}// 注册任务void Register(task_t t){_tasks.emplace_back(t);}// 返回任务码int TaskCode() const{int ret = rand() % _tasks.size();return ret;}void ExecuteTask(int taskcode) const{if (taskcode < 0 || taskcode >= _tasks.size()){std::cout << "读取的任务码无效" << std::endl;return;}printf("进程：%d执行任务，taskcode为:%d\n",getpid(),taskcode);_tasks[taskcode]();printf("任务：%d 执行完毕\n",taskcode);}~TaskManager(){}private:std::vector<Task> _tasks;
};

main.cc

#include <iostream>
#include <unistd.h>
#include <sys/wait.h>
#include "ProcessPool.hpp"int main()
{pid_t pid = fork();if (pid == 0){ProcessPool pool;pool.Create(5);int cnt = 5;while (cnt--){sleep(1);pool.Run();}sleep(10);pool.Close();exit(0);}int status;waitpid(pid, &status, 0);std::cout << "main exit code:" << ((status >> 8) & 0xFF) << " signal code:" << (status & 0x7F) << std::endl;return 0;
}