Linux学习六(进程间通信)

Linux学习六 : 进程间通信

进程间通信

进程间通信（IPC）：指多个进程之间传输数据或共享信息的机制，在操作系统中每个进程的地址空间和资源是独立的，为了实现多个进程间的数据交换和协作，需要使用IPC机制。最终结果就是进程能够访问相同的内存区域。

进程间通信的方法：

管道：有名管道，无名管道。
消息队列
信号量
共享内存：内存映射实现，共享内存传递数据。
信号：通过特定信号执行处理情况。
socket：主要是网络通信中用到。

管道

无名管道：有亲缘关系的进程间单向通信。管道的本质其实就是内核中的一块内存(或者叫内核缓冲区)，这块缓冲区中的数据存储在一个环形队列中，因为管道在内核里边，因此我们不能直接对其进行任何操作。

半双工，数据单向流动，要实现全双工通信，要用两个管道。
字节流通信，数据格式由用户自行定义
多用于父子进程间
管道对应的内核缓冲区大小是固定的，默认为4k

无名管道实现原理：
* 父进程调用pipe函数会创建两个文件分别用作读和写，对应节点为pipe inode。
* 父进程调用fork创建子进程，子进程拷贝父进程的文件表，由于父子进程文件表内容相同，指向的file相同，所以最终父子进程操作的pipe管道相同。
* fork函数成功后，父子进程不能同时保留读写文件描述符，需要关闭读或写文件描述符，防止父子进程同时读写引发数据错误。
在这里插入图片描述
创建无名管道进行父子进程的通信：

#include<unistd.h>
#include<iostream>

using namespace std;

int pipe_test(){
    int fd[2];
    int ret = pipe(fd);
    if(ret == -1){
        perror("pipe");
        return -1;
    }
    ret = fork();
    if(ret == -1){
        perror("fork");
        return -1;
    }else if(ret == 0){
        // 子进程
        close(fd[0]); // 关闭读端
        string s = "123456";
        while(1){
            write(fd[1],s.c_str(),s.size());
            sleep(1);
        }
    }else if(ret>0){
        // 父进程
        close(fd[1]); // 关闭写端
        while(1){
            char buf[1024] = {0};
            read(fd[0],buf,1024);
            cout<<buf<<endl;
        }
        wait(NULL);
    }
}

int main(){
    pipe_test();
    return 0;
}

有名管道：（FIFO文件）是一种特殊类型的文件，在磁盘上有实体文件, 文件类型为p ，有名管道文件大小永远为0，因为有名管道也是将数据存储到内存的缓冲区中，打开这个磁盘上的管道文件就可以得到操作有名管道的文件描述符，通过文件描述符读写管道存储在内核中的数据。

可以通过名称进行识别和访问，而不仅仅依赖于文件描述符，因此相比于无名管道，有名管道可以用于没有亲缘关系的进程间通信。
可以像其他文件一样进行访问和管理，文件类型为p。
半双工通信，同时写入和读取操作，但需要多个fifo文件。

有名管道的创建方式有两种：

通过命令：mkfifo 文件名
通过函数：int mkfifo(const char *pathname, mode_t mode); pathname是要创建的管道的名字，mode为文件权限。

有名管道实现原理：管道创建成功以后，进程调用open打开FIFO文件，多个进程都可以打开相同的inode节点，进程间都可以看到管道内存空间，所以进程间能够正常通信。多个进程同时读写一个命名管道可能会出现数据异常，所以进程调用open函数时需要指定打开标志为O_RDONLY或者O_WRONLY。

#include<unistd.h>
#include<iostream>
#include <sys/wait.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <string.h>
using namespace std;

#define FIFO_PATH "./testfifo"
int fifo_read(){

    int rfd = open(FIFO_PATH, O_RDONLY);
    if(rfd == -1)
    {
        perror("open");
        return -1;
    }
    while(1){
        char buf[1024] = {0};
        read(rfd,buf,1024);
        cout<<buf<<endl;
    }
    close(rfd);
    return 0;
}

