Linux 进程间通信详解：管道、共享内存与内核机制

Linux 进程间通信详解：管道、共享内存与内核机制 | 极客日志

特性	管道	System V IPC	POSIX IPC
标准化	Unix 传统	System V Unix	POSIX 标准
对象标识	文件描述符 (匿名管道)	键值 (Key)	文件系统路径名
生命周期	随进程结束	需显式删除	可配置为随进程结束
访问控制	文件权限	IPC 权限	文件权限
跨平台性	有限	有限	较好
性能	中等	高 (特别是共享内存)	高

#include <unistd.h>
int pipe(int pipefd[2]); // 成功：返回 0，并填充 pipefd
// 失败：返回 -1，并设置 errno 表示错误原因

#include <iostream>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <cstdio>
#include <string.h>

void ChildWrite(int wfd) {
    char buffer[1024];
    int cnt = 0;
    while(true) {
        snprintf(buffer, sizeof(buffer), "I am child, pid: %d, cnt: %d", getpid(), cnt++);
        write(wfd, buffer, strlen(buffer));
        sleep(1); // 每隔一秒写入一行
    }
}

void FatherRead(int rfd) {
    char buffer[1024];
    while(true) {
        buffer[0] = 0;
        size_t n = read(rfd, buffer, sizeof(buffer)-1);
        if(n > 0) {
            buffer[n] = 0;
            std::cout << "Child say: " << buffer << std::endl;
        }
    }
}

int main() {
    int fds[2] = {0};
    int n = pipe(fds);
    if (n < 0) {
        std::cerr << "pipe error!" << std::endl;
        return -1;
    }

    pid_t id = fork();
    if (id == 0) {
        close(fds[0]);
        ChildWrite(fds[1]);
        close(fds[1]);
        exit(0);
    }

    close(fds[1]);
    FatherRead(fds[0]);
    waitpid(id, nullptr, 0);
    close(fds[0]);
    return 0;
}

情况	读端 `read()`	写端 `write()`
管道空	阻塞（直到有数据）	正常写入
管道满	正常读取	阻塞（直到有空间）
所有写端关闭	`read()` 返回 `0`（EOF）	-
所有读端关闭	-	SIGPIPE

#pragma once
#include <iostream>
#include <vector>
#include <ctime>
typedef void (*task_t)();

void PrintLog() { std::cout << "我是一个打印日志的任务" << std::endl; }
void DownLoad() { std::cout << "我是一个下载的任务" << std::endl; }
void Upload() { std::cout << "我是一个上传的任务" << std::endl; }

class TaskManager {
private:
    std::vector<task_t> _tasks;
public:
    TaskManager() { srand(time(nullptr)); }
    void Register(task_t t) { _tasks.push_back(t); }
    int Code() { return rand() % _tasks.size(); }
    void Execute(int code) {
        if(code >= 0 && code < _tasks.size()) {
            _tasks[code]();
        }
    }
    ~TaskManager() {}
};

#pragma once
#include <iostream>
#include <vector>
#include <unistd.h>
#include <cstdlib>
#include <string>
#include <sys/wait.h>
#include "Task.hpp"

class Channel {
private:
    int _wfd;
    pid_t _subid;
    std::string _name;
public:
    Channel(int wfd, int subid) : _wfd(wfd), _subid(subid) {
        _name = "channel- " + std::to_string(wfd) + " - " + std::to_string(subid);
    }
    int Fd() { return _wfd; }
    pid_t Subid() { return _subid; }
    void Send(int code) {
        int n = write(_wfd, &code, sizeof(code));
        (void)n;
    }
    void Close() { close(_wfd); }
    void Wait() { waitpid(_subid, nullptr, 0); }
    ~Channel() {};
};

class ChannelManager {
private:
    std::vector<Channel> _channels;
    int _next;
public:
    ChannelManager() : _next(0) {};
    void InsertChannel(int wfd, pid_t subid) {
        _channels.emplace_back(wfd, subid);
    }
    Channel &Select() {
        auto &c = _channels[_next];
        _next++;
        _next %= _channels.size();
        return c;
    }
    void StopSubProcess() {
        for(auto &channel : _channels) {
            channel.Close();
        }
    }
    void WaitSubProcess() {
        for(auto &channel : _channels) {
            channel.Wait();
        }
    }
    ~ChannelManager() {};
};

