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

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

一、匿名管道通信的四种情况和五种特性

1.1、四种情况

  1. 管道内部没有数据且子进程不关闭自己的写端文件fd,读端(父进程)就要阻塞等待,直到管道里有数据。
  2. 管道内部被写满了且父进程(读端)不关闭自己的读端fd,写端(子进程)写满以后就要阻塞等待。
  3. 对于写端而言,如果写端不写了且关闭了写端fd,读端就会将管道中的数据读完,最后会读到返回值为0,表示读结束,类似于读到了文件的结尾。
  4. 读端关闭了,操作系统就会发送信号直接杀死进行写入的进程,因为没有读端写入也就没有了意义。

1.2、五种特性

  1. 管道自带同步机制,参照上面四种情况中的1,2,3。
  2. 具有血缘关系的进程进行通信,常见于父子。
  3. 管道是面向字节流的。
  4. 父子进程退出,管道自动释放,因为内存中的文件的生命周期是随进程的
  5. 管道只能进行单向通信。

二、匿名管道实现简单的进程池

       这个进程池可以分配我们想要的进程的个数,用命令行的方式来控制进程的个数,任务由我们自己定好,每次随机选择一个任务指派给一个进程去完成,进程的选派采用轮询的方式按顺序指派,这其中还有一些实现的细节,会在代码中以注释的方式给出。

头文件

#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;
}

makefile

myprocesspool:processpool.cc
  g++ -o $@ $^
.PHONY:clean
clean:
  rm -f myprocesspool
相关文章
|
1天前
|
Linux 数据处理
深入了解Linux命令kill:终止进程的艺术
**Linux的`kill`命令详解:高效管理进程的工具** `kill`命令在Linux中用于向进程发送信号,如SIGTERM(默认)和SIGKILL,以终止或影响进程行为。它通过进程ID(PID)操作,支持多种信号和选项,如`-l`列出信号,`-9`强制杀进程。例如,`kill 1234`发送TERM信号,`kill -9 1234`发送KILL信号。使用时注意,SIGKILL是不可忽视的,可能导致数据丢失。配合`pgrep`和`pkill`能更灵活管理进程。了解进程依赖和使用其他命令如`ps`和`top`可优化系统资源管理。
|
2天前
|
Linux
Linux如何创建进程扇和进程链
Linux如何创建进程扇和进程链
|
4天前
|
Linux Shell C语言
Linux进程控制——Linux进程程序替换
Linux进程控制——Linux进程程序替换
12 2
|
4天前
|
Linux 调度
Linux进程控制——Linux进程等待
Linux进程控制——Linux进程等待
11 2
|
4天前
|
存储 缓存 Linux
Linux进程控制——Linux进程终止
Linux进程控制——Linux进程终止
13 2
|
4天前
|
安全 Linux 编译器
Linux进程——进程地址空间
Linux进程——进程地址空间
14 2
|
4天前
|
Linux Shell 编译器
Linux进程——Linux环境变量
Linux进程——Linux环境变量
9 3
|
4天前
|
Linux 调度
Linux进程——Linux进程间切换与命令行参数
Linux进程——Linux进程间切换与命令行参数
8 1
|
4天前
|
Linux 调度
Linux进程——Linux进程与进程优先级
Linux进程——Linux进程与进程优先级
7 1