blob: 6573fc3b7743bac08eb09eaa28f81b95bf93d546 [file] [log] [blame]
Marat Dukhan0a312192015-08-22 17:46:29 -04001/* Standard C headers */
2#include <stdint.h>
3#include <stdbool.h>
Marat Dukhan3a45d9a2015-08-23 22:25:19 -04004#include <stdlib.h>
Marat Dukhan0a312192015-08-22 17:46:29 -04005#include <string.h>
6#include <assert.h>
7
8/* POSIX headers */
9#include <pthread.h>
10#include <unistd.h>
11
Marat Dukhan1325d6e2016-07-03 13:13:16 -040012/* Dependencies */
13#include <fxdiv.h>
14
Marat Dukhan0a312192015-08-22 17:46:29 -040015/* Library header */
16#include <pthreadpool.h>
17
18#define PTHREADPOOL_CACHELINE_SIZE 64
19#define PTHREADPOOL_CACHELINE_ALIGNED __attribute__((__aligned__(PTHREADPOOL_CACHELINE_SIZE)))
Marat Dukhanaf6468b2015-08-25 12:16:57 -040020
Marat Dukhana04943a2015-08-25 12:41:05 -040021#if defined(__clang__)
22 #if __has_extension(c_static_assert) || __has_feature(c_static_assert)
23 #define PTHREADPOOL_STATIC_ASSERT(predicate, message) _Static_assert((predicate), message)
24 #else
25 #define PTHREADPOOL_STATIC_ASSERT(predicate, message)
26 #endif
27#elif defined(__GNUC__) && ((__GNUC__ > 4) || (__GNUC__ == 4) && (__GNUC_MINOR__ >= 6))
28 /* Static assert is supported by gcc >= 4.6 */
Marat Dukhanaf6468b2015-08-25 12:16:57 -040029 #define PTHREADPOOL_STATIC_ASSERT(predicate, message) _Static_assert((predicate), message)
30#else
Marat Dukhanaf6468b2015-08-25 12:16:57 -040031 #define PTHREADPOOL_STATIC_ASSERT(predicate, message)
32#endif
Marat Dukhan0a312192015-08-22 17:46:29 -040033
Marat Dukhanad0ca6a2015-10-16 03:15:19 -040034static inline size_t multiply_divide(size_t a, size_t b, size_t d) {
35 #if defined(__SIZEOF_SIZE_T__) && (__SIZEOF_SIZE_T__ == 4)
36 return (size_t) (((uint64_t) a) * ((uint64_t) b)) / ((uint64_t) d);
37 #elif defined(__SIZEOF_SIZE_T__) && (__SIZEOF_SIZE_T__ == 8)
38 return (size_t) (((__uint128_t) a) * ((__uint128_t) b)) / ((__uint128_t) d);
39 #else
40 #error "Unsupported platform"
41 #endif
42}
43
44static inline size_t divide_round_up(size_t dividend, size_t divisor) {
45 if (dividend % divisor == 0) {
46 return dividend / divisor;
47 } else {
48 return dividend / divisor + 1;
49 }
50}
51
52static inline size_t min(size_t a, size_t b) {
53 return a < b ? a : b;
54}
55
Marat Dukhan0a312192015-08-22 17:46:29 -040056enum thread_state {
57 thread_state_idle,
58 thread_state_compute_1d,
59 thread_state_shutdown,
60};
61
62struct PTHREADPOOL_CACHELINE_ALIGNED thread_info {
63 /**
64 * Index of the first element in the work range.
65 * Before processing a new element the owning worker thread increments this value.
66 */
67 volatile size_t range_start;
68 /**
69 * Index of the element after the last element of the work range.
70 * Before processing a new element the stealing worker thread decrements this value.
71 */
72 volatile size_t range_end;
73 /**
74 * The number of elements in the work range.
75 * Due to race conditions range_length <= range_end - range_start.
76 * The owning worker thread must decrement this value before incrementing @a range_start.
77 * The stealing worker thread must decrement this value before decrementing @a range_end.
78 */
79 volatile size_t range_length;
80 /**
81 * The active state of the thread.
82 */
83 volatile enum thread_state state;
84 /**
85 * Thread number in the 0..threads_count-1 range.
86 */
87 size_t thread_number;
88 /**
89 * The pthread object corresponding to the thread.
90 */
91 pthread_t thread_object;
92 /**
93 * Condition variable used to wake up the thread.
94 * When the thread is idle, it waits on this condition variable.
95 */
96 pthread_cond_t wakeup_condvar;
97};
98
99PTHREADPOOL_STATIC_ASSERT(sizeof(struct thread_info) % PTHREADPOOL_CACHELINE_SIZE == 0, "thread_info structure must occupy an integer number of cache lines (64 bytes)");
100
101struct PTHREADPOOL_CACHELINE_ALIGNED pthreadpool {
102 /**
103 * The number of threads that signalled completion of an operation.
104 */
105 volatile size_t checkedin_threads;
106 /**
107 * The function to call for each item.
108 */
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400109 volatile void* function;
Marat Dukhan0a312192015-08-22 17:46:29 -0400110 /**
111 * The first argument to the item processing function.
112 */
113 void *volatile argument;
114 /**
115 * Serializes concurrent calls to @a pthreadpool_compute_* from different threads.
116 */
117 pthread_mutex_t execution_mutex;
118 /**
119 * Guards access to the @a checkedin_threads variable.
120 */
121 pthread_mutex_t barrier_mutex;
122 /**
123 * Condition variable to wait until all threads check in.
124 */
125 pthread_cond_t barrier_condvar;
126 /**
127 * Guards access to the @a state variables.
128 */
129 pthread_mutex_t state_mutex;
130 /**
131 * Condition variable to wait for change of @a state variable.
132 */
133 pthread_cond_t state_condvar;
134 /**
135 * The number of threads in the thread pool. Never changes after initialization.
136 */
137 size_t threads_count;
138 /**
139 * Thread information structures that immediately follow this structure.
140 */
141 struct thread_info threads[];
142};
143
144PTHREADPOOL_STATIC_ASSERT(sizeof(struct pthreadpool) % PTHREADPOOL_CACHELINE_SIZE == 0, "pthreadpool structure must occupy an integer number of cache lines (64 bytes)");
145
146static void checkin_worker_thread(struct pthreadpool* threadpool) {
147 pthread_mutex_lock(&threadpool->barrier_mutex);
148 const size_t checkedin_threads = threadpool->checkedin_threads + 1;
149 threadpool->checkedin_threads = checkedin_threads;
150 if (checkedin_threads == threadpool->threads_count) {
151 pthread_cond_signal(&threadpool->barrier_condvar);
152 }
153 pthread_mutex_unlock(&threadpool->barrier_mutex);
154}
155
156static void wait_worker_threads(struct pthreadpool* threadpool) {
157 if (threadpool->checkedin_threads != threadpool->threads_count) {
158 pthread_mutex_lock(&threadpool->barrier_mutex);
159 while (threadpool->checkedin_threads != threadpool->threads_count) {
160 pthread_cond_wait(&threadpool->barrier_condvar, &threadpool->barrier_mutex);
161 };
162 pthread_mutex_unlock(&threadpool->barrier_mutex);
163 }
164}
165
166static void wakeup_worker_threads(struct pthreadpool* threadpool) {
167 pthread_mutex_lock(&threadpool->state_mutex);
168 threadpool->checkedin_threads = 0; /* Locking of barrier_mutex not needed: readers are sleeping */
169 pthread_cond_broadcast(&threadpool->state_condvar);
170 pthread_mutex_unlock(&threadpool->state_mutex); /* Do wake up */
171}
172
173inline static bool atomic_decrement(volatile size_t* value) {
174 size_t actual_value = *value;
175 if (actual_value != 0) {
176 size_t expected_value;
177 do {
178 expected_value = actual_value;
179 const size_t new_value = actual_value - 1;
180 actual_value = __sync_val_compare_and_swap(value, expected_value, new_value);
181 } while ((actual_value != expected_value) && (actual_value != 0));
182 }
183 return actual_value != 0;
184}
185
186static void thread_compute_1d(struct pthreadpool* threadpool, struct thread_info* thread) {
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400187 const pthreadpool_function_1d_t function = (pthreadpool_function_1d_t) threadpool->function;
Marat Dukhan0a312192015-08-22 17:46:29 -0400188 void *const argument = threadpool->argument;
189 /* Process thread's own range of items */
190 size_t range_start = thread->range_start;
191 while (atomic_decrement(&thread->range_length)) {
192 function(argument, range_start++);
193 }
194 /* Done, now look for other threads' items to steal */
195 const size_t thread_number = thread->thread_number;
196 const size_t threads_count = threadpool->threads_count;
197 for (size_t tid = (thread_number + 1) % threads_count; tid != thread_number; tid = (tid + 1) % threads_count) {
198 struct thread_info* other_thread = &threadpool->threads[tid];
199 if (other_thread->state != thread_state_idle) {
200 while (atomic_decrement(&other_thread->range_length)) {
201 const size_t item_id = __sync_sub_and_fetch(&other_thread->range_end, 1);
202 function(argument, item_id);
203 }
204 }
205 }
206}
207
208static void* thread_main(void* arg) {
209 struct thread_info* thread = (struct thread_info*) arg;
210 struct pthreadpool* threadpool = ((struct pthreadpool*) (thread - thread->thread_number)) - 1;
211
212 /* Check in */
213 checkin_worker_thread(threadpool);
214
215 /* Monitor the state changes and act accordingly */
216 for (;;) {
217 /* Lock the state mutex */
218 pthread_mutex_lock(&threadpool->state_mutex);
219 /* Read the state */
220 enum thread_state state;
221 while ((state = thread->state) == thread_state_idle) {
222 /* Wait for state change */
223 pthread_cond_wait(&threadpool->state_condvar, &threadpool->state_mutex);
224 }
225 /* Read non-idle state */
226 pthread_mutex_unlock(&threadpool->state_mutex);
227 switch (state) {
228 case thread_state_compute_1d:
229 thread_compute_1d(threadpool, thread);
230 break;
231 case thread_state_shutdown:
232 return NULL;
233 case thread_state_idle:
234 /* To inhibit compiler warning */
235 break;
236 }
237 /* Notify the master thread that we finished processing */
238 thread->state = thread_state_idle;
239 checkin_worker_thread(threadpool);
240 };
241}
242
243struct pthreadpool* pthreadpool_create(size_t threads_count) {
244 if (threads_count == 0) {
245 threads_count = (size_t) sysconf(_SC_NPROCESSORS_ONLN);
246 }
Marat Dukhan3a45d9a2015-08-23 22:25:19 -0400247 struct pthreadpool* threadpool = NULL;
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400248 if (posix_memalign((void**) &threadpool, 64, sizeof(struct pthreadpool) + threads_count * sizeof(struct thread_info)) != 0) {
249 return NULL;
250 }
Marat Dukhan0a312192015-08-22 17:46:29 -0400251 memset(threadpool, 0, sizeof(struct pthreadpool) + threads_count * sizeof(struct thread_info));
252 threadpool->threads_count = threads_count;
253 pthread_mutex_init(&threadpool->execution_mutex, NULL);
254 pthread_mutex_init(&threadpool->barrier_mutex, NULL);
255 pthread_cond_init(&threadpool->barrier_condvar, NULL);
256 pthread_mutex_init(&threadpool->state_mutex, NULL);
257 pthread_cond_init(&threadpool->state_condvar, NULL);
258
259 for (size_t tid = 0; tid < threads_count; tid++) {
260 threadpool->threads[tid].thread_number = tid;
261 pthread_create(&threadpool->threads[tid].thread_object, NULL, &thread_main, &threadpool->threads[tid]);
262 }
263
264 /* Wait until all threads initialize */
265 wait_worker_threads(threadpool);
266 return threadpool;
267}
268
Marat Dukhan7b1f6e52015-08-25 11:24:08 -0400269size_t pthreadpool_get_threads_count(struct pthreadpool* threadpool) {
Marat Dukhan0a312192015-08-22 17:46:29 -0400270 return threadpool->threads_count;
271}
272
Marat Dukhan0a312192015-08-22 17:46:29 -0400273void pthreadpool_compute_1d(
274 struct pthreadpool* threadpool,
275 pthreadpool_function_1d_t function,
276 void* argument,
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400277 size_t range)
Marat Dukhan0a312192015-08-22 17:46:29 -0400278{
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400279 if (threadpool == NULL) {
280 /* No thread pool provided: execute function sequentially on the calling thread */
281 for (size_t i = 0; i < range; i++) {
282 function(argument, i);
283 }
284 } else {
285 /* Protect the global threadpool structures */
286 pthread_mutex_lock(&threadpool->execution_mutex);
Marat Dukhan0a312192015-08-22 17:46:29 -0400287
Marat Dukhanfa98b4b2015-11-06 18:19:31 -0500288 /* Lock the state variables to ensure that threads don't start processing before they observe complete state */
289 pthread_mutex_lock(&threadpool->state_mutex);
290
291 /* Setup global arguments */
292 threadpool->function = function;
293 threadpool->argument = argument;
294
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400295 /* Spread the work between threads */
296 for (size_t tid = 0; tid < threadpool->threads_count; tid++) {
297 struct thread_info* thread = &threadpool->threads[tid];
298 thread->range_start = multiply_divide(range, tid, threadpool->threads_count);
299 thread->range_end = multiply_divide(range, tid + 1, threadpool->threads_count);
300 thread->range_length = thread->range_end - thread->range_start;
301 thread->state = thread_state_compute_1d;
302 }
303
Marat Dukhanfa98b4b2015-11-06 18:19:31 -0500304 /* Unlock the state variables before waking up the threads for better performance */
305 pthread_mutex_unlock(&threadpool->state_mutex);
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400306
307 /* Wake up the threads */
308 wakeup_worker_threads(threadpool);
309
310 /* Wait until the threads finish computation */
311 wait_worker_threads(threadpool);
312
313 /* Unprotect the global threadpool structures */
314 pthread_mutex_unlock(&threadpool->execution_mutex);
Marat Dukhan0a312192015-08-22 17:46:29 -0400315 }
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400316}
Marat Dukhan0a312192015-08-22 17:46:29 -0400317
Marat Dukhane76282f2015-11-02 17:47:04 -0500318struct compute_1d_tiled_context {
319 pthreadpool_function_1d_tiled_t function;
320 void* argument;
321 size_t range;
322 size_t tile;
323};
324
325static void compute_1d_tiled(const struct compute_1d_tiled_context* context, size_t linear_index) {
326 const size_t tile_index = linear_index;
327 const size_t index = tile_index * context->tile;
328 const size_t tile = min(context->tile, context->range - index);
329 context->function(context->argument, index, tile);
330}
331
332void pthreadpool_compute_1d_tiled(
333 pthreadpool_t threadpool,
334 pthreadpool_function_1d_tiled_t function,
335 void* argument,
336 size_t range,
337 size_t tile)
338{
339 const size_t tile_range = divide_round_up(range, tile);
340 struct compute_1d_tiled_context context = {
341 .function = function,
342 .argument = argument,
343 .range = range,
344 .tile = tile
345 };
346 pthreadpool_compute_1d(threadpool, (pthreadpool_function_1d_t) compute_1d_tiled, &context, tile_range);
347}
348
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400349struct compute_2d_context {
350 pthreadpool_function_2d_t function;
351 void* argument;
Marat Dukhan1325d6e2016-07-03 13:13:16 -0400352 struct fxdiv_divisor_size_t range_j;
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400353};
Marat Dukhan0a312192015-08-22 17:46:29 -0400354
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400355static void compute_2d(const struct compute_2d_context* context, size_t linear_index) {
Marat Dukhan1325d6e2016-07-03 13:13:16 -0400356 const struct fxdiv_divisor_size_t range_j = context->range_j;
357 const struct fxdiv_result_size_t index = fxdiv_divide_size_t(linear_index, range_j);
358 context->function(context->argument, index.quotient, index.remainder);
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400359}
Marat Dukhan0a312192015-08-22 17:46:29 -0400360
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400361void pthreadpool_compute_2d(
362 struct pthreadpool* threadpool,
363 pthreadpool_function_2d_t function,
364 void* argument,
365 size_t range_i,
366 size_t range_j)
367{
368 struct compute_2d_context context = {
369 .function = function,
370 .argument = argument,
Marat Dukhan1325d6e2016-07-03 13:13:16 -0400371 .range_j = fxdiv_init_size_t(range_j)
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400372 };
373 pthreadpool_compute_1d(threadpool, (pthreadpool_function_1d_t) compute_2d, &context, range_i * range_j);
374}
Marat Dukhan0a312192015-08-22 17:46:29 -0400375
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400376struct compute_2d_tiled_context {
377 pthreadpool_function_2d_tiled_t function;
378 void* argument;
Marat Dukhan1325d6e2016-07-03 13:13:16 -0400379 struct fxdiv_divisor_size_t tile_range_j;
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400380 size_t range_i;
381 size_t range_j;
382 size_t tile_i;
383 size_t tile_j;
384};
385
386static void compute_2d_tiled(const struct compute_2d_tiled_context* context, size_t linear_index) {
Marat Dukhan1325d6e2016-07-03 13:13:16 -0400387 const struct fxdiv_divisor_size_t tile_range_j = context->tile_range_j;
388 const struct fxdiv_result_size_t tile_index = fxdiv_divide_size_t(linear_index, tile_range_j);
389 const size_t max_tile_i = context->tile_i;
390 const size_t max_tile_j = context->tile_j;
391 const size_t index_i = tile_index.quotient * max_tile_i;
392 const size_t index_j = tile_index.remainder * max_tile_j;
393 const size_t tile_i = min(max_tile_i, context->range_i - index_i);
394 const size_t tile_j = min(max_tile_j, context->range_j - index_j);
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400395 context->function(context->argument, index_i, index_j, tile_i, tile_j);
396}
397
398void pthreadpool_compute_2d_tiled(
399 pthreadpool_t threadpool,
400 pthreadpool_function_2d_tiled_t function,
401 void* argument,
402 size_t range_i,
403 size_t range_j,
404 size_t tile_i,
405 size_t tile_j)
406{
407 const size_t tile_range_i = divide_round_up(range_i, tile_i);
408 const size_t tile_range_j = divide_round_up(range_j, tile_j);
409 struct compute_2d_tiled_context context = {
410 .function = function,
411 .argument = argument,
Marat Dukhan1325d6e2016-07-03 13:13:16 -0400412 .tile_range_j = fxdiv_init_size_t(tile_range_j),
Marat Dukhanad0ca6a2015-10-16 03:15:19 -0400413 .range_i = range_i,
414 .range_j = range_j,
415 .tile_i = tile_i,
416 .tile_j = tile_j
417 };
418 pthreadpool_compute_1d(threadpool, (pthreadpool_function_1d_t) compute_2d_tiled, &context, tile_range_i * tile_range_j);
Marat Dukhan0a312192015-08-22 17:46:29 -0400419}
420
421void pthreadpool_destroy(struct pthreadpool* threadpool) {
422 /* Update threads' states */
423 for (size_t tid = 0; tid < threadpool->threads_count; tid++) {
424 threadpool->threads[tid].state = thread_state_shutdown;
425 }
426
427 /* Wake up the threads */
428 wakeup_worker_threads(threadpool);
429
430 /* Wait until all threads return */
431 for (size_t tid = 0; tid < threadpool->threads_count; tid++) {
432 pthread_join(threadpool->threads[tid].thread_object, NULL);
433 }
434
435 /* Release resources */
436 pthread_mutex_destroy(&threadpool->execution_mutex);
437 pthread_mutex_destroy(&threadpool->barrier_mutex);
438 pthread_cond_destroy(&threadpool->barrier_condvar);
439 pthread_mutex_destroy(&threadpool->state_mutex);
440 pthread_cond_destroy(&threadpool->state_condvar);
441 free(threadpool);
442}