const int defaultnum = 5;
class ProcessPool {
private:
    ChannelManager _cm;
    int _process_num;
    TaskManager _tm;
public:
    ProcessPool(int num) : _process_num(num) {
        _tm.Register(PrintLog);
        _tm.Register(DownLoad);
        _tm.Register(Upload);
    }
    void Work(int rfd) {
        while (true) {
            int code = 0;
            ssize_t n = read(rfd, &code, sizeof(code));
            if (n > 0) {
                std::cout << "子进程 [" << getpid() << "] 收到一个任务码：" << code << std::endl;
                _tm.Execute(code);
            } else if (n == 0) {
                break;
            } else {
                break;
            }
        }
    }
    bool Start() {
        for (int i = 0; i < _process_num; i++) {
            int pipefd[2] = {0};
            int n = pipe(pipefd);
            if (n < 0) return false;
            pid_t subid = fork();
            if (subid < 0) return false;
            else if (subid == 0) {
                close(pipefd[1]);
                Work(pipefd[0]);
                close(pipefd[0]);
                exit(0);
            } else {
                close(pipefd[0]);
                _cm.InsertChannel(pipefd[1], subid);
            }
        }
        return true;
    }
    void Run() {
        int taskcode = _tm.Code();
        auto &c = _cm.Select();
        c.Send(taskcode);
    }
    void Stop() {
        _cm.StopSubProcess();
        _cm.WaitSubProcess();
    }
    ~ProcessPool() {};
};

#include "ProcessPool.hpp"
int main() {
    ProcessPool pp(defaultnum);
    pp.Start();
    int cnt = 5;
    while (cnt--) {
        pp.Run();
        sleep(1);
    }
    pp.Stop();
    return 0;
}

process_pool:Main.cc
g++ -o $@ $^
.PHONY:clean
clean:
	rm -f process_pool

void StopAndWait() {
    for (int i = _channels.size() - 1; i >= 0; i--) {
        _channels[i].Close();
        _channels[i].Wait();
    }
}

void CloseAll() {
    for (auto &channel : _channels) {
        channel.Close();
    }
}

$ mkfifo fifo
$ ll total 8
prw-rw-r-- 1 zyt zyt 0 May 22 16:24 fifo|

# 终端 1（写入端）
$ echo "hello fifo">fifo
# 终端 2（读取端）
$ cat < fifo
hello fifo

#include <sys/types.h>
#include <sys/stat.h>
int mkfifo(const char *pathname, mode_t mode);

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <string>
#include <iostream>
#include <unistd.h>
#include "comm.hpp"

int main() {
    umask(0);
    int n = mkfifo(FIFO_FILE, 0666);
    if(n < 0) {
        std::cerr << "mkfifo error" << std::endl;
        return 1;
    }
    int fd = open(FIFO_FILE, O_RDONLY);
    if(fd < 0) {
        std::cerr << "open fifo error" << std::endl;
        return 2;
    }
    while(true) {
        char buffer[1024];
        int number = read(fd, buffer, sizeof(buffer)-1);
        if(number > 0) {
            buffer[number] = 0;
            std::cout << "client say: " << buffer << std::endl;
        } else if(number == 0) {
            break;
        } else {
            break;
        }
    }
    unlink(FIFO_FILE);
    return 0;
}

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <string>
#include <iostream>
#include <unistd.h>
#include "comm.hpp"

int main() {
    int fd = open(FIFO_FILE, O_WRONLY);
    if(fd < 0) {
        std::cerr << "open fifo error" << std::endl;
        return 1;
    }
    std::string message;
    int cnt = 1;
    pid_t id = getpid();
    while(true) {
        std::cout << "Please Enter# ";
        std::getline(std::cin, message);
        message += ", message number " + std::to_string(cnt++) + ", [" + std::to_string(id) + "]";
        write(fd, message.c_str(), message.size());
    }
    close(fd);
    return 0;
}

