Add cookies to thread pool

5 files changed, 35 insertions, 9 deletions

diff --git a/src/calibrate_detector.c b/src/calibrate_detector.c
index 1f413b44..ccf577dd 100644
--- a/src/calibrate_detector.c
+++ b/src/calibrate_detector.c
@@ -157,7 +157,7 @@ static void sum_threshold(struct image *image, double *sum, double threshold)
 }
 
 
-static void add_image(void *args)
+static void add_image(void *args, int cookie)
 {
 	struct sum_args *pargs = args;
 	struct hdfile *hdfile;
@@ -178,7 +178,7 @@ static void add_image(void *args)
 	image.orientation.y = 0.0;
 	image.orientation.z = 0.0;
 
-	STATUS("Processing '%s'\n", pargs->filename);
+	STATUS("%3i: Processing '%s'\n", cookie, pargs->filename);
 
 	hdfile = hdfile_open(pargs->filename);
 	if ( hdfile == NULL ) {
diff --git a/src/cubeit.c b/src/cubeit.c
index 98e6711a..a8e969ff 100644
--- a/src/cubeit.c
+++ b/src/cubeit.c
@@ -200,7 +200,7 @@ static void interpolate_onto_grid(double *vals, double v,
 }
 
 
-static void sum_image(void *pg)
+static void sum_image(void *pg, int cookie)
 {
 	struct sum_args *apargs = pg;
 	struct static_sum_args *pargs = &apargs->static_args;
diff --git a/src/reintegrate.c b/src/reintegrate.c
index 5b9cf0f2..50e73877 100644
--- a/src/reintegrate.c
+++ b/src/reintegrate.c
@@ -97,7 +97,7 @@ static void show_help(const char *s)
 }
 
 
-static void process_image(void *pg)
+static void process_image(void *pg, int cookie)
 {
 	struct integration_args *apargs = pg;
 	struct static_integration_args *pargs = &apargs->static_args;
diff --git a/src/thread-pool.c b/src/thread-pool.c
index 7781f883..3a39f1ca 100644
--- a/src/thread-pool.c
+++ b/src/thread-pool.c
@@ -21,6 +21,7 @@
 #include <string.h>
 #include <unistd.h>
 #include <pthread.h>
+#include <assert.h>
 
 #include "utils.h"
 
@@ -145,10 +146,12 @@ struct task_queue
 	int              n_started;
 	int              n_completed;
 	int              max;
+	int              n_cookies;
+	int             *cookies;
 
 	void *(*get_task)(void *);
 	void *queue_args;
-	void (*work)(void *);
+	void (*work)(void *, int);
 };
 
 
@@ -159,6 +162,9 @@ static void *task_worker(void *pargsv)
 	do {
 
 		void *task;
+		int i;
+		int mycookie = -1;
+		int found = 0;
 
 		/* Get a task */
 		pthread_mutex_lock(&q->lock);
@@ -174,14 +180,26 @@ static void *task_worker(void *pargsv)
 			break;
 		}
 
+		/* Find a cookie */
+		for ( i=0; i<q->n_cookies; i++ ) {
+			if ( q->cookies[i] == 0 ) {
+				mycookie = i;
+				found = 1;
+				q->cookies[i] = 1;
+				break;
+			}
+		}
+		assert(found);
+
 		q->n_started++;
 		pthread_mutex_unlock(&q->lock);
 
-		q->work(task);
+		q->work(task, mycookie);
 
-		/* Update totals etc */
+		/* Update totals, release cookie etc */
 		pthread_mutex_lock(&q->lock);
 		q->n_completed++;
+		q->cookies[mycookie] = 0;
 		pthread_mutex_unlock(&q->lock);
 
 	} while ( 1 );
@@ -190,7 +208,7 @@ static void *task_worker(void *pargsv)
 }
 
 
-int run_threads(int n_threads, void (*work)(void *),
+int run_threads(int n_threads, void (*work)(void *, int),
                 void *(*get_task)(void *), void *queue_args, int max)
 {
 	pthread_t *workers;
@@ -206,6 +224,13 @@ int run_threads(int n_threads, void (*work)(void *),
 	q.n_started = 0;
 	q.n_completed = 0;
 	q.max = max;
+	q.n_cookies = n_threads;
+	q.cookies = malloc(q.n_cookies * sizeof(int));
+
+
+	for ( i=0; i<n_threads; i++ ) {
+		q.cookies[i] = 0;
+	}
 
 	/* Start threads */
 	for ( i=0; i<n_threads; i++ ) {
@@ -224,6 +249,7 @@ int run_threads(int n_threads, void (*work)(void *),
 	}
 
 	free(workers);
+	free(q.cookies);
 
 	return q.n_completed;
 }
diff --git a/src/thread-pool.h b/src/thread-pool.h
index 11123493..3376b8fe 100644
--- a/src/thread-pool.h
+++ b/src/thread-pool.h
@@ -30,7 +30,7 @@ extern void run_thread_range(int n_tasks, int n_threads, const char *text,
  * be passed to work().  Work will stop after 'max' tasks have been processed.
  * get_task() does not need to be re-entrant.
  * Returns: the number of tasks processed. */
-extern int run_threads(int n_threads, void (*work)(void *),
+extern int run_threads(int n_threads, void (*work)(void *, int),
                        void *(*get_task)(void *), void *queue_args, int max);