c - 多进程多线程的同步问题

Question

我有一个代码。有2个过程。 Parent 是文件 a.txt 的编写者。子级是 a.txt 的阅读器。父级有 2 个线程，子级有 2 个线程。父级的第一个线程打开文件parent1.txt。读取 128 个字符。写入a.txt.Parent的第二个线程打开文件parent2.txt.读取128个字符。写入 a.txt。 Child 的第一个线程从 a.txt 读取 128 个字符并写入 child1.txt。 child 的第二个线程从 a.txt 和 child2.txt 读取 128 个字符。任何父线程在写入后都应该生成事件并调用子读取器线程。我已经使用互斥锁和条件变量实现了一个解决方案。父级写入器线程在写入 a.txt 后生成 pthread_cond_signal。 1>但是 child 的阅读器线程在那之后没有运行。两个父线程都在loop.2>Parent读取frmparent1.txt中运行。读取成功。但是当它写入a.txt时，却没有成功。 a.txt 文件始终为空。我认为互斥锁不能在多个进程之间使用。这可能是 1 个问题

My code is as follows


  #include <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <sys/mman.h>
#include <sys/stat.h>        /* For mode constants */
#include <fcntl.h>           /* For O_* constants */
#include <sys/types.h>
#include <sys/wait.h>
using namespace std;

FILE*fd,*fdRead1,*fdRead2,*fdWrite1,*fdWrite2;

pthread_mutex_t *mut1;
pthread_mutexattr_t attrmutex;

pthread_cond_t  *cond_var;
pthread_condattr_t attrcond;

#define OKTOWRITE "/condwrite"
#define MESSAGE "/msg"
#define MUTEX "/mutex_lock"

void* R1(void *)
{
    char buf[128];
    int size;
    fdWrite1 = fopen("child1.txt","w+");
    cout<<"R1Thread"<<endl; 
    for(int i=0;i<10;i++)
    {
    cout<<"R1Thread-1"<<endl; 
        pthread_mutex_lock(mut1);
        pthread_cond_wait(cond_var,mut1);
    cout<<"R1Thread-2"<<endl; 
        size = fread(buf,128,1,fd);
        fwrite(buf,size,1,fdWrite1);
        pthread_mutex_unlock(mut1);
    }
    fclose(fdWrite1);
}

void* R2(void *)
{
    char buf[128];
    int size;
    fdWrite2 = fopen("child2.txt","w+");
    cout<<"R2Thread"<<endl; 
    for(int i=0;i<10;i++)
    {
    cout<<"R2Thread-1"<<endl; 
        pthread_mutex_lock(mut1);
        pthread_cond_wait(cond_var,mut1);
    cout<<"R2Thread-2"<<endl; 
        size = fread(buf,128,1,fd);
        fwrite(buf,size,1,fdWrite2);
        pthread_mutex_unlock(mut1);

    }
    fclose(fdWrite2);

}


void* W1(void *)
{
    char buf[128];
    int size;
    fdRead1 = fopen("parent1.txt","r");
    for(int i=0;i<10;i++)
    {
        pthread_mutex_lock(mut1);
        size = fread(buf,128,1,fdRead1);
        fwrite(buf,size,1,fd);
        pthread_cond_signal(cond_var);
    cout<<"W2Thread-1"<<endl; 
        pthread_mutex_unlock(mut1);
        sleep(10);
    }
    fclose(fdRead1);

}

void* W2(void *)
{
    char buf[128];
    int size;
    fdRead2 = fopen("parent2.txt","r");
    for(int i=0;i<10;i++)
    {
        pthread_mutex_lock(mut1);
        size = fread(buf,128,1,fdRead2);
        fwrite(buf,size,1,fd);
        pthread_cond_signal(cond_var);
    cout<<"W2Thread-1"<<endl; 
        pthread_mutex_unlock(mut1);
        sleep(1000);

    }
    fclose(fdRead2);

}


