实现匿名管道多进程任务派发

mypipe

mypipe.cc

#include <iostream>
#include <string>
#include <unistd.h>
#include <cstring>
#include <cassert>
#include <sys/types.h>
#include <sys/wait.h>
#include <algorithm>
#include "Task.hpp"
#include "Log.hpp"
using namespace std;

#define PROCESS_NUM 10

int waitCommand(int fd, bool &quit)
{
    u_int32_t command = 0;
    ssize_t ret = read(fd, &command, sizeof command);
    Log("子进程读取任务", READ) << endl;
    if (0 == ret)
    {
        quit = true;
        return -1;
    }

    assert(ret == sizeof command);
    return command;
}

int main()
{
    // 装载任务
    load();

    // 父进程保留每个子进程的pid与对应管道文件写端描述符
    vector<pair<pid_t, int>> slots;

    for (size_t i = 0; i < PROCESS_NUM; i++)
    {

        int pipefd[2] = {0};
        int n = pipe(pipefd);

        Log("创建管道文件", CREATE) << endl;

        Log("打开管道文件", OPEN) << endl;

        assert(n != -1);
        // void(n);

        pid_t id = fork();
        assert(id != -1);

        if (0 == id)
        {
            // 子进程
            // 接收任务
            // 保留读，关闭1

            close(pipefd[1]);

            while (true)
            {
                bool quit = false;

                // 等待命令
                // 内部read阻塞等待
                int command = waitCommand(pipefd[0], quit);

                // 执行命令

                if (quit)
                {
                    break;
                }
                if (command > 0 && command < TASK_NUM)
                {

                    // 执行任务
                    callbacks[command]();
                }
                else
                {
                    cout << "非法命令: " << command + 1 << endl;
                }
            }

            exit(0);
            Log("关闭管道文件", CLOSE) << endl;
        }

        // 父进程
        // 保留写，关闭0
        // 保留每个管道对应子进程pid和写端

        close(pipefd[0]);
        slots.push_back(make_pair(id, pipefd[1]));
    }

    // 父进程
    // 分配任务

    srand((unsigned)time(nullptr));

    while (true)
    {

        printf("######################################\n");
        printf("#   1.show task      2.send task     #\n");
        printf("#   3.exit           4.....          #\n");
        printf("######################################\n");

        int command = -1;
        cin >> command;
        if (1 == command)
        {
            // auto it = desc.rbegin();

            // reverse(desc.cbegin(), desc.cend());
            for (auto e : desc)
            {
                cout << e.second << endl;
            }
        }
        else if (2 == command)
        {
            printf("Please chose task: ");

            int task = -1;
            cin >> task;

            // 派发任务给谁？
            int process = rand() % PROCESS_NUM;

            task--;

            write(slots[process].second, &task, sizeof task);

            Log("父进程发送任务", WRITE) << endl;
        }
        else if (3 == command)
        {
            break;
        }
        else
        {
            cout << "非法选项: " << command << endl;
        }

        sleep(2);
    }

    // 关闭fd
    for (auto e : slots)
    {
        close(e.second);
    }

    for (auto e : slots)
    {
        waitpid(e.first, nullptr, 0);
    }

    return 0;
}

Task

Task.hpp

#pragma once
#ifndef _TASK_H_
#define _TASK_H_
#include <unordered_map>
#include <iostream>
#include <functional>
#include <vector>
#include <unistd.h>
using namespace std;

#define TASK_NUM 10

unordered_map<int, string> desc;

typedef function<void()> func;
vector<func> callbacks;
void task_1()
{
    printf("Sub Process [ %d ] 执行任务 1 ......\n", getpid());
}
void task_2()
{
    printf("Sub Process [ %d ] 执行任务 2 ......\n", getpid());
}
void task_3()
{
    printf("Sub Process [ %d ] 执行任务 3 ......\n", getpid());
}
void task_4()
{
    printf("Sub Process [ %d ] 执行任务 4 ......\n", getpid());
}
void task_5()
{
    printf("Sub Process [ %d ] 执行任务 5 ......\n", getpid());
}
void task_6()
{
    printf("Sub Process [ %d ] 执行任务 6 ......\n", getpid());
}
void task_7()
{
    printf("Sub Process [ %d ] 执行任务 7 ......\n", getpid());
}

void task_8()
{
    printf("Sub Process [ %d ] 执行任务 8 ......\n", getpid());
}
void task_9()
{
    printf("Sub Process [ %d ] 执行任务 9 ......\n", getpid());
}
void task_10()
{
    printf("Sub Process [ %d ] 执行任务 10 ......\n", getpid());
}

void load()
{
    desc.insert(make_pair(callbacks.size(), " task_1 "));
    callbacks.push_back(task_1);

    desc.insert(make_pair(callbacks.size(), " task_2 "));
    callbacks.push_back(task_2);

    desc.insert(make_pair(callbacks.size(), " task_3 "));
    callbacks.push_back(task_3);

    desc.insert(make_pair(callbacks.size(), " task_4 "));
    callbacks.push_back(task_4);

    desc.insert(make_pair(callbacks.size(), " task_5 "));
    callbacks.push_back(task_5);

    desc.insert(make_pair(callbacks.size(), " task_6 "));
    callbacks.push_back(task_6);

    desc.insert(make_pair(callbacks.size(), " task_7 "));
    callbacks.push_back(task_7);

    desc.insert(make_pair(callbacks.size(), " task_8 "));
    callbacks.push_back(task_8);

    desc.insert(make_pair(callbacks.size(), " task_9 "));
    callbacks.push_back(task_9);

    desc.insert(make_pair(callbacks.size(), " task_10 "));
    callbacks.push_back(task_10);
}

#endif

Log

#ifndef _LOG_H_
#define _LOG_H_
#pragma once

#include <iostream>
#include <string>

#define CREATE 0
#define OPEN 1
#define WRITE 2
#define READ 3
#define CLOSE 4

std::ostream &Log(std::string message, int mode)
{
    std::cout << "| " << message << " | " << "step " << mode << " | ";
}

#endif

makefile

mypipe:mypipe.cc
	g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
	rm -f mypipe