mandelbrot: initial multi-consumer calculation

mandelbrot/config.h

@@ -1,5 +1,5 @@
-#define BAIL_OUT 32
-#define MAX_ITER 64
+#define BAIL_OUT 8192
+#define MAX_ITER 16384
 
 #define WIDTH 1024
 #define HEIGHT 1024

mandelbrot/mandelbrot.c

@@ -70,6 +70,7 @@ void *produce(void *b)
 			enqueue(buf->q, m, buf->lock);
 		}
 	}
+	buf->produce_end = 1;
 	return NULL;
 }
 
@@ -83,8 +84,9 @@ void *consume(void *b)
 	buf = (mandel_buf_t *) b;
 
 
-	/* This will not work on a multi-consumer program */
-	while (i < buf->prop.width * buf->prop.height) {
+	/* 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);
@@ -97,6 +99,9 @@ void *consume(void *b)
 			free(m);
 			i++;
 		}
+		if(is_empty(buf->q) && buf->produce_end){
+			return NULL;
+		}
 	}
 	return NULL;
 }
@@ -104,7 +109,7 @@ void *consume(void *b)
 int main(int argc, char **argv)
 {
 
-	pthread_t consumer;
+	pthread_t consumer1, consumer2;
 	pthread_t producer;
 	pthread_attr_t attr;
 	pthread_mutex_t lock;
@@ -129,20 +134,24 @@ int main(int argc, char **argv)
 	queue = initialize_queue();
 	pthread_mutex_init(&lock, NULL);
 	buf.lock = &lock;
+	buf.produce_end = 0;
 	buf.result_field = malloc(sizeof(int) * WIDTH * HEIGHT);
 
 	buf.q = queue;
 
 	/* Create and start threads */
 	pthread_create(&producer, &attr, produce, &buf);
-	pthread_create(&consumer, &attr, consume, &buf);
+	pthread_create(&consumer1, &attr, consume, &buf);
+	pthread_create(&consumer2, &attr, consume, &buf);
 
 
 
 	pthread_join(producer, NULL);
 	printf("\nFinished producer!");
-	pthread_join(consumer, NULL);
-	printf("\nFinished consumer!");
+	pthread_join(consumer1, NULL);
+	printf("\nFinished consumer 1!");
+	pthread_join(consumer2, NULL);
+	printf("\nFinished consumer 2!");
 
 	/* Create the png image with the result buffer */
 	pix_row rows[WIDTH];

mandelbrot/mandelbrot.h

@@ -33,6 +33,7 @@ typedef struct mandel_buf_t {
 	int *result_field;
 	mandel_prop_t prop;
 	pthread_mutex_t *lock;
+	int produce_end;
 } mandel_buf_t;
 
 queue_t *initialize_queue();