kenmirrors/asterisk
1
0
Fork 0
mirror of https://github.com/asterisk/asterisk.git synced 2025-09-19 11:42:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add some doxygen and rearrange code.

Browse Source 
git-svn-id: https://origsvn.digium.com/svn/asterisk/team/mmichelson/threadpool@377209 65c4cc65-6c06-0410-ace0-fbb531ad65f3
This commit is contained in:
Mark Michelson
2012-12-04 21:11:34 +00:00
parent 47a9abdeda
commit a37fb2e8c8
2 changed files with 551 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

689
main/threadpool.c
View File

@@ -26,85 +26,116 @@
#define THREAD_BUCKETS 89
static int id_counter;
/*!
* \brief An opaque threadpool structure
*
* A threadpool is a collection of threads that execute
* tasks from a common queue.
*/
struct ast_threadpool {
/*! Threadpool listener */
struct ast_threadpool_listener *listener;
/*!
* \brief The container of active threads.
* Active threads are those that are currently running tasks
*/
struct ao2_container *active_threads;
/*!
* \brief The container of idle threads.
* Idle threads are those that are currenly waiting to run tasks
*/
struct ao2_container *idle_threads;
/*!
* \brief The container of zombie threads.
* Zombie threads may be running tasks, but they are scheduled to die soon
*/
struct ao2_container *zombie_threads;
/*!
* \brief The main taskprocessor
*
* Tasks that are queued in this taskprocessor are
* doled out to the worker threads. Worker threads that
* execute tasks from the threadpool are executing tasks
* in this taskprocessor.
*
* The threadpool itself is actually the private data for
* this taskprocessor's listener. This way, as taskprocessor
* changes occur, the threadpool can alert its listeners
* appropriately.
*/
struct ast_taskprocessor *tps;
/*!
* \brief The control taskprocessor
*
* This is a standard taskprocessor that uses the default
* taskprocessor listener. In other words, all tasks queued to
* this taskprocessor have a single thread that executes the
* tasks.
*
* All tasks that modify the state of the threadpool and all tasks
* that call out to threadpool listeners are pushed to this
* taskprocessor.
*
* For instance, when the threadpool changes sizes, a task is put
* into this taskprocessor to do so. When it comes time to tell the
* threadpool listener that worker threads have changed state,
* the task is placed in this taskprocessor.
*
* This is done for three main reasons
* 1) It ensures that listeners are given an accurate portrayal
* of the threadpool's current state. In other words, when a listener
* gets told a count of active, idle and zombie threads, it does not
* need to worry that internal state of the threadpool might be different
* from what it has been told.
* 2) It minimizes the locking required in both the threadpool and in
* threadpool listener's callbacks.
* 3) It ensures that listener callbacks are called in the same order
* that the threadpool had its state change.
*/
struct ast_taskprocessor *control_tps;
};
/*!
* \brief states for worker threads
*/
enum worker_state {
/*! The worker is either active or idle */
ALIVE,
/*!
* The worker has been asked to shut down but
* may still be in the process of executing tasks.
* This transition happens when the threadpool needs
* to shrink and needs to kill active threads in order
* to do so.
*/
ZOMBIE,
/*!
* The worker has been asked to shut down. Typically
* only idle threads go to this state directly, but
* active threads may go straight to this state when
* the threadpool is shut down.
*/
DEAD,
};
struct worker_thread {
int id;
ast_cond_t cond;
ast_mutex_t lock;
pthread_t thread;
struct ast_threadpool *pool;
enum worker_state state;
int wake_up;
};
static int worker_thread_hash(const void *obj, int flags)
{
const struct worker_thread *worker = obj;
return worker->id;
}
static int worker_thread_cmp(void *obj, void *arg, int flags)
{
struct worker_thread *worker1 = obj;
struct worker_thread *worker2 = arg;
return worker1->id == worker2->id ? CMP_MATCH : 0;
}
static void worker_thread_destroy(void *obj)
{
struct worker_thread *worker = obj;
ast_mutex_destroy(&worker->lock);
ast_cond_destroy(&worker->cond);
}
static int worker_active(struct worker_thread *worker);
static void *worker_start(void *arg)
{
struct worker_thread *worker = arg;
worker_active(worker);
return NULL;
}
static struct worker_thread *worker_thread_alloc(struct ast_threadpool *pool)
{
struct worker_thread *worker = ao2_alloc(sizeof(*worker), worker_thread_destroy);
if (!worker) {
return NULL;
}
worker->id = ast_atomic_fetchadd_int(&id_counter, 1);
ast_mutex_init(&worker->lock);
ast_cond_init(&worker->cond, NULL);
worker->pool = pool;
worker->thread = AST_PTHREADT_NULL;
worker->state = ALIVE;
if (ast_pthread_create(&worker->thread, NULL, worker_start, worker) < 0) {
ast_log(LOG_ERROR, "Unable to start worker thread!\n");
ao2_ref(worker, -1);
return NULL;
}
return worker;
}
/* Worker thread forward declarations. See definitions for documentation */
struct worker_thread;
static int worker_thread_hash(const void *obj, int flags);
static int worker_thread_cmp(void *obj, void *arg, int flags);
static void worker_thread_destroy(void *obj);
static void worker_active(struct worker_thread *worker);
static void *worker_start(void *arg);
static struct worker_thread *worker_thread_alloc(struct ast_threadpool *pool);
static int worker_idle(struct worker_thread *worker);
static void worker_set_state(struct worker_thread *worker, enum worker_state state);
static int worker_shutdown(void *obj, void *arg, int flags);
/*!
* \brief Notify the threadpool listener that the state has changed.
*
* This notifies the threadpool listener via its state_changed callback.
* \param pool The threadpool whose state has changed
*/
static void threadpool_send_state_changed(struct ast_threadpool *pool)
{
int active_size = ao2_container_count(pool->active_threads);
@@ -114,11 +145,19 @@ static void threadpool_send_state_changed(struct ast_threadpool *pool)
pool->listener->callbacks->state_changed(pool->listener, active_size, idle_size, zombie_size);
}
/*!
* \brief Struct used for queued operations involving worker state changes
*/
struct thread_worker_pair {
/*! Threadpool that contains the worker whose state has changed */
struct ast_threadpool *pool;
/*! Worker whose state has changed */
struct worker_thread *worker;
};
/*!
* \brief Destructor for thread_worker_pair
*/
static void thread_worker_pair_destructor(void *obj)
{
struct thread_worker_pair *pair = obj;
@@ -126,6 +165,11 @@ static void thread_worker_pair_destructor(void *obj)
ao2_ref(pair->worker, -1);
}
/*!
* \brief Allocate and initialize a thread_worker_pair
* \param pool Threadpool to assign to the thread_worker_pair
* \param worker Worker thread to assign to the thread_worker_pair
*/
static struct thread_worker_pair *thread_worker_pair_alloc(struct ast_threadpool *pool,
struct worker_thread *worker)
{
@@ -140,6 +184,13 @@ static struct thread_worker_pair *thread_worker_pair_alloc(struct ast_threadpool
return pair;
}
/*!
* \brief Move a worker thread from the active container to the idle container.
*
* This function is called from the threadpool's control taskprocessor thread.
* \param data A thread_worker_pair containing the threadpool and the worker to move.
* \return 0
*/
static int queued_active_thread_idle(void *data)
{
struct thread_worker_pair *pair = data;
@@ -153,6 +204,14 @@ static int queued_active_thread_idle(void *data)
return 0;
}
/*!
* \brief Queue a task to move a thread from the active list to the idle list
*
* This is called by a worker thread when it runs out of tasks to perform and
* goes idle.
* \param pool The threadpool to which the worker belongs
* \param worker The worker thread that has gone idle
*/
static void threadpool_active_thread_idle(struct ast_threadpool *pool,
struct worker_thread *worker)
{
@@ -163,17 +222,31 @@ static void threadpool_active_thread_idle(struct ast_threadpool *pool,
ast_taskprocessor_push(pool->control_tps, queued_active_thread_idle, pair);
}
/*!
* \brief Kill a zombie thread
*
* This runs from the threadpool's control taskprocessor thread.
*
* \param data A thread_worker_pair containing the threadpool and the zombie thread
* \return 0
*/
static int queued_zombie_thread_dead(void *data)
{
struct thread_worker_pair *pair = data;
ao2_unlink(pair->pool->zombie_threads, pair->worker);
ao2_unlink(pair->pool, pair->worker);
threadpool_send_state_changed(pair->pool);
ao2_ref(pair, -1);
return 0;
}
/*!
* \brief Queue a task to kill a zombie thread
*
* This is called by a worker thread when it acknowledges that it is time for
* it to die.
*/
static void threadpool_zombie_thread_dead(struct ast_threadpool *pool,
struct worker_thread *worker)
{
@@ -184,83 +257,56 @@ static void threadpool_zombie_thread_dead(struct ast_threadpool *pool,
ast_taskprocessor_push(pool->control_tps, queued_zombie_thread_dead, pair);
}
static int worker_idle(struct worker_thread *worker)
{
SCOPED_MUTEX(lock, &worker->lock);
if (worker->state != ALIVE) {
return 0;
}
threadpool_active_thread_idle(worker->pool, worker);
while (!worker->wake_up) {
ast_cond_wait(&worker->cond, lock);
}
worker->wake_up = 0;
return worker->state == ALIVE;
}
/*!
* \brief Execute a task in the threadpool
*
* This is the function that worker threads call in order to execute tasks
* in the threadpool
*
* \param pool The pool to which the tasks belong.
* \retval 0 Either the pool has been shut down or there are no tasks.
* \retval 1 There are still tasks remaining in the pool.
*/
static int threadpool_execute(struct ast_threadpool *pool)
{
return ast_taskprocessor_execute(pool->tps);
}
static int worker_active(struct worker_thread *worker)
/*!
* \brief Destroy a threadpool's components.
*
* This is the destructor called automatically when the threadpool's
* reference count reaches zero. This is not to be confused with
* threadpool_destroy.
*
* By the time this actually gets called, most of the cleanup has already
* been done in the pool. The only thing left to do is to release the
* final reference to the threadpool listener.
*
* \param obj The pool to destroy
*/
static void threadpool_destructor(void *obj)
{
int alive = 1;
while (alive) {
if (threadpool_execute(worker->pool)) {
alive = worker_idle(worker);
}
}
/* Reaching this portion means the thread is
* on death's door. It may have been killed while
* it was idle, in which case it can just die
* peacefully. If it's a zombie, though, then
* it needs to let the pool know so
* that the thread can be removed from the
* list of zombie threads.
*/
if (worker->state == ZOMBIE) {
threadpool_zombie_thread_dead(worker->pool, worker);
}
return 0;
}
static void worker_set_state(struct worker_thread *worker, enum worker_state state)
{
SCOPED_MUTEX(lock, &worker->lock);
worker->state = state;
worker->wake_up = 1;
ast_cond_signal(&worker->cond);
}
static int worker_shutdown(void *obj, void *arg, int flags)
{
struct worker_thread *worker = obj;
worker_set_state(worker, DEAD);
if (worker->thread != AST_PTHREADT_NULL) {
pthread_join(worker->thread, NULL);
worker->thread = AST_PTHREADT_NULL;
}
return 0;
}
static void threadpool_destructor(void *private_data)
{
struct ast_threadpool *pool = private_data;
struct ast_threadpool *pool = obj;
/* XXX Probably should let the listener know we're being destroyed? */
/* Threads should all be shut down by now, so this should be a painless
* operation
*/
ao2_cleanup(pool->active_threads);
ao2_cleanup(pool->idle_threads);
ao2_cleanup(pool->zombie_threads);
ao2_cleanup(pool->listener);
}
/*
* \brief Allocate a threadpool
*
* This is implemented as a taskprocessor listener's alloc callback. This
* is because the threadpool exists as the private data on a taskprocessor
* listener.
*
* \param listener The taskprocessor listener where the threadpool will live.
* \retval NULL Could not initialize threadpool properly
* \retval non-NULL The newly-allocated threadpool
*/
static void *threadpool_alloc(struct ast_taskprocessor_listener *listener)
{
RAII_VAR(struct ast_threadpool *, pool,
@@ -289,17 +335,32 @@ static void *threadpool_alloc(struct ast_taskprocessor_listener *listener)
return pool;
}
/*!
* \brief helper used for queued task when tasks are pushed
*/
struct task_pushed_data {
/*! Pool into which a task was pushed */
struct ast_threadpool *pool;
/*! Indicator of whether the pool had no tasks prior to the new task being added */
int was_empty;
};
/*!
* \brief Destructor for task_pushed_data
*/
static void task_pushed_data_destroy(void *obj)
{
struct task_pushed_data *tpd = obj;
ao2_ref(tpd->pool, -1);
}
/*!
* \brief Allocate and initialize a task_pushed_data
* \param pool The threadpool to set in the task_pushed_data
* \param was_empty The was_empty value to set in the task_pushed_data
* \retval NULL Unable to allocate task_pushed_data
* \retval non-NULL The newly-allocated task_pushed_data
*/
static struct task_pushed_data *task_pushed_data_alloc(struct ast_threadpool *pool,
int was_empty)
{
@@ -315,6 +376,18 @@ static struct task_pushed_data *task_pushed_data_alloc(struct ast_threadpool *po
return tpd;
}
/*!
* \brief Activate idle threads
*
* This function always returns CMP_MATCH because all threads that this
* function acts on need to be seen as matches so they are unlinked from the
* list of idle threads.
*
* Called as an ao2_callback in the threadpool's control taskprocessor thread.
* \param obj The worker to activate
* \param arg The pool where the worker belongs
* \retval CMP_MATCH
*/
static int activate_threads(void *obj, void *arg, int flags)
{
struct worker_thread *worker = obj;
@@ -322,9 +395,18 @@ static int activate_threads(void *obj, void *arg, int flags)
ao2_link(pool->active_threads, worker);
worker_set_state(worker, ALIVE);
return 0;
return CMP_MATCH;
}
/*!
* \brief Queue task called when tasks are pushed into the threadpool
*
* This function first calls into the threadpool's listener to let it know
* that a task has been pushed. It then wakes up all idle threads and moves
* them into the active thread container.
* \param data A task_pushed_data
* \return 0
*/
static int handle_task_pushed(void *data)
{
struct task_pushed_data *tpd = data;
@@ -332,11 +414,21 @@ static int handle_task_pushed(void *data)
int was_empty = tpd->was_empty;
pool->listener->callbacks->tps_task_pushed(pool->listener, was_empty);
ao2_callback(pool->idle_threads, OBJ_UNLINK, activate_threads, pool);
ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
activate_threads, pool);
ao2_ref(tpd, -1);
return 0;
}
/*!
* \brief Taskprocessor listener callback called when a task is added
*
* The threadpool uses this opportunity to queue a task on its control taskprocessor
* in order to activate idle threads and notify the threadpool listener that the
* task has been pushed.
* \param listener The taskprocessor listener. The threadpool is the listener's private data
* \param was_empty True if the taskprocessor was empty prior to the task being pushed
*/
static void threadpool_tps_task_pushed(struct ast_taskprocessor_listener *listener,
int was_empty)
{
@@ -350,6 +442,13 @@ static void threadpool_tps_task_pushed(struct ast_taskprocessor_listener *listen
ast_taskprocessor_push(pool->control_tps, handle_task_pushed, tpd);
}
/*!
* \brief Queued task that handles the case where the threadpool's taskprocessor is emptied
*
* This simply lets the threadpool's listener know that the threadpool is devoid of tasks
* \param data The pool that has become empty
* \return 0
*/
static int handle_emptied(void *data)
{
struct ast_threadpool *pool = data;
@@ -359,6 +458,13 @@ static int handle_emptied(void *data)
return 0;
}
/*!
* \brief Taskprocessor listener emptied callback
*
* The threadpool queues a task to let the threadpool listener know that
* the threadpool no longer contains any tasks.
* \param listener The taskprocessor listener. The threadpool is the listener's private data.
*/
static void threadpool_tps_emptied(struct ast_taskprocessor_listener *listener)
{
struct ast_threadpool *pool = listener->private_data;
@@ -367,34 +473,44 @@ static void threadpool_tps_emptied(struct ast_taskprocessor_listener *listener)
ast_taskprocessor_push(pool->control_tps, handle_emptied, pool);
}
/*!
* \brief Taskprocessor listener shutdown callback
*
* The threadpool will shut down and destroy all of its worker threads when
* this is called back. By the time this gets called, the taskprocessor's
* control taskprocessor has already been destroyed. Therefore there is no risk
* in outright destroying the worker threads here.
* \param listener The taskprocessor listener. The threadpool is the listener's private data.
*/
static void threadpool_tps_shutdown(struct ast_taskprocessor_listener *listener)
{
/*
* The threadpool triggers the taskprocessor to shut down. As a result,
* we have the freedom of shutting things down in three stages:
*
* 1) Before the tasprocessor is shut down
* 2) During taskprocessor shutdown (here)
* 3) After taskprocessor shutdown
*
* In the spirit of the taskprocessor shutdown, this would be
* where we make sure that all the worker threads are no longer
* executing. We could just do this before we even shut down
* the taskprocessor, but this feels more "right".
*/
struct ast_threadpool *pool = listener->private_data;
ao2_callback(pool->active_threads, 0, worker_shutdown, NULL);
ao2_callback(pool->idle_threads, 0, worker_shutdown, NULL);
ao2_callback(pool->zombie_threads, 0, worker_shutdown, NULL);
int flags = (OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE);
ao2_cleanup(pool->active_threads);
ao2_cleanup(pool->idle_threads);
ao2_cleanup(pool->zombie_threads);
}
/*!
* \brief Taskprocessor listener destroy callback
*
* Since the threadpool is an ao2 object, all that is necessary is to
* decrease the refcount. Since the control taskprocessor should already
* be destroyed by this point, this should be the final reference to the
* threadpool.
*
* \param private_data The threadpool to destroy
*/
static void threadpool_destroy(void *private_data)
{
struct ast_threadpool *pool = private_data;
ao2_cleanup(pool);
}
/*!
* \brief Table of taskprocessor listener callbacks for threadpool's main taskprocessor
*/
static struct ast_taskprocessor_listener_callbacks threadpool_tps_listener_callbacks = {
.alloc = threadpool_alloc,
.task_pushed = threadpool_tps_task_pushed,
@@ -403,6 +519,13 @@ static struct ast_taskprocessor_listener_callbacks threadpool_tps_listener_callb
.destroy = threadpool_destroy,
};
/*!
* \brief Add threads to the threadpool
*
* This function is called from the threadpool's control taskprocessor thread.
* \param pool The pool that is expanding
* \delta The number of threads to add to the pool
*/
static void grow(struct ast_threadpool *pool, int delta)
{
int i;
@@ -415,19 +538,49 @@ static void grow(struct ast_threadpool *pool, int delta)
}
}
/*!
* \brief ao2 callback to kill a set number of threads.
*
* Threads will be unlinked from the container as long as the
* counter has not reached zero. The counter is decremented with
* each thread that is removed.
* \param obj The worker thread up for possible destruction
* \param arg The counter
* \param flags Unused
* \retval CMP_MATCH The counter has not reached zero, so this flag should be removed.
* \retval CMP_STOP The counter has reached zero so no more threads should be removed.
*/
static int kill_threads(void *obj, void *arg, int flags)
{
struct worker_thread *worker = obj;
int *num_to_kill = arg;
if ((*num_to_kill)-- > 0) {
worker_shutdown(worker, arg, flags);
return CMP_MATCH;
} else {
return CMP_STOP;
}
}
/*!
* \brief ao2 callback to zombify a set number of threads.
*
* Threads will be zombified as long as as the counter has not reached
* zero. The counter is decremented with each thread that is zombified.
*
* Zombifying a thread involves removing it from its current container,
* adding it to the zombie container, and changing the state of the
* worker to a zombie
*
* This callback is called from the threadpool control taskprocessor thread.
*
* \param obj The worker thread that may be zombified
* \param arg The pool to which the worker belongs
* \param data The counter
* \param flags Unused
* \retval CMP_MATCH The zombified thread should be removed from its current container
* \retval CMP_STOP Stop attempting to zombify threads
*/
static int zombify_threads(void *obj, void *arg, void *data, int flags)
{
struct worker_thread *worker = obj;
@@ -443,6 +596,18 @@ static int zombify_threads(void *obj, void *arg, void *data, int flags)
}
}
/*!
* \brief Remove threads from the threadpool
*
* The preference is to kill idle threads. However, if there are
* more threads to remove than there are idle threads, then active
* threads will be zombified instead.
*
* This function is called from the threadpool control taskprocessor thread.
*
* \param pool The threadpool to remove threads from
* \param delta The number of threads to remove
*/
static void shrink(struct ast_threadpool *pool, int delta)
{
/*
@@ -454,24 +619,38 @@ static void shrink(struct ast_threadpool *pool, int delta)
int idle_threads_to_kill = MIN(delta, idle_threads);
int active_threads_to_zombify = delta - idle_threads_to_kill;
ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE | OBJ_NOLOCK,
ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK,
kill_threads, &idle_threads_to_kill);
ao2_callback_data(pool->active_threads, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE | OBJ_NOLOCK,
ao2_callback_data(pool->active_threads, OBJ_UNLINK | OBJ_NOLOCK,
zombify_threads, pool, &active_threads_to_zombify);
}
/*!
* \brief Helper struct used for queued operations that change the size of the threadpool
*/
struct set_size_data {
/*! The pool whose size is to change */
struct ast_threadpool *pool;
/*! The requested new size of the pool */
unsigned int size;
};
/*!
* \brief Destructor for set_size_data
* \param obj The set_size_data to destroy
*/
static void set_size_data_destroy(void *obj)
{
struct set_size_data *ssd = obj;
ao2_ref(ssd->pool, -1);
}
/*!
* \brief Allocate and initialize a set_size_data
* \param pool The pool for the set_size_data
* \param size The size to store in the set_size_data
*/
static struct set_size_data *set_size_data_alloc(struct ast_threadpool *pool,
unsigned int size)
{
@@ -486,6 +665,17 @@ static struct set_size_data *set_size_data_alloc(struct ast_threadpool *pool,
return ssd;
}
/*!
* \brief Change the size of the threadpool
*
* This can either result in shrinking or growing the threadpool depending
* on the new desired size and the current size.
*
* This function is run from the threadpool control taskprocessor thread
*
* \param data A set_size_data used for determining how to act
* \return 0
*/
static int queued_set_size(void *data)
{
struct set_size_data *ssd = data;
@@ -548,12 +738,209 @@ struct ast_threadpool *ast_threadpool_create(struct ast_threadpool_listener *lis
return pool;
}
int ast_threadpool_push(struct ast_threadpool *pool, int (*task)(void *data), void *data)
{
return ast_taskprocessor_push(pool->tps, task, data);
}
void ast_threadpool_shutdown(struct ast_threadpool *pool)
{
/* Pretty simple really. We just shut down the
* taskprocessors and everything else just
/* Shut down the taskprocessors and everything else just
* takes care of itself via the taskprocessor callbacks
*/
ast_taskprocessor_unreference(pool->control_tps);
ast_taskprocessor_unreference(pool->tps);
}
/*!
* A thread that executes threadpool tasks
*/
struct worker_thread {
/*! A unique (within a run of Asterisk) ID for the thread. Used for hashing and searching */
int id;
/*! Condition used in conjunction with state changes */
ast_cond_t cond;
/*! Lock used alongside the condition for state changes */
ast_mutex_t lock;
/*! The actual thread that is executing tasks */
pthread_t thread;
/*! A pointer to the threadpool. Needed to be able to execute tasks */
struct ast_threadpool *pool;
/*! The current state of the worker thread */
enum worker_state state;
/*! A boolean used to determine if an idle thread should become active */
int wake_up;
};
/*!
* A monotonically increasing integer used for worker
* thread identification.
*/
static int worker_id_counter;
static int worker_thread_hash(const void *obj, int flags)
{
const struct worker_thread *worker = obj;
return worker->id;
}
static int worker_thread_cmp(void *obj, void *arg, int flags)
{
struct worker_thread *worker1 = obj;
struct worker_thread *worker2 = arg;
return worker1->id == worker2->id ? CMP_MATCH : 0;
}
/*!
* \brief shut a worker thread down
*
* Set the worker dead and then wait for its thread
* to finish executing.
*
* \param worker The worker thread to shut down
*/
static void worker_shutdown(struct worker_thread *worker)
{
struct worker_thread *worker = obj;
worker_set_state(worker, DEAD);
if (worker->thread != AST_PTHREADT_NULL) {
pthread_join(worker->thread, NULL);
worker->thread = AST_PTHREADT_NULL;
}
}
/*!
* \brief Worker thread destructor
*
* Called automatically when refcount reaches 0. Shuts
* down the worker thread and destroys its component
* parts
*/
static void worker_thread_destroy(void *obj)
{
struct worker_thread *worker = obj;
worker_shutdown(worker);
ast_mutex_destroy(&worker->lock);
ast_cond_destroy(&worker->cond);
}
/*!
* \brief start point for worker threads
*
* Worker threads start in the active state but may
* immediately go idle if there is no work to be
* done
*
* \param arg The worker thread
* \retval NULL
*/
static void *worker_start(void *arg)
{
struct worker_thread *worker = arg;
worker_active(worker);
return NULL;
}
/*!
* \brief Allocate and initialize a new worker thread
*
* This will create, initialize, and start the thread.
*
* \param pool The threadpool to which the worker will be added
* \retval NULL Failed to allocate or start the worker thread
* \retval non-NULL The newly-created worker thread
*/
static struct worker_thread *worker_thread_alloc(struct ast_threadpool *pool)
{
struct worker_thread *worker = ao2_alloc(sizeof(*worker), worker_thread_destroy);
if (!worker) {
return NULL;
}
worker->id = ast_atomic_fetchadd_int(&worker_id_counter, 1);
ast_mutex_init(&worker->lock);
ast_cond_init(&worker->cond, NULL);
worker->pool = pool;
worker->thread = AST_PTHREADT_NULL;
worker->state = ALIVE;
if (ast_pthread_create(&worker->thread, NULL, worker_start, worker) < 0) {
ast_log(LOG_ERROR, "Unable to start worker thread!\n");
ao2_ref(worker, -1);
return NULL;
}
return worker;
}
/*!
* \brief Active loop for worker threads
*
* The worker will stay in this loop for its lifetime,
* executing tasks as they become available. If there
* are no tasks currently available, then the thread
* will go idle.
*
* \param worker The worker thread executing tasks.
*/
static void worker_active(struct worker_thread *worker)
{
int alive = 1;
while (alive) {
if (threadpool_execute(worker->pool)) {
alive = worker_idle(worker);
}
}
/* Reaching this portion means the thread is
* on death's door. It may have been killed while
* it was idle, in which case it can just die
* peacefully. If it's a zombie, though, then
* it needs to let the pool know so
* that the thread can be removed from the
* list of zombie threads.
*/
if (worker->state == ZOMBIE) {
threadpool_zombie_thread_dead(worker->pool, worker);
}
}
/*!
* \brief Idle function for worker threads
*
* The worker waits here until it gets told by the threadpool
* to wake up.
*
* \param worker The idle worker
* \retval 0 The thread is being woken up so that it can conclude.
* \retval non-zero The thread is being woken up to do more work.
*/
static int worker_idle(struct worker_thread *worker)
{
SCOPED_MUTEX(lock, &worker->lock);
if (worker->state != ALIVE) {
return 0;
}
threadpool_active_thread_idle(worker->pool, worker);
while (!worker->wake_up) {
ast_cond_wait(&worker->cond, lock);
}
worker->wake_up = 0;
return worker->state == ALIVE;
}
/*!
* \brief Change a worker's state
*
* The threadpool calls into this function in order to let a worker know
* how it should proceed.
*/
static void worker_set_state(struct worker_thread *worker, enum worker_state state)
{
SCOPED_MUTEX(lock, &worker->lock);
worker->state = state;
worker->wake_up = 1;
ast_cond_signal(&worker->cond);
}