System-V共享内存和基于管道通信实现的进程池

参考Linux内核源码版本——linux-2.4.3

一.进程间通信:

• 操作系统中,为了保证安全性,进程之间具有严格的独立性(独立的PCB,独立的虚拟地址空间mm_struct和页表…等各种独立的系统资源),即便是父子进程之间也通过数据的写时拷贝保证了两者之间的的数据独立性.因此要实现进程间通信和任务协作,就要让不同进程的共同读写同一份信息资源.由于违背了进程独立性的原则,要实现进程间共享资源就需要一定的技术成本.
• 进程间的独立性:
  

进程间通信的本质:

• 不同的进程对同一块内存资源进行的一系列的读写操作.

二.Linux管道通信

• 将同一个管道文件的文件结构体指针分别填入两个进程的文件信息列表(通过父子进程的继承关系或者open接口实现),之后两个进程便可以对管道文件的内核级读写缓冲区(本质上是一块内存)进行读写操作实现通信.
• 管道通信是一种单向通信手段,有固定的读端进程和写端进程.

匿名管道:

• 匿名管道通信是父子进程间通信的一种方式.
• 匿名管道通信机制图解:
  
  
• 内核视角下管道文件结构体内部结构:

关于管道通信的要点:

• 管道通信可用于进程间协同,提供访问控制(同步与互斥):
  • 管道读写端正常,如果管道中缓冲区为空,则读端进程进入阻塞状态
  • 管道读写端正常,如果管道中缓冲区被写满,则写端进程进入阻塞状态
  • 管道写端先关闭,管道读端read接口返回0,标识读取结束
  • 管道读端先关闭,操作系统会终止写端进程.

基于匿名管道构建进程池:

• Task.hpp模拟任务列表

#include <iostream&gt;
#include <cstdio>
#include <string>
#include <vector>
#include <sys/types.h>
#include <sys/wait.h>
#include <cstdlib>
#include <ctime>
#include <cassert>
#include <unistd.h>

//重定义函数指针
typedef void (*task_t)();

void task1()
{
    std::cout << "执行任务1:矩阵计算" << std::endl;
}
void task2()
{
    std::cout << "执行任务2:pid控制算法" << std::endl;
}
void task3()
{
    std::cout << "执行任务3:图像计算" << std::endl;
}
void task4()
{
    std::cout << "执行任务4:人脸识别算法" << std::endl;
}

//向数组中加载任务
void LoadTask(std::vector<task_t> &amp;tasks)
{
    tasks.push_back(task1);
    tasks.push_back(task2);
    tasks.push_back(task3);
    tasks.push_back(task4);
}

#include "Task.hpp"
using namespace std;
#define ChildNum 5

//子进程信息结构体
class Channel
{
public: 
    Channel(){}
    Channel(const string & Name,pid_t Childpid,int Pipefd)
        : _Childpid(Childpid),
          _Pipefd(Pipefd),
          _Name(Name)
    {}
    
    pid_t getpid() const {return _Childpid;}
    string PrintName() const {return _Name;}
    int Getfd() const {return _Pipefd;}
private:
    pid_t _Childpid;
    int _Pipefd;    //保存管道的写入端
    string _Name;
};


//子进程任务执行函数
void slaver(const vector<task_t>&amp;Taskarr)
{
    //任务码
    int CommandCode = 0;
    while(true)
    {
        int check = read(0,&CommandCode,sizeof(CommandCode));
        //若管道中没有数据且写入段没有关闭,子进程就会阻塞
        assert(check!=-1);
        if(check > 0)
        {
            std::cout <<"slaver say@ get a command: "<< getpid() << " : CommandCode: " <<  CommandCode << std::endl;
            //子进程解析并执行命令
            if(CommandCode < 0 || CommandCode >= Taskarr.size())
            {
                cout << "CommandCode Error! slaver exit!" << endl;
                exit(0);
            }
            //子进程根据任务码执行任务
            Taskarr[CommandCode]();
        }
        else 
        {
            //一旦父进程关闭管道写入端,check就会接收到0,子进程退出
            break;
        }
    }
}

