Unable to send signal using pthread_cond_signal to another process in C

Question

I am trying to implement producer-consumer problem using a two-process scenario.

Process1 — Producer, and Process2 — Consumer. The consumer process is waiting on a condition variable (pthread_cond_wait(cond)) and producer will send a signal to consumer via pthread_cond_signal(cond).

I have gone through these links shared mutex and condition variable across process pthread_mutexattr_setpshared, everywhere it was said it is possible to use mutex and condition variable across multiple processes,

pthread_mutexattr_setpshared(&mutexAttr, PTHREAD_PROCESS_SHARED);
pthread_condattr_setpshared(&condAttr, PTHREAD_PROCESS_SHARED);

In this link shared mutexes, as void's recommendation, it was said to check whether my system supports or not? I have checked and I got 200809 as return value of sysconf(_SC_THREAD_PROCESS_SHARED) which denotes my system supports PTHREAD_PROCESS_SHARED.

I am trying to send pthread_cond_signal from producer (process-1) to consumer (Process-2). Both producer and consumer are using the same mutex/condition variable initialized.

However, consumer is not receiving the signal. It seems either signal is not sent or it is lost.

Where I am making mistakes? I am using Ubuntu, gcc-4.6.3.

Here is my code.

Producer.c:

    #include <pthread.h>
    #include <stdlib.h>
    #include <stdio.h>
    #include <unistd.h>
    #include <string.h>
    #include <sys/types.h>
    #include <sys/ipc.h>
    #include <sys/shm.h>
    #include <sys/wait.h>
    #include <pthread.h>
    #include <sched.h>
    #include <syscall.h>
    #include <sys/stat.h>
    #include <errno.h>
    #include <sys/mman.h>
    #include <fcntl.h>
    #include <stdbool.h>

    pthread_cond_t* condition;
    pthread_mutex_t* mutex;

    #define OKTOWRITE "/oktowrite"
    #define MESSAGE "/message"
    #define MUTEX "/lock"


    struct shared_use_st 
    {
    bool conditionSatisfied;
    };

    struct shared_use_st *shared_stuff;

    void create_shared_memory()
    {
        int shmid;
        void *shared_memory=(void *)0;
        shmid =shmget( (key_t)1234, 4096, 0666 | IPC_CREAT );

        if (shmid == -1)
        {
            fprintf(stderr,"shmget failed\n");
            exit(EXIT_FAILURE);
        }

        shared_memory =shmat(shmid, (void *)0,0);

        if(shared_memory == (void *)-1)
        {
            fprintf(stderr,"shmat failed\n");
            exit(EXIT_FAILURE); 
        }

        shared_stuff = (struct shared_use_st *)shared_memory;
    }


    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);
        }

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

        if (mutex == 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);
        }

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

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

        /* set mutex shared between processes */
        pthread_mutexattr_t mutexAttr;
        pthread_mutexattr_init(&mutexAttr);
        pthread_mutexattr_setpshared(&mutexAttr, PTHREAD_PROCESS_SHARED);
        pthread_mutex_init(mutex, &mutexAttr);

        /* set condition shared between processes */
        pthread_condattr_t condAttr;
        pthread_condattr_init(&condAttr);
        pthread_condattr_setpshared(&condAttr, PTHREAD_PROCESS_SHARED);
        pthread_cond_init(condition, &condAttr);

        create_shared_memory();
        shared_stuff->conditionSatisfied=0;

        int count=0;
        while(count++<10)
        {
            pthread_mutex_lock(mutex);
            shared_stuff->conditionSatisfied=1;
            pthread_mutex_unlock(mutex);

            pthread_cond_signal(condition);
            printf("signal sent to consumer, %d\n",count);

            sleep(3);
        }

        pthread_condattr_destroy(&condAttr);
        pthread_mutexattr_destroy(&mutexAttr);
        pthread_mutex_destroy(mutex);
        pthread_cond_destroy(condition);

        shm_unlink(OKTOWRITE);
        shm_unlink(MESSAGE);
        shm_unlink(MUTEX);

        return 0;
    }

Consumer.c:

#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/wait.h>
#include <pthread.h>
#include <sched.h>
#include <syscall.h>
#include <sys/stat.h>
#include <errno.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <stdbool.h>

pthread_cond_t* condition;
pthread_mutex_t* mutex;

#define OKTOWRITE "/oktowrite"
#define MESSAGE "/message"
#define MUTEX "/lock"

struct shared_use_st 
{
bool conditionSatisfied;
};

struct shared_use_st *shared_stuff;

void create_shared_memory()
{
    int shmid;
    void *shared_memory=(void *)0;
    shmid =shmget( (key_t)1234, 4096, 0666 | IPC_CREAT );

    if (shmid == -1)
    {
        fprintf(stderr,"shmget failed\n");
        exit(EXIT_FAILURE);
    }

    shared_memory =shmat(shmid, (void *)0,0);

    if(shared_memory == (void *)-1)
    {
        fprintf(stderr,"shmat failed\n");
        exit(EXIT_FAILURE); 
    }

    shared_stuff = (struct shared_use_st *)shared_memory;
}

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);
    }

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

    if (mutex == 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);
    }

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

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

    /* set mutex shared between processes */
    pthread_mutexattr_t mutexAttr;
    pthread_mutexattr_init(&mutexAttr);
    pthread_mutexattr_setpshared(&mutexAttr, PTHREAD_PROCESS_SHARED);
    pthread_mutex_init(mutex, &mutexAttr);

    /* set condition shared between processes */
    pthread_condattr_t condAttr;
    pthread_condattr_init(&condAttr);
    pthread_condattr_setpshared(&condAttr, PTHREAD_PROCESS_SHARED);
    pthread_cond_init(condition, &condAttr);

    create_shared_memory();
    shared_stuff->conditionSatisfied=0;

    while(1)
    {
        printf("Receiver waits on for signal from hello1.c \n");

        pthread_mutex_lock(mutex);
        while(!shared_stuff->conditionSatisfied)
            pthread_cond_wait(condition, mutex);
        pthread_mutex_unlock(mutex);

        printf("Signal received, wake up!!!!!!!!\n");

        //reset
        pthread_mutex_lock(mutex);
        shared_stuff->conditionSatisfied=0;
        pthread_mutex_unlock(mutex);
    }

}

