【Linux】匿名管道实现简单进程池

#include <iostream>
#include <string>
#include <unistd.h>
#include <sys/types.h>
#include <vector>
#include <sys/wait.h>
#include <vector>
using namespace std;

typedef void (*work_t)(int pipefd);//pipefd为写端文件描述符
typedef void (*task_t)(int pipefd, pid_t fd);//fd为子进程pid

//三个任务
void PrintLog(int pipefd, pid_t fd)
{
    cout <<"pipefd:" << pipefd <<" fd:" << fd <<  " task:printf log task" << endl;
}

void ReloadConf(int pipefd, pid_t fd)
{
     cout <<"pipefd:" << pipefd <<" fd:" << fd <<  " task:Reload Conf" << endl;
}

void ConnectMysql(int pipefd, pid_t fd)
{
     cout <<"pipefd:" << pipefd <<" fd:" << fd <<  " task:Connect Mysql" << endl;
}

task_t task[3] = {PrintLog,ReloadConf,ConnectMysql };

//随机在三个任务中选择一个
uint32_t nextTask()
{
    return rand() % 3;
}


void worker(int pipefd)
{
    while(true)
    {
        uint32_t code = 0;
        //子进程会阻塞在此读，读到零就证明写端已经关闭了
        int n = read(0, &code, sizeof(code));

        if(n == sizeof(code))
        {
            if(code >= 3) continue;
            task[code](pipefd, getpid());
            sleep(1);
        }
        else if(n == 0)
        {
            cout << "child process quit now!" << endl;
            break;
        }
    }
}

#include "processpool.hpp"

enum
{
    UsageError = 1,
    ArgError,
    PipeError
};

void Usage(char *proc)
{
    cout << "Usage:" << proc << " num" << endl;
}

class Channel
{
private:
    //管道保存写端文件描述符，管道名和子进程pid。
    int _wfd;
    string _name;
    pid_t _processid;

public:
    Channel(int wfd, string name, pid_t id)
        : _wfd(wfd), _name(name), _processid(id)
    {
    }

    void print()
    {
        cout << "_wfd:" << _wfd << " _name:" << _name << " _processid:" << _processid << endl;
    }

    const string name() const
    {
        return _name;
    }

    const int wfd() const
    {
        return _wfd;
    }

    const pid_t processid() const
    {
        return _processid;
    }

    void Close()
    {
        close(_wfd);
    }

    ~Channel()
    {
    }
};

//进程池
class processPool
{
private:
    int _sum_child_process;//进程池中进程的个数
    vector<Channel> _channelVect;

public:
    processPool(int sum_child_process)
        : _sum_child_process(sum_child_process)
    {
    }

    // 创建信道和子进程
    int CreateProcess(work_t work)//任务以函数指针的方式传入
    {
        vector<int> fds;
        for (int i = 0; i < _sum_child_process; i++)
        {
            //创建匿名管道
            int pipefd[2];
            int n = pipe(pipefd);
            if (n == -1)
                return PipeError;
            pid_t id = fork();
            if (id == 0)
            {
                //关闭多余的写端描述符，因为父进程在创建子进程的同时会将父进程的文件描述符表也给子进程拷贝一份，
                //这样子进程的文件描述符表就会保存了之前的子进程的写端文件描述符，必须要把之前的子进程的写端文件描述符关闭，
                //否则子进程在退出的时候会出异常
                if (!fds.empty())
                {
                    for (auto e : fds)
                        close(e);
                }
                close(pipefd[1]);
                dup2(pipefd[0], 0);//输入重定向
                work(pipefd[0]);
                exit(0);
            }
            string name = "Channel" + to_string(i);
            close(pipefd[0]);
            _channelVect.push_back(Channel(pipefd[1], name, id));
            fds.push_back(pipefd[1]);
        }
        return 0;
    }
    //发送某个任务给某个管道
    void SendTaskCode(const Channel *channel, uint32_t code)
    {
        cout << "send code: " << code << " to " << channel->name() << " sub prorcess id: " << channel->processid() << endl;
        write(channel->wfd(), &code, sizeof(code));
    }

    // 控制子进程
    void CtrlProcessPool(processPool *processPoolPtr, int cnt)//cnt表示任务的个数
    {
        while (cnt)
        {
            // 1、选择一个进程和信道
            const Channel *cur = processPoolPtr->nextChannel();
            cout << cur->name() << endl;

            // 2、选择一个任务
            uint32_t n = nextTask();

            // 派送任务
            processPoolPtr->SendTaskCode(cur, n);
            sleep(1);
            cnt--;
        }
    }

    //按顺序选择一个管道
    const Channel *nextChannel()
    {
        static int n = 0;
        n %= _sum_child_process;
        return &_channelVect[n++];
    }

    void Print()
    {
        for (auto &channel : _channelVect)
        {
            channel.print();
        }
    }

    // 退出所有子进程
    void killAll()
    {
        for (auto &Channel : _channelVect)
        {
            Channel.Close();
            pid_t rid = Channel.processid();
            pid_t quitid = waitpid(rid, nullptr, 0);
            if (rid == quitid)
                cout << rid << " quit succeed!" << endl;
        }
    }   
    ~processPool()
    {
    }
};

int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        Usage(argv[0]);
        return ArgError;
    }

    // 创建信道和子进程
    int sum_child_process = stoi(argv[1]);
    processPool *processPoolPtr = new processPool(sum_child_process);
    // vector<Channel> channelVect;
    processPoolPtr->CreateProcess(worker);

    srand(time(nullptr));

    // 控制子进程
    int cnt = 10;//10个任务
    processPoolPtr->CtrlProcessPool(processPoolPtr, cnt);

    // 回收子进程
    processPoolPtr->killAll();
    //processPoolPtr->Wait();

    delete processPoolPtr;
    return 0;
}

myprocesspool:processpool.cc
	g++ -o $@ $^
.PHONY:clean
clean:
	rm -f myprocesspool