libresman
view src/threadpool.c @ 26:6b9974a8bdae
started BSD/mac support for file watching
| author | John Tsiombikas <nuclear@member.fsf.org> |
|---|---|
| date | Mon, 31 Mar 2014 19:51:00 +0300 |
| parents | 711698580eb0 |
| children |
line source
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <string.h>
4 #include <pthread.h>
5 #include "threadpool.h"
7 struct work_item {
8 int id; /* just for debugging messages */
9 void *data;
10 struct work_item *next;
11 };
13 struct thread_pool {
14 pthread_t *workers;
15 int num_workers;
17 pthread_mutex_t work_lock;
18 pthread_cond_t work_cond;
20 int start;
21 pthread_cond_t start_cond;
23 tpool_work_func work_func;
24 void *cls;
26 struct work_item *work_list, *work_list_tail;
27 int work_count;
28 };
31 static void *thread_func(void *tp);
32 static struct work_item *alloc_node(void);
33 static void free_node(struct work_item *node);
34 static int get_processor_count(void);
38 int tpool_init(struct thread_pool *tpool, int num_threads)
39 {
40 int i;
42 memset(tpool, 0, sizeof *tpool);
45 pthread_mutex_init(&tpool->work_lock, 0);
46 pthread_cond_init(&tpool->work_cond, 0);
49 if(num_threads <= 0) {
50 num_threads = get_processor_count();
51 }
52 tpool->num_workers = num_threads;
54 printf("initializing thread pool with %d worker threads\n", num_threads);
56 if(!(tpool->workers = malloc(num_threads * sizeof *tpool->workers))) {
57 fprintf(stderr, "failed to create array of %d threads\n", num_threads);
58 return -1;
59 }
61 /* this start condvar is pretty useless */
62 pthread_cond_init(&tpool->start_cond, 0);
64 for(i=0; i<num_threads; i++) {
65 if(pthread_create(tpool->workers + i, 0, thread_func, tpool) == -1) {
66 fprintf(stderr, "%s: failed to create thread %d\n", __FUNCTION__, i);
67 tpool_destroy(tpool);
68 return -1;
69 }
70 }
71 tpool->start = 1;
72 pthread_cond_broadcast(&tpool->start_cond);
73 return 0;
74 }
76 void tpool_destroy(struct thread_pool *tpool)
77 {
78 int i;
79 for(i=0; i<tpool->num_workers; i++) {
80 void *ret;
81 pthread_join(tpool->workers[i], &ret);
82 }
84 pthread_mutex_destroy(&tpool->work_lock);
85 pthread_cond_destroy(&tpool->work_cond);
86 }
88 struct thread_pool *tpool_create(int num_threads)
89 {
90 struct thread_pool *tpool = malloc(sizeof *tpool);
91 if(!tpool) {
92 return 0;
93 }
94 if(tpool_init(tpool, num_threads) == -1) {
95 free(tpool);
96 return 0;
97 }
98 return tpool;
99 }
101 void tpool_free(struct thread_pool *tpool)
102 {
103 if(tpool) {
104 tpool_destroy(tpool);
105 free(tpool);
106 }
107 }
109 void tpool_set_work_func(struct thread_pool *tpool, tpool_work_func func, void *cls)
110 {
111 tpool->work_func = func;
112 tpool->cls = cls;
113 }
115 int tpool_add_work(struct thread_pool *tpool, void *data)
116 {
117 struct work_item *node;
118 static int jcounter;
120 if(!(node = alloc_node())) {
121 fprintf(stderr, "%s: failed to allocate new work item node\n", __FUNCTION__);
122 return -1;
123 }
124 node->data = data;
125 node->next = 0;
127 pthread_mutex_lock(&tpool->work_lock);
128 node->id = jcounter++;
130 printf("TPOOL: adding work item: %d\n", node->id);
132 if(!tpool->work_list) {
133 tpool->work_list = tpool->work_list_tail = node;
134 } else {
135 tpool->work_list_tail->next = node;
136 tpool->work_list_tail = node;
137 }
138 pthread_mutex_unlock(&tpool->work_lock);
140 /* wakeup all threads, there's work to do */
141 pthread_cond_broadcast(&tpool->work_cond);
142 return 0;
143 }
146 static void *thread_func(void *tp)
147 {
148 int i, tidx = -1;
149 struct work_item *job;
150 struct thread_pool *tpool = tp;
151 pthread_t tid = pthread_self();
153 /* wait for the start signal :) */
154 pthread_mutex_lock(&tpool->work_lock);
155 while(!tpool->start) {
156 pthread_cond_wait(&tpool->start_cond, &tpool->work_lock);
157 }
158 pthread_mutex_unlock(&tpool->work_lock);
160 for(i=0; i<tpool->num_workers; i++) {
161 if(pthread_equal(tpool->workers[i], tid)) {
162 tidx = i;
163 break;
164 }
165 }
167 for(;;) {
168 int job_id;
169 void *data;
171 pthread_mutex_lock(&tpool->work_lock);
172 /* while there aren't any work items to do go to sleep on the condvar */
173 while(!tpool->work_list) {
174 pthread_cond_wait(&tpool->work_cond, &tpool->work_lock);
175 }
177 job = tpool->work_list;
178 tpool->work_list = tpool->work_list->next;
180 job_id = job->id;
181 data = job->data;
182 free_node(job);
183 pthread_mutex_unlock(&tpool->work_lock);
185 printf("TPOOL: worker %d start job: %d\n", tidx, job_id);
186 tpool->work_func(data, tpool->cls);
187 printf("TPOOL: worker %d completed job: %d\n", tidx, job_id);
188 }
189 return 0;
190 }
192 /* TODO: custom allocator */
193 static struct work_item *alloc_node(void)
194 {
195 return malloc(sizeof(struct work_item));
196 }
198 static void free_node(struct work_item *node)
199 {
200 free(node);
201 }
203 /* The following highly platform-specific code detects the number
204 * of processors available in the system. It's used by the thread pool
205 * to autodetect how many threads to spawn.
206 * Currently works on: Linux, BSD, Darwin, and Windows.
207 */
209 #if defined(__APPLE__) && defined(__MACH__)
210 # ifndef __unix__
211 # define __unix__ 1
212 # endif /* unix */
213 # ifndef __bsd__
214 # define __bsd__ 1
215 # endif /* bsd */
216 #endif /* apple */
218 #if defined(unix) || defined(__unix__)
219 #include <unistd.h>
221 # ifdef __bsd__
222 # include <sys/sysctl.h>
223 # endif
224 #endif
226 #if defined(WIN32) || defined(__WIN32__)
227 #include <windows.h>
228 #endif
231 static int get_processor_count(void)
232 {
233 #if defined(unix) || defined(__unix__)
234 # if defined(__bsd__)
235 /* BSD systems provide the num.processors through sysctl */
236 int num, mib[] = {CTL_HW, HW_NCPU};
237 size_t len = sizeof num;
239 sysctl(mib, 2, &num, &len, 0, 0);
240 return num;
242 # elif defined(__sgi)
243 /* SGI IRIX flavour of the _SC_NPROC_ONLN sysconf */
244 return sysconf(_SC_NPROC_ONLN);
245 # else
246 /* Linux (and others?) have the _SC_NPROCESSORS_ONLN sysconf */
247 return sysconf(_SC_NPROCESSORS_ONLN);
248 # endif /* bsd/sgi/other */
250 #elif defined(WIN32) || defined(__WIN32__)
251 /* under windows we need to call GetSystemInfo */
252 SYSTEM_INFO info;
253 GetSystemInfo(&info);
254 return info.dwNumberOfProcessors;
255 #endif
256 }