// comm.hpp
#pragma once
#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#define FIFO_FILE "fifo"
#define PATH "."
#define FILENAME "fifo"
#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); } while (0)

struct Namedfifo {
public:
    Namedfifo(const std::string path, const std::string name) : _path(path), _name(name) {
        _fifoname = _path + "/" + _name;
        umask(0);
        int n = mkfifo(_fifoname.c_str(), 0666);
        if (n < 0) ERR_EXIT("mkfifo");
    }
    ~Namedfifo() {
        unlink(_fifoname.c_str());
    }
private:
    std::string _path;
    std::string _name;
    std::string _fifoname;
};

class FileOper {
public:
    FileOper(const std::string &path, const std::string &name) : _path(path), _name(name), _fd(-1) {
        _fifoname = _path + "/" + _name;
    }
    void OpenForRead() {
        _fd = open(_fifoname.c_str(), O_RDONLY);
        if (_fd < 0) ERR_EXIT("open");
    }
    void OpenForWrite() {
        _fd = open(_fifoname.c_str(), O_WRONLY);
        if (_fd < 0) ERR_EXIT("open");
    }
    void Write() {
        std::string message;
        int cnt = 1;
        pid_t id = getpid();
        while (true) {
            std::cout << "Please Enter# ";
            std::getline(std::cin, message);
            message += ", message number " + std::to_string(cnt++) + ", [" + std::to_string(id) + "]";
            write(_fd, message.c_str(), message.size());
        }
    }
    void Read() {
        while (true) {
            char buffer[1024];
            int number = read(_fd, buffer, sizeof(buffer) - 1);
            if (number > 0) {
                buffer[number] = 0;
                std::cout << "client say: " << buffer << std::endl;
            } else if (number == 0) {
                break;
            } else {
                ERR_EXIT("read");
            }
        }
    }
    void Close() { if (_fd > 0) close(_fd); }
    ~FileOper() {}
private:
    std::string _path;
    std::string _name;
    std::string _fifoname;
    int _fd;
};

struct shmid_ds {
    struct ipc_perm shm_perm;
    int shm_segsz;
    __kernel_time_t shm_atime;
    __kernel_time_t shm_dtime;
    __kernel_time_t shm_ctime;
    __kernel_ipc_pid_t shm_cpid;
    __kernel_ipc_pid_t shm_lpid;
    unsigned short shm_nattch;
};

struct shmid_kernel {
    struct shmid_ds u;
    struct file *shm_file;
    unsigned long shm_nattch;
};

#include <sys/types.h>
key_t ftok(const char *path, int id);

问题	判断依据
共享内存是否存在？	`shmget(key, 0, 0)` 是否成功
是否同一共享内存？	两个进程是否使用相同的 `key`

#include <sys/ipc.h>
#include <sys/shm.h>
int shmget(key_t key, size_t size, int shmflg);

#include <sys/ipc.h>
#include <sys/shm.h>
int shmctl(int shmid, int cmd, struct shmid_ds *buf);

#include <sys/shm.h>
void *shmat(int shmid, const void *shmaddr, int shmflg);

#include <sys/shm.h>
int shmdt(const void *shmaddr);

#pragma once
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/types.h>
#include <iostream>
#include <string>
#include <cstdio>
#include <unistd.h>
#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); } while (0)
const int defaultid = -1;
const int gsize = 4096;
const std::string pathname = ".";
const int projid = 0x66;

class Shm {
public:
    Shm() : _shmid(defaultid), _size(gsize), _start_mem(nullptr) {}
    void Create() {
        key_t k = ftok(pathname.c_str(), projid);
        _shmid = shmget(k, _size, IPC_CREAT | IPC_EXCL);
    }
    void Destory() {
        if (_shmid == defaultid) return;
        shmctl(_shmid, IPC_RMID, nullptr);
    }
    void Attach() {
        _start_mem = shmat(_shmid, nullptr, 0);
    }
    void *VirtualAddr() { return _start_mem; }
    ~Shm() {}
private:
    int _shmid;
    int _size;
    void *_start_mem;
};