int fifo_write(){
    int ret = mkfifo(FIFO_PATH,0644);
    if((ret == -1)&& errno!= EEXIST){
        perror("mkfifo");
        return -1;
    }

    int wfd = open(FIFO_PATH, O_WRONLY);
    if(wfd == -1)
    {
        perror("open");
        return -1;
    }
    int i = 0;
    while(i<100)
    {
        char buf[1024];
        sprintf(buf, "hello, fifo, 我在写管道...%d\n", i);
        write(wfd, buf, strlen(buf));
        i++;
        sleep(1);
    }
    close(wfd);
    return 0;
}

int fifo_test(){
    int ret = fork();
    if(ret == 0){
        fifo_read();
    }else if(ret > 0){
        fifo_write();
    }else{
        perror("fork");
        return -1;
    }
    return 0;
}

int main(){
    fifo_test();
    return 0;
}

消息队列

System V消息队列：允许在同一系统上运行的不同进程间进行消息传递。

可以实现独立的进程间通信，不受进程的启动和结束顺序的影响。
允许多个进程同时向消息队列中写入和读取消息，实现了并发处理。
通过消息优先级机制，可以优先处理重要的消息。

消息队列实现原理：具有相同IPC命名空间的进程可以同时访问IPC命名空间相同内存。
在这里插入图片描述
IPC对象是消息队列，信号量，共享内存的父对象

ftok函数用于产生Sytem V键值。
msgget函数用于创建或获取一个System V消息队列，返回标识ID，后续根据标识ID查找消息队列进行进程间通信。
msgsnd函数将一个消息添加到一个指定的消息队列中。msgsnd函数的参数非常重要，需要仔细查阅用法。
msgrvc函数接收缓冲区中的消息。
msgctl函数对消息队列中进行控制操作。

ipcs命令可用于查看System V IPC对象信息（消息队列，信号量，共享内存）
ipcs -a // 查看消息队列，信号量，共享内存
ipcs -q // 查看消息队列
ipcmk主要用于创建IPC对象信息。
ipcmk -Q 创建消息队列
ipcrm -q 删除消息队列

// 打开两个终端进行读写即可看到消息队列的交互
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <iostream>
#include <string.h>
#include <unistd.h>

using std::cout;
using std::endl;

struct message {
    long mtype;
    char mtext[100];
};

int main(int argc, char *argv[]){

    int mode = atoi(argv[1]);

    key_t key = 1234;//也可以用ftok
    int id = msgget(key,0644|IPC_CREAT); //通过key值创建消息队列

    while(1){
        if(mode == 0){
            message msg;
            msg.mtype = atoi(argv[2]);
            int ret = msgrcv(id, &msg, sizeof(msg), msg.mtype, 0); // 接收消息
            if(ret == -1){
                perror("msgrcv");
                break;
            }
            cout<<" type: "<<msg.mtype <<" mtext: "<<msg.mtext<<endl;
        }else{
            message msg;
            msg.mtype = atoi(argv[2]);
            strcpy(msg.mtext, "Hello Message Queue"); //可以手动输入
            int ret = msgsnd(id,&msg,sizeof(msg),0); // 发送消息
            if(ret == -1){
                perror("msgsnd");
                break;
            }
            sleep(1);
        }
    }
    msgctl(id,IPC_RMID,0); // 通过id删除消息队列
    return 0;
}

POSIX消息队列：POSIX消息队列是一种基于文件的消息队列，system v进程间通信实现方式不能和文件兼容，Linux一切皆文件，所以选择posix进程间通信方式会更好些。
在这里插入图片描述
POSIX消息队列是基于mqueue文件系统实现，POSIX消息队列其实就是mqueue文件系统的一个inode节点。
mount | grep “mqueue” 查看mqueue文件系统挂载点，挂在路径为/dev/mqueue。
POSIX消息队列底层实现为mqueue inode节点，基于红黑树存储信息。
在这里插入图片描述

mq_open：打开或创建一个消息队列
mq_send：将消息太耐到消息队列，如果消息队列已满，则阻塞、
mq_receive：接收消息队列中的消息。
mq_notify：消息队列有通知的系统调用，可以用于进程发送通知，告诉它有新的消息到达了消息队列。
mq_cloas：关闭一个消息队列。
mq_unlink：删除一个消息队列，新使用一个消息队列，删除旧的。

POSIX消息队列编译的时候要加上 -lrt

