ThreadPool.c revision cf3a6383a9bc9a94ca5b424ea97313293ee0dcb0
1/* 2 * Copyright (C) 2010 The Android Open Source Project 3 * 4 * Licensed under the Apache License, Version 2.0 (the "License"); 5 * you may not use this file except in compliance with the License. 6 * You may obtain a copy of the License at 7 * 8 * http://www.apache.org/licenses/LICENSE-2.0 9 * 10 * Unless required by applicable law or agreed to in writing, software 11 * distributed under the License is distributed on an "AS IS" BASIS, 12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 * See the License for the specific language governing permissions and 14 * limitations under the License. 15 */ 16 17/* ThreadPool */ 18 19#include "sles_allinclusive.h" 20 21// Entry point for each worker thread 22 23static void *ThreadPool_start(void *context) 24{ 25 ThreadPool *tp = (ThreadPool *) context; 26 assert(NULL != tp); 27 for (;;) { 28 Closure *pClosure = ThreadPool_remove(tp); 29 // closure is NULL when thread pool is being destroyed 30 if (NULL == pClosure) { 31 break; 32 } 33 // make a copy of parameters, then free the parameters 34 const Closure closure = *pClosure; 35 free(pClosure); 36 // extract parameters and call the right method depending on kind 37 ClosureKind kind = closure.mKind; 38 void *context1 = closure.mContext1; 39 void *context2 = closure.mContext2; 40 int parameter1 = closure.mParameter1; 41 switch (kind) { 42 case CLOSURE_KIND_PPI: 43 { 44 ClosureHandler_ppi handler_ppi; 45 handler_ppi = closure.mHandler.mHandler_ppi; 46 assert(NULL != handler_ppi); 47 (*handler_ppi)(context1, context2, parameter1); 48 } 49 break; 50 case CLOSURE_KIND_PPII: 51 { 52 ClosureHandler_ppii handler_ppii; 53 handler_ppii = closure.mHandler.mHandler_ppii; 54 assert(NULL != handler_ppii); 55 int parameter2 = closure.mParameter2; 56 (*handler_ppii)(context1, context2, parameter1, parameter2); 57 } 58 break; 59 default: 60 SL_LOGE("Unexpected callback kind %d", kind); 61 assert(false); 62 break; 63 } 64 } 65 return NULL; 66} 67 68#define INITIALIZED_NONE 0 69#define INITIALIZED_MUTEX 1 70#define INITIALIZED_CONDNOTFULL 2 71#define INITIALIZED_CONDNOTEMPTY 4 72#define INITIALIZED_ALL 7 73 74static void ThreadPool_deinit_internal(ThreadPool *tp, unsigned initialized, unsigned nThreads); 75 76// Initialize a ThreadPool 77// maxClosures defaults to CLOSURE_TYPICAL if 0 78// maxThreads defaults to THREAD_TYPICAL if 0 79 80SLresult ThreadPool_init(ThreadPool *tp, unsigned maxClosures, unsigned maxThreads) 81{ 82 assert(NULL != tp); 83 memset(tp, 0, sizeof(ThreadPool)); 84 tp->mShutdown = SL_BOOLEAN_FALSE; 85 unsigned initialized = INITIALIZED_NONE; // which objects were successfully initialized 86 unsigned nThreads = 0; // number of threads successfully created 87 int err; 88 SLresult result; 89 90 // initialize mutex and condition variables 91 err = pthread_mutex_init(&tp->mMutex, (const pthread_mutexattr_t *) NULL); 92 result = err_to_result(err); 93 if (SL_RESULT_SUCCESS != result) 94 goto fail; 95 initialized |= INITIALIZED_MUTEX; 96 err = pthread_cond_init(&tp->mCondNotFull, (const pthread_condattr_t *) NULL); 97 result = err_to_result(err); 98 if (SL_RESULT_SUCCESS != result) 99 goto fail; 100 initialized |= INITIALIZED_CONDNOTFULL; 101 err = pthread_cond_init(&tp->mCondNotEmpty, (const pthread_condattr_t *) NULL); 102 result = err_to_result(err); 103 if (SL_RESULT_SUCCESS != result) 104 goto fail; 105 initialized |= INITIALIZED_CONDNOTEMPTY; 106 107 // use default values for parameters, if not specified explicitly 108 tp->mWaitingNotFull = 0; 109 tp->mWaitingNotEmpty = 0; 110 if (0 == maxClosures) 111 maxClosures = CLOSURE_TYPICAL; 112 tp->mMaxClosures = maxClosures; 113 if (0 == maxThreads) 114 maxThreads = THREAD_TYPICAL; 115 tp->mMaxThreads = maxThreads; 116 117 // initialize circular buffer for closures 118 if (CLOSURE_TYPICAL >= maxClosures) { 119 tp->mClosureArray = tp->mClosureTypical; 120 } else { 121 tp->mClosureArray = (Closure **) malloc((maxClosures + 1) * sizeof(Closure *)); 122 if (NULL == tp->mClosureArray) { 123 result = SL_RESULT_RESOURCE_ERROR; 124 goto fail; 125 } 126 } 127 tp->mClosureFront = tp->mClosureArray; 128 tp->mClosureRear = tp->mClosureArray; 129 130 // initialize thread pool 131 if (THREAD_TYPICAL >= maxThreads) { 132 tp->mThreadArray = tp->mThreadTypical; 133 } else { 134 tp->mThreadArray = (pthread_t *) malloc(maxThreads * sizeof(pthread_t)); 135 if (NULL == tp->mThreadArray) { 136 result = SL_RESULT_RESOURCE_ERROR; 137 goto fail; 138 } 139 } 140 unsigned i; 141 for (i = 0; i < maxThreads; ++i) { 142 int err = pthread_create(&tp->mThreadArray[i], (const pthread_attr_t *) NULL, 143 ThreadPool_start, tp); 144 result = err_to_result(err); 145 if (SL_RESULT_SUCCESS != result) 146 goto fail; 147 ++nThreads; 148 } 149 tp->mInitialized = initialized; 150 151 // done 152 return SL_RESULT_SUCCESS; 153 154 // here on any kind of error 155fail: 156 ThreadPool_deinit_internal(tp, initialized, nThreads); 157 return result; 158} 159 160static void ThreadPool_deinit_internal(ThreadPool *tp, unsigned initialized, unsigned nThreads) 161{ 162 int ok; 163 164 assert(NULL != tp); 165 // Destroy all threads 166 if (0 < nThreads) { 167 assert(INITIALIZED_ALL == initialized); 168 ok = pthread_mutex_lock(&tp->mMutex); 169 assert(0 == ok); 170 tp->mShutdown = SL_BOOLEAN_TRUE; 171 ok = pthread_cond_broadcast(&tp->mCondNotEmpty); 172 assert(0 == ok); 173 ok = pthread_cond_broadcast(&tp->mCondNotFull); 174 assert(0 == ok); 175 ok = pthread_mutex_unlock(&tp->mMutex); 176 assert(0 == ok); 177 unsigned i; 178 for (i = 0; i < nThreads; ++i) { 179 ok = pthread_join(tp->mThreadArray[i], (void **) NULL); 180 assert(ok == 0); 181 } 182 183 // Empty out the circular buffer of closures 184 ok = pthread_mutex_lock(&tp->mMutex); 185 assert(0 == ok); 186 assert(0 == tp->mWaitingNotEmpty); 187 Closure **oldFront = tp->mClosureFront; 188 while (oldFront != tp->mClosureRear) { 189 Closure **newFront = oldFront; 190 if (++newFront == &tp->mClosureArray[tp->mMaxClosures + 1]) 191 newFront = tp->mClosureArray; 192 Closure *pClosure = *oldFront; 193 assert(NULL != pClosure); 194 *oldFront = NULL; 195 tp->mClosureFront = newFront; 196 ok = pthread_mutex_unlock(&tp->mMutex); 197 assert(0 == ok); 198 free(pClosure); 199 ok = pthread_mutex_lock(&tp->mMutex); 200 assert(0 == ok); 201 } 202 ok = pthread_mutex_unlock(&tp->mMutex); 203 assert(0 == ok); 204 // Note that we can't be sure when mWaitingNotFull will drop to zero 205 } 206 207 // destroy the mutex and condition variables 208 if (initialized & INITIALIZED_CONDNOTEMPTY) { 209 ok = pthread_cond_destroy(&tp->mCondNotEmpty); 210 assert(0 == ok); 211 } 212 if (initialized & INITIALIZED_CONDNOTFULL) { 213 ok = pthread_cond_destroy(&tp->mCondNotFull); 214 assert(0 == ok); 215 } 216 if (initialized & INITIALIZED_MUTEX) { 217 ok = pthread_mutex_destroy(&tp->mMutex); 218 assert(0 == ok); 219 } 220 tp->mInitialized = INITIALIZED_NONE; 221 222 // release the closure circular buffer 223 if (tp->mClosureTypical != tp->mClosureArray && NULL != tp->mClosureArray) { 224 free(tp->mClosureArray); 225 tp->mClosureArray = NULL; 226 } 227 228 // release the thread pool 229 if (tp->mThreadTypical != tp->mThreadArray && NULL != tp->mThreadArray) { 230 free(tp->mThreadArray); 231 tp->mThreadArray = NULL; 232 } 233 234} 235 236void ThreadPool_deinit(ThreadPool *tp) 237{ 238 ThreadPool_deinit_internal(tp, tp->mInitialized, tp->mMaxThreads); 239} 240 241// Enqueue a closure to be executed later by a worker thread. 242// Note that this raw interface requires an explicit "kind" and full parameter list. 243// There are convenience methods below that make this easier to use. 244SLresult ThreadPool_add(ThreadPool *tp, ClosureKind kind, ClosureHandler_ppii handler, 245 void *context1, void *context2, int parameter1, int parameter2) 246{ 247 assert(NULL != tp); 248 assert(NULL != handler); 249 Closure *closure = (Closure *) malloc(sizeof(Closure)); 250 if (NULL == closure) { 251 return SL_RESULT_RESOURCE_ERROR; 252 } 253 closure->mKind = kind; 254 // note this will automagically assign to mHandler_ppi also 255 closure->mHandler.mHandler_ppii = handler; 256 closure->mContext1 = context1; 257 closure->mContext2 = context2; 258 closure->mParameter1 = parameter1; 259 closure->mParameter2 = parameter2; 260 int ok; 261 ok = pthread_mutex_lock(&tp->mMutex); 262 assert(0 == ok); 263 // can't enqueue while thread pool shutting down 264 if (tp->mShutdown) { 265 ok = pthread_mutex_unlock(&tp->mMutex); 266 assert(0 == ok); 267 free(closure); 268 return SL_RESULT_PRECONDITIONS_VIOLATED; 269 } 270 for (;;) { 271 Closure **oldRear = tp->mClosureRear; 272 Closure **newRear = oldRear; 273 if (++newRear == &tp->mClosureArray[tp->mMaxClosures + 1]) 274 newRear = tp->mClosureArray; 275 // if closure circular buffer is full, then wait for it to become non-full 276 if (newRear == tp->mClosureFront) { 277 ++tp->mWaitingNotFull; 278 ok = pthread_cond_wait(&tp->mCondNotFull, &tp->mMutex); 279 assert(0 == ok); 280 // can't enqueue while thread pool shutting down 281 if (tp->mShutdown) { 282 assert(0 < tp->mWaitingNotFull); 283 --tp->mWaitingNotFull; 284 ok = pthread_mutex_unlock(&tp->mMutex); 285 assert(0 == ok); 286 free(closure); 287 return SL_RESULT_PRECONDITIONS_VIOLATED; 288 } 289 continue; 290 } 291 assert(NULL == *oldRear); 292 *oldRear = closure; 293 tp->mClosureRear = newRear; 294 // if a worker thread was waiting to dequeue, then suggest that it try again 295 if (0 < tp->mWaitingNotEmpty) { 296 --tp->mWaitingNotEmpty; 297 ok = pthread_cond_signal(&tp->mCondNotEmpty); 298 assert(0 == ok); 299 } 300 break; 301 } 302 ok = pthread_mutex_unlock(&tp->mMutex); 303 assert(0 == ok); 304 return SL_RESULT_SUCCESS; 305} 306 307// Called by a worker thread when it is ready to accept the next closure to execute 308Closure *ThreadPool_remove(ThreadPool *tp) 309{ 310 Closure *pClosure; 311 int ok; 312 ok = pthread_mutex_lock(&tp->mMutex); 313 assert(0 == ok); 314 for (;;) { 315 Closure **oldFront = tp->mClosureFront; 316 // if closure circular buffer is empty, then wait for it to become non-empty 317 if (oldFront == tp->mClosureRear) { 318 ++tp->mWaitingNotEmpty; 319 ok = pthread_cond_wait(&tp->mCondNotEmpty, &tp->mMutex); 320 assert(0 == ok); 321 // fail if thread pool is shutting down 322 if (tp->mShutdown) { 323 assert(0 < tp->mWaitingNotEmpty); 324 --tp->mWaitingNotEmpty; 325 pClosure = NULL; 326 break; 327 } 328 // try again 329 continue; 330 } 331 // dequeue the closure at front of circular buffer 332 Closure **newFront = oldFront; 333 if (++newFront == &tp->mClosureArray[tp->mMaxClosures + 1]) 334 newFront = tp->mClosureArray; 335 pClosure = *oldFront; 336 assert(NULL != pClosure); 337 *oldFront = NULL; 338 tp->mClosureFront = newFront; 339 // if a client thread was waiting to enqueue, then suggest that it try again 340 if (0 < tp->mWaitingNotFull) { 341 --tp->mWaitingNotFull; 342 ok = pthread_cond_signal(&tp->mCondNotFull); 343 assert(0 == ok); 344 } 345 break; 346 } 347 ok = pthread_mutex_unlock(&tp->mMutex); 348 assert(0 == ok); 349 return pClosure; 350} 351 352// Convenience methods for applications 353SLresult ThreadPool_add_ppi(ThreadPool *tp, ClosureHandler_ppi handler, 354 void *context1, void *context2, int parameter1) 355{ 356 // function pointers are the same size so this is a safe cast 357 return ThreadPool_add(tp, CLOSURE_KIND_PPI, (ClosureHandler_ppii) handler, 358 context1, context2, parameter1, 0); 359} 360 361SLresult ThreadPool_add_ppii(ThreadPool *tp, ClosureHandler_ppii handler, 362 void *context1, void *context2, int parameter1, int parameter2) 363{ 364 // note that no cast is needed for handler because it is already the right type 365 return ThreadPool_add(tp, CLOSURE_KIND_PPII, handler, context1, context2, parameter1, 366 parameter2); 367} 368