multithreading - consumers producer

[fonsde]

τι ειναι λαθος?? και περνω consume ενω δεν πρεπει

 

 

>

[P1] Producing 0 ...
[P0] Producing 0 ...
=====>[C1] Consuming 0 ...
=====>[C2] Consuming 0 ...
[P2] Producing 0 ...
=====>[C0] Consuming 0 ...
[P1] Producing 1 ...
[P0] Producing 1 ...
=====>[C1] Consuming 1 ...
=====>[C2] Consuming 1 ...
[P2] Producing 1 ...
=====>[C0] Consuming 1 ...
=====>[C2] Consuming 0 ...
[P1] Producing 2 ...
[P0] Producing 2 ...
[P2] Producing 2 ...
=====>[C1] Consuming 2 ...
=====>[C0] Consuming 2 ...
=====>[C2] Consuming 1 ...
[P1] Producing 3 ...
[P0] Producing 3 ...
[P2] Producing 3 ...
=====>[C1] Consuming 3 ...
=====>[C0] Consuming 3 ...

 

>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include </usr/include/semaphore.h>

#define BUFF_SIZE   5 /* total number of slots */
#define NP          3 /* total number of producers */
#define NC          3 /* total number of consumers */
#define NITERS      4 /* number of items produced/consumed */

typedef struct {
   int buf[bUFF_SIZE];   /* shared var */
   int in;            /* buf[in%BUFF_SIZE] is the first empty slot */
   int out;           /* buf[out%BUFF_SIZE] is the first full slot */
   sem_t full;        /* keep track of the number of full spots */
   sem_t empty;       /* keep track of the number of empty spots */
   sem_t mutex;       /* enforce mutual exclusion to shared data */
} sbuf_t;

sbuf_t shared;

void *Producer(void *arg)
{
   int i, item, index;

   index = (int)arg;

   for (i=0; i < NITERS; i++) {

       /* Produce item */
       item = i; 

       /* Prepare to write item to buf */

       /* If there are no empty slots, wait */
       sem_wait(&shared.empty);
       /* If another thread uses the buffer, wait */
       sem_wait(&shared.mutex);
       shared.buf[shared.in] = item;
       shared.in = (shared.in+1)%BUFF_SIZE;
       printf("[P%d] Producing %d ...\n", index, item); fflush(stdout);
       /* Release the buffer */
       sem_post(&shared.mutex);
       /* Increment the number of full slots */
       sem_post(&shared.full);

       /* Interleave  producer and consumer execution */
       if (i % 2 == 1) sleep(1);
   }
   pthread_exit(0);
}

void *Consumer(void *arg)
{
   int i, item, index;

   index = (int)arg;

   for (i=0; i < NITERS; i++) {


       /* Prepare to write item to buf */

       /* If there are no used slots, wait */
       sem_wait(&shared.full);
       /* If another thread uses the buffer, wait */
       sem_wait(&shared.mutex);

       item = shared.buf[shared.out];
       shared.out = (shared.out + 1) % BUFF_SIZE;
       printf("=====>[C%d] Consuming %d ...\n", index, item); fflush(stdout);
       /* Release the buffer */
       sem_post(&shared.mutex);
       /* Increment the number of full slots */
       sem_post(&shared.empty);

       /* Interleave  producer and consumer execution */
       if (i % 2 == 1) sleep(1);

   }
   pthread_exit(0);
}

int main()
{
   pthread_t idP, idC;
   int index;

   sem_open(&shared.full, 0, 0);
   sem_open(&shared.empty, 0, BUFF_SIZE);

   /* Insert code here to initialize mutex*/
   sem_open(&shared.mutex, 0, 1);


   for (index = 0; index < NP; index++)
   {  
      /* Create a new producer */
      pthread_create(&idP, NULL, (void*) Producer, (void*) index);
   }

   /* Insert code here to create NC consumers */
   for (index = 0; index < NC; index++)
   {  
      /* Create a new consumer */
      pthread_create(&idC, NULL, (void*) Consumer, (void*) index);
   }

   sleep(6);


   sem_close(&shared.full);
   sem_close(&shared.empty);
   sem_close(&shared.mutex);

   exit(0);
}

Επεξ/σία 16 Νοεμβρίου 2012 από fonsde

[migf1]

Εγώ σε cygwin παίρνω αυτή την έξοδο..

 

>
[P1] Producing 0 ...
[P1] Producing 1 ...
[P2] Producing 0 ...
[P2] Producing 1 ...
[P2] Producing 0 ...
[P2] Producing 1 ...
=====>[C1] Consuming 1 ...
=====>[C1] Consuming 1 ...
=====>[C2] Consuming 0 ...
=====>[C2] Consuming 1 ...
=====>[C3] Consuming 0 ...
=====>[C3] Consuming 1 ...

 