#include "comm.hpp"
int main() {
    Shm shm;
    shm.Create();
    sleep(3);
    shm.Attach();
    shm.VirtualAddr();
    sleep(3);
    shm.Destory();
    return 0;
}

// Shm.hpp
#pragma once
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/types.h>
#include <iostream>
#include <string>
#include <cstdio>
#include <unistd.h>
#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); } while (0)
const int defaultid = -1;
const int gsize = 4096;
const std::string pathname = ".";
const int projid = 0x66;
const int gmode = 0666;
#define CREATER "creater"
#define USER "user"

class Shm {
private:
    void CreateHelper(int flag) {
        _shmid = shmget(_key, _size, flag);
    }
    void Create() { CreateHelper(IPC_CREAT | IPC_EXCL | gmode); }
    void Attach() { _start_mem = shmat(_shmid, nullptr, 0); }
    void Destory() { shmctl(_shmid, IPC_RMID, nullptr); }
public:
    Shm(const std::string &pathname, int projid, const std::string &usertype) : _shmid(defaultid), _size(gsize), _start_mem(nullptr), _usertype(usertype) {
        _key = ftok(pathname.c_str(), projid);
        if(_usertype == CREATER) Create();
        else if(_usertype == USER) CreateHelper(IPC_CREAT);
        Attach();
    }
    void *VirtualAddr() { return _start_mem; }
    ~Shm() { if(_usertype == CREATER) Destory(); }
private:
    int _shmid;
    key_t _key;
    int _size;
    void *_start_mem;
    std::string _usertype;
};

#include "shm.hpp"
int main() {
    Shm shm(pathname, projid, CREATER);
    char * mem = (char*)shm.VirtualAddr();
    while (true) {
        printf("%s\n", mem);
        sleep(1);
    }
    return 0;
}

#include "shm.hpp"
int main() {
    Shm shm(pathname, projid, USER);
    char *mem = (char*)shm.VirtualAddr();
    int index = 0;
    for (char c = 'A'; c <= 'Z'; index += 2) {
        mem[index] = c;
        sleep(1);
        mem[index + 1] = c;
        sleep(1);
    }
    return 0;
}

#pragma once
#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#define FIFO_FILE "fifo"
#define PATH "."
#define FILENAME "fifo"
#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); } while (0)

struct Namedfifo {
public:
    Namedfifo(const std::string path, const std::string name) : _path(path), _name(name) {
        _fifoname = _path + "/" + _name;
        umask(0);
        int n = mkfifo(_fifoname.c_str(), 0666);
        if (n < 0) ERR_EXIT("mkfifo");
    }
    ~Namedfifo() { unlink(_fifoname.c_str()); }
private:
    std::string _path;
    std::string _name;
    std::string _fifoname;
};

class FileOper {
public:
    FileOper(const std::string &path, const std::string &name) : _path(path), _name(name), _fd(-1) {
        _fifoname = _path + "/" + _name;
    }
    void OpenForRead() { _fd = open(_fifoname.c_str(), O_RDONLY); }
    void OpenForWrite() { _fd = open(_fifoname.c_str(), O_WRONLY); }
    void Wakeup() { char c = 'c'; write(_fd, &c, 1); }
    bool Wait() { char c; int number = read(_fd, &c, 1); return number > 0; }
    void Close() { if (_fd > 0) close(_fd); }
    ~FileOper() { unlink(_fifoname.c_str()); }
private:
    std::string _path;
    std::string _name;
    std::string _fifoname;
    int _fd;
};

#include "shm.hpp"
#include "Fifo.hpp"
int main() {
    Shm shm(pathname, projid, CREATER);
    Namedfifo fifo(PATH, FILENAME);
    FileOper readerfile(PATH, FILENAME);
    readerfile.OpenForRead();
    char *mem = (char *)shm.VirtualAddr();
    while (true) {
        if (readerfile.Wait()) {
            printf("%s\n", mem);
        } else break;
    }
    readerfile.Close();
    return 0;
}

