libresman

view src/threadpool.c @ 10:4d18498a0078

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