/*
|
* Copyright (C) 2010 The Android Open Source Project
|
*
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
* you may not use this file except in compliance with the License.
|
* You may obtain a copy of the License at
|
*
|
* http://www.apache.org/licenses/LICENSE-2.0
|
*
|
* Unless required by applicable law or agreed to in writing, software
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
* See the License for the specific language governing permissions and
|
* limitations under the License.
|
*/
|
|
/* ThreadPool */
|
|
#include "sles_allinclusive.h"
|
|
// Entry point for each worker thread
|
|
static void *ThreadPool_start(void *context)
|
{
|
ThreadPool *tp = (ThreadPool *) context;
|
assert(NULL != tp);
|
for (;;) {
|
Closure *pClosure = ThreadPool_remove(tp);
|
// closure is NULL when thread pool is being destroyed
|
if (NULL == pClosure) {
|
break;
|
}
|
// make a copy of parameters, then free the parameters
|
const Closure closure = *pClosure;
|
free(pClosure);
|
// extract parameters and call the right method depending on kind
|
ClosureKind kind = closure.mKind;
|
void *context1 = closure.mContext1;
|
void *context2 = closure.mContext2;
|
int parameter1 = closure.mParameter1;
|
switch (kind) {
|
case CLOSURE_KIND_PPI:
|
{
|
ClosureHandler_ppi handler_ppi = closure.mHandler.mHandler_ppi;
|
assert(NULL != handler_ppi);
|
(*handler_ppi)(context1, context2, parameter1);
|
}
|
break;
|
case CLOSURE_KIND_PPII:
|
{
|
ClosureHandler_ppii handler_ppii = closure.mHandler.mHandler_ppii;
|
assert(NULL != handler_ppii);
|
int parameter2 = closure.mParameter2;
|
(*handler_ppii)(context1, context2, parameter1, parameter2);
|
}
|
break;
|
case CLOSURE_KIND_PIIPP:
|
{
|
ClosureHandler_piipp handler_piipp = closure.mHandler.mHandler_piipp;
|
assert(NULL != handler_piipp);
|
int parameter2 = closure.mParameter2;
|
void *context3 = closure.mContext3;
|
(*handler_piipp)(context1, parameter1, parameter2, context2, context3);
|
}
|
break;
|
default:
|
SL_LOGE("Unexpected callback kind %d", kind);
|
assert(false);
|
break;
|
}
|
}
|
return NULL;
|
}
|
|
#define INITIALIZED_NONE 0
|
#define INITIALIZED_MUTEX 1
|
#define INITIALIZED_CONDNOTFULL 2
|
#define INITIALIZED_CONDNOTEMPTY 4
|
#define INITIALIZED_ALL 7
|
|
static void ThreadPool_deinit_internal(ThreadPool *tp, unsigned initialized, unsigned nThreads);
|
|
// Initialize a ThreadPool
|
// maxClosures defaults to CLOSURE_TYPICAL if 0
|
// maxThreads defaults to THREAD_TYPICAL if 0
|
|
SLresult ThreadPool_init(ThreadPool *tp, unsigned maxClosures, unsigned maxThreads)
|
{
|
assert(NULL != tp);
|
memset(tp, 0, sizeof(ThreadPool));
|
tp->mShutdown = SL_BOOLEAN_FALSE;
|
unsigned initialized = INITIALIZED_NONE; // which objects were successfully initialized
|
unsigned nThreads = 0; // number of threads successfully created
|
int err;
|
SLresult result;
|
|
// initialize mutex and condition variables
|
err = pthread_mutex_init(&tp->mMutex, (const pthread_mutexattr_t *) NULL);
|
result = err_to_result(err);
|
if (SL_RESULT_SUCCESS != result)
|
goto fail;
|
initialized |= INITIALIZED_MUTEX;
|
err = pthread_cond_init(&tp->mCondNotFull, (const pthread_condattr_t *) NULL);
|
result = err_to_result(err);
|
if (SL_RESULT_SUCCESS != result)
|
goto fail;
|
initialized |= INITIALIZED_CONDNOTFULL;
|
err = pthread_cond_init(&tp->mCondNotEmpty, (const pthread_condattr_t *) NULL);
|
result = err_to_result(err);
|
if (SL_RESULT_SUCCESS != result)
|
goto fail;
|
initialized |= INITIALIZED_CONDNOTEMPTY;
|
|
// use default values for parameters, if not specified explicitly
|
tp->mWaitingNotFull = 0;
|
tp->mWaitingNotEmpty = 0;
|
if (0 == maxClosures)
|
maxClosures = CLOSURE_TYPICAL;
|
tp->mMaxClosures = maxClosures;
|
if (0 == maxThreads)
|
maxThreads = THREAD_TYPICAL;
|
tp->mMaxThreads = maxThreads;
|
|
// initialize circular buffer for closures
|
if (CLOSURE_TYPICAL >= maxClosures) {
|
tp->mClosureArray = tp->mClosureTypical;
|
} else {
|
tp->mClosureArray = (Closure **) malloc((maxClosures + 1) * sizeof(Closure *));
|
if (NULL == tp->mClosureArray) {
|
result = SL_RESULT_RESOURCE_ERROR;
|
goto fail;
|
}
|
}
|
tp->mClosureFront = tp->mClosureArray;
|
tp->mClosureRear = tp->mClosureArray;
|
|
// initialize thread pool
|
if (THREAD_TYPICAL >= maxThreads) {
|
tp->mThreadArray = tp->mThreadTypical;
|
} else {
|
tp->mThreadArray = (pthread_t *) malloc(maxThreads * sizeof(pthread_t));
|
if (NULL == tp->mThreadArray) {
|
result = SL_RESULT_RESOURCE_ERROR;
|
goto fail;
|
}
|
}
|
unsigned i;
|
for (i = 0; i < maxThreads; ++i) {
|
int err = pthread_create(&tp->mThreadArray[i], (const pthread_attr_t *) NULL,
|
ThreadPool_start, tp);
|
result = err_to_result(err);
|
if (SL_RESULT_SUCCESS != result)
|
goto fail;
|
++nThreads;
|
}
|
tp->mInitialized = initialized;
|
|
// done
|
return SL_RESULT_SUCCESS;
|
|
// here on any kind of error
|
fail:
|
ThreadPool_deinit_internal(tp, initialized, nThreads);
|
return result;
|
}
|
|
static void ThreadPool_deinit_internal(ThreadPool *tp, unsigned initialized, unsigned nThreads)
|
{
|
int ok;
|
|
assert(NULL != tp);
|
// Destroy all threads
|
if (0 < nThreads) {
|
assert(INITIALIZED_ALL == initialized);
|
ok = pthread_mutex_lock(&tp->mMutex);
|
assert(0 == ok);
|
tp->mShutdown = SL_BOOLEAN_TRUE;
|
ok = pthread_cond_broadcast(&tp->mCondNotEmpty);
|
assert(0 == ok);
|
ok = pthread_cond_broadcast(&tp->mCondNotFull);
|
assert(0 == ok);
|
ok = pthread_mutex_unlock(&tp->mMutex);
|
assert(0 == ok);
|
unsigned i;
|
for (i = 0; i < nThreads; ++i) {
|
ok = pthread_join(tp->mThreadArray[i], (void **) NULL);
|
assert(ok == 0);
|
}
|
|
// Empty out the circular buffer of closures
|
ok = pthread_mutex_lock(&tp->mMutex);
|
assert(0 == ok);
|
Closure **oldFront = tp->mClosureFront;
|
while (oldFront != tp->mClosureRear) {
|
Closure **newFront = oldFront;
|
if (++newFront == &tp->mClosureArray[tp->mMaxClosures + 1])
|
newFront = tp->mClosureArray;
|
Closure *pClosure = *oldFront;
|
assert(NULL != pClosure);
|
*oldFront = NULL;
|
tp->mClosureFront = newFront;
|
ok = pthread_mutex_unlock(&tp->mMutex);
|
assert(0 == ok);
|
free(pClosure);
|
ok = pthread_mutex_lock(&tp->mMutex);
|
assert(0 == ok);
|
}
|
ok = pthread_mutex_unlock(&tp->mMutex);
|
assert(0 == ok);
|
// Note that we can't be sure when mWaitingNotFull will drop to zero
|
}
|
|
// destroy the mutex and condition variables
|
if (initialized & INITIALIZED_CONDNOTEMPTY) {
|
ok = pthread_cond_destroy(&tp->mCondNotEmpty);
|
assert(0 == ok);
|
}
|
if (initialized & INITIALIZED_CONDNOTFULL) {
|
ok = pthread_cond_destroy(&tp->mCondNotFull);
|
assert(0 == ok);
|
}
|
if (initialized & INITIALIZED_MUTEX) {
|
ok = pthread_mutex_destroy(&tp->mMutex);
|
assert(0 == ok);
|
}
|
tp->mInitialized = INITIALIZED_NONE;
|
|
// release the closure circular buffer
|
if (tp->mClosureTypical != tp->mClosureArray && NULL != tp->mClosureArray) {
|
free(tp->mClosureArray);
|
tp->mClosureArray = NULL;
|
}
|
|
// release the thread pool
|
if (tp->mThreadTypical != tp->mThreadArray && NULL != tp->mThreadArray) {
|
free(tp->mThreadArray);
|
tp->mThreadArray = NULL;
|
}
|
|
}
|
|
void ThreadPool_deinit(ThreadPool *tp)
|
{
|
ThreadPool_deinit_internal(tp, tp->mInitialized, tp->mMaxThreads);
|
}
|
|
// Enqueue a closure to be executed later by a worker thread.
|
// Note that this raw interface requires an explicit "kind" and full parameter list.
|
// There are convenience methods below that make this easier to use.
|
SLresult ThreadPool_add(ThreadPool *tp, ClosureKind kind, ClosureHandler_generic handler,
|
void *context1, void *context2, void *context3, int parameter1, int parameter2)
|
{
|
assert(NULL != tp);
|
assert(NULL != handler);
|
Closure *closure = (Closure *) malloc(sizeof(Closure));
|
if (NULL == closure) {
|
return SL_RESULT_RESOURCE_ERROR;
|
}
|
closure->mKind = kind;
|
switch (kind) {
|
case CLOSURE_KIND_PPI:
|
closure->mHandler.mHandler_ppi = (ClosureHandler_ppi)handler;
|
break;
|
case CLOSURE_KIND_PPII:
|
closure->mHandler.mHandler_ppii = (ClosureHandler_ppii)handler;
|
break;
|
case CLOSURE_KIND_PIIPP:
|
closure->mHandler.mHandler_piipp = (ClosureHandler_piipp)handler;
|
break;
|
default:
|
SL_LOGE("ThreadPool_add() invalid closure kind %d", kind);
|
assert(false);
|
}
|
closure->mContext1 = context1;
|
closure->mContext2 = context2;
|
closure->mContext3 = context3;
|
closure->mParameter1 = parameter1;
|
closure->mParameter2 = parameter2;
|
int ok;
|
ok = pthread_mutex_lock(&tp->mMutex);
|
assert(0 == ok);
|
// can't enqueue while thread pool shutting down
|
if (tp->mShutdown) {
|
ok = pthread_mutex_unlock(&tp->mMutex);
|
assert(0 == ok);
|
free(closure);
|
return SL_RESULT_PRECONDITIONS_VIOLATED;
|
}
|
for (;;) {
|
Closure **oldRear = tp->mClosureRear;
|
Closure **newRear = oldRear;
|
if (++newRear == &tp->mClosureArray[tp->mMaxClosures + 1])
|
newRear = tp->mClosureArray;
|
// if closure circular buffer is full, then wait for it to become non-full
|
if (newRear == tp->mClosureFront) {
|
++tp->mWaitingNotFull;
|
ok = pthread_cond_wait(&tp->mCondNotFull, &tp->mMutex);
|
assert(0 == ok);
|
// can't enqueue while thread pool shutting down
|
if (tp->mShutdown) {
|
assert(0 < tp->mWaitingNotFull);
|
--tp->mWaitingNotFull;
|
ok = pthread_mutex_unlock(&tp->mMutex);
|
assert(0 == ok);
|
free(closure);
|
return SL_RESULT_PRECONDITIONS_VIOLATED;
|
}
|
continue;
|
}
|
assert(NULL == *oldRear);
|
*oldRear = closure;
|
tp->mClosureRear = newRear;
|
// if a worker thread was waiting to dequeue, then suggest that it try again
|
if (0 < tp->mWaitingNotEmpty) {
|
--tp->mWaitingNotEmpty;
|
ok = pthread_cond_signal(&tp->mCondNotEmpty);
|
assert(0 == ok);
|
}
|
break;
|
}
|
ok = pthread_mutex_unlock(&tp->mMutex);
|
assert(0 == ok);
|
return SL_RESULT_SUCCESS;
|
}
|
|
// Called by a worker thread when it is ready to accept the next closure to execute
|
Closure *ThreadPool_remove(ThreadPool *tp)
|
{
|
Closure *pClosure;
|
int ok;
|
ok = pthread_mutex_lock(&tp->mMutex);
|
assert(0 == ok);
|
for (;;) {
|
// fail if thread pool is shutting down
|
if (tp->mShutdown) {
|
pClosure = NULL;
|
break;
|
}
|
Closure **oldFront = tp->mClosureFront;
|
// if closure circular buffer is empty, then wait for it to become non-empty
|
if (oldFront == tp->mClosureRear) {
|
++tp->mWaitingNotEmpty;
|
ok = pthread_cond_wait(&tp->mCondNotEmpty, &tp->mMutex);
|
assert(0 == ok);
|
// try again
|
continue;
|
}
|
// dequeue the closure at front of circular buffer
|
Closure **newFront = oldFront;
|
if (++newFront == &tp->mClosureArray[tp->mMaxClosures + 1]) {
|
newFront = tp->mClosureArray;
|
}
|
pClosure = *oldFront;
|
assert(NULL != pClosure);
|
*oldFront = NULL;
|
tp->mClosureFront = newFront;
|
// if a client thread was waiting to enqueue, then suggest that it try again
|
if (0 < tp->mWaitingNotFull) {
|
--tp->mWaitingNotFull;
|
ok = pthread_cond_signal(&tp->mCondNotFull);
|
assert(0 == ok);
|
}
|
break;
|
}
|
ok = pthread_mutex_unlock(&tp->mMutex);
|
assert(0 == ok);
|
return pClosure;
|
}
|
|
// Convenience methods for applications
|
SLresult ThreadPool_add_ppi(ThreadPool *tp, ClosureHandler_ppi handler,
|
void *context1, void *context2, int parameter1)
|
{
|
// function pointers are the same size so this is a safe cast
|
return ThreadPool_add(tp, CLOSURE_KIND_PPI, (ClosureHandler_generic) handler,
|
context1, context2, NULL, parameter1, 0);
|
}
|
|
SLresult ThreadPool_add_ppii(ThreadPool *tp, ClosureHandler_ppii handler,
|
void *context1, void *context2, int parameter1, int parameter2)
|
{
|
// function pointers are the same size so this is a safe cast
|
return ThreadPool_add(tp, CLOSURE_KIND_PPII, (ClosureHandler_generic) handler,
|
context1, context2, NULL, parameter1, parameter2);
|
}
|
|
SLresult ThreadPool_add_piipp(ThreadPool *tp, ClosureHandler_piipp handler,
|
void *cntxt1, int param1, int param2, void *cntxt2, void *cntxt3)
|
{
|
// function pointers are the same size so this is a safe cast
|
return ThreadPool_add(tp, CLOSURE_KIND_PIIPP, (ClosureHandler_generic) handler,
|
cntxt1, cntxt2, cntxt3, param1, param2);
|
}
|