// 消息队列，用mq_notify多线程的方式去读数据

#include<fcntl.h>
#include<sys/stat.h>
#include<mqueue.h>
#include<iostream>
#include<unistd.h>
#include<signal.h>

using std::cout;
using std::endl;

#define TEST_STRING "123456"

int do_notify(int fd);
void test_proc(sigval_t val);

// 处理函数
void test_proc(sigval_t val){
    int fd = val.sival_int; //进程收到通知后，注册信息失效，需要重新注册
    do_notify(fd);
    char rbuf[2048] = {0};
    unsigned int prio = 0;
    int ret = mq_receive(fd,rbuf,2048,&prio);
    if(ret == -1 ){
        perror("mq_receive");
        return;
    }
    cout<<"ret: "<<ret <<" prio: "<<prio<<" rbuf: "<<rbuf<<endl;
}

// mq_notify多线程的方式去读数据
int do_notify(int fd){
    struct sigevent ev;
    ev.sigev_value.sival_int = fd;
    ev.sigev_notify = SIGEV_THREAD; // 线程
    ev.sigev_notify_function = test_proc; //绑定函数
    ev.sigev_notify_attributes = NULL;
    int ret = mq_notify(fd,&ev);
    if(ret == -1 ){
        perror("mq_notify");
        return -1;
    }
    return 0;
}

int main(int argc, char *argv[]){

    int mode = atoi(argv[1]);

    if(mode == -1){
        mq_unlink("/posixMq1"); // 删除之前的消息队列，如果消息队列有更新，不清除的话，会占用
        return 0;
    }

    struct mq_attr attr; //设置mq属性
    attr.mq_flags = 0;
    attr.mq_maxmsg = 10;
    attr.mq_msgsize = 1500; 

    mqd_t fd = mq_open("/posixMq1", O_RDWR|O_CREAT, 0664, &attr);
    if(fd == -1 ){
        perror("mq_open");
        return -1;
    }

    if(mode==0)
        do_notify(fd); //完成注册
    while(1){
        if(mode == 0){
            sleep(1);
#if 0
            char rbuf[2048] = {0};
            unsigned int prio = 0;
            int ret = mq_receive(fd,rbuf,2048,&prio);
            if(ret == -1 ){
                perror("mq_receive");
                break;
            }
            cout<<"ret: "<<ret <<" prio: "<<prio<<" rbuf: "<<rbuf<<endl;
#endif
        }else{
            int ret = mq_send(fd,TEST_STRING,sizeof(TEST_STRING),1);
            if(ret == -1 ){
                perror("mq_send");
                break;
            }
            sleep(1);
        }
    }
    return 0;
}

信号量

System V 信号量（System V Semaphores）和 POSIX 信号量（POSIX Semaphores）都是用于多进程或多线程之间进行进程同步和互斥的机制。

System V 信号量是一种在系统级别上维护的计数器，用于控制对共享资源的访问。它使用三个基本操作来操作信号量：创建、初始化和执行 P（proberen）和 V（verhogen）操作。P 操作用于申请资源，如果资源不可用，则进程会等待。V 操作用于释放资源，允许其他进程继续访问该资源。System V 信号量可以在不同进程间共享，并且可以持久化存储在系统中。

P操作：等待操作或者减操作，用于申请资源，当信号量的值大于0时，将其减一，当信号量的值等于0时，进程将被阻塞。
V操作：释放操作或加操作，用于释放资源，将信号量的值加一，并唤醒等待的进程。

实现原理：具有相同IPC命名空间的进程能够同时访问IPC命名空间相同内存空间，命名空间内维护信号量，和消息队列大同小异。
在这里插入图片描述

semget：创建或打开一个信号量集
semop：对信号量进行操作，semval：信号量值。
semctl：对一个已经存在的信号量集值进行各种操作，比如获取信号量集值的信息，设置信号量集的值，删除信号量等。
ipcs-a：查看消息队列，信号量，共享内存
ipcs-s：查看信号量
ipcmk -S 2 -p 0644 //创建信号量，权限为0644
ipcrm -s 6 //删除标识ID 6的信号量。

#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include <iostream>
#include <unistd.h>
#include <signal.h>

