🪴 Quartz 4.0

02_进程间通信

20 min read

定义

进程间通信必须通过内核。

Linux进程间通信机制分三类：数据交互，同步，信号。理解了这些机制才能灵活运用操作系统提供的IPC工具。

  • 管道(包括有名管道和无名管道)
  • 信号
  • System V IPC（消息队列，共享内存，信号量）
  • Socket（UNIX域套接字和网络套接字）

多进程内存共享问题

竞争条件（Race Condition）、数据同步、死锁（Deadlock）

  • 竞争条件（Race Condition）：当多个进程同时访问和修改共享内存时，由于执行顺序的不确定性，可能导致数据不一致或不正确的结果。
  • 数据同步问题：不同的进程可能以不同的速度访问共享内存，导致数据在读取和更新之间的时间差异，进而引发数据不一致的问题。
  • 死锁（Deadlock）：如果多个进程在访问共享内存时发生互相等待的情况，可能导致死锁，使得进程无法继续执行。

解决措施

互斥锁、信号量、条件变量、进程间通信（IPC)

  • 互斥锁（Mutex）：通过在访问共享内存之前获取互斥锁，并在访问完成后释放锁，可以确保同一时间只有一个进程访问共享内存，从而避免竞争条件。
  • 信号量（Semaphore）：通过使用信号量来同步进程的访问，可以控制同时访问共享内存的进程数量，从而避免数据同步问题和死锁。
  • 条件变量（Condition Variable）：条件变量可以用于进程间的通信和同步，它可以在特定条件满足时唤醒等待的进程，从而避免忙等待和减少资源消耗。
  • 进程间通信机制（IPC）：使用操作系统提供的进程间通信机制，如管道、消息队列、共享内存、套接字等，可以实现进程间的数据传输和同步，确保共享数据的正确性和一致性。

多进程、多线程同步(通讯)方法⭐⭐⭐⭐

同步/通信方式进程通信进程同步描述
互斥锁（Mutex）⭐️保护共享资源的互斥访问。
信号量（Semaphore）⭐️⭐️控制临界资源的访问数量。
条件变量（Condition Variable）⭐️进程等待和唤醒。
屏障（Barrier）⭐️⭐️同步多个进程或线程的执行。
读写锁（Read-Write Lock）⭐️允许多读单写的访问模式。
事件（Event）⭐️⭐️进程间的通知和同步。
管程⭐️组合互斥锁、条件变量和共享数据结构，确保线程的互斥和同步。
管道（Pipe）⭐️适用于有亲缘关系的进程，单向通信。
命名管道（Named Pipe）⭐️适用于没有亲缘关系的进程，通过特殊文件实现通信。
共享内存（Shared Memory）⭐️⭐️多个进程访问同一块物理内存实现数据共享。
消息队列（Message Queue）⭐️通过消息队列发送和接收消息，实现异步通信。
套接字（Socket）⭐️通用的进程间通信机制，可在不同主机间通信。
文件⭐️通过文件操作实现进程间通信和数据交换。

管道 ⭐⭐⭐

管道是一种用于两个进程间同一时刻进行单向通信的机制，也被称为半双工管道。

管道传输的数据是无格式的流且大小受限。 pipe_buf：内核定义管道的最大容量，通常为4KB~64KB。

管道优势

管道（pipe）在Unix和类Unix操作系统中是一种基本的进程间通信（IPC）机制

  • 简单：无需特殊权限，通过标准系统调用创建。
  • 高效：不需要额外的系统调用或上下文切换，提高数据传输效率。
  • 同步：写操作会阻塞直到读操作准备好读取数据，可以避免数据丢失。
  • 流式：可以处理任意长度的数据流，不需要预先定义数据大小。

实现原理

操作系统在内核中开辟一块缓冲区（管道），用于进程之间的通信。由于其特性，同一时刻只能有一个进程进行读或写操作，所以称为半双工。

管道破裂

一个进程向另一个已经被关闭的管道写数据时产生的错误，通常发生在写端没有检查读端是否有效。

