1

I have two codes : PRODUCER (PR) and CONSUMER (CO). There is a block of memory (Mat) (a 3D matrix to be precise) which needs to be shared between both the programs. I am currently using shared memory based IPC functions to share the memory space between the two codes.

Constraints:

  1. PR is the owner of the Mat and performs iterations changing the values of the matrix. CO is the user of the Mat and only reads the values and uses for some further calculation
  2. PR should write the data first and then should wait for the CO to read and use the values of the Matrix then signalling PR to proceed with further iteration and it should continue like this.

What I am currently using is -

PRODUCER

#include <unistd.h>
#include  <stdio.h>
#include  <stdlib.h>
#include  <sys/types.h>
#include  <sys/ipc.h>
#include  <sys/shm.h>

#define  NOT_READY  -1
#define  FILLED     0
#define  TAKEN      1

#define  nx (400)
#define  ny (400)
#define  nz (400)

struct Memory {
     int      status;
     double   u_x[nx+1][ny+2][nz+2];
}

int 
main(int  argc, char *argv[])
{
     key_t          ShmKEY;
     int            ShmID;
     struct Memory  *ShmPTR;
     int            i, j, k;
     int            niter = 5;

     int            sumX[niter],sumY[niter],sumZ[niter];

     ShmKEY = ftok(".", 'x'); // getting the unique identifier key from directory location
     ShmID = shmget(ShmKEY, sizeof(struct Memory), IPC_CREAT | 0666);
     if (ShmID < 0) {
          printf("*** shmget error (server) ***\n");
          exit(1);
     }

     ShmPTR = (struct Memory *) shmat(ShmID, NULL, 0);
     if ((int) ShmPTR == -1) {
          printf("*** shmat error (server) ***\n");
          exit(1);
     }
     printf("Server attached the memory to its virtual space...\n");

     ShmPTR->status  = NOT_READY; // setting the status to be not ready before filling it

     for (int m = 0; m < niter; m++){
        for (i=0; i<=nx; i++) for (j=0; j<=ny+1; j++) for (k=0; k<=nz+1; k++) 
           ShmPTR->u_x[i][j][k] = m; // filling the array with iteration number (just for depiction purpose)
        ShmPTR->status = FILLED; // change the status to Filled
        //printf("Please start the client in another window...\n");
        while (ShmPTR->status != TAKEN)
             sleep(1);
     }

     printf("Server has detected the completion of its child...\n");
     shmdt((void *) ShmPTR);
     printf("Server has detached its shared memory...\n");
     shmctl(ShmID, IPC_RMID, NULL);
     printf("Server has removed its shared memory...\n");
     printf("Server exits...\n");
     exit(0);
}

CONSUMER

#include  <stdio.h>
#include  <stdlib.h>
#include  <sys/types.h>
#include  <sys/ipc.h>
#include  <sys/shm.h>

#define  NOT_READY  -1
#define  FILLED     0
#define  TAKEN      1

#define  nx (400)
#define  ny (400)
#define  nz (400)

struct Memory {
    int      status;
    double   u_x[nx+1][ny+2][nz+2];
}

