libresman

view src/threadpool.c @ 11:bebc065a941f

find changesets by author, revision, files, or words in the commit message
doesn't work yet
author John Tsiombikas <nuclear@member.fsf.org>
date Fri, 07 Feb 2014 07:50:02 +0200
parents 4d18498a0078
children 84f55eab27cb
line source
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <string.h>
4 #include <pthread.h>
5 #include "threadpool.h"
6
7 struct work_item {
8 void *data;
9 struct work_item *next;
10 };
11
12 struct thread_pool {
13 pthread_t *workers;
14 int num_workers;
15
16 pthread_mutex_t work_lock;
17 pthread_cond_t work_cond;
18
19 tpool_work_func work_func;
20 void *cls;
21
22 struct work_item *work_list, *work_list_tail;
23 int work_count;
24 };
25
26
27 static void *thread_func(void *tp);
28 static struct work_item *alloc_node(void);
29 static void free_node(struct work_item *node);
30 static int get_processor_count(void);
31
32
33
34 int tpool_init(struct thread_pool *tpool, int num_threads)
35 {
36 int i;
37
38 memset(tpool, 0, sizeof *tpool);
39
40
41 pthread_mutex_init(&tpool->work_lock, 0);
42 pthread_cond_init(&tpool->work_cond, 0);
43
44
45 if(num_threads <= 0) {
46 num_threads = get_processor_count();
47 }
48 tpool->num_workers = num_threads;
49
50 printf("initializing thread pool with %d worker threads\n", num_threads);
51
52 if(!(tpool->workers = malloc(num_threads * sizeof *tpool->workers))) {
53 fprintf(stderr, "failed to create array of %d threads\n", num_threads);
54 return -1;
55 }
56
57 for(i=0; i<num_threads; i++) {
58 if(pthread_create(tpool->workers + i, 0, thread_func, tpool) == -1) {
59 fprintf(stderr, "%s: failed to create thread %d\n", __FUNCTION__, i);
60 tpool_destroy(tpool);
61 return -1;
62 }
63 }
64 return 0;
65 }
66
67 void tpool_destroy(struct thread_pool *tpool)
68 {
69 int i;
70 for(i=0; i<tpool->num_workers; i++) {
71 void *ret;
72 pthread_join(tpool->workers[i], &ret);
73 }
74
75 pthread_mutex_destroy(&tpool->work_lock);
76 pthread_cond_destroy(&tpool->work_cond);
77 }
78
79 struct thread_pool *tpool_create(int num_threads)
80 {
81 struct thread_pool *tpool = malloc(sizeof *tpool);
82 if(!tpool) {
83 return 0;
84 }
85 if(tpool_init(tpool, num_threads) == -1) {
86 free(tpool);
87 return 0;
88 }
89 return tpool;
90 }
91
92 void tpool_free(struct thread_pool *tpool)
93 {
94 if(tpool) {
95 tpool_destroy(tpool);
96 free(tpool);
97 }
98 }
99
100 void tpool_set_work_func(struct thread_pool *tpool, tpool_work_func func, void *cls)
101 {
102 tpool->work_func = func;
103 tpool->cls = cls;
104 }
105
106 int tpool_add_work(struct thread_pool *tpool, void *data)
107 {
108 struct work_item *node;
109
110 if(!(node = alloc_node())) {
111 fprintf(stderr, "%s: failed to allocate new work item node\n", __FUNCTION__);
112 return -1;
113 }
114 node->data = data;
115 node->next = 0;
116
117 pthread_mutex_lock(&tpool->work_lock);
118
119 if(!tpool->work_list) {
120 tpool->work_list = tpool->work_list_tail = node;
121 } else {
122 tpool->work_list_tail->next = node;
123 tpool->work_list_tail = node;
124 }
125
126 pthread_mutex_unlock(&tpool->work_lock);
127 return 0;
128 }
129
130
131 static void *thread_func(void *tp)
132 {
133 struct work_item *job;
134 struct thread_pool *tpool = tp;
135
136 pthread_mutex_lock(&tpool->work_lock);
137 for(;;) {
138 /* while there aren't any work items to do go to sleep on the condvar */
139 pthread_cond_wait(&tpool->work_cond, &tpool->work_lock);
140 if(!tpool->work_list) {
141 continue; /* spurious wakeup, go back to sleep */
142 }
143
144 job = tpool->work_list;
145 tpool->work_list = tpool->work_list->next;
146
147 tpool->work_func(job->data, tpool->cls);
148
149 free_node(job);
150 }
151 pthread_mutex_unlock(&tpool->work_lock);
152 return 0;
153 }
154
155 /* TODO: custom allocator */
156 static struct work_item *alloc_node(void)
157 {
158 return malloc(sizeof(struct work_item));
159 }
160
161 static void free_node(struct work_item *node)
162 {
163 free(node);
164 }
165
166 /* The following highly platform-specific code detects the number
167 * of processors available in the system. It's used by the thread pool
168 * to autodetect how many threads to spawn.
169 * Currently works on: Linux, BSD, Darwin, and Windows.
170 */
171
172 #if defined(__APPLE__) && defined(__MACH__)
173 # ifndef __unix__
174 # define __unix__ 1
175 # endif /* unix */
176 # ifndef __bsd__
177 # define __bsd__ 1
178 # endif /* bsd */
179 #endif /* apple */
180
181 #if defined(unix) || defined(__unix__)
182 #include <unistd.h>
183
184 # ifdef __bsd__
185 # include <sys/sysctl.h>
186 # endif
187 #endif
188
189 #if defined(WIN32) || defined(__WIN32__)
190 #include <windows.h>
191 #endif
192
193
194 static int get_processor_count(void)
195 {
196 #if defined(unix) || defined(__unix__)
197 # if defined(__bsd__)
198 /* BSD systems provide the num.processors through sysctl */
199 int num, mib[] = {CTL_HW, HW_NCPU};
200 size_t len = sizeof num;
201
202 sysctl(mib, 2, &num, &len, 0, 0);
203 return num;
204
205 # elif defined(__sgi)
206 /* SGI IRIX flavour of the _SC_NPROC_ONLN sysconf */
207 return sysconf(_SC_NPROC_ONLN);
208 # else
209 /* Linux (and others?) have the _SC_NPROCESSORS_ONLN sysconf */
210 return sysconf(_SC_NPROCESSORS_ONLN);
211 # endif /* bsd/sgi/other */
212
213 #elif defined(WIN32) || defined(__WIN32__)
214 /* under windows we need to call GetSystemInfo */
215 SYSTEM_INFO info;
216 GetSystemInfo(&info);
217 return info.dwNumberOfProcessors;
218 #endif
219 }