【Linux】进程间通信——进程池

enum{
    OK = 0,
    UsageError,
    PipeError,
    ForkError,
};

void Usage(string proc){
    cout << "Usage: " << proc << " <number_of_processes>" << endl;
    cout << "Example: " << proc << " 4" << endl;
    exit(UsageError);
}

// main函数的第一个参数是数组元素个数，第二个元素是创建子进程个数
int main(int argc, char *argv[]){
    if(argc != 2){
        Usage(argv[0]);
        return UsageError;
    }
    int num = atoi(argv[1]);
    // 接下来创建管道和子进程
}

using work_t = function<void>;

class ProcessPool{
public:
    ProcessPool(int n, work_t w) : num(n), work(w) {}
    ~ProcessPool() {}
    int InitProcesspool();
    void DisPatchTasks();
    void CleanProcessPool();
private:
    vector<Channel> channels; // 管道
    int num; // 进程的个数
    work_t work; // 工作类型（void()）
};

class Channel{
public:
    Channel(int wfd, pid_t who) : _wfd(wfd), _who(who) {
        _name = "Channel-" + to_string(wfd) + '-' + to_string(who);
    }
    string Name() { return _name; }
    void Send(int cmd) { write(_wfd, &cmd, sizeof(cmd)); }
    void Close() { close(_wfd); }
    pid_t Id() { return _who; }
    ~Channel() {}
private:
    int _wfd; // master的写端
    string _name; // 对应worker的名字
    pid_t _who; // 记录哪个子进程的管道
};

void Download() { cout << "Downloading..." << endl; }
void Log() { cout << "Logging..." << endl; }
void Sql() { cout << "Executing SQL..." << endl; }

using task_t = function<void>;

class TaskManager{
public:
    TaskManager(){
        srand(time(nullptr));
        InsertTask(Download);
        InsertTask(Log);
        InsertTask(Sql);
    }
    ~TaskManager() {}
    void InsertTask(task_t t) { tasks[number++] = t; }
    int SelectTask() { return rand() % number; }
    void Excute(int number) {
        if(tasks.find(number) == tasks.end()) return;
        tasks[number]();
    }
private:
    unordered_map<int, task_t> tasks;
};

static int number = 0;
TaskManager tmp;

void Usage(string proc){
    cout << "Usage: " << proc << " <number_of_processes>" << endl;
    cout << "Example: " << proc << " 4" << endl;
    exit(UsageError);
}

int InitProcesspool(){
    for (int i = 0; i < num; i++) {
        int pipefd[2];
        if (pipe(pipefd) < 0) {
            perror("pipe");
            return PipeError;
        }
        pid_t pid = fork();
        if (pid < 0) {
            perror("fork");
            return ForkError;
        } else if (pid == 0) {
            close(pipefd[1]);
            dup2(pipefd[0], 0);
            close(pipefd[0]);
            work();
            exit(OK);
        } else {
            close(pipefd[0]);
            channels.push_back(Channel(pipefd[1], pid));
        }
    }
    return OK;
}

void DisPatchTasks(){
    int who = 0;
    int num = 20;
    while (num--) {
        int task = tmp.SelectTask();
        Channel &curr = channels[who];
        who++;
        who %= channels.size();
        cout << "Dispatch task " << task << " to " << curr.Name() << endl;
        curr.Send(task);
    }
}

void CleanProcessPool() {
    for (auto &c : channels) {
        c.Close();
    }
    for (auto &e : channels) {
        pid_t rid = waitpid(e.Id(), nullptr, 0);
        if (rid > 0) {
            cout << "子进程 " << rid << " 已回收" << endl;
        }
    }
}

#include "ProcessPool.hpp"
#include "Task.hpp"

int main(int argc, char *argv[]){
    if(argc != 2) {
        Usage(argv[0]);
        return UsageError;
    }
    int num = atoi(argv[1]);
    ProcessPool *pp = new ProcessPool(num, Worker);
    pp->InitProcesspool();
    pp->DisPatchTasks();
    pp->CleanProcessPool();
    delete pp;
    return 0;
}

// Channel.hpp
#ifndef __CHANNEL_HPP__
#define __CHANNEL_HPP__
#include <iostream>
#include <string>
#include <unistd.h>
using namespace std;
// ... (Channel类定义)
#endif

// ProcessPool.hpp
#include <string>
#include <unistd.h>
#include <cstdlib>
#include <vector>
#include <sys>
#include <functional>
#include <sys>
#include "Channel.hpp"
#include "Task.hpp"
// ... (ProcessPool类定义)

// Task.hpp
#pragma once
#include <iostream>
#include <unordered_map>
#include <functional>
#include <ctime>
using namespace std;
// ... (TaskManager类定义)

// Makefile
BIN=processpool
cc=g++
FLAGS=-c -Wall -std=c++11
LDFLAGS=-o
SRC=$(shell ls *.cc)
OBJ=$(SRC:.cc=.o)
$(BIN):$(OBJ)
    $(cc) $(LDFLAGS) $@ $^
%.o:%.cc
    $(cc) $(FLAGS) $<
clean:
    rm -f $(OBJ) $(BIN)