Answer 1

Mutex shared among processes should be initialised by only one of them

The error is that you are initialising the mutex and the condition on both processes. However as they are shared, you should initialise them only on the consumer.

I also suggest not to mix the old System V shmget/shmat functions with the new POSIX shm_open.

I defined a new shared integer shint and which points to a shared memory initialised with shm_open and attached via mmap. shint is used as a flag to wait on the pthread condition.

In my view the producer, which starts after the consumer, doesn't need to truncate the memory segment and can open the shared memory with only O_RDWR. The producer takes the lock first and set shint to 1.

On the other hand the consumer, which starts first, has to create the shared memory segments and resize them with ftruncate. It also set the shared flag shint to 0 and wait on it. Also, before calling shm_open it is better to unlink (via shm_unlik) all used shared memory segments, to clean eventual errors on previous calls (for example if the code crash before unlinking).

I moved the initialisation of attributes for the mutex and condition at the beginning, because it seems more clear and correct.

---

Producer.c

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

    /* set mutex shared between processes */
    pthread_mutexattr_t mutexAttr;
    pthread_mutexattr_init(&mutexAttr);
    pthread_mutexattr_setpshared(&mutexAttr, PTHREAD_PROCESS_SHARED);

    /* set condition shared between processes */
    pthread_condattr_t condAttr;
    pthread_condattr_init(&condAttr);
    pthread_condattr_setpshared(&condAttr, PTHREAD_PROCESS_SHARED);

    des_mutex = shm_open(MUTEX, O_RDWR, mode);

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

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

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

    des_cond = shm_open(OKTOWRITE, O_RDWR, mode);

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

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

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

    int fd = shm_open(MESSAGE, O_RDWR, 0644);
    if (fd < 0) {
        perror("failure on shm_open on fd");
        exit(1);
    }

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

    int *shint;
    shint = (int *) mmap(NULL, sizeof(int), PROT_WRITE|PROT_READ, MAP_SHARED, fd, 0);

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


    // set ot 0
    *shint = 0;


    int count=0;
    while(count++<10)
    {
        pthread_mutex_lock(mutex);
        *shint = 1;
        pthread_mutex_unlock(mutex);

        pthread_cond_signal(condition);
        printf("signal sent to consumer, %d\n",count);

        sleep(3);
    }

    pthread_condattr_destroy(&condAttr);
    pthread_mutexattr_destroy(&mutexAttr);
    pthread_mutex_destroy(mutex);
    pthread_cond_destroy(condition);

    shm_unlink(OKTOWRITE);
    shm_unlink(MESSAGE);
    shm_unlink(MUTEX);

    return 0;
}

Consumer.c

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

    // Unlink first to clean  
    shm_unlink(MUTEX);
    shm_unlink(OKTOWRITE);
    shm_unlink(MESSAGE);

    pthread_mutexattr_t mutexAttr;
    pthread_mutexattr_init(&mutexAttr);
    pthread_mutexattr_setpshared(&mutexAttr, PTHREAD_PROCESS_SHARED);

    pthread_condattr_t condAttr;
    pthread_condattr_init(&condAttr);
    pthread_condattr_setpshared(&condAttr, PTHREAD_PROCESS_SHARED);


    des_mutex = shm_open(MUTEX, O_CREAT | O_RDWR, 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);
    }

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

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

    des_cond = shm_open(OKTOWRITE, O_CREAT | O_RDWR, 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);
    }

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

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

    int fd = shm_open(MESSAGE, O_CREAT|O_RDWR, 0644);
    if (fd < 0) {
        perror("failure on shm_open on fd");
        exit(1);
    }

    if(ftruncate(fd, 16) == -1) {
        perror("Error on ftruncate to sizeof ftruncate fd\n");
        exit(-1);
    }
    int *shint;
    shint = (int *) mmap(NULL, sizeof(int), PROT_WRITE|PROT_READ, MAP_SHARED, fd, 0);

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


    *shint = 0;

        if (pthread_mutex_init(mutex, &mutexAttr) != 0)    {printf("Error initi mutex"); exit(111);}
        if (pthread_cond_init(condition, &condAttr) != 0) {printf("Error initi cond");  exit(111);}

    while(1)
    {
        printf("Receiver waits on for signal from hello1.c \n");

        pthread_mutex_lock(mutex);
        while(shint == 0)
            pthread_cond_wait(condition, mutex);
        printf("Waiting"); sleep(1);
        pthread_mutex_unlock(mutex);

        printf("Signal received, wake up!!!!!!!!\n");

        break;
        //reset
        pthread_mutex_lock(mutex);
        shared_stuff->conditionSatisfied=0;
        pthread_mutex_unlock(mutex);
    }

    return 0;
}