#include "shm.hpp"
#include "Fifo.hpp"
int main() {
    FileOper writerfile(PATH, FILENAME);
    writerfile.OpenForWrite();
    Shm shm(pathname, projid, USER);
    char *mem = (char*)shm.VirtualAddr();
    int index = 0;
    for (char c = 'A'; c <= 'Z'; index += 2) {
        mem[index] = c;
        sleep(1);
        mem[index + 1] = c;
        sleep(1);
        mem[index + 2] = 0;
        writerfile.Wakeup();
    }
    writerfile.Close();
    return 0;
}

接收模式	参数设置	行为
先进先出 (FIFO)	`msgtyp=0`	读取队列中最早的消息
指定类型	`msgtyp>0`	读取队列中第一个该类型消息
类型阈值	`msgtyp<0`	读取类型≤

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgget(key_t key, int msgflg);

int msgctl(int msqid, int cmd, struct msqid_ds *buf);

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

操作	名称	行为
P (Proberen)	等待/获取	尝试减少信号量值。若值 ≥ 1，则减 1 并继续；否则进程阻塞。
V (Verhogen)	释放/唤醒	增加信号量值。若有进程在等待该信号量，则唤醒其中一个。

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
int semget(key_t key, int nsems, int semflg);

#include <sys/ipc.h>
#include <sys/sem.h>
int semctl(int semid, int semnum, int cmd, ... /* union semun arg */);

#include <sys/ipc.h>
#include <sys/sem.h>
int semop(int semid, struct sembuf *sops, size_t nsops);

struct semid_ds {
    struct ipc_perm sem_perm;
    time_t sem_otime;
    time_t sem_ctime;
    unsigned short sem_nsems;
};

struct ipc_namespace {
    struct ipc_ids ids[3];
    struct shmem_info shm_info;
};

struct ipc_ids {
    struct kern_ipc_perm *entries;
    int in_use;
    unsigned short seq;
    struct rw_semaphore rwsem;
};

struct kern_ipc_perm {
    key_t key;
    uid_t uid;
    gid_t gid;
    mode_t mode;
    int id;
};

Linux 进程间通信详解：管道、共享内存与内核机制

1. 什么是进程间通信

2. 进程间通信的发展和分类（简单介绍）

3. 进程间通信的目的

二、管道

1. 什么是管道？

2. 匿名管道

（1）核心特点

（2）创建匿名管道

（3）fork 共享管道原理

（4）内核角度 - 管道本质

（5）管道代码样例

（6）管道文件的特点

（7）基于匿名管道 --- 进程池

① Task.hpp

② ProcessPool.hpp

③ Main.cc

④ Makefile

⑤ BUG --- 文件描述符的继承和泄漏

3. 命名管道

（1）创建命名管道

① 命令行创建

② 系统调用函数

（2）实例

① 示例：简单实现一个客户端与服务端的通信

② 优化：用类封装

三、System V

1. System V 共享内存

（1）原理

① 物理内存共享

② 虚拟地址映射

③ 直接访问

④ 工作流程

（2）内核数据结构

① struct shmid_ds（用户可见的元信息）

② struct shmid_kernel（内核实际使用的扩展结构）

③ struct file 与物理页帧

④ 操作系统如何跟踪共享内存的使用状态？

（3）标识符和键

① 标识符 shmid

② 键 key

③ 总结

（4）共享内存接口函数

① shmget()

② shmctl()

③ shmat()

④ shmdt()

（5）共享内存使用全流程

① comm.hpp

② server.cc

（6）基于共享内存实现进程间通信

① 无同步机制的进程间通信

② 共享内存 + 管道

2. System V 消息队列

（1）基本概念

（2）接口函数

① msgget()

② msgctl()

③ msgsnd()

④ msgrcv()

3. System V 信号量

（1）并发编程，概念铺垫

（2）原子性

（3）什么是信号量？

（4）信号量接口函数

① semget()

② semctl()

③ semop()

（5）内核信号量集属性

四、内核是如何组织管理 IPC 资源的？

（1）核心数据结构物理布局

（2）内核源代码关系图示

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具