Add some doxygen and rearrange code.
authorMark Michelson <mmichelson@digium.com>
Tue, 4 Dec 2012 21:11:34 +0000 (21:11 +0000)
committerMark Michelson <mmichelson@digium.com>
Tue, 4 Dec 2012 21:11:34 +0000 (21:11 +0000)
git-svn-id: https://origsvn.digium.com/svn/asterisk/team/mmichelson/threadpool@377209 65c4cc65-6c06-0410-ace0-fbb531ad65f3

include/asterisk/threadpool.h
main/threadpool.c

index d4c10b8..8ff2184 100644 (file)
@@ -94,6 +94,19 @@ struct ast_threadpool *ast_threadpool_create(struct ast_threadpool_listener *lis
 void ast_threadpool_set_size(struct ast_threadpool *threadpool, unsigned int size);
 
 /*!
+ * \brief Push a task to the threadpool
+ *
+ * Tasks pushed into the threadpool will be automatically taken by
+ * one of the threads within
+ * \param pool The threadpool to add the task to
+ * \param task The task to add
+ * \param data The parameter for the task
+ * \retval 0 success
+ * \retval -1 failure
+ */
+int ast_threadpool_push(struct ast_threadpool *pool, int (*task)(void *data), void *data);
+
+/*!
  * \brief Shut down a threadpool and destroy it
  *
  * \param pool The pool to shut down
index aa38ee5..1c2e963 100644 (file)
 
 #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)
-{
-       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)
+/*!
+ * \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)
 {
-       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);
+}
+