mandelbrot: finished multi-consumer queue

mandelbrot/.gitignore

@@ -58,5 +58,5 @@ dkms.conf
 # Remove vscode metafiles
 .vscode/*
 
-
+*.png
 bin/*

mandelbrot/config.h

@@ -4,5 +4,5 @@
 #define WIDTH 1024
 #define HEIGHT 1024
 
-#define THREAD_COUNT 16
+#define THREAD_COUNT 36
 #define BUFFER_SIZE  256

mandelbrot/mandelbrot.c

@@ -87,21 +87,17 @@ void *consume(void *b)
 	/* This will not work on a multi-consumer program 
 	while (i < buf->prop.width * buf->prop.height) { */
 	for(;;){
-		if (!is_empty(buf->q)) {
-			m = dequeue(buf->q, buf->lock);
-			x = map(m->x, 0, buf->prop.width, -1, 1);
-			y = map(m->y, 0, buf->prop.height, -1, 1);
-			//fprintf(stdout, "\nSending %d %d", m->x, m->y, k);
-
-			k = mandelbrot_pix(x, y, buf->prop.bail_out,
-					buf->prop.max_iter);
-			buf->result_field[m->y + m->x * buf->prop.height] = k;
-			free(m);
-			i++;
-		}
-		if(is_empty(buf->q) && buf->produce_end){
-			return NULL;
-		}
+		m = dequeue(buf->q, buf->lock);
+		if (m == NULL) return NULL; // if m is null then the queue is empty ?
+		x = map(m->x, 0, buf->prop.width, -1, 1);
+		y = map(m->y, 0, buf->prop.height, -1, 1);
+		//fprintf(stdout, "\nSending %d %d", m->x, m->y, k);
+
+		k = mandelbrot_pix(x, y, buf->prop.bail_out,
+				buf->prop.max_iter);
+		buf->result_field[m->y + m->x * buf->prop.height] = k;
+		free(m);
+		i++;
 	}
 	return NULL;
 }
@@ -109,7 +105,7 @@ void *consume(void *b)
 int main(int argc, char **argv)
 {
 
-	pthread_t consumer1, consumer2;
+	pthread_t consumer[THREAD_COUNT];
 	pthread_t producer;
 	pthread_attr_t attr;
 	pthread_mutex_t lock;
@@ -141,17 +137,16 @@ int main(int argc, char **argv)
 
 	/* Create and start threads */
 	pthread_create(&producer, &attr, produce, &buf);
-	pthread_create(&consumer1, &attr, consume, &buf);
-	pthread_create(&consumer2, &attr, consume, &buf);
-
-
 
 	pthread_join(producer, NULL);
-	printf("\nFinished producer!");
-	pthread_join(consumer1, NULL);
-	printf("\nFinished consumer 1!");
-	pthread_join(consumer2, NULL);
-	printf("\nFinished consumer 2!");
+	
+	for (int i=0; i<THREAD_COUNT; i++) {
+		pthread_create(&consumer[i], &attr, consume, &buf);
+	}
+
+	for (int i=0; i<THREAD_COUNT; i++) {
+		pthread_join(consumer[i], NULL);
+	}
 
 	/* Create the png image with the result buffer */
 	pix_row rows[WIDTH];

mandelbrot/mandelbrot.h

@@ -40,4 +40,4 @@ queue_t *initialize_queue();
 void enqueue(queue_t * q, mandel_t * w, pthread_mutex_t * lock);
 mandel_t *dequeue(queue_t * q, pthread_mutex_t * lock);
 void destroy_queue(queue_t * q);
-int is_empty(queue_t * q);
+int is_empty(queue_t * q, pthread_mutex_t * lock);

mandelbrot/queue.c

@@ -39,23 +39,35 @@ mandel_t *dequeue(queue_t * q, pthread_mutex_t * lock)
 {
 	struct queue_node_t *aux;
 	mandel_t *ret_val;
-
-	if (q->head != NULL) {
-		pthread_mutex_lock(lock);
-		ret_val = q->head->node;
+	pthread_mutex_lock(lock);
+	
+	if (q->head == NULL) {
+		ret_val = NULL;
+	}
+	else if (q->head == q->tail) {
 		aux = q->head;
-		q->head = q->head->next;
-		pthread_mutex_unlock(lock);
-		free(aux);
-		return (ret_val);
+		ret_val = aux->node;
+		q->head = NULL;
+		q->tail = NULL;
 	}
-	return NULL;
+	else {
+		aux = q->head;
+		ret_val = aux->node;
+		q->head = aux->next;
+	}
+	pthread_mutex_unlock(lock);
+	return ret_val;
 
 }
 
-int is_empty(queue_t * q)
+int is_empty(queue_t * q, pthread_mutex_t * lock)
 {
-	return (q->head == NULL);
+	int ret = 0;
+	pthread_mutex_lock(lock);
+	if (q->head == NULL) ret = 1;
+	pthread_mutex_unlock(lock);
+	
+	return (ret);
 }
 
 void destroy_queue(queue_t * q)