using std::cout;
using std::endl;

union semun {
    int val;                // 用于 SETVAL 操作，设置信号量的初始值
    struct semid_ds *buf;    // 用于 IPC_STAT 和 IPC_SET 操作，获取和设置信号量的状态信息
    unsigned short *array;   // 用于 GETALL 和 SETALL 操作，获取和设置信号量的值数组
    struct seminfo *__buf;   // 用于 IPC_INFO 操作，获取系统的信号量信息
    void *__pad;
};

// 初始化信号量,用setval命令
int sem_init_value(int id,int num,int value){
    union semun un;
    un.val = value;
    int ret = semctl(id,num,SETVAL,un);
    if(ret == -1){
        perror("semctl SETVAL");
        return -1;
    }
    cout<<"sem init value:"<<value<<endl;
    return 0;
}

// 删除信号量
int sem_del_sem(int id) {
    union semun un;
    if (semctl(id,0,IPC_RMID,un)==-1) {
        perror("semctl IPC_RMID");
        exit(1);
    }
    return 0;
}

// 获取信号量值
int sem_get_value(int id,int num){
    union semun un;
    int ret = semctl(id,num,GETVAL,un);
    if(ret == -1){
        perror("semctl GETVAL");
        return -1;
    }
    cout<<"sem get value:"<<ret<<endl;
    return 0;
}

int sem_p(int id){
    struct sembuf buf;
    buf.sem_num = 0; // 信号量编号
    buf.sem_op = -1; // P操作代表符号
    buf.sem_flg |= SEM_UNDO;
    int ret=semop(id,&buf,1);
    if(ret == -1){
        perror("semop p");
        return -1;
    }
    return 0;
}

int sem_w(int id){
    struct sembuf buf;
    buf.sem_num = 0; //信号量编号
    buf.sem_op = 0; // w操作代表符号
    buf.sem_flg |= SEM_UNDO; //系统退出前未释放信号量，系统自动释放
    int ret=semop(id,&buf,1);
    if(ret == -1){
        perror("semop w");
        return -1;
    }
    return 0;
}

// 释放资源
int sem_v(int id){
    struct sembuf buf;
    buf.sem_num = 0; //信号量编号
    buf.sem_op = 1; // v操作代表符号
    buf.sem_flg |= SEM_UNDO; //系统退出前未释放信号量，系统自动释放
    int ret=semop(id,&buf,1);
    if(ret == -1){
        perror("semop v");
        return -1;
    }
    return 0;
}

int main(int argc, char *argv[]){
    int op = atoi(argv[1]);
    int value = atoi(argv[2]); //信号量的值初始化为value

    key_t key = ftok("./system_v_msg",1); // 产生键值
    if(key == -1){
        perror("ftok");
        return -1;
    }

    int id = semget(key,5,0644|IPC_CREAT); //创建信号量
    cout<<id<<endl;
    
    if(value>=0){
        sem_init_value(id,0,value);
    }
    sem_get_value(id,0);


    if(op == -1){
        // 申请资源 P操作
        sem_p(id);
    }else if(op == 0){
        // wait
        sem_w(id);
    }else{
        // 释放资源 V操作
        sem_v(id);
    }

    sem_get_value(id,0);
    sleep(2);
    sem_del_sem();
    return 0;
}

POSIX信号量：
* POSIX 信号量是与 System V 信号量相似，但使用更简单的 API。编译的时候需要链接 -pthread
* POSIX 信号量也是计数器，但它使用两个基本操作来操作信号量：初始化和执行 wait 和 post 操作。wait 操作类似于 P 操作，用于申请资源并等待，而 post 操作类似于 V 操作，用于释放资源。

POSIX信号量有两种类型：命名信号量和无名信号量，分别用于进程和线程的通信。命名信号量是用tmp文件来实现，无名信号量用全局变量实现。
POSIX信号量由tmpfs文件系统和mmap内存映射共同实现，sem_open函数会创建一个tmpfs文件，然后通过mmap函数将文件进行内存映射，mmap函数调用成功后，将虚拟地址以sem_t*的形式返回给应用层。

为什么无名信号量只能用于线程中? 使用无名信号量一般会定义一个全局变量，别的进程是无法访问的，只有线程能用。

