|
|
|
@ -1,3 +1,8 @@ |
|
|
|
|
|
|
|
|
|
|
|
// 用C语言实现“线程池+共享内存”的进程间通信demo（模拟驱动与上层应用交互）；
|
|
|
|
// 进程间通信是父进程和子进程之间通信，子进程通过fork创建
|
|
|
|
|
|
|
|
#include <stdio.h> |
|
|
|
#include <stdlib.h> |
|
|
|
#include <unistd.h> |
|
|
|
@ -18,6 +23,11 @@ |
|
|
|
#define SEM_KEY 0x5678 // 信号量键值
|
|
|
|
#define PIPE_TIMEOUT_SEC 5 // 管道超时时间（秒）
|
|
|
|
|
|
|
|
|
|
|
|
// 共享内存中是一个消息队列，这个队列是环形的
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// 任务状态枚举
|
|
|
|
typedef enum { |
|
|
|
TASK_INIT = 0, // 初始态
|
|
|
|
@ -47,10 +57,28 @@ typedef struct { |
|
|
|
int pool_running; // 线程池运行标记（1：运行，0：停止）
|
|
|
|
} ThreadPool; |
|
|
|
|
|
|
|
|
|
|
|
// - 创建包含3个信号量的信号量集
|
|
|
|
// - 互斥信号量（索引0）：初始值1表示互斥，用于保护共享内存的互斥访问
|
|
|
|
// - 空信号量（索引1）：初始值0，用于表示队列中的任务数量
|
|
|
|
// - 满信号量（索引2）：初始值MAX_TASK，用于表示队列中的空闲位置数量
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// 将信号量的值设置为不同的数值代表不同的用法
|
|
|
|
// - 0，同步信号量，出食物可用资源，所有申请P操作的进程都会阻塞，直到有进程执行C操作释放资源
|
|
|
|
// - 1，互斥信号量（二值信号量），初始一个资源，仅允许1个进程进入临界访问共享资源
|
|
|
|
// - N（N>1，如2，3，5）技术信号量，初始N个资源，允许最多N个进程同时进入临界区（多进程共享有限资源）
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// ====================== 信号量操作封装 ======================
|
|
|
|
// 信号量初始化（创建/获取信号量集）
|
|
|
|
int sem_init(int sem_key) { |
|
|
|
int sem_id = semget(sem_key, 3, IPC_CREAT | 0666); |
|
|
|
int sem_id = semget(sem_key, 3, IPC_CREAT | 0666); // 创建三个信号量，0表示互斥信号量，用来互斥共享内存；1表示空信号量，2表示满信号量，用来表示队列中的任务数量
|
|
|
|
// 信号量集是内核资源，生命周期独立于创建它的进程
|
|
|
|
// 所以在整个程序中或者其他程序中都可以通过键值访问这个信号量集
|
|
|
|
if (sem_id == -1) { |
|
|
|
perror("semget failed"); |
|
|
|
return -1; |
|
|
|
@ -84,6 +112,24 @@ int sem_init(int sem_key) { |
|
|
|
return sem_id; |
|
|
|
} |
|
|
|
|
|
|
|
// sembuf结构体变量的sem_op的值表示是P操作还是V操作
|
|
|
|
// - P操作（减1）：sem_op = -1
|
|
|
|
// - V操作（加1）：sem_op = 1
|
|
|
|
|
|
|
|
// struct sembuf {
|
|
|
|
// unsigned short sem_num; // 信号量集中的信号量编号（从0开始）
|
|
|
|
// short sem_op; // 操作类型：-1=P操作（减1），1=V操作（加1），0=等待信号量值为0
|
|
|
|
// short sem_flg; // 操作标志：SEM_UNDO/IPC_NOWAIT等
|
|
|
|
// };
|
|
|
|
|
|
|
|
|
|
|
|
// int semop(int semid, struct sembuf *sops, size_t nsops);
|
|
|
|
// 参数 类型 含义与关键说明
|
|
|
|
// semid int 信号量集的 ID（由 semget 函数返回，唯一标识一个信号量集）。
|
|
|
|
// sops struct sembuf* 指向 信号量操作结构体数组 的指针，每个元素定义一个对信号量的操作（如 P/V）。
|
|
|
|
// nsops size_t 操作数组 sops 的长度（即要执行的信号量操作数量）；最常用值为 1（单操作，如仅 P 或仅 V）。
|
|
|
|
|
|
|
|
|
|
|
|
// 信号量P操作（减1，阻塞）
|
|
|
|
void sem_p(int sem_id, int sem_idx) { |
|
|
|
struct sembuf sem_buf = {sem_idx, -1, SEM_UNDO}; |
|
|
|
@ -104,11 +150,39 @@ void sem_v(int sem_id, int sem_idx) { |
|
|
|
// 创建/挂载共享内存（父进程用）
|
|
|
|
SharedMem* shm_create() { |
|
|
|
// 1. 创建共享内存
|
|
|
|
|
|
|
|
// int shmget(key_t key, size_t size, int shmflg);
|
|
|
|
// 参数 类型 核心作用
|
|
|
|
// key key_t 共享内存段的「唯一键值」，用于标识系统中唯一的共享内存资源：
|
|
|
|
// ① 手动指定（如 IPC_PRIVATE，仅当前进程亲缘关系可用）；
|
|
|
|
// ② 通过 ftok() 生成（跨无亲缘进程通信）。
|
|
|
|
// size size_t 共享内存段的大小（字节）：
|
|
|
|
// ① 创建新段时：需指定非 0 值（会按系统「页大小」对齐，如 4096 字节）；
|
|
|
|
// ② 获取已存在段时：可设为 0（无需匹配大小）。
|
|
|
|
// shmflg int 标志位（权限 + 行为），组合使用（按位或 ` ）：<br>▷ 权限位：如 0666（同文件权限，r/w/x 对应6=rw）；<br>▷ 行为位：<br>- IPC_CREAT：无则创建，有则获取；<br>- IPC_EXCL：仅与 IPC_CREAT搭配，若段已存在则报错（避免重复创建）；<br>-IPC_EXCL IPC_CREAT`：保证创建「全新」共享内存段。
|
|
|
|
int shm_id = shmget(SHM_KEY, sizeof(SharedMem), IPC_CREAT | 0666); |
|
|
|
if (shm_id == -1) { |
|
|
|
perror("shmget failed"); |
|
|
|
return NULL; |
|
|
|
} |
|
|
|
|
|
|
|
// shmat函数作用是将shmget创建/获取的共享内存段挂载（关联）到当前进程的地址空间，是共享内存从“系统资源”变成进程可读写内存的关键步骤
|
|
|
|
// void *shmat(int shmid, const void *shmaddr, int shmflg);
|
|
|
|
// 执行结果 返回值含义
|
|
|
|
// 成功 返回 void* 类型的指针，指向该共享内存段挂载到当前进程地址空间的起始虚拟地址；
|
|
|
|
// 失败 返回 (void *) -1（注意：不是普通的 int 型 -1，是强制转换为 void* 类型的 -1），同时设置全局变量 errno 标识错误原因。
|
|
|
|
|
|
|
|
// 参数 类型 核心作用
|
|
|
|
// shmid int 由 shmget 返回的共享内存标识符，唯一标识要挂载的共享内存段。
|
|
|
|
// shmaddr const void* 指定共享内存挂载到进程地址空间的起始地址：
|
|
|
|
// ✅ 推荐设为 NULL：由系统自动分配合适的地址（最常用，无需关心具体地址）；
|
|
|
|
// ❌ 非 NULL：需结合 shmflg 的 SHM_RND 标志，否则地址必须严格对齐到系统页大小（不推荐手动指定，易出错）。
|
|
|
|
// shmflg int 挂载标志（核心是权限控制）：
|
|
|
|
// ▷ 0：默认值，以读写权限挂载（需 shmget 时权限匹配）；
|
|
|
|
// ▷ SHM_RDONLY：以只读权限挂载（仅能读，不能写）；
|
|
|
|
// ▷ SHM_RND：仅配合非 NULL 的 shmaddr 使用，将地址向下对齐到 SHMLBA（共享内存低边界地址）的整数倍。
|
|
|
|
|
|
|
|
|
|
|
|
// 2. 挂载共享内存到进程地址空间
|
|
|
|
SharedMem *shm = (SharedMem*)shmat(shm_id, NULL, 0); |
|
|
|
if (shm == (void*)-1) { |
|
|
|
@ -119,6 +193,24 @@ SharedMem* shm_create() { |
|
|
|
memset(shm, 0, sizeof(SharedMem)); |
|
|
|
shm->front = 0; |
|
|
|
shm->rear = 0; |
|
|
|
// 为什么此处要创建三个信号量
|
|
|
|
// 信号量类型 索引 初始值 作用
|
|
|
|
// 互斥信号量 0 1 保护共享内存的互斥访问
|
|
|
|
// 空信号量 1 0 表示队列中的任务数量
|
|
|
|
// 满信号量 2 MAX_TASK 表示队列中的空闲位置数量
|
|
|
|
|
|
|
|
// 互斥信号量只能保证同时只有一个进程/线程访问共享内存，而空信号量和满信号量可以确保共享内存不溢出和队列为空时不获取队列的任务
|
|
|
|
// ### 1. 生产者在队列满时继续生产
|
|
|
|
// 如果只有互斥信号量，当队列已满时，生产者仍然可以获取互斥锁并尝试向队列添加任务，这会导致：
|
|
|
|
|
|
|
|
// - 队列溢出（环形队列可能覆盖旧任务）
|
|
|
|
// - 数据丢失
|
|
|
|
// - 程序逻辑错误
|
|
|
|
// ### 2. 消费者在队列空时继续消费
|
|
|
|
// 如果只有互斥信号量，当队列为空时，消费者仍然可以获取互斥锁并尝试从队列获取任务，这会导致：
|
|
|
|
|
|
|
|
// - 读取无效任务数据
|
|
|
|
// - 程序崩溃或异常行为
|
|
|
|
shm->sem_id = sem_init(SEM_KEY); // 初始化信号量
|
|
|
|
if (shm->sem_id == -1) { |
|
|
|
shmdt(shm); |
|
|
|
@ -148,14 +240,11 @@ void shm_detach_only(SharedMem *shm) { |
|
|
|
shmdt(shm); // 仅分离，不删除
|
|
|
|
} |
|
|
|
|
|
|
|
// 修复后的 shm_destroy 函数
|
|
|
|
// 销毁共享内存（父进程用，删除内核资源）
|
|
|
|
void shm_destroy(SharedMem *shm) { |
|
|
|
if (shm == NULL) return; |
|
|
|
|
|
|
|
// 关键：先保存sem_id到临时变量（在shmdt之前）
|
|
|
|
int sem_id_temp = shm->sem_id; |
|
|
|
|
|
|
|
// 1. 分离共享内存（分离后shm指针失效，不可再访问）
|
|
|
|
// 1. 分离共享内存
|
|
|
|
shmdt(shm); |
|
|
|
|
|
|
|
// 2. 删除共享内存（检查资源是否存在）
|
|
|
|
@ -164,9 +253,9 @@ void shm_destroy(SharedMem *shm) { |
|
|
|
shmctl(shm_id, IPC_RMID, NULL); |
|
|
|
} |
|
|
|
|
|
|
|
// 3. 删除信号量（用临时变量判断，避免访问失效的shm指针）
|
|
|
|
if (sem_id_temp != -1) { |
|
|
|
semctl(sem_id_temp, 0, IPC_RMID); |
|
|
|
// 3. 删除信号量（检查资源是否存在）
|
|
|
|
if (shm->sem_id != -1) { |
|
|
|
semctl(shm->sem_id, 0, IPC_RMID); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
@ -249,7 +338,7 @@ void thread_pool_destroy(ThreadPool *pool) { |
|
|
|
for (int i = 0; i < THREAD_POOL_SIZE; i++) { |
|
|
|
sem_v(pool->shm->sem_id, 1); |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// 等待所有线程退出
|
|
|
|
for (int i = 0; i < THREAD_POOL_SIZE; i++) { |
|
|
|
pthread_join(pool->tid[i], NULL); |
|
|
|
@ -312,6 +401,8 @@ int pipe_read_with_timeout(int fd, char *buf, int len, int timeout_sec) { |
|
|
|
// ====================== 主函数（进程分支） ======================
|
|
|
|
int main() { |
|
|
|
// 创建管道（父子进程同步用）
|
|
|
|
// pipe_fd[0]读，子进程只读这个fd，pipe_fd[1]写，父进程只写这个fd
|
|
|
|
// 使用管道确保在共享内存创建完成后子进程再挂载共享内存
|
|
|
|
int pipe_fd[2]; |
|
|
|
if (pipe(pipe_fd) == -1) { |
|
|
|
perror("pipe create failed"); |
|
|
|
@ -325,7 +416,7 @@ int main() { |
|
|
|
return -1; |
|
|
|
} |
|
|
|
|
|
|
|
// 子进程：工作进程（线程池 + 消费任务）
|
|
|
|
// 子进程：工作进程（线程池 + 消费任务），子进程多线程处理父进程提交过来的任务
|
|
|
|
if (pid == 0) { |
|
|
|
// 关闭管道写端（子进程只读）
|
|
|
|
close(pipe_fd[1]); |
|
|
|
@ -367,7 +458,7 @@ int main() { |
|
|
|
return 0; |
|
|
|
} |
|
|
|
|
|
|
|
// 父进程：任务提交进程
|
|
|
|
// 父进程：任务提交进程，创建任务。父进程单线程提交任务
|
|
|
|
else { |
|
|
|
// 关闭管道读端（父进程只写）
|
|
|
|
close(pipe_fd[0]); |
gaorui
1 week ago