producer/consumer code (working)

threads-sema/Makefile

@@ -7,6 +7,7 @@ SRCS   := dining_philosophers_deadlock.c \
 	dining_philosophers_no_deadlock_print.c \
 	join.c \
 	binary.c \
+	producer_consumer_works.c \
 	zemaphore.c
 
 OBJS   := ${SRCS:c=o}

threads-sema/README.md

@@ -24,6 +24,25 @@ prompt> make
 prompt> ./binary
 ```
 
+# Producer/Consumer
+
+Code for the working producer/consumer solution from the text,
+found in `producer_consumer.c`.
+
+Run `make` to build, and `producer_consumer` to test it.
+The program takes a few different arguments:
+- The number of buffers between the producer/consumer
+- The number of times a producer should produce something
+- The number of consumer threads
+
+```sh
+prompt> make
+prompt> ./producer_consumer 1 1000 1
+```
+
+The output should print each produced item once, and show which
+consumer consumed each produced item.
+
 
 # Dining Philosophers
 

threads-sema/producer_consumer_works.c

@@ -0,0 +1,107 @@
+#include <stdio.h>
+#include <unistd.h>
+#include <assert.h>
+#include <pthread.h>
+#include <stdlib.h>
+#include <semaphore.h>
+
+#include "common.h"
+#include "common_threads.h"
+
+int max;
+int loops;
+int *buffer;
+
+int use  = 0;
+int fill = 0;
+
+sem_t empty;
+sem_t full;
+sem_t mutex;
+
+#define CMAX (10)
+int consumers = 1;
+
+void do_fill(int value) {
+    buffer[fill] = value;
+    fill++;
+    if (fill == max)
+	fill = 0;
+}
+
+int do_get() {
+    int tmp = buffer[use];
+    use++;
+    if (use == max)
+	use = 0;
+    return tmp;
+}
+
+void *producer(void *arg) {
+    int i;
+    for (i = 0; i < loops; i++) {
+	Sem_wait(&empty);
+	Sem_wait(&mutex);
+	do_fill(i);
+	Sem_post(&mutex);
+	Sem_post(&full);
+    }
+
+    // end case
+    for (i = 0; i < consumers; i++) {
+	Sem_wait(&empty);
+	Sem_wait(&mutex);
+	do_fill(-1);
+	Sem_post(&mutex);
+	Sem_post(&full);
+    }
+
+    return NULL;
+}
+                                                                               
+void *consumer(void *arg) {
+    int tmp = 0;
+    while (tmp != -1) {
+	Sem_wait(&full);
+	Sem_wait(&mutex);
+	tmp = do_get();
+	Sem_post(&mutex);
+	Sem_post(&empty);
+	printf("%lld %d\n", (long long int) arg, tmp);
+    }
+    return NULL;
+}
+
+int main(int argc, char *argv[]) {
+    if (argc != 4) {
+	fprintf(stderr, "usage: %s <buffersize> <loops> <consumers>\n", argv[0]);
+	exit(1);
+    }
+    max   = atoi(argv[1]);
+    loops = atoi(argv[2]);
+    consumers = atoi(argv[3]);
+    assert(consumers <= CMAX);
+
+    buffer = (int *) malloc(max * sizeof(int));
+    assert(buffer != NULL);
+    int i;
+    for (i = 0; i < max; i++) {
+	buffer[i] = 0;
+    }
+
+    Sem_init(&empty, max); // max are empty 
+    Sem_init(&full, 0);    // 0 are full
+    Sem_init(&mutex, 1);   // mutex
+
+    pthread_t pid, cid[CMAX];
+    Pthread_create(&pid, NULL, producer, NULL); 
+    for (i = 0; i < consumers; i++) {
+	Pthread_create(&cid[i], NULL, consumer, (void *) (long long int) i); 
+    }
+    Pthread_join(pid, NULL); 
+    for (i = 0; i < consumers; i++) {
+	Pthread_join(cid[i], NULL); 
+    }
+    return 0;
+}
+