sem_open：创建或打开一个命名的信号量的系统调用。命名信号量（用于进程间通信）。
sem_init：初始化一个无名信号量。（用于线程间通信）。
sem_close：关闭一个打开的信号量。
sem_unlink：删除一个已命名的信号量。

命名信号量

#include <iostream>
#include <semaphore.h>
#include <unistd.h>
#include<fcntl.h>
#include<sys/stat.h>

using std::cout;
using std::cerr;
using std::endl;

int main(int argc, char* argv[]) {
    int mode = atoi(argv[1]);

    sem_t *sem= sem_open("/posixSem",O_RDWR|O_CREAT, 0664,0);
    if(sem == SEM_FAILED){
        cerr << "SEM_FAILED"<<endl;
        return -1;
    }

    while(1){
        if(mode==0){
            sem_wait(sem); // P
            cout<<"sem--"<<endl;
        }else{
            sleep(1);
            sem_post(sem); // V
            cout<<"sem++"<<endl;
        }
    }

    // 关闭信号量
    sem_close(sem);
    sem_unlink("/posixSem");

    return 0;
}

无名信号量

#include <iostream>
#include <semaphore.h>

using std::cout;
using std::cerr;
using std::endl;

int main() {
    sem_t sem;

    // 初始化信号量，第二个参数为0表示信号量在进程内共享，否则在进程间共享
    if (sem_init(&sem, 0, 0) == -1) {
        std::cerr << "Failed to initialize semaphore\n";
        return 1;
    }

    // 以一定的方式使用信号量，如：
    // sem_wait(&sem)  // 等待信号量
    // sem_post(&sem)  // 发送信号量

    // 销毁信号量
    if (sem_destroy(&sem) == -1) {
        std::cerr << "Failed to destroy semaphore\n";
        return 1;
    }

    return 0;
}

共享内存

SystemV 内存映射：内存映射是一种将文件系统中的文件映射到进程内存空间的技术。可以像操作内存一样操作文件，而不需要读写操作，可以提高文件的读写效率，减少IO操作，提高系统性能。

#include <sys/mman.h>
// 创建内存映射区
void *mmap(void *addr, size_t length, int prot, int flags, int fd, off_t offset);
// 解除内存映射
int munmap(void *addr, size_t length);

共享内存不同于内存映射区，它不属于任何进程，并且不受进程生命周期的影响。

共享内存不同于内存映射区，它不属于任何进程，并且不受进程生命周期的影响。
通过调用Linux提供的系统函数就可得到这块共享内存。使用之前需要让进程和共享内存进行关联，得到共享内存的起始地址之后就可以直接进行读写操作了，进程也可以和这块共享内存解除关联, 解除关联之后就不能操作这块共享内存了。
在所有进程间通信的方式中共享内存的效率是最高的。

shmget：创建或获取共享内存
shmat：与共享内存做内存映射的函数
shmdt：将共享内存与当前进程分离。
shmctl：对共享内存段进行操作，如创建、删除获取和修改属性

ipcs -m 查看共享内存
ipcmk - M 4096 共享内存为4096字节
ipcrm -m 2 删除标识为2的共享内存。

内存映射区和共享内存的区别：

共享内存只需要一块共享内存区就可以进行进程间通信，内存映射区位于每个进程的虚拟地址空间中，需要关联磁盘文件才能实现进程间数据通信。
共享内存效率更高，内存映射区需要文件数据同步，效率较低。
共享内存独立于进程，内存映射区属于进程。
突发情况下，内存映射区可以数据同步到文件中，共享内存不可以，因为共享内存在内存中。

SystemV 共享内存

#include <iostream>
#include <sys/ipc.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/shm.h>
#include <unistd.h>
#include <string.h>

using std::cout;
using std::cerr;
using std::endl;