int main()
{
int des_cond, des_msg, des_mutex;
int mode = S_IRWXU | S_IRWXG;

des_mutex = shm_open(MUTEX, O_CREAT | O_RDWR | O_TRUNC, mode);

if (des_mutex < 0) {
    perror("failure on shm_open on des_mutex");
    exit(1);
}

if (ftruncate(des_mutex, sizeof(pthread_mutex_t)) == -1) {
    perror("Error on ftruncate to sizeof pthread_cond_t\n");
    exit(-1);
}

mut1 = (pthread_mutex_t*) mmap(NULL, sizeof(pthread_mutex_t),
        PROT_READ | PROT_WRITE, MAP_SHARED, des_mutex, 0);

if (mut1 == MAP_FAILED ) {
    perror("Error on mmap on mutex\n");
    exit(1);
}

des_cond = shm_open(OKTOWRITE, O_CREAT | O_RDWR | O_TRUNC, mode);

if (des_cond < 0) {
    perror("failure on shm_open on des_cond");
    exit(1);
}

if (ftruncate(des_cond, sizeof(pthread_cond_t)) == -1) {
    perror("Error on ftruncate to sizeof pthread_cond_t\n");
    exit(-1);
}

cond_var = (pthread_cond_t*) mmap(NULL, sizeof(pthread_cond_t),
        PROT_READ | PROT_WRITE, MAP_SHARED, des_cond, 0);

if (cond_var == MAP_FAILED ) {
    perror("Error on mmap on condition\n");
    exit(1);
}

/* Initialise attribute to mutex. */
pthread_mutexattr_init(&attrmutex);
pthread_mutexattr_setpshared(&attrmutex, PTHREAD_PROCESS_SHARED);

/* Allocate memory to pmutex here. */

/* Initialise mutex. */
pthread_mutex_init(mut1, &attrmutex);
/* Initialise attribute to condition. */
pthread_condattr_init(&attrcond);
pthread_condattr_setpshared(&attrcond, PTHREAD_PROCESS_SHARED);

/* Allocate memory to pcond here. */

/* Initialise condition. */
pthread_cond_init(cond_var, &attrcond);



    pthread_t thR1,thR2,thW1,thW2;
    fd = fopen("a.txt","w+");
    int res = fork();
    if(res<0) perror("error forking\n");
    if(res==0)//child
    {
        cout<<"child created"<<endl;
        pthread_create(&thR1,0,R1,0);
        //pthread_create(&thR2,0,R2,0);
        pthread_join(thR1,0);
        //pthread_join(thR2,0);
        fclose(fd);
    }
    else//parent
    {
        //fdRead = fopen("parent.txt","r");
        pthread_create(&thW1,0,W1,0);
        //pthread_create(&thW2,0,W2,0);
        pthread_join(thW1,0);
        //pthread_join(thW2,0);
        fclose(fd);
        wait(0);
    } 
}

输出如下-

child created
W2Thread-1
R1Thread
R1Thread-1
W2Thread-1
W2Thread-1
W2Thread-1
W2Thread-1
W2Thread-1
W2Thread-1
W2Thread-1
W2Thread-1

child中的condition_wait永远不会出来。

Answer 1

使用多个进程(每个进程都有多个线程)存在潜在问题，但它们主要围绕 fork 时的程序状态。由于您的程序在创建任何其他线程之前进行 fork ，因此您可以对其当时的状态充满信心，特别是，您可以确信它的一个线程当时不在关键部分中执行。这很好。

但是，您缺少两个关键细节:

1. 尽管您将互斥锁设置为进程共享，但您最初提供的代码版本未能对条件变量执行相同的操作。

2. 将 pthread_* 同步对象设置为进程共享是必要的，但对于进程间使用来说还不够。为此，您需要将同步对象驻留在所有参与进程访问的共享内存中。只有这样所有进程才能访问同一个对象。