//父进程向子进程发送任务的接口
void ctrlSlaver(const std::vector<Channel> & channels,const vector<task_t>&Taskarr)
{
    int count = 10;
    while(count--)
    {
        sleep(1);
        //随机选择子进程发送任务码
        int choseSlaver = rand()%channels.size();
        int Task = rand()%Taskarr.size();
        cout << "父进程向子进程" << channels[choseSlaver].PrintName() << "写入命令:" << Task << endl;
        write(channels[choseSlaver].Getfd(),&Task,sizeof(Task));
    }
    sleep(1);
    cout << "n所有任务执行完毕,系统准备退出n" << endl;
    sleep(2);
}


//构建进程池接口
void InitProcessPool(vector<Channel>& ChildProc,const vector<task_t>&Taskarr)
{
    for(int i = 0; i < ChildNum; ++i)
    {
        int pipefd[2];
        int check = pipe(pipefd);
        assert(!check); (void)check;
        pid_t pid = fork();
        assert(pid != -1);
        //父进程写 子进程读
        if(pid == 0)
        {
            //子进程执行流
            close(pipefd[1]);
            //将stdin对应文件指针修改为管道的读入端
            dup2(pipefd[0],0);
            //将文件信息列表中对应的指针位置空
            close(pipefd[0]);
            slaver(Taskarr);
            close(0);
            exit(0);
        }
        close(pipefd[0]);
        //将管道的写入端存入channel对象中
        ChildProc.push_back(Channel(string("Process ") + to_string(pid),pid,pipefd[1]));      
    }
}


//父进程轮询等待子进程退出
void WaitChildProc(const std::vector<Channel> & channels)
{
    //先关闭各个管道的写入端,相应的子进程会自动退出
    for(auto& e : channels)
    {
        close(e.Getfd());
    }
    //等待各个子进程退出
    for(auto & e : channels)
    {
        int Status = 0;
        waitpid(e.getpid(),&Status,0);
        cout << "写入端关闭,子进程:" << e.getpid() << "退出,退出码:"<< WIFEXITED(Status) << endl;
    }
}


int main()
{
    vector<task_t>Taskarr;
    LoadTask(Taskarr);
    srand(time(nullptr)^getpid()^1023);
    vector<Channel> ChildProc;
    InitProcessPool(ChildProc,Taskarr);
    ctrlSlaver(ChildProc,Taskarr);
    WaitChildProc(ChildProc);
    return 0;
}

• 命名管道和匿名管道的内核原理相同

三.System-V共享内存

• 共享内存通信原理:
  
• 构建共享内存通信环境的系统接口:
  • int shmget(key_t key, size_t size, int shmflg);
    • key是用户自定义共享内存标识键,用ftok接口获取
    • size是申请共享内存的大小
    • shmflg:取IPC_CREAT时,接口可以申请共享内存并获取共享内存的key,若参数指定的共享内存已存在则直接返回共享内存的key;取IPC_CREAT | IPC_EXCL时,接口只能用于申请新的共享内存.
  • void *shmat(int shmid, const void *shmaddr, int shmflg);
    • 接口作用:在当前进程的虚拟地址空间的共享区中为指定的共享内存块编址,并建立页表映射.
  • int shmdt(const void *shmaddr);
    • 接口作用:共享内存块与当前进程的虚拟地址空间取消关联,进程将无法再访问指定的共享内存
  • int shmctl(int shmid, int cmd, struct shmid_ds *buf);
    • 接口作用:对共享内存块进行cmd码指定的控制操作(比如释放操作),也可以用于获取共享内存块在内核中的描述信息
• 共享内存通信环境中,由于多个进程可以对同一个内存块直接进行读写操作,因此,共享内存通信缺少同步互斥机制,无法保证数据的读写安全,为此,可以借助命名管道为共享内存通信提供读写控制.

共享内存和命名管道协同通信

• 构建通信环境的接口头文件:

#ifndef __COMM_HPP__
#define __COMM_HPP__

#include "log.hpp"
#include <sys/sem.h>
#include <sys/ipc.h>
#include <sys/shm.h>