int
main(void)
{
    key_t          ShmKEY;
    int            ShmID;
    struct Memory  *ShmPTR;
    int            i, j, k;

    int            niter = 5;
    int            sumX[niter];

    ShmKEY = ftok(".", 'x');
    ShmID = shmget(ShmKEY, sizeof(struct Memory), 0666);
    if (ShmID < 0) {
          printf("*** shmget error (client) ***\n");
        exit(1);
    }
    printf("Client has received a shared memory...\n");

    ShmPTR = (struct Memory *) shmat(ShmID, NULL, 0);
    if ((int) ShmPTR == -1) {
        printf("*** shmat error (client) ***\n");
        exit(1);
    }
    printf("Client has attached the shared memory to it's virtual memory space...\n");

    for (int m =0; m<niter; m++){
        sumX[m] = 0;
        while (ShmPTR->status != FILLED)
            ;
        printf("Client found the data is ready, performing sanity check...\n");
        // read the integers and check for the sum

        for (i=0; i<=nx; i++) for (j=0; j<=ny+1; j++) for (k=0; k<=nz+1; k++) 
            sumX[m] +=  ShmPTR->u_x[i][j][k];
        printf("Cycle %d : sumX-> %d\n", m,sumX[m);

        ShmPTR->status = TAKEN;
        printf("Client has informed server data have been taken...\n");
    }
    shmdt((void *) ShmPTR);
    printf("Client has detached its shared memory...\n");
    printf("Client exits...\n");
    exit(0);
}

What I am doing right now is using a member of the struct called status to kind of prevent race conditions. From what I have read till now, semaphores allow a similar thing in IPC.

Question: How to use semaphore in this such that the memory space which needs to be shared is just the array(s) and not wrapping it in a struct with a self defined flag ?

Edit1: Or maybe mutex for that matter if it is better than semaphore for this application.

Edit2: Following @Stargateur answer which works for this code but in the production code where nx, ny and nz are variables, how does one define a shared memory for the struct consisting of a member which is a variable length multidimensional array ? ( ofcourse it would live until the call shmdt and shmctl )

Improve this question
0

2 Answers 2

Reset to default
3

I will advice you to implement your feature with two semaphores, one for unlock produser, one for unlock consumer.

How to use semaphore in this such that the memory space which needs to be shared is just the array(s) and not wrapping it in a struct with a self defined flag ?

Yes but why separate the data and the semaphore associate with the data ?

I would do the following:

struct Memory {
     sem_t    prod;
     sem_t    cons;
     double   u_x[nx+1][ny+2][nz+2];
};

// produser
sem_init(&ShmPTR->cons, !0, 0);
sem_init(&ShmPTR->prod, !0, 1);

 for (int m = 0; m < niter; m++) {
     sem_wait(&ShmPTR->prod);
     // ...
     sem_post(&ShmPTR->cons);
 }

// consumer
for (int m =0; m<niter; m++) {
     sem_wait(&ShmPTR->cons);
     // ...
     sem_post(&ShmPTR->prod);
}

Or maybe mutex for that matter if it is better than semaphore for this application.

mutex can't be share between process.

By the way, you use an int to iterate on a array, you should use size_t

Improve this answer
Sign up to request clarification or add additional context in comments.

9 Comments

bit struggling with the clubbing the semaphore and the data ( there are more such arrays, i have just showed one such array u_x there exists u_y and u_z ). is there a particular place where you can refer me to , to understand a bit more about semaphore.
@datapanda Well, I'm don't know if there are such resource. My method is to read the documentation, this is generally enough for me, man7.org/linux/man-pages/man7/sem_overview.7.html, when you finish to understand the mechanics the only limit is your imagination. If you need a more detailed answer you can update or create a new question.
This did worked out seamlessly in the example code, let me give it a try on the production code. On a side note, why do I have to link pthread library while compiling the program with semaphores in it (semaphore.h)
@datapanda History reason, "Programs using the POSIX semaphores API must be compiled with cc -pthread to link against the real-time library, librt.", not very important but it's needed.
and hope they are portable on HPC clusters running Linux which are indeed POSIX compliant. Thanks @Stargateur
|
0

If you are planning to keep PR and CO as separate processes, you can try to fork one of the processes from the other to synchronize them. In this particular case, my suggestion is to fork CO from PR process. Here is briefly my idea:

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/shm.h>
#include <sys/stat.h>
#include <sys/mman.h>


int main(int argc, char* argv[])
{
    const char *name="SHARED";
    const int SIZE = 4096;


    pid_t pidA;
    pidA = fork();

    if (pidA < 0)
    {
        printf("forkA Failed" );
        return 1;
    }
    else if (pidA == 0) // Child process A
    {


        // Read from the shared memory object.


        exit(0);
    }
    else // Parent process
    {

        int shm_fd;
        /* pointer to shared memory obect */
        void* ptr;
        /* create the shared memory object */
        shm_fd = shm_open(name, O_CREAT | O_RDWR, 0666);
        /* configure the size of the shared memory object */
        ftruncate(shm_fd, SIZE);
        /* memory map the shared memory object */
        ptr = mmap(0, SIZE, PROT_WRITE, MAP_SHARED, shm_fd, 0);



        /* write to the shared memory object */



        // Wait for child to read matrix
        wait(NULL);


        printf("Program finished------\n");
    }
}

But we all know that threads are more lightweight, hence preferable. You can use semaphore with two threads as below:

#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h>

sem_t can_read,can_write; // declare two global semaphore

void* threadPR(void* arg)
{
  while(true)
  {
    //wait
    sem_wait(&can_write);


    //Write to matrix


    //signal
    sem_post(&can_read);
  }
}

void* threadCO(void* arg)
{
  while(true)
  {
    //wait
    sem_wait(&can_read);


    //Read the matrix


    //signal
    sem_post(&can_write);
  }
}


int main()
{
    // initialize the semaphore
    sem_init(&mutex, 0, 1);
    // declare two threads
    pthread_t t1,t2;
    pthread_create(&t1,NULL,threadPR,NULL); // Run the PR thread

    // do whatever needed before running CO

    pthread_create(&t2,NULL,threadCO,NULL); // Run the CO thread

    // wait for threads to join
    pthread_join(t1,NULL);
    pthread_join(t2,NULL);

    // free the semaphore
    sem_destroy(&mutex);
    return 0;
}

You can add required initializations as globals to this implementation.

Improve this answer

8 Comments

In my case the intended use case will not allow me to fork as there are two separate codes one which is openMP parallelised and the other one is an MPI parallelised code. MPI code needs read access to a memory block which is created and modified by openMP code in each iteration. :(
You need a shared common storage for your both openMP and MPI parallelised programs then. Or you can change one of the program if it is okay for you. Or you can force same context for both program, which i do not suggest at all for parallelised programs, it will bring huge overhead.
the problem is the OpenMP code is a legacy code which is very difficult to port to MPI in a straightforward manner and the MPI code heavily depends on an opensource library which needs MPI to run by default. So I am a bit stuck in an odd situation leaving the block of memory that I need in my MPI code to be shared by the OpenMP code.
I see. If i were you, I would use a shared file system and handle the synchronization on the file system level.
Could you elaborate on it ? (Maybe it will help me in deciding the course of solution ) I presume a filesystem would lead to disk IO. To give you more context both of the codes in the discussion are CFD codes where I am interested in sharing the velocity matrices which are huge ones.
|

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.