Και πολλά warnings πως η sem_open() περιμένει 1ο όρισμα (const char *) δηλαδή το όνομα του sempahore, το οποίο αν τα θυμάμαι σωστά πρέπει να αρχίζει με / ... (εσύ περνάς στην sem_open() sem_t ** ως 1ο όρισμα).

[fonsde]

ναι γιατι ειμαι σε mac και δεν παιζει η sem_init

 

λαθος ειναι και αυτο που περνεις. ( τα index ειναι λαθος)

 

γιατι τοσο λιγα output sου βγαζει?

[migf1]

Ουπς, λάθος... την έξοδο που δίνεις κι εσύ παίρνω.

 

ΥΓ. Οπότε πως να σου δουλέψει σωστά άμα της περνάς sem_t **? Βασικά έχω πολλά χρόνια να ασχοληθώ, δεν θυμάμαι καν πως ορίζεται το sem_t (έχει const char * ως 1ο field? χλωμό το κόβω).

 

Αλλά άσχετα από αυτό, δεν πρέπει να περάσεις O_CREATE ως 2ο όρισμα την 1η φορά στην sem_open ώστε να δημιουργήσεις το semaphore?

[fonsde]

εχειςδικαιο.

works fine

 

>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include </usr/include/semaphore.h>
#define BUFF_SIZE   5  /* total number of slots */
#define NP		  3  /* total number of producers */
#define NC		  3  /* total number of consumers */
#define NITERS	  4  /* number of items produced/consumed */
typedef struct {
   int buf[bUFF_SIZE];   /* shared var */
   int in;		    /* buf[in%BUFF_SIZE] is the first empty slot */
   int out;		   /* buf[out%BUFF_SIZE] is the first full slot */
   sem_t* full;	    /* keep track of the number of full spots */
   sem_t* empty;	   /* keep track of the number of empty spots */
   sem_t* mutex;	   /* enforce mutual exclusion to shared data */
} sbuf_t;
sbuf_t shared;
void *Producer(void *arg)
{
   int i, item, index;
   index = (int)arg;
   for (i=0; i < NITERS; i++) {
    /* Produce item */
    item = i;
    /* Prepare to write item to buf */
    /* If there are no empty slots, wait */
    sem_wait(shared.empty);
    /* If another thread uses the buffer, wait */
    sem_wait(shared.mutex);
    shared.buf[shared.in] = item;
    shared.in = (shared.in+1)%BUFF_SIZE;
    printf("[P%d] Producing %d ...\n", index, item); fflush(stdout);
    /* Release the buffer */
    sem_post(shared.mutex);
    /* Increment the number of full slots */
    sem_post(shared.full);
    /* Interleave  producer and consumer execution */
    if (i % 2 == 1) sleep(1);
   }
   pthread_exit(0);
}
void *Consumer(void *arg)
{
   int i, item, index;
   index = (int)arg;
   for (i=0; i < NITERS; i++) {

    /* Prepare to write item to buf */
    /* If there are no used slots, wait */
    sem_wait(shared.full);
    /* If another thread uses the buffer, wait */
    sem_wait(shared.mutex);
   
    item = shared.buf[shared.out];
    shared.out = (shared.out + 1) % BUFF_SIZE;
    printf("=====>[C%d] Consuming %d ...\n", index, item); fflush(stdout);
    /* Release the buffer */
    sem_post(shared.mutex);
    /* Increment the number of full slots */
    sem_post(shared.empty);
    /* Interleave  producer and consumer execution */
    if (i % 2 == 1) sleep(1);
   }
   pthread_exit(0);
}
int main()
{
   pthread_t idP, idC;
   int index;
   shared.full = sem_open("/mysemaphore", O_CREAT, S_IRUSR | S_IWUSR, 0);
   shared.empty = sem_open("/mysemaphore2", O_CREAT, S_IRUSR | S_IWUSR, BUFF_SIZE);
   /* Insert code here to initialize mutex*/
   shared.mutex = sem_open("/mysemaphore3", O_CREAT, S_IRUSR | S_IWUSR, 1);

   for (index = 0; index < NP; index++)
   { 
   /* Create a new producer */
   pthread_create(&idP, NULL, (void*) Producer, (void*) index);
   }
   /* Insert code here to create NC consumers */
   for (index = 0; index < NC; index++)
   { 
   /* Create a new consumer */
   pthread_create(&idC, NULL, (void*) Consumer, (void*) index);
   }
  
   sleep(6);

   sem_close(shared.full);
   sem_close(shared.empty);
   sem_close(shared.mutex);
  
   exit(0);
}