解决方式：

  • 忽略SIGPIPE信号：设置信号处理函数来忽略SIGPIPE信号，这样即使管道破裂，进程也不会被终止。
  • 使用getsockopt()检查连接状态：调用getsockopt()函数可以主动检测套接字的状态，如果检测到与服务器的连接已经断开，可以采取相应的措施，如重新连接或存储数据以便稍后发送。
  • 优化代码逻辑：在写数据之前检查管道的读端是否关闭，如果关闭则避免写入数据，或者使用非阻塞的管道操作来避免进程被阻塞。
  • 增加管道缓冲区大小：通过ulimit命令查看和修改管道缓冲区的大小，以避免因缓冲区满而导致的管道破裂错误。

无名管道

无名管道的读端由描述符fd[0]表示，写端由描述符fd[1]表示。

  • 只能用于有关联的进程间数据交互，如父子进程，兄弟进程，子孙进程，在目录中看不到文件节点，读写文件描述符存在一个int型数组中。
  • 只能单向传输数据，即管道创建好后，一个进程只能进行读操作，另一个进程只能进行写操作，读出来字节顺序和写入的顺序一样。

使用步骤

  1. 调用pipe()函数创建管道，获取管道的读端和写端的文件描述符。
  2. 调用fork()创建子进程，此时子进程会继承父进程的管道描述符。
  3. 在父进程中关闭管道的不需要的端口，例如关闭管道的读端，保留管道的写端；
  4. 在子进程中关闭管道的另一个端口，即关闭管道的写端，保留管道的读端。
  5. 父进程通过保留的管道写端，使用write()函数向管道写入数据。
  6. 子进程通过保留的管道读端，使用read()函数从管道中读取数据。
  7. 父进程和子进程根据需求进行数据的交换和通信。
  8. 当通信结束后，父进程和子进程分别关闭其管道端口，即父进程关闭管道的写端，子进程关闭管道的读端。
  9. 父进程通过调用wait()等待子进程的结束，确保子进程正确退出。
#include <iostream>
#include <unistd.h>
#include <sys/wait.h>
#include <cstring>
 
int main() {
    int pipefd[2]; // 用于存储管道读端和写端的文件描述符
 
    // 创建管道
    if (pipe(pipefd) == -1) {
        perror("pipe");
        exit(EXIT_FAILURE);
    }
 
    pid_t pid = fork();
 
    if (pid == -1) {
        perror("fork");
        exit(EXIT_FAILURE);
    }
    else if (pid == 0) {
        // 子进程读取管道中的数据
        close(pipefd[1]); // 关闭管道写端
        
        char buffer[100];
        ssize_t bytesRead = read(pipefd[0], buffer, sizeof(buffer));
 
        if (bytesRead == -1) {
            perror("read");
            exit(EXIT_FAILURE);
        }
        else if (bytesRead == 0) {
            std::cout << "管道已关闭" << std::endl;
        }
        else {
            std::cout << "子进程读取到的数据：" << buffer << std::endl;
        }
 
        close(pipefd[0]); // 关闭管道读端
        exit(EXIT_SUCCESS);
    }
    else {
        // 父进程向管道写入数据
        close(pipefd[0]); // 关闭管道读端
 
        const char message[] = "Hello, World!";
        ssize_t bytesWritten = write(pipefd[1], message, strlen(message));
 
        if (bytesWritten == -1) {
            perror("write");
            exit(EXIT_FAILURE);
        }
 
        close(pipefd[1]); // 关闭管道写端
        wait(NULL); // 等待子进程退出
        exit(EXIT_SUCCESS);
    }
}

有名管道

命名管道允许无亲缘关系的进程间通信，通过文件系统中的特殊文件进行。

  • 可以使无关联的进程通过fifo文件描述符进行数据传递；
  • 单向传输有一个写入端和一个读出端，操作方式和无名管道相同。

使用步骤

  • 使用mkfifo()创建fifo文件描述符。
  • 打开管道文件描述符。
  • 通过读写文件描述符进行单向数据传输。
mkfifo fifo
ls
ls -al

信号 ⭐⭐⭐

常见信号及含义

信号含义
SIGHUP终端挂起信号
SIGINT中止前台进程
SIGFPE致命算术运算错误
SIGKILL立即结束程序
SIGALRM时钟定时信号
SIGTERM正常结束进程
SIGCHLD子进程结束信号
SIGCONT让暂停进程恢复执行
SIGSTOP暂停前台进程

信号发送

信号是Linux系统响应某些条件而产生的一个事件，接收到该信号的进程会执行相应的操作。

信号产生的三种方式

  • 由硬件产生，如从键盘输入Ctrl+C可以终止当前进程
  • 由其他进程发送，如可在shell进程下，使用命令 kill -信号标号 PID，向指定进程发送信号。
  • 异常，进程异常时会发送信号

信号接收

信号处理程序

一个用于处理接收到的信号的特殊函数。当进程接收到一个信号时，操作系统会将控制权转移到相应的信号处理程序，该程序执行与信号相关的操作。信号处理程序可以执行各种操作，如捕获和处理异常、记录日志、执行特定的操作等。

信号的作用

  • 异常处理：当进程遇到错误或异常情况时，操作系统可以向进程发送相应的信号，使得进程能够捕获并处理这些异常，保证程序的稳定性和可靠性。
  • 事件通知：操作系统可以向进程发送信号，通知其发生了某个特定的事件，如键盘输入、鼠标点击等。进程可以捕获这些信号并作出相应的响应。
  • 进程间通信：进程可以使用信号进行通信。一个进程可以向另一个进程发送信号，请求某种操作或提醒对方发生了某个事件。

System V IPC

  • System V Interprocess Communication

共享内存

共享内存允许多个进程直接访问同一块内存区域，就是拿出一块虚拟地址空间来，映射到相同的物理内存中，进程间的数据传输不再涉及内核。

  • 无需数据复制。
  • 需要开辟一块物理内存，映射到不同进程的内存空间。
  • 需要借助同步机制避免数据竞争。
  • 生命周期跟随内核。

共享内存如何实现数据同步

借助同步机制避免数据竞争。

消息队列

消息队列是保存在内核中的消息链表

  • 消息队列生命周期随内核
  • 消息队列不适合比较大数据的传输
  • 消息队列通信过程中，存在用户态与内核态之间的数据拷贝开销

信号量

信号量其实是一个整型的计数器，主要用于实现进程间的互斥与同步，而不是用于缓存进程间通信的数据。

  • 信号量的主要作用是保护临界资源，使得在一个时刻只有一定数量的进程或线程可以访问。
  • 如果其他进程已经对其进行上锁，那么当前进程会进入睡眠状态，等待其他人对信号量进行解锁。

原理

基于 P(sv) 和 V(sv) 两种原子操作，P操作减一，V操作加一

  • P 操作是用在进入共享资源之前，V 操作是用在离开共享资源之后，这两个操作是必须成对出现的。
  • 信号初始化为 0，就代表着是同步信号量，它可以保证进程 A 应在进程 B 之前执行。
  • P(sv) 操作会将信号量的值减1，如果信号量的值大于零，进程或线程可以继续访问共享资源；如果信号量的值为零，进程或线程会被挂起，直到其他进程或线程通过 V(sv) 操作释放信号量。
  • V(sv) 操作会将信号量的值加1，如果有进程或线程因等待信号量而被挂起，它们中的一个会被唤醒继续执行；如果没有进程或线程等待信号量，信号量的值会增加。

struct semaphore

struct semaphore{
	raw_spinlok_t lock;//自旋锁，用于多核CPU同步
	unsigned int count;//信号量的计数器，上锁时`count`减一，如果结果`count`大于0，则表示上锁成功。
	struct list_head wait_list;//正在等待信号量解锁的进程队列。
};

信号量 上锁通过 down() 函数实现

void down(struct semaphore *sem)
{
	unsigned long flags;
	spin_lock_irqsave(&sem->lock, flags);
	if (likely(sem->count > 0)){
		sem->count--;
	}else{
		__down(sem);
	}
	spin_unlock_irqrestore(&sem->lock, flags);
}
  1. down() 函数首先对信号量进行自旋锁操作（为了避免多核CPU竞争）。
  2. 然后比较计数器是否大于0，如果是对计数器进行减一操作，并且返回。
  3. 否则调用 __down() 函数进行下一步操作。
static noinline void __sched __down(struct semaphore *sem)
{
	__down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}
static inline int __down_common(struct semaphore *sem,long state, long timeout)
{
	struct task_struct *task = current;
	struct semaphore_waiter waiter;
	
	// 把当前进程添加到等待队列中
	list_add_tail(&waiter.list, &sem->wait_list);
	waiter.task = task;
	waiter.up = 0;
	for (;;) {
		...
		__set_task_state(task, state);
		spin_unlock_irq(&sem->lock);
		timeout = schedule_timeout(timeout);
		spin_lock_irq(&sem->lock);
		// 当前进程是否获得信号量锁?
		if (waiter.up){
			return 0;
		}
	}
	...
}

__down() 函数最终调用 __down_common() 函数

__down_common()的操作流程：

  1. 把当前进程添加到信号量的等待队列中。
  2. 切换到其他进程运行，直到被其他进程唤醒。
  3. 如果当前进程获得信号量锁（由解锁进程传递），那么函数返回。

解锁过程主要通过 up() 函数实现

void up(struct semaphore *sem)
{
	unsigned long flags;
	raw_spin_lock_irqsave(&sem->lock, flags);
	if (likely(list_empty(&sem->wait_list))) // 如果没有等待的进程, 直接对计数器加一操作
		sem->count++;
	else
		__up(sem); // 如果有等待进程, 那么调用 __up() 函数进行唤醒
	raw_spin_unlock_irqrestore(&sem->lock, flags);
}
static noinline void __sched __up(struct semaphore *sem)
{
	// 获取到等待队列的第一个进程
	struct semaphore_waiter *waiter = list_first_entry(
	&sem->wait_list, struct semaphore_waiter, list);
	list_del(&waiter->list); // 把进程从等待队列中删除
	waiter->up = 1; // 告诉进程已经获得信号量锁
	wake_up_process(waiter->task); // 唤醒进程
}

解锁流程：

  1. 判断当前信号量是否有等待的进程，如果没有等待的进程, 直接对计数器加以操作
  2. 如果有等待的进程，那么获取到等待队列的第一个进程。
  3. 把进程从等待队列中删除。
  4. 告诉进程已经获得信号量锁。
  5. 唤醒进程。

Socket

跨网络与不同主机上的进程之间通信，就需要 Socket 通信。

  • 实现 TCP 字节流通信： Socket 类型是 AFINET 和 SOCKSTREAM；
  • 实现UDP协议通信。
  • 本地进程间通信。

Socket 通信建立流程

服务端调用 accept 时，连接成功了会返回一个已完成连接的 socket，后续用来传输数据。

监听的 socket 和真正用来传送数据的 socket，是「两个」 socket，一个叫作监听 socket，一个叫作已完成连接 socket。

  1. 服务端和客户端初始化 socket，得到文件描述符；
  2. 服务端调用 bind，将绑定在 IP 地址和端口;
  3. 服务端调用 listen，进行监听；
  4. 服务端调用 accept，等待客户端连接；
  5. 客户端调用 connect，向服务器端的地址和端口发起连接请求；
  6. 服务端 accept 返回用于传输的 socket 的文件描述符；
  7. 客户端调用 write 写入数据；
  8. 服务端调用 read 读取数据；
  9. 客户端断开连接时，会调用 close，那么服务端 read 读取数据的时候，就会读取到了 EOF，待处理完数据后，服务端调用 close，表示连接关闭。

Graph View

Backlinks