int main(int argc, char* argv[]) {

    int mode = atoi(argv[1]);
    key_t shmkey = ftok("./system_v_shm",1); // 产生键值，这个名字一定要和程序名字一样。。
    if(shmkey == -1){
        perror("ftok");
        return -1;
    }

    // 键值 共享内存大小 权限
    int shmid = shmget(shmkey,1024,0644|IPC_CREAT);
    void* addr = shmat(shmid,NULL,0);
    if(addr == (void*)-1){
        cerr<<"shmat"<<endl;
        return -1;
    }

    int i=0;
    while(1){
        if(mode==0){
            char rbuf[1024] = {0};
            memcpy(rbuf,addr,1024);
            cout<<rbuf<<endl;
        }else{
            char wbuf[1024] = {0};
            sprintf(wbuf,"%d",i++);
            memset(addr,0,1024);//清空
            memcpy(addr,wbuf,strlen(wbuf));
            sleep(1);
        }
    }

    shmctl(shmid,IPC_RMID,NULL); //删除共享内存
    return 0;
}

POSIX共享内存：也是通过mmap函数实现，编译的时候需要加一个rt的库，-lrt。

shm_open函数创建一个共享内存对象。
shm_unlink关闭共享内存。
ftruncate函数设置共享内存大小。
mmap函数将共享内存映射到进程的地址空间中

#include <iostream>
#include <sys/ipc.h>
#include <sys/types.h>
#include <unistd.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/stat.h>        /* For mode constants */
#include <fcntl.h>           /* For O_* constants */

using std::cout;
using std::cerr;
using std::endl;

#define SHM_PATH "./system_v_shm"

int main(int argc, char* argv[]) {

    int mode = atoi(argv[1]);

    int fd = shm_open(SHM_PATH,O_RDWR|O_CREAT,0644); //打开或创建共享内存
    if(fd==-1){
        perror("shm_open");
        return -1;
    }

    int ret = ftruncate(fd,4096); //设置共享内存大小
    if(ret==-1){
        perror("ftruncate");
        return -1;
    }

    void* addr = mmap(NULL,4096,PROT_READ|PROT_WRITE,MAP_SHARED,fd,0); // 内存映射
    if(addr == (void*)-1){
        cerr<<"mmap"<<endl;
        return -1;
    }

    int i=0;
    while(1){
        if(mode==0){
            //读内存
            char rbuf[1024] = {0};
            memcpy(rbuf,addr,1024);
            cout<<rbuf<<endl;
        }else{
            char wbuf[1024] = {0};
            sprintf(wbuf,"%d",i++);
            memset(addr,0,1024);//清空
            memcpy(addr,wbuf,strlen(wbuf));
        }
        sleep(1);
    }

    munmap(addr,4096); //解除内存映射
    shm_unlink(SHM_PATH);// 删除内存映射
    return 0;
}

信号

Linux内核中实现信号的关键是信号处理函数和信号传递，每个进程都有一个信号来表示该进程对不同信号的处理情况。
当一个进程向另一个进程发送信号时，内核会将信号添加到目标进程的信号队列中。
在这里插入图片描述

信号产生：可以由多种事件触发，如硬件中断，软件异常，用户自定义信号灯。产生信号以后就会进行信号传递，处理，终止。
Linux每个进程都会维护一张信号表，信号表记录了每个信号和信号处理方法，用户调用signal或sigaction函数可以修改函数处理方法。

signal函数：捕捉产生的信号，并将信号的处理函数设置为handler
sigaction：捕捉产生的信号，注册和处理信号处理器
SIGCHLD 信号：子进程退出，暂停或暂停恢复运行的时候，会给父进程一个SIGCHLD信号。

Linux信号的三种状态：产生，未决，递达。

kill：杀死进程的信号
raise：给当前进程发送指定的信号
abort：给当前进程发送一个固定信号 (SIGABRT)，杀死当前进程。
alarm：定时器，单词定时信号，完成时发射一个信号。
settimer：周期性触发信号。

信号集函数：
sigprocmask：将信号加到阻塞信号集 or 解除 or 覆盖
sigpending:读取未决信号集。

参考列表
https://www.bilibili.com/video/BV1ob4y1u7ZL/
https://subingwen.cn/linux/pipe/
https://zhuanlan.zhihu.com/p/672264623

Linux

#C++ #Linux

Linux学习六(进程间通信)

https://cauccliu.github.io/2024/03/26/Linux学习六(进程间通信)/

Author

Liuchang

Posted on

March 26, 2024

Licensed under

Linux学习七(线程) Previous

Linux学习五(进程) Next