aboutsummaryrefslogtreecommitdiff
path: root/src/thread-pool.c
blob: b0d3d8c2df3b4a5faf9efd2d0a5eff26a4f71d7f (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
/*
 * thread-pool.c
 *
 * A thread pool implementation
 *
 * (c) 2006-2010 Thomas White <taw@physics.org>
 *
 * Part of CrystFEL - crystallography with a FEL
 *
 */


#ifdef HAVE_CONFIG_H
#include <config.h>
#endif


#include <stdarg.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>

#include "utils.h"


struct task_queue
{
	pthread_mutex_t  lock;

	int              n_tasks;
	int             *done;
	int              n_done;

	void (*work)(int, void *);
	void *work_args;

	const char      *text;
};


static void *worker_thread(void *pargsv)
{
	struct task_queue *q = pargsv;

	do {

		int i;
		int found = 0;
		int mytask = -1;

		/* Get a task */
		pthread_mutex_lock(&q->lock);
		for ( i=0; i<q->n_tasks; i++ ) {
			if ( q->done[i] == 0 ) {
				mytask = i;
				found = 1;
			}
		}
		pthread_mutex_unlock(&q->lock);

		/* No more tasks? */
		if ( !found ) break;

		q->work(mytask, q->work_args);

		/* Mark this task as done, update totals etc */
		pthread_mutex_lock(&q->lock);
		q->done[mytask] = 1;
		q->n_done++;
		progress_bar(q->n_done, q->n_tasks, q->text);
		pthread_mutex_unlock(&q->lock);

	} while ( 1 );

	return NULL;
}


void munch_threads(int n_tasks, int n_threads, const char *text,
                   void (*work)(int, void *), void *work_args)
{
	pthread_t *workers;
	int i;
	struct task_queue q;

	/* The nation of CrystFEL prides itself on having 0% unemployment. */
	if ( n_threads > n_tasks ) n_threads = n_tasks;

	workers = malloc(n_threads * sizeof(pthread_t));

	q.done = malloc(n_tasks * sizeof(int));
	pthread_mutex_init(&q.lock, NULL);
	q.n_tasks = n_tasks;
	q.work = work;
	q.work_args = work_args;
	q.n_done = 0;
	q.text = text;

	for ( i=0; i<n_tasks; i++ ) {
		q.done[i] = 0;
	}

	/* Start threads */
	for ( i=0; i<n_threads; i++ ) {

		if ( pthread_create(&workers[i], NULL, worker_thread, &q) ) {
			ERROR("Couldn't start thread %i\n", i);
			n_threads = i;
			break;
		}

	}

	/* Join threads */
	for ( i=0; i<n_threads; i++ ) {
		pthread_join(workers[i], NULL);
	}

	free(q.done);
	free(workers);
}