文章目录
- 一、进程通信的介绍
- 1.1进程间为什么需要通信
- 1.2进程如何通信
- 二、管道
- 2.1匿名管道
- 2.1.1文件描述符理解管道
- 2.1.2接口使用
- 2.1.3管道的4种情况
- 2.1.4管道的五种特征
- 2.2管道的使用场景
- 2.2.1命令行中的管道
- 2.2.2进程池
- 2.命名管道
- 2.1.1原理
- 2.2.2接口
- 2.2.3代码实例
一、进程通信的介绍
1.1进程间为什么需要通信
进程之间需要协同。 例如,学校里面的各个管理层之间都是互相联系的,不能只是纵向管理。正是因为进程之间需要协同,协同的前提条件是进程之间需要通信,数据是有类别的,有的数据是通知就绪的,有些数据是单纯所传递数据,有的是控制相关的数据。
事实:进程是具有独立性的,进程=内核数据结构+进程的代码和数据
进程通信的目的:
- 数据传输:一个进程需要将它的数据发送给另一个进程
- 资源共享:多个进程之间共享同样的资源。
- 通知事件:一个进程需要向另一个或一组进程发送消息,通知它(它们)发生了某种事件(如进程终止时要通知父进程)。
- 进程控制:有些进程希望完全控制另一个进程的执行(如Debug进程),此时控制进程希望能够拦截另一个进程的所有陷入和异常,并能够及时知道它的状态改变。
1.2进程如何通信
进程之间的通信,成本可能稍微高一些:进程是具有独立性的,任何一个进程开辟的资源,另一个进程是看不到的。之前在聊父子进程的时候,父进程的数据被子进程继承下去,这不属于通信,能继承但是不是一直继承,能传递信息和一直能传递信息是不一样的。
进程间的通信的前提:先让不同的进程看到同一份(操作系统)资源(“一段内存”)。两个进程之间是独立的,要实现通信,需要一个工具,即操作系统,使得两个进程之间有一片相同的内存。操作系统这样做的原因是用户决定的。
如何让操作系统创建资源:
- 一定是,某一个进程先需要通信,让OS创建一个共享资源
- OS必须提供很多的系统调用,让进程以系统调用的方式申请系统的资源
OS创建的共享资源的不同、系统调用接口的不同决定进程间通信会有不同的种类。
二、管道
2.1匿名管道
2.1.1文件描述符理解管道
管道本质上是一种内存级文件,它不用往磁盘上进行刷新
首先父进程以读写方式分两次打开一个文件,分两次的原因是为了获得两个 struct file 对象,这样对一个文件就有两个读写指针,让读写操作使用各自独立的指针,这样读写之间就不会相互影响。读写指针记录了当前文件读取或写入的位置,一个 struct file
中只有一个读写指针,在向文件写入(或读取)的时候,读写指针会发生移动,然后再去读取(写入),此时读写指针已经不再最初的位置,无法将刚写入的内容读取上来,因此这里需要分两次以不同的方式打开同一个文件。接着创建子进程,子进程会继承父进程中打开的文件,也就是继承父进程的文件描述符表,此时父子进程就会共享同一个文件资源,子进程可以通过4号文件描述符向文件中进行写入,父进程就可以通过3号文件描述符从文件中进程读取,此时父子进程就实现了数据传输,也就是通信。父子进程看到同一段内存缓冲区,这里我们称之为管道文件。管道只允许单向通信,因为简单。
为什么父子进程会向同一个显示器终端打印数据?
因为对应的子进程会继承父进程对应的文件描述符表,进而会指向同一个文件,也就意味着父进程往一个文件里面打,子进程也会往一个文件里面打,都会写到同样的一个缓冲区里,操作系统就会刷新到同一个显示器。
进程会默认打开三个标准输入标准输出:0,1,2…如何默认打开0,1,2?
所有的,命令都是bash
的子进程,只要bash
打开了,默认的子进程就都打开了。
为什么子进程主动clos(0/1/2),不影响父进程继续使用显示器文件?
内存级引用计数会--
,当内存级引用计数减到0,就释放文件资源。
父子进程关闭不需要的文件描述符,为什么之前需要打开?
为了让子进程继承下去。可以不关闭,建议关了,防止万一误写了。
为什么管道是单向通信的?
方式简单,减少开发成本,只让它进行单向通信,任何一个文件刷新到缓冲区里,再把数据刷新到文件里,这个过程本身就是单向的。
生活中我们见到的简单管道都是单向的,比如自来水管道,一个入口一个出口,符合管道的特点。
2.1.2接口使用
可以使用pipe
来创建一个无名管道,参数不需要文件路径和文件名
int pipe(int pipefd[2]);
fd
:文件描述符数组,其中fd[0]
表示读端, fd[1]
表示写端
返回值:成功返回0
,失败返回错误代码
测试管道接口:
#include<iostream>
#include<cerrno>
#include<cstring>
#include<unistd.h>
int main()
{
int pipefd[2]; // 用于存储管道的两个文件描述符
int n=pipe(pipefd); // 创建管道,返回值为0表示成功,-1表示失败
if(n!=0) //如果 pipe() 函数返回值不为0,表示管道创建失败
{
std::cerr<<"errno: "<<errno<<": "<<"errstring: "<<strerror(errno)<<std::endl;
return 1;
}
std::cout<<"pipefd[0]"<<",pipefd[1]: "<<pipefd[1]<<std::endl;
//如果管道创建成功,使用 std::cout 打印管道的两个文件描述符 pipefd[0] 和 pipefd[1] 的值
return 0;
}
上述代码的主要作用是演示如何在C++中使用 pipe()
函数创建管道,并进行简单的错误处理和输出操作。
pipefd[0]->0->管道文件R(读)端
、pipefd[1]->1->管道文件W(写)端
上面我们创建好管道,接下来创建子进程
让子进程能和父进程进行通信:
#include<iostream>
#include<cerrno>
#include<cstring>
#include<unistd.h>
#include<sys/wait.h>
#include<sys/types.h>
#include<string>
const int size=1024;
std::string getOtherMessage()
{
static int cnt=0;
std::string messageid=std::to_string(cnt);
cnt++;
pid_t self_id=getpid();
std::string stringpid=std::to_string(self_id);
std::string message="messageid: ";
message+=messageid;
message+="my pid is: ";
message+=stringpid;
return message;
}
//子进程进行写入
void subProcessWrite(int wfd)
{
std::string message="father,I am your son process!";
while(true)
{
std::string info=message+getOtherMessage(); //子进程发送给父进程的消息
write(wfd,info.c_str(),info.size()); //写入管道没有写入\0,没有必要写进去
sleep(1);
}
}
//父进程进行读取
void fatherProcessRead(int rfd)
{
char inbuffer[size];
while(true)
{
ssize_t n=read(rfd,inbuffer,sizeof(inbuffer)-1);
if(n>0)
{
inbuffer[n]=0; //==\0
std::cout<<"father get message: "<<inbuffer<<std::endl;
}
}
}
int main()
{
int pipefd[2];
int n=pipe(pipefd);
if(n!=0)
{
std::cerr<<"errno: "<<errno<<": "<<"errstring: "<<strerror(errno)<<std::endl;
return 1;
}
std::cout<<"pipefd[0]"<<",pipefd[1]: "<<pipefd[1]<<std::endl;
sleep(1);
//创建子进程
pid_t id=fork();
if(id==0)
{
std::cout<<"子进程关闭不需要的fd,准备发消息"<<std::endl;
sleep(1);
//子进程 write
//关闭不需要的fd
close(pipefd[0]);
subProcessWrite(pipefd[1]);
close(pipefd[1]);
exit(0);
}
std::cout<<"父进程关闭不需要的fd,准备收消息"<<std::endl;
sleep(1);
//父进程 read
//关闭不需要的fd
close(pipefd[1]);
fatherProcessRead(pipefd[0]);
close(pipefd[0]);
pid_t rid=waitpid(id,nullptr,0);
if(rid>0)
{
std::cout<<"wait child process done"<<std::endl;
}
return 0;
}
getOtherMessage()
函数生成一个带有进程ID信息的消息字符串,用于子进程向父进程发送消息。
subProcessWrite(int wfd)
函数用于子进程,它不断地生成消息并写入管道 (wfd) 中,每隔一秒发送一次消息
fatherProcessRead(int rfd)
函数用于父进程,它不断地从管道 (rfd) 中读取消息并输出到控制台
子进程关闭不需要的管道读取端 (pipefd[0]
),调用 subProcessWrite()
发送消息,然后关闭写入端 (pipefd[1]
)。
父进程关闭不需要的管道写入端 (pipefd[1]
),调用 fatherProcessRead()
接收消息,然后关闭读取端 (pipefd[0]
)。
2.1.3管道的4种情况
-
如果管道是空的,并且写端文件描述符没有关闭,读取条件不具备,读进程(父进程)会被阻塞,自动等待读取条件具备(写入进程再重新写入)。
在sleep(1)
时间内,管道内部没有数据,父进程就在阻塞等待。 -
如果管道被写满了,读端不进行读写但是没有关闭,此时写进程会被阻塞(管道被写满,即写条件不具备),直到写条件具备(读取数据)。
-
管道一直在读并且写端关闭了
fd
,读端会读到0,表示读到了文件结尾。
//子进程进行写入
void subProcessWrite(int wfd)
{
std::string message="father,I am your son process!";
int pipesize=0;
char c='A';
while(true)
{
write(wfd,&c,1);
std::cout << "pipesize: " << ++pipesize << " write charator is : "<< c++ << std::endl;
if(c=='G') break;
sleep(1);
}
std::cout<<"child quit..."<<std::endl;
}
//父进程进行读取
void fatherProcessRead(int rfd)
{
char inbuffer[size];
while(true)
{
ssize_t n=read(rfd,inbuffer,sizeof(inbuffer)-1);
if(n>0)
{
inbuffer[n]=0; //==\0
std::cout<<"father get message: "<<inbuffer<<std::endl;
}
else if(n==0)
{
// 如果read的返回值是0,表示写端直接关闭了,我们读到了文件的结尾
std::cout << "client quit, father get return val: " << n << " father quit too!" << std::endl;
break;
}
else if(n < 0)
{
std::cerr << "read error" << std::endl;
break;
}
}
}
- 读端
fd
直接关闭,写端fd
一直进行写入,这个管道称之为坏的管道,操作系统会杀掉对应的进程,属于异常情况,操作系统会给目标发送信号(13号:SIGPIPE)。写端进程会被操作系统直接使用13号信号关掉,相当于进程出现了异常。
2.1.4管道的五种特征
- 匿名管道:只能用来进行具有血缘关系的进程之间通信(常用于父子进程),因为子进程是对父进程的写时拷贝,不能用于毫不相关的两个进程。
- 管道内部自带进程之间同步机制,同步:多执行流执行代码的时候具有明显的顺序性。在上述代码中,子进程写一个,父进程读一个。
- 文件的声明周期是随进程的
- 管道文件在通信的时候,是面向字节流的。写的次数和读取的次数不是一一匹配的
- 管道的通信模式是一种特殊的半双工
2.2管道的使用场景
2.2.1命令行中的管道
管道 |
在这里用于串联命令,实现对进程信息的过滤、筛选和显示,使得可以实时监视和管理特定的进程活动。
2.2.2进程池
当前有一个父进程(master),提前创建好几个子进程(子进程A、子进程B、子进程C、子进程D),每一个子进程还对应一个管道,用于和父进程进行通信。当父进程需要某一个子进程的时候,只需要将信息传入对应管道的写端,然后对应的子进程从管道读端读取数据。像这种提前创建好多个子进程,我们称之为进程池,这样可以大大减少创建进程的成本,只需要把任务交付给对应的子进程。
如果管道里面没有数据,当前对应的worker进程就在阻塞等待,直到任务的到来。
管道里一旦有数据,对应的子进程就被系统唤醒来处理任务。
对于父进程的任务,要进行后端任务划分的负载均衡。
代码实现:
对信道的一个一个管理转化成对vector
的增删查改,将父进程的文件描述符为_wfd
,写给对应的子程序为_subprocessid
#include <iostream>
#include <string>
#include <vector>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include "Task.hpp"
// master
class Channel
{
public:
Channel(int wfd, pid_t id, const std::string &name)
: _wfd(wfd), _subprocessid(id), _name(name)
{
}
int GetWfd() { return _wfd; }
pid_t GetProcessId() { return _subprocessid; }
std::string GetName() { return _name; }
void CloseChannel()
{
close(_wfd);
}
void Wait()
{
pid_t rid = waitpid(_subprocessid, nullptr, 0);
if (rid > 0)
{
std::cout << "wait " << rid << " success" << std::endl;
}
}
~Channel()
{
}
private:
int _wfd;
pid_t _subprocessid;
std::string _name;
};
void CreateChannelAndSub(int num, std::vector<Channel> *channels, task_t task)
{
// BUG? --> fix bug
for (int i = 0; i < num; i++)
{
// 1. 创建管道
int pipefd[2] = {0};
int n = pipe(pipefd);
if (n < 0)
exit(1);
// 2. 创建子进程
pid_t id = fork();
if (id == 0)
{
if (!channels->empty())
{
// 第二次之后,开始创建的管道
for(auto &channel : *channels) channel.CloseChannel();
}
// child - read
close(pipefd[1]);
dup2(pipefd[0], 0); // 将管道的读端,重定向到标准输入
task();
close(pipefd[0]);
exit(0);
}
// 3.构建一个channel名称
std::string channel_name = "Channel-" + std::to_string(i);
// 父进程
close(pipefd[0]);
// a. 子进程的pid b. 父进程关心的管道的w端
channels->push_back(Channel(pipefd[1], id, channel_name));
}
}
// 0 1 2 3 4 channelnum
int NextChannel(int channelnum)
{
static int next = 0;
int channel = next;
next++;
next %= channelnum;
return channel;
}
void SendTaskCommand(Channel &channel, int taskcommand)
{
write(channel.GetWfd(), &taskcommand, sizeof(taskcommand));
}
void ctrlProcessOnce(std::vector<Channel> &channels)
{
sleep(1);
// a. 选择一个任务
int taskcommand = SelectTask();
// b. 选择一个信道和进程
int channel_index = NextChannel(channels.size());
// c. 发送任务
SendTaskCommand(channels[channel_index], taskcommand);
std::cout << std::endl;
std::cout << "taskcommand: " << taskcommand << " channel: "
<< channels[channel_index].GetName() << " sub process: " << channels[channel_index].GetProcessId() << std::endl;
}
void ctrlProcess(std::vector<Channel> &channels, int times = -1)
{
if (times > 0)
{
while (times--)
{
ctrlProcessOnce(channels);
}
}
else
{
while (true)
{
ctrlProcessOnce(channels);
}
}
}
void CleanUpChannel(std::vector<Channel> &channels)
{
// int num = channels.size() -1;
// while(num >= 0)
// {
// channels[num].CloseChannel();
// channels[num--].Wait();
// }
for (auto &channel : channels)
{
channel.CloseChannel();
channel.Wait();
}
// // 注意
// for (auto &channel : channels)
// {
// channel.Wait();
// }
}
// ./processpool 5
int main(int argc, char *argv[])
{
if (argc != 2)
{
std::cerr << "Usage: " << argv[0] << " processnum" << std::endl;
return 1;
}
int num = std::stoi(argv[1]);
LoadTask();
std::vector<Channel> channels;
// 1. 创建信道和子进程
CreateChannelAndSub(num, &channels, work1);
// 2. 通过channel控制子进程
ctrlProcess(channels, 5);
// 3. 回收管道和子进程. a. 关闭所有的写端 b. 回收子进程
CleanUpChannel(channels);
// sleep(100);
return 0;
}
Task.hpp:
#pragma once
#include <iostream>
#include <ctime>
#include <cstdlib>
#include <sys/types.h>
#include <unistd.h>
#define TaskNum 3
typedef void (*task_t)(); // task_t 函数指针类型
void Print()
{
std::cout << "I am print task" << std::endl;
}
void DownLoad()
{
std::cout << "I am a download task" << std::endl;
}
void Flush()
{
std::cout << "I am a flush task" << std::endl;
}
task_t tasks[TaskNum];
void LoadTask()
{
srand(time(nullptr) ^ getpid() ^ 17777);
tasks[0] = Print;
tasks[1] = DownLoad;
tasks[2] = Flush;
}
void ExcuteTask(int number)
{
if (number < 0 || number > 2)
return;
tasks[number]();
}
int SelectTask()
{
return rand() % TaskNum;
}
void work()
{
while (true)
{
int command = 0;
int n = read(0, &command, sizeof(command));
if (n == sizeof(int))
{
std::cout << "pid is : " << getpid() << " handler task" << std::endl;
ExcuteTask(command);
}
else if (n == 0)
{
std::cout << "sub process : " << getpid() << " quit" << std::endl;
break;
}
}
}
void work1()
{
while (true)
{
int command = 0;
int n = read(0, &command, sizeof(command));
if (n == sizeof(int))
{
std::cout << "pid is : " << getpid() << " handler task" << std::endl;
ExcuteTask(command);
}
else if (n == 0)
{
std::cout << "sub process : " << getpid() << " quit" << std::endl;
break;
}
}
}
void work2()
{
while (true)
{
int command = 0;
int n = read(0, &command, sizeof(command));
if (n == sizeof(int))
{
std::cout << "pid is : " << getpid() << " handler task" << std::endl;
ExcuteTask(command);
}
else if (n == 0)
{
std::cout << "sub process : " << getpid() << " quit" << std::endl;
break;
}
}
}
2.命名管道
2.1.1原理
如何保证两个毫不相连的两个进程打开同一个文件?每一个文件都有文件路劲(唯一性路劲),两个进程使用同一个文件路劲。
磁盘中的文件是一个特殊文件,经过路劲标识,命名管道本质上就是系统中的一个内存级文件,它和匿名管道一样,不会向磁盘中刷新,但是它有文件名。路径+文件名,唯一标识了一个命名管道。
2.2.2接口
认识mkfifo
:
创建一个管道:
mkfifo 文件名
2.2.3代码实例
namedPipe.hpp:
#pragma once
#include <iostream>
#include <cstdio>
#include <cerrno>
#include <string>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
const std::string comm_path = "./myfifo";
#define DefaultFd -1
#define Creater 1
#define User 2
#define Read O_RDONLY
#define Write O_WRONLY
#define BaseSize 4096
class NamePiped
{
private:
bool OpenNamedPipe(int mode)
{
_fd = open(_fifo_path.c_str(), mode);
if (_fd < 0)
return false;
return true;
}
public:
NamePiped(const std::string &path, int who)
: _fifo_path(path), _id(who), _fd(DefaultFd)
{
if (_id == Creater)
{
int res = mkfifo(_fifo_path.c_str(), 0666);
if (res != 0)
{
perror("mkfifo");
}
std::cout << "creater create named pipe" << std::endl;
}
}
bool OpenForRead()
{
return OpenNamedPipe(Read);
}
bool OpenForWrite()
{
return OpenNamedPipe(Write);
}
// const &: const std::string &XXX
// * : std::string *
// & : std::string &
int ReadNamedPipe(std::string *out)
{
char buffer[BaseSize];
int n = read(_fd, buffer, sizeof(buffer));
if(n > 0)
{
buffer[n] = 0;
*out = buffer;
}
return n;
}
int WriteNamedPipe(const std::string &in)
{
return write(_fd, in.c_str(), in.size());
}
~NamePiped()
{
if (_id == Creater)
{
int res = unlink(_fifo_path.c_str());
if (res != 0)
{
perror("unlink");
}
std::cout << "creater free named pipe" << std::endl;
}
if(_fd != DefaultFd) close(_fd);
}
private:
const std::string _fifo_path;
int _id;
int _fd;
};
client.cc:
#include "namedPipe.hpp"
// write
int main()
{
NamePiped fifo(comm_path, User);
if (fifo.OpenForWrite())
{
std::cout << "client open namd pipe done" << std::endl;
while (true)
{
std::cout << "Please Enter> ";
std::string message;
std::getline(std::cin, message);
fifo.WriteNamedPipe(message);
}
}
return 0;
}
server.cc:
#include "namedPipe.hpp"
// server read: 管理命名管道的整个生命周期
int main()
{
NamePiped fifo(comm_path, Creater);
// 对于读端而言,如果我们打开文件,但是写还没来,我会阻塞在open调用中,直到对方打开
// 进程同步
if (fifo.OpenForRead())
{
std::cout << "server open named pipe done" << std::endl;
sleep(3);
while (true)
{
std::string message;
int n = fifo.ReadNamedPipe(&message);
if (n > 0)
{
std::cout << "Client Say> " << message << std::endl;
}
else if(n == 0)
{
std::cout << "Client quit, Server Too!" << std::endl;
break;
}
else
{
std::cout << "fifo.ReadNamedPipe Error" << std::endl;
break;
}
}
}
return 0;
}
运行结果: