多进程多线程场景下Mutex同步字符串传输的问题排查

你的核心问题是普通pthread互斥量仅在进程内部线程有效，父进程创建的互斥量无法被子进程的线程识别，导致跨进程的线程同步失效。要解决这个问题，必须使用共享内存存储同步原语（互斥量、条件变量），并为这些同步原语设置跨进程共享属性。

关键修改步骤 #

1. 创建跨进程共享内存区域：用来存放共享缓冲区、互斥量、条件变量以及缓冲区状态（写入/读取索引、元素数量）。
2. 初始化跨进程属性的同步原语：给互斥量和条件变量设置PTHREAD_PROCESS_SHARED属性，确保父子进程的线程都能访问。
3. 调整线程逻辑：父线程写入缓冲区后通过条件变量通知子线程，子线程等待条件变量触发后读取数据写入文件。
4. SIGINT信号处理：读取文件内容时确保资源正确清理，避免泄漏。

修改后的完整代码 #

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <signal.h>
#include <string.h>
#include <sys/wait.h>

#define BUFFER_SIZE 10
#define MAX_STR_LEN 256
#define SHM_NAME "/shared_buf_sync"

// 共享内存结构体：包含缓冲区、同步原语和状态
typedef struct {
    char buffer[BUFFER_SIZE][MAX_STR_LEN];
    int in;          // 写入位置索引
    int out;         // 读取位置索引
    int count;       // 缓冲区元素数
    pthread_mutex_t mutex;
    pthread_cond_t cond;
} SharedData;

SharedData *shared_data;
int shm_fd;
FILE *out_file;

// SIGINT信号处理：输出文件内容并清理资源
void handle_sigint(int sig) {
    rewind(out_file);
    char line[MAX_STR_LEN];
    printf("\n=== 文件内容 ===\n");
    while (fgets(line, MAX_STR_LEN, out_file)) {
        printf("%s", line);
    }

    // 清理同步原语与共享内存
    pthread_mutex_destroy(&shared_data->mutex);
    pthread_cond_destroy(&shared_data->cond);
    munmap(shared_data, sizeof(SharedData));
    shm_unlink(SHM_NAME);
    fclose(out_file);
    exit(EXIT_SUCCESS);
}

// 父进程线程：读取标准输入写入共享缓冲区
void *parent_thread(void *arg) {
    char input[MAX_STR_LEN];
    while (1) {
        if (!fgets(input, MAX_STR_LEN, stdin)) break;
        // 移除换行符
        input[strcspn(input, "\n")] = '\0';

        pthread_mutex_lock(&shared_data->mutex);
        // 缓冲区满则等待
        while (shared_data->count >= BUFFER_SIZE) {
            pthread_cond_wait(&shared_data->cond, &shared_data->mutex);
        }
        // 写入缓冲区
        strcpy(shared_data->buffer[shared_data->in], input);
        shared_data->in = (shared_data->in + 1) % BUFFER_SIZE;
        shared_data->count++;
        // 通知子线程有数据可读
        pthread_cond_signal(&shared_data->cond);
        pthread_mutex_unlock(&shared_data->mutex);
    }
    return NULL;
}

// 子进程线程：读取共享缓冲区写入文件
void *child_thread(void *arg) {
    char content[MAX_STR_LEN];
    while (1) {
        pthread_mutex_lock(&shared_data->mutex);
        // 缓冲区空则等待
        while (shared_data->count == 0) {
            pthread_cond_wait(&shared_data->cond, &shared_data->mutex);
        }
        // 读取缓冲区
        strcpy(content, shared_data->buffer[shared_data->out]);
        shared_data->out = (shared_data->out + 1) % BUFFER_SIZE;
        shared_data->count--;
        // 通知父线程缓冲区有空位
        pthread_cond_signal(&shared_data->cond);
        pthread_mutex_unlock(&shared_data->mutex);

        // 写入文件并刷新缓冲区
        fprintf(out_file, "%s\n", content);
        fflush(out_file);
    }
    return NULL;
}

int main(int argc, char *argv[]) {
    if (argc != 3) {
        fprintf(stderr, "用法: %s <线程数量N> <输出文件名>\n", argv[0]);
        exit(EXIT_FAILURE);
    }
    int thread_num = atoi(argv[1]);
    char *filename = argv[2];

    // 注册SIGINT信号处理
    signal(SIGINT, handle_sigint);

    // 创建并映射共享内存
    shm_fd = shm_open(SHM_NAME, O_CREAT | O_RDWR, 0666);
    if (shm_fd == -1) { perror("shm_open"); exit(EXIT_FAILURE); }
    if (ftruncate(shm_fd, sizeof(SharedData)) == -1) { perror("ftruncate"); exit(EXIT_FAILURE); }
    shared_data = mmap(NULL, sizeof(SharedData), PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
    if (shared_data == MAP_FAILED) { perror("mmap"); exit(EXIT_FAILURE); }

    // 初始化共享数据状态
    shared_data->in = 0;
    shared_data->out = 0;
    shared_data->count = 0;

    // 初始化跨进程共享的互斥量
    pthread_mutexattr_t mutex_attr;
    pthread_mutexattr_init(&mutex_attr);
    pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED);
    pthread_mutex_init(&shared_data->mutex, &mutex_attr);
    pthread_mutexattr_destroy(&mutex_attr);

    // 初始化跨进程共享的条件变量
    pthread_condattr_t cond_attr;
    pthread_condattr_init(&cond_attr);
    pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED);
    pthread_cond_init(&shared_data->cond, &cond_attr);
    pthread_condattr_destroy(&cond_attr);

    // 打开输出文件
    out_file = fopen(filename, "w");
    if (!out_file) { perror("fopen"); exit(EXIT_FAILURE); }

    pid_t pid = fork();
    if (pid == -1) { perror("fork"); exit(EXIT_FAILURE); }

    if (pid == 0) {
        // 子进程：创建N个读取线程
        pthread_t threads[thread_num];
        for (int i = 0; i < thread_num; i++) {
            if (pthread_create(&threads[i], NULL, child_thread, NULL) != 0) {
                perror("pthread_create child");
                exit(EXIT_FAILURE);
            }
        }
        // 等待子线程结束
        for (int i = 0; i < thread_num; i++) {
            pthread_join(threads[i], NULL);
        }
    } else {
        // 父进程：创建N个写入线程
        pthread_t threads[thread_num];
        for (int i = 0; i < thread_num; i++) {
            if (pthread_create(&threads[i], NULL, parent_thread, NULL) != 0) {
                perror("pthread_create parent");
                exit(EXIT_FAILURE);
            }
        }
        // 等待父线程结束
        for (int i = 0; i < thread_num; i++) {
            pthread_join(threads[i], NULL);
        }
        // 等待子进程退出
        wait(NULL);
    }

    // 正常退出时清理资源
    pthread_mutex_destroy(&shared_data->mutex);
    pthread_cond_destroy(&shared_data->cond);
    munmap(shared_data, sizeof(SharedData));
    shm_unlink(SHM_NAME);
    fclose(out_file);
    return EXIT_SUCCESS;
}

核心要点说明 #

• 共享内存：通过shm_open创建系统级共享内存对象，mmap将其映射到进程地址空间，确保父子进程访问同一块内存。
• 跨进程同步属性：PTHREAD_PROCESS_SHARED属性让互斥量和条件变量可以被不同进程的线程共享，这是解决问题的核心。
• 条件变量的正确使用：用while循环等待条件（而非if），避免虚假唤醒；每次写入/读取后发送信号通知对方线程。
• 文件即时写入：调用fflush确保写入的数据立即同步到磁盘，避免SIGINT触发时看不到最新输入的内容。
• 资源清理：信号处理函数和正常退出路径都要清理同步原语、共享内存和文件，防止资源泄漏。

编译时需要链接pthread库：gcc -o sync_threads sync_threads.c -lpthread -lrt

内容的提问来源于stack exchange，提问作者Vanack Sabbadium