using namespace std;
const int SIZE = 4096;
const string pathname = "/home/user1/LinuxLearning/sharedMEM";
const int pro_id = 0x123456;

log LOG;


//获取自定义共享内存key
key_t GetKey(const string pathname,const int pro_id)
{
    //KEY生成器
    key_t K = ftok(pathname.c_str(),pro_id);
    if(K < 0)
    {
        LOG(Fatal,"GetKey Error, message: %sn",strerror(errno));
        exit(-1);
    }
    LOG(Info,"key generated, message: %sn",strerror(errno));
    return K;
}

//调用系统接口申请共享内存
int GetShareMemHelper(int flag)
{
    key_t KEY = GetKey(pathname,pro_id);
    //系统调用接口shmget申请共享内存或返回已存在的共享内存id
    int shmid = shmget(KEY,SIZE,flag);
    if(shmid == -1)
    {
        LOG(Fatal,"Get ShareMem failed, message: %sn",strerror(errno));
        exit(-1);
    }
    LOG(Info,"Get ShareMem completed, message: %sn",strerror(errno));
    return shmid;
}

//申请新的共享内存
int CreateShm()
{
    return GetShareMemHelper(IPC_CREAT | IPC_EXCL | 0666);
}

//获取已存在的共享内存的id
int GetShm()
{
    return GetShareMemHelper(IPC_CREAT); 
}

#define FIFO_FILE "./myfifo"
#define MODE 0664

enum
{
    FIFO_CREATE_ERR = 1,
    FIFO_DELETE_ERR,
    FIFO_OPEN_ERR
};

class Init
{
public:
    Init()
    {
        // 创建管道
        int n = mkfifo(FIFO_FILE, MODE);
        if (n == -1)
        {
            perror("mkfifo");
            exit(FIFO_CREATE_ERR);
        }
    }
    ~Init()
    {
        //管道去链接,若引用计数为0则删除管道文件
        int m = unlink(FIFO_FILE);
        if (m == -1)
        {
            perror("unlink");
            exit(FIFO_DELETE_ERR);
        }
    }
};

#endif

• 自制日志类log:

#pragma once
#include <time.h>
#include <stdarg.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <iostream>
#include <sys/types.h>
#include <sys/wait.h>
#include <string>
#include <cstdio>
#include <cstring>
#include <cassert>
#include <unistd.h>
#include <stdlib.h>

//日志等级
#define Info 0
#define Debug 1
#define Warning 2
#define Error 3
#define Fatal 4
//日志写入方式
#define Screen 1
#define Onefile 2
#define Classfile 3
#define LogFile "log.txt"


using std :: string;
class log
{
public:
    log()
    {
        printMethod = Screen;
        path = "./log/";
    }
    void Enable(int method)
    {
        printMethod = method;
    }

    string LeveltoString(int level)
    {
        switch (level)
        {
        case 0:
            return string("Info");
            break;
        case 1:
            return string("Debug");
            break;
        case 2:
            return string("Warning");
            break;
        case 3:
            return string("Error");
            break;
        case 4:
            return string("Fatal");
            break;
        default:
            break;
        }
    }


    void operator()(int level,char * format,...)
    {
        //将时间格式化存入tm结构体中
        time_t t = time(nullptr);
        struct tm* ctime = localtime(&t);
        char leftbuffer[1024];
        snprintf(leftbuffer,sizeof(leftbuffer),"[%s][%d-%d-%d %d:%d:%d]",LeveltoString(level).c_str(),
                 ctime->tm_year+1900,ctime->tm_mon+1,ctime->tm_mday,ctime->tm_hour,ctime->tm_min,ctime->tm_sec);

        //解析可变参数
        va_list vls;
        va_start(vls,format);
        char rightbuffer[1024];
        vsnprintf(rightbuffer,sizeof(rightbuffer),format,vls);
        va_end(vls);

        //合并时间和可变参数
        char logtxt[2048];
        snprintf(logtxt,sizeof(logtxt),"%s %sn",leftbuffer,rightbuffer);

        //执行日志记录
        printLog(level,string(logtxt));
    }

    //日志信息写出接口
    void printLog(int level, const std::string &logtxt)
    {
        switch (printMethod)
        {
        case Screen:
            //将日志信息打印到标准输出
            std::cout << logtxt << std::endl;
            break;
        case Onefile:
            //将日志信息存入log.txt
            printOneFile(LogFile, logtxt);
            break;
        case Classfile:
            //将日志信息存入指定的分类日志文件
            printClassFile(level, logtxt);
            break;
        default:
            break;
        }
    }
    //日志信息写到log.txt中
    void printOneFile(const std::string &logname, const std::string &logtxt)
    {
        //path-->日志保存路径  logname-->日志文件名
        std::string _logname = path + logname;
        //打开"log.txt"日志文件
        int fd = open(_logname.c_str(), O_WRONLY | O_CREAT | O_APPEND, 0666); 
        if (fd < 0)
            return;
        write(fd, logtxt.c_str(), logtxt.size());
        close(fd);
    }
    //日志信息写到log.txt.level中
    void printClassFile(int level, const std::string &logtxt)
    {
        //对日志文件名进行修改,根据日志等级分出多个日志文件
        std::string filename = LogFile;
        filename += ".";
        // "log.txt.Debug(Warning)(Fatal)"
        filename += LeveltoString(level); 
        printOneFile(filename, logtxt);
    }

    ~log(){}
private:
    int printMethod;
    std::string path;
};

• 读端进程示例:

#include "log.hpp"
#include "ShareMemBuild.hpp"

extern log LOG;

int main()
{
    //创建管道和共享内存
    Init pipeCreate;
    int shmid = CreateShm();
    //建立共享内存与进程虚拟地址空间之间的映射,并获取共享内存的虚拟地址
    char * shmaddr = (char *)shmat(shmid,NULL,0);

    //打开管道文件
    int fd = open(FIFO_FILE,O_RDONLY);
    if(fd == -1)
    {
        LOG(Fatal, "error string: %s, error code: %d", strerror(errno), errno);
        exit(FIFO_OPEN_ERR);
    }

    while(true)
    {
        //借助管道进行共享内存的读写控制,若写端没有给信号,则读端保持阻塞状态
        char c;
        int RSize = read(fd,&c,sizeof(c));
        if(RSize <=0) break;

        //直接访问共享内存,实现高效通信
        cout << "client say@ " << shmaddr << endl; 
        sleep(1);
    }

    //进程与共享内存断开连接
    shmdt(shmaddr);
    //将共享内存标记为已销毁
    shmctl(shmid,IPC_RMID,nullptr);

    close(fd);
    return 0;
}

• 写端进程示例:

#include "log.hpp"
#include "ShareMemBuild.hpp"

extern log LOG;

int main()
{   
    //获取共享内存标识
    int shmid = GetShm();
    //建立共享内存与进程虚拟地址空间之间的映射,并获取共享内存的虚拟地址
    char * shmaddr = (char *)shmat(shmid,NULL,0);

    //打开管道文件
    int fd = open(FIFO_FILE,O_WRONLY);
    if(fd == -1)
    {
        LOG(Fatal, "error string: %s, error code: %d", strerror(errno), errno);
        exit(FIFO_OPEN_ERR);
    }
    while(true)
    {
        cout << "Please Enter@ ";
        //将信息写入共享内存
        fgets(shmaddr, 4096, stdin);
        //管道写入信号,解除读端的阻塞状态
        write(fd, "c", 1); 
    }
    //进程与共享内存断开连接
    shmdt(shmaddr);

    close(fd);
    return 0;
}

• 多个进程直接通过各自的虚拟地址空间对同一个内存块进行访问使得共享内存通信具有很高的通信效率.管道通信过程中,数据至少要经过两次拷贝(用户读写缓冲区和内核读写缓冲区之间的拷贝),而共享内存通信不存在通信数据拷贝问题
• 共享内存,消息队列,信号量等通信内存资源(称为ipc资源)统一由操作系统描述为各种数据结构统一进行管理,在Linux内核中,描述共享内存,消息队列,信号量的结构体形成继承体系:(C语言实现的继承体系)
  

代码007普通

打赏 收藏 海报 链接