OLD | NEW |
(Empty) | |
| 1 // Copyright (c) 2013 The Chromium Authors. All rights reserved. |
| 2 // Use of this source code is governed by a BSD-style license that can be |
| 3 // found in the LICENSE file. |
| 4 |
| 5 #include "threadpool.h" |
| 6 |
| 7 #include <pthread.h> |
| 8 #include <semaphore.h> |
| 9 #include <stdio.h> |
| 10 #include <stdlib.h> |
| 11 |
| 12 // TODO(nfullagar): Switch DecCounter to use atomic decrement. |
| 13 |
| 14 // Initializes mutex, semaphores and a pool of threads. If 0 is passed for |
| 15 // num_threads, all work will be performed on the dispatch thread. |
| 16 ThreadPool::ThreadPool(int num_threads) |
| 17 : threads_(NULL), counter_(0), num_threads_(num_threads), exiting_(false), |
| 18 user_data_(NULL), user_work_function_(NULL) { |
| 19 if (num_threads_ > 0) { |
| 20 int status; |
| 21 status = pthread_mutex_init(&mutex_, NULL); |
| 22 if (0 != status) { |
| 23 fprintf(stderr, "Failed to initialize mutex!\n"); |
| 24 exit(-1); |
| 25 } |
| 26 status = sem_init(&work_sem_, 0, 0); |
| 27 if (-1 == status) { |
| 28 fprintf(stderr, "Failed to initialize semaphore!\n"); |
| 29 exit(-1); |
| 30 } |
| 31 status = sem_init(&done_sem_, 0, 0); |
| 32 if (-1 == status) { |
| 33 fprintf(stderr, "Failed to initialize semaphore!\n"); |
| 34 exit(-1); |
| 35 } |
| 36 threads_ = new pthread_t[num_threads_]; |
| 37 for (int i = 0; i < num_threads_; i++) { |
| 38 status = pthread_create(&threads_[i], NULL, WorkerThreadEntry, this); |
| 39 if (0 != status) { |
| 40 fprintf(stderr, "Failed to create thread!\n"); |
| 41 exit(-1); |
| 42 } |
| 43 } |
| 44 } |
| 45 } |
| 46 |
| 47 // Post exit request, wait for all threads to join, and cleanup. |
| 48 ThreadPool::~ThreadPool() { |
| 49 if (num_threads_ > 1) { |
| 50 PostExitAndJoinAll(); |
| 51 delete[] threads_; |
| 52 sem_destroy(&done_sem_); |
| 53 sem_destroy(&work_sem_); |
| 54 pthread_mutex_destroy(&mutex_); |
| 55 } |
| 56 } |
| 57 |
| 58 // Setup work parameters. This function is called from the dispatch thread, |
| 59 // when all worker threads are sleeping. |
| 60 void ThreadPool::Setup(int counter, WorkFunction work, void *data) { |
| 61 counter_ = counter; |
| 62 user_work_function_ = work; |
| 63 user_data_ = data; |
| 64 } |
| 65 |
| 66 // Decrement and get the value of the mutex protected counter. This function |
| 67 // can be called from multiple threads at any given time. |
| 68 int ThreadPool::DecCounter() { |
| 69 int v; |
| 70 pthread_mutex_lock(&mutex_); |
| 71 { |
| 72 v = --counter_; |
| 73 } |
| 74 pthread_mutex_unlock(&mutex_); |
| 75 return v; |
| 76 } |
| 77 |
| 78 // Set exit flag, post and join all the threads in the pool. This function is |
| 79 // called only from the dispatch thread, and only when all worker threads are |
| 80 // sleeping. |
| 81 void ThreadPool::PostExitAndJoinAll() { |
| 82 exiting_ = true; |
| 83 // Wake up all the sleeping worker threads. |
| 84 for (int i = 0; i < num_threads_; ++i) |
| 85 sem_post(&work_sem_); |
| 86 void* retval; |
| 87 for (int i = 0; i < num_threads_; ++i) |
| 88 pthread_join(threads_[i], &retval); |
| 89 } |
| 90 |
| 91 // Main work loop - one for each worker thread. |
| 92 void ThreadPool::WorkLoop() { |
| 93 while (true) { |
| 94 // Wait for work. If no work is availble, this thread will sleep here. |
| 95 sem_wait(&work_sem_); |
| 96 if (exiting_) break; |
| 97 while (true) { |
| 98 // Grab a task index to work on from the counter. |
| 99 int task_index = DecCounter(); |
| 100 if (task_index < 0) |
| 101 break; |
| 102 user_work_function_(task_index, user_data_); |
| 103 } |
| 104 // Post to dispatch thread work is done. |
| 105 sem_post(&done_sem_); |
| 106 } |
| 107 } |
| 108 |
| 109 // pthread entry point for a worker thread. |
| 110 void* ThreadPool::WorkerThreadEntry(void* thiz) { |
| 111 static_cast<ThreadPool*>(thiz)->WorkLoop(); |
| 112 return NULL; |
| 113 } |
| 114 |
| 115 // DispatchMany() will dispatch a set of tasks across worker threads. |
| 116 // Note: This function will block until all work has completed. |
| 117 void ThreadPool::DispatchMany(int num_tasks, WorkFunction work, void* data) { |
| 118 // On entry, all worker threads are sleeping. |
| 119 Setup(num_tasks, work, data); |
| 120 |
| 121 // Wake up the worker threads & have them process tasks. |
| 122 for (int i = 0; i < num_threads_; i++) |
| 123 sem_post(&work_sem_); |
| 124 |
| 125 // Worker threads are now awake and busy. |
| 126 |
| 127 // This dispatch thread will now sleep-wait for the worker threads to finish. |
| 128 for (int i = 0; i < num_threads_; i++) |
| 129 sem_wait(&done_sem_); |
| 130 // On exit, all tasks are done and all worker threads are sleeping again. |
| 131 } |
| 132 |
| 133 // DispatchHere will dispatch all tasks on this thread. |
| 134 void ThreadPool::DispatchHere(int num_tasks, WorkFunction work, void* data) { |
| 135 for (int i = 0; i < num_tasks; i++) |
| 136 work(i, data); |
| 137 } |
| 138 |
| 139 // Dispatch() will invoke the user supplied work function across |
| 140 // one or more threads for each task. |
| 141 // Note: This function will block until all work has completed. |
| 142 void ThreadPool::Dispatch(int num_tasks, WorkFunction work, void* data) { |
| 143 if (num_threads_ > 0) |
| 144 DispatchMany(num_tasks, work, data); |
| 145 else |
| 146 DispatchHere(num_tasks, work, data); |
| 147 } |
| 148 |
OLD | NEW |