【Linux】进程间通信之管道实现进程池

#include <iostream>
#include <vector>
#include <string>
#include <unistd.h>
#include "task.hpp"
#include <sys/stat.h>
#include <sys/wait.h>
#include <cstdio>

#define PROCESSNUM 10
std::vector<task_t> tasks;

int main()
{
	//加载任务
	LoadTask(&tasks);
	//定义一个vector管理所有的管道，channel是描述，channels是组织
    std::vector<channel> channels;
    //初始化
    InitProcessPool(&channels);
    //开始进行
    StartProcessPool(channels);
    //清理
    CleanProcessPool(channels);
    return 0;
}

#pragma once

#include <iostream>
#include <vector>
//定义一个函数指针task_t指向返回值为void，没有参数的函数
typedef void (*task_t)();

void task1()
{
    std::cout << "this is task1 running" << std::endl;
}
void task2()
{
    std::cout << "this is task2 running" << std::endl;
}
void task3()
{
    std::cout << "this is task3 running" << std::endl;
}
void task4()
{
    std::cout << "this is task4 running" << std::endl;
}
//加载任务函数，将任务pushback到vector中
void LoadTask(std::vector<task_t> *tasks)
{
    tasks->push_back(task1);
    tasks->push_back(task2);
    tasks->push_back(task3);
    tasks->push_back(task4);
}

class channel
{
public:
    // 描述父进程的fd，对应子进程的pid，子进程的名字
    channel(int cmdfd, int slaverid, const std::string &processname)
    :_cmdfd(cmdfd),_slaverid(slaverid),_processname(processname)
    {}

    int _cmdfd;
    pid_t _slaverid;
    std::string _processname;
};

void slaver()
{
    while(true)
    {
        // 用于存储从标准输入读取的命令码
        int cmdcode = 0;
        // 从标准输入（管道）读取数据，尝试读取sizeof(int)字节的数据到cmdcode中
        // 如果父进程不给子进程发送数据子进程就会进入阻塞等待
        int n = read(0, &cmdcode, sizeof(int));    
        if(n == sizeof(int))
        {
            // read的返回值与sizeof(int)相等，就输出子进程pid和获得命令码
            // 如果命令码有效就调用task任务，无效就退出
            std::cout <<"slaver say@ get a command: "<< getpid() << " : cmdcode: " 
            <<  cmdcode << std::endl;
            if(cmdcode >= 0 && cmdcode < tasks.size()) tasks[cmdcode]();
        }
        else break;
    }
}

void InitProcessPool(std::vector<channel>* channels)
{
    //用于存储之前创建的管道的写端文件描述符
    //目的是让后续创建的子进程可以关闭这些旧的写端文件描述符，避免资源泄漏
    std::vector<int> oldfds;
    //循环创建子进程
    for(int i = 0; i < PROCESSNUM; i++)
    {
        int pipefd[2] = {0};
        int n = pipe(pipefd);
        if(n < 0)
        {
            return;
        }

        pid_t id = fork();
        if(id < 0)
        {
            return;
        }
        if(id == 0)
        {
            //打印子进程pid，打印并关闭上一个管道写端文件描述符
            std::cout << "child : " << getpid() << " close history fd: ";
            for(auto fd : oldfds)
            {
                std::cout << fd << " ";
                close(fd);
            }
            std::cout << std::endl;
            //关闭写端通道
            close(pipefd[1]);
            //将当前管道的读端文件描述符复制到标准输入
            //这样子进程就可以通过标准输入从管道读取数据
            dup2(pipefd[0],0);
            // 读取完关闭管道读端
            close(pipefd[0]);
            // 子进程主要业务
            slaver();
            //打印子进程要退出了
            std::cout << "process : " << getpid() << " quit" << std::endl;
            exit(0);
        }
        //父进程开始
        //关闭读端
        close(pipefd[0]);
        //将当前channel信息添加到channels进行组织
        std::string name = "process-" + std::to_string(i);
        channels->push_back(channel(pipefd[1],id,name));
        //添加这个写端的文件描述符，方便后面的进程关闭它
        oldfds.push_back(pipefd[1]);
        sleep(1);
    }
}

//打印一个选择任务的菜单
void Menu()
{
    std::cout << "################################################" << std::endl;
    std::cout << "# 1. 任务一             2. 任务二               #" << std::endl;
    std::cout << "# 3. 任务三             4. 任务四               #" << std::endl;
    std::cout << "#                   0. 退出                     #" << std::endl;
    std::cout << "#################################################" << std::endl;
}

void StartProcessPool(std::vector<channel>* channels)
{
    while(true)
    {	
        int select = 0;
        Menu();
        sleep(1);
        //输入选项
        std::cout << "Please Enter>> ";
        std::cin >> select;

        if(select <= 0 || select >= 5) break;
		//将控制码也就是选择的数字select1234转化为0123，因为vector下标从0开始，所以要-1
        int cmdcode = select - 1;
		//通过管道写入信息，等待slaver()读取
        write(channels[select]._cmdfd, &cmdcode, sizeof(cmdcode));
        sleep(1);
    }
}

void CleanProcessPool(std::vector<channel> &channels)
{
    //每个channel对象的左边为父进程的fd，右边为子进程fd，断开父进程fd，然后进程等待
    //父进程断开后子进程会在管道中读到0，即文件结束，然后子进程就会终止
    //然后被父进程回收
    for(const auto &c : channels){
        close(c._cmdfd);
        waitpid(c._slaverid, nullptr, 0);
    }
}