Add threadpool support to Asterisk.
authorMark Michelson <mmichelson@digium.com>
Fri, 18 Jan 2013 15:42:10 +0000 (15:42 +0000)
committerMark Michelson <mmichelson@digium.com>
Fri, 18 Jan 2013 15:42:10 +0000 (15:42 +0000)
This commit consists of two parts.

Part one changes the taskprocessor API to be less self-contained.
Instead, the taskprocessor is now more of a task queue that informs
a listener of changes to the queue. The listener then has the responsibility
of executing the tasks as it pleases. There is a default listener implementation
that functions the same way as "classic" taskprocessors, in that it creates
a single thread for tasks to execute in. Old users of taskprocessors have
not been altered and still function the same way.

Part two introduces the threadpool API. A threadpool is a special type of
taskprocessor listener that has multiple threads associated with it. The threadpool
also has an optional listener that can adjust the threadpool as conditions change.
In addition the threadpool has a set of options that can allow for the threadpool
to grow and shrink on its own as tasks are added and executed.

Both set of changes contain accompanying unit tests.

(closes issue ASTERISK-20691)
Reported By: Matt Jordan

Review: https://reviewboard.asterisk.org/r/2242

git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@379432 65c4cc65-6c06-0410-ace0-fbb531ad65f3

include/asterisk/taskprocessor.h
include/asterisk/threadpool.h [new file with mode: 0644]
main/taskprocessor.c
main/threadpool.c [new file with mode: 0644]
tests/test_taskprocessor.c [new file with mode: 0644]
tests/test_threadpool.c [new file with mode: 0644]

index 0f1876e..2191663 100644 (file)
@@ -1,7 +1,7 @@
 /*
  * Asterisk -- An open source telephony toolkit.
  *
- * Copyright (C) 2007-2008, Digium, Inc.
+ * Copyright (C) 2007-2013, Digium, Inc.
  *
  * Dwayne M. Hubbard <dhubbard@digium.com>
  *
  *
  * \author Dwayne M. Hubbard <dhubbard@digium.com>
  *
- * \note A taskprocessor is a named singleton containing a processing thread and
- * a task queue that serializes tasks pushed into it by [a] module(s) that reference the taskprocessor.  
- * A taskprocessor is created the first time its name is requested via the ast_taskprocessor_get()
- * function and destroyed when the taskprocessor reference count reaches zero.
+ * \note A taskprocessor is a named object containing a task queue that
+ * serializes tasks pushed into it by [a] module(s) that reference the taskprocessor.
+ * A taskprocessor is created the first time its name is requested via the
+ * ast_taskprocessor_get() function or the ast_taskprocessor_create_with_listener()
+ * function and destroyed when the taskprocessor reference count reaches zero. A
+ * taskprocessor also contains an accompanying listener that is notified when changes
+ * in the task queue occur.
  *
- * Modules that obtain a reference to a taskprocessor can queue tasks into the taskprocessor
- * to be processed by the singleton processing thread when the task is popped off the front 
- * of the queue.  A task is a wrapper around a task-handling function pointer and a data
- * pointer.  It is the responsibility of the task handling function to free memory allocated for
- * the task data pointer.  A task is pushed into a taskprocessor queue using the 
+ * A task is a wrapper around a task-handling function pointer and a data
+ * pointer.  A task is pushed into a taskprocessor queue using the
  * ast_taskprocessor_push(taskprocessor, taskhandler, taskdata) function and freed by the
- * taskprocessor after the task handling function returns.  A module releases its reference to a
- * taskprocessor using the ast_taskprocessor_unreference() function which may result in the
- * destruction of the taskprocessor if the taskprocessor's reference count reaches zero.  Tasks waiting
- * to be processed in the taskprocessor queue when the taskprocessor reference count reaches zero
- * will be purged and released from the taskprocessor queue without being processed.
+ * taskprocessor after the task handling function returns.  A module releases its
+ * reference to a taskprocessor using the ast_taskprocessor_unreference() function which
+ * may result in the destruction of the taskprocessor if the taskprocessor's reference
+ * count reaches zero. When the taskprocessor's reference count reaches zero, its
+ * listener's shutdown() callback will be called. Any further attempts to execute tasks
+ * will be denied.
+ *
+ * The taskprocessor listener has the flexibility of doling out tasks to best fit the
+ * module's needs. For instance, a taskprocessor listener may have a single dispatch
+ * thread that handles all tasks, or it may dispatch tasks to a thread pool.
+ *
+ * There is a default taskprocessor listener that will be used if a taskprocessor is
+ * created without any explicit listener. This default listener runs tasks sequentially
+ * in a single thread. The listener will execute tasks as long as there are tasks to be
+ * processed. When the taskprocessor is shut down, the default listener will stop
+ * processing tasks and join its execution thread.
  */
 
 #ifndef __AST_TASKPROCESSOR_H__
@@ -48,9 +59,9 @@ struct ast_taskprocessor;
 /*!
  * \brief ast_tps_options for specification of taskprocessor options
  *
- * Specify whether a taskprocessor should be created via ast_taskprocessor_get() if the taskprocessor 
- * does not already exist.  The default behavior is to create a taskprocessor if it does not already exist 
- * and provide its reference to the calling function.  To only return a reference to a taskprocessor if 
+ * Specify whether a taskprocessor should be created via ast_taskprocessor_get() if the taskprocessor
+ * does not already exist.  The default behavior is to create a taskprocessor if it does not already exist
+ * and provide its reference to the calling function.  To only return a reference to a taskprocessor if
  * and only if it exists, use the TPS_REF_IF_EXISTS option in ast_taskprocessor_get().
  */
 enum ast_tps_options {
@@ -60,13 +71,86 @@ enum ast_tps_options {
        TPS_REF_IF_EXISTS = (1 << 0),
 };
 
+struct ast_taskprocessor_listener;
+
+struct ast_taskprocessor_listener_callbacks {
+       /*!
+        * \brief The taskprocessor has started completely
+        *
+        * This indicates that the taskprocessor is fully set up and the listener
+        * can now start interacting with it.
+        *
+        * \param listener The listener to start
+        */
+       int (*start)(struct ast_taskprocessor_listener *listener);
+       /*!
+        * \brief Indicates a task was pushed to the processor
+        *
+        * \param listener The listener
+        * \param was_empty If non-zero, the taskprocessor was empty prior to the task being pushed
+        */
+       void (*task_pushed)(struct ast_taskprocessor_listener *listener, int was_empty);
+       /*!
+        * \brief Indicates the task processor has become empty
+        *
+        * \param listener The listener
+        */
+       void (*emptied)(struct ast_taskprocessor_listener *listener);
+       /*!
+        * \brief Indicates the taskprocessor wishes to die.
+        *
+        * All operations on the task processor must to be stopped in
+        * this callback. This is an opportune time to free the listener's
+        * user data if it is not going to be used anywhere else.
+        *
+        * After this callback returns, it is NOT safe to operate on the
+        * listener's reference to the taskprocessor.
+        *
+        * \param listener The listener
+        */
+       void (*shutdown)(struct ast_taskprocessor_listener *listener);
+};
+
+/*!
+ * \brief Get a reference to the listener's taskprocessor
+ *
+ * This will return the taskprocessor with its reference count increased. Release
+ * the reference to this object by using ast_taskprocessor_unreference()
+ *
+ * \param listener The listener that has the taskprocessor
+ * \return The taskprocessor
+ */
+struct ast_taskprocessor *ast_taskprocessor_listener_get_tps(const struct ast_taskprocessor_listener *listener);
+
+/*!
+ * \brief Get the user data from the listener
+ * \param listener The taskprocessor listener
+ * \return The listener's user data
+ */
+void *ast_taskprocessor_listener_get_user_data(const struct ast_taskprocessor_listener *listener);
+
+/*!
+ * \brief Allocate a taskprocessor listener
+ *
+ * \since 12.0.0
+ *
+ * This will result in the listener being allocated with the specified
+ * callbacks.
+ *
+ * \param callbacks The callbacks to assign to the listener
+ * \param user_data The user data for the listener
+ * \retval NULL Failure
+ * \retval non-NULL The newly allocated taskprocessor listener
+ */
+struct ast_taskprocessor_listener *ast_taskprocessor_listener_alloc(const struct ast_taskprocessor_listener_callbacks *callbacks, void *user_data);
+
 /*!
  * \brief Get a reference to a taskprocessor with the specified name and create the taskprocessor if necessary
  *
  * The default behavior of instantiating a taskprocessor if one does not already exist can be
  * disabled by specifying the TPS_REF_IF_EXISTS ast_tps_options as the second argument to ast_taskprocessor_get().
  * \param name The name of the taskprocessor
- * \param create Use 0 by default or specify TPS_REF_IF_EXISTS to return NULL if the taskprocessor does 
+ * \param create Use 0 by default or specify TPS_REF_IF_EXISTS to return NULL if the taskprocessor does
  * not already exist
  * return A pointer to a reference counted taskprocessor under normal conditions, or NULL if the
  * TPS_REF_IF_EXISTS reference type is specified and the taskprocessor does not exist
@@ -75,6 +159,22 @@ enum ast_tps_options {
 struct ast_taskprocessor *ast_taskprocessor_get(const char *name, enum ast_tps_options create);
 
 /*!
+ * \brief Create a taskprocessor with a custom listener
+ *
+ * \since 12.0.0
+ *
+ * Note that when a taskprocessor is created in this way, it does not create
+ * any threads to execute the tasks. This job is left up to the listener.
+ * The listener's start() callback will be called during this function.
+ *
+ * \param name The name of the taskprocessor to create
+ * \param listener The listener for operations on this taskprocessor
+ * \retval NULL Failure
+ * \reval non-NULL success
+ */
+struct ast_taskprocessor *ast_taskprocessor_create_with_listener(const char *name, struct ast_taskprocessor_listener *listener);
+
+/*!
  * \brief Unreference the specified taskprocessor and its reference count will decrement.
  *
  * Taskprocessors use astobj2 and will unlink from the taskprocessor singleton container and destroy
@@ -97,6 +197,17 @@ void *ast_taskprocessor_unreference(struct ast_taskprocessor *tps);
 int ast_taskprocessor_push(struct ast_taskprocessor *tps, int (*task_exe)(void *datap), void *datap);
 
 /*!
+ * \brief Pop a task off the taskprocessor and execute it.
+ *
+ * \since 12.0.0
+ *
+ * \param tps The taskprocessor from which to execute.
+ * \retval 0 There is no further work to be done.
+ * \retval 1 Tasks still remain in the taskprocessor queue.
+ */
+int ast_taskprocessor_execute(struct ast_taskprocessor *tps);
+
+/*!
  * \brief Return the name of the taskprocessor singleton
  * \since 1.6.1
  */
diff --git a/include/asterisk/threadpool.h b/include/asterisk/threadpool.h
new file mode 100644 (file)
index 0000000..e18ba6c
--- /dev/null
@@ -0,0 +1,180 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2012-2013, Digium, Inc.
+ *
+ * Mark Michelson <mmmichelson@digium.com>
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE file
+ * at the top of the source tree.
+ */
+
+
+#ifndef _ASTERISK_THREADPOOL_H
+#define _ASTERISK_THREADPOOL_H
+
+struct ast_threadpool;
+struct ast_taskprocessor;
+struct ast_threadpool_listener;
+
+struct ast_threadpool_listener_callbacks {
+       /*!
+        * \brief Indicates that the state of threads in the pool has changed
+        *
+        * \param pool The pool whose state has changed
+        * \param listener The threadpool listener
+        * \param active_threads The number of active threads in the pool
+        * \param idle_threads The number of idle threads in the pool
+        */
+       void (*state_changed)(struct ast_threadpool *pool,
+                       struct ast_threadpool_listener *listener,
+                       int active_threads,
+                       int idle_threads);
+       /*!
+        * \brief Indicates that a task was pushed to the threadpool
+        *
+        * \param pool The pool that had a task pushed
+        * \param listener The threadpool listener
+        * \param was_empty Indicates whether there were any tasks prior to adding the new one.
+        */
+       void (*task_pushed)(struct ast_threadpool *pool,
+                       struct ast_threadpool_listener *listener,
+                       int was_empty);
+       /*!
+        * \brief Indicates the threadpool's taskprocessor has become empty
+        *
+        * \param pool The pool that has become empty
+        * \param listener The threadpool's listener
+        */
+       void (*emptied)(struct ast_threadpool *pool, struct ast_threadpool_listener *listener);
+
+       /*!
+        * \brief The threadpool is shutting down
+        *
+        * This would be an opportune time to free the listener's user data
+        * if one wishes. However, it is acceptable to not do so if the user data
+        * should persist beyond the lifetime of the pool.
+        *
+        * \param listener The threadpool's listener
+        */
+       void (*shutdown)(struct ast_threadpool_listener *listener);
+};
+
+struct ast_threadpool_options {
+#define AST_THREADPOOL_OPTIONS_VERSION 1
+       /*! Version of threadpool options in use */
+       int version;
+       /*!
+        * \brief Time limit in seconds for idle threads
+        *
+        * A time of 0 or less will mean no timeout.
+        */
+       int idle_timeout;
+       /*!
+        * \brief Number of threads to increment pool by
+        *
+        * If a task is added into a pool and no idle thread is
+        * available to activate, then the pool can automatically
+        * grow by the given amount.
+        *
+        * Zero is a perfectly valid value to give here if you want
+        * to control threadpool growth yourself via your listener.
+        */
+       int auto_increment;
+       /*!
+        * \brief Number of threads the pool will start with
+        *
+        * When the threadpool is allocated, it will immediately size
+        * itself to have this number of threads in it.
+        *
+        * Zero is a valid value if the threadpool should start
+        * without any threads allocated.
+        */
+       int initial_size;
+       /*!
+        * \brief Maximum number of threads a pool may have
+        *
+        * When the threadpool's size increases, it can never increase
+        * beyond this number of threads.
+        */
+       int max_size;
+};
+
+/*!
+ * \brief Allocate a threadpool listener
+ *
+ * This function will call back into the alloc callback for the
+ * listener.
+ *
+ * \param callbacks Listener callbacks to assign to the listener
+ * \param user_data User data to be stored in the threadpool listener
+ * \retval NULL Failed to allocate the listener
+ * \retval non-NULL The newly-created threadpool listener
+ */
+struct ast_threadpool_listener *ast_threadpool_listener_alloc(
+               const struct ast_threadpool_listener_callbacks *callbacks, void *user_data);
+
+/*!
+ * \brief Get the threadpool listener's user data
+ * \param listener The threadpool listener
+ * \return The user data
+ */
+void *ast_threadpool_listener_get_user_data(const struct ast_threadpool_listener *listener);
+
+/*!
+ * \brief Create a new threadpool
+ *
+ * This function creates a threadpool. Tasks may be pushed onto this thread pool
+ * in and will be automatically acted upon by threads within the pool.
+ *
+ * Only a single threadpool with a given name may exist. This function will fail
+ * if a threadpool with the given name already exists.
+ *
+ * \param name The unique name for the threadpool
+ * \param listener The listener the threadpool will notify of changes. Can be NULL.
+ * \param options The behavioral options for this threadpool
+ * \retval NULL Failed to create the threadpool
+ * \retval non-NULL The newly-created threadpool
+ */
+struct ast_threadpool *ast_threadpool_create(const char *name,
+               struct ast_threadpool_listener *listener,
+               const struct ast_threadpool_options *options);
+
+/*!
+ * \brief Set the number of threads for the thread pool
+ *
+ * This number may be more or less than the current number of
+ * threads in the threadpool.
+ *
+ * \param threadpool The threadpool to adjust
+ * \param size The new desired size of the threadpool
+ */
+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
+ */
+void ast_threadpool_shutdown(struct ast_threadpool *pool);
+#endif /* ASTERISK_THREADPOOL_H */
index d8f1af3..b603e57 100644 (file)
@@ -1,7 +1,7 @@
 /*
  * Asterisk -- An open source telephony toolkit.
  *
- * Copyright (C) 2007-2008, Digium, Inc.
+ * Copyright (C) 2007-2013, Digium, Inc.
  *
  * Dwayne M. Hubbard <dhubbard@digium.com>
  *
@@ -38,7 +38,6 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/cli.h"
 #include "asterisk/taskprocessor.h"
 
-
 /*!
  * \brief tps_task structure is queued to a taskprocessor
  *
@@ -67,14 +66,6 @@ struct tps_taskprocessor_stats {
 struct ast_taskprocessor {
        /*! \brief Friendly name of the taskprocessor */
        const char *name;
-       /*! \brief Thread poll condition */
-       ast_cond_t poll_cond;
-       /*! \brief Taskprocessor thread */
-       pthread_t poll_thread;
-       /*! \brief Taskprocessor lock */
-       ast_mutex_t taskprocessor_lock;
-       /*! \brief Taskprocesor thread run flag */
-       unsigned char poll_thread_run;
        /*! \brief Taskprocessor statistics */
        struct tps_taskprocessor_stats *stats;
        /*! \brief Taskprocessor current queue size */
@@ -83,7 +74,30 @@ struct ast_taskprocessor {
        AST_LIST_HEAD_NOLOCK(tps_queue, tps_task) tps_queue;
        /*! \brief Taskprocessor singleton list entry */
        AST_LIST_ENTRY(ast_taskprocessor) list;
+       struct ast_taskprocessor_listener *listener;
+       /*! Indicates if the taskprocessor is in the process of shuting down */
+       unsigned int shutting_down:1;
 };
+
+/*!
+ * \brief A listener for taskprocessors
+ *
+ * \since 12.0.0
+ *
+ * When a taskprocessor's state changes, the listener
+ * is notified of the change. This allows for tasks
+ * to be addressed in whatever way is appropriate for
+ * the module using the taskprocessor.
+ */
+struct ast_taskprocessor_listener {
+       /*! The callbacks the taskprocessor calls into to notify of state changes */
+       const struct ast_taskprocessor_listener_callbacks *callbacks;
+       /*! The taskprocessor that the listener is listening to */
+       struct ast_taskprocessor *tps;
+       /*! Data private to the listener */
+       void *user_data;
+};
+
 #define TPS_MAX_BUCKETS 7
 /*! \brief tps_singletons is the astobj2 container for taskprocessor singletons */
 static struct ao2_container *tps_singletons;
@@ -122,6 +136,94 @@ static struct ast_cli_entry taskprocessor_clis[] = {
        AST_CLI_DEFINE(cli_tps_report, "List instantiated task processors and statistics"),
 };
 
+struct default_taskprocessor_listener_pvt {
+       pthread_t poll_thread;
+       ast_mutex_t lock;
+       ast_cond_t cond;
+       int wake_up;
+       int dead;
+};
+
+
+static void default_tps_wake_up(struct default_taskprocessor_listener_pvt *pvt, int should_die)
+{
+       SCOPED_MUTEX(lock, &pvt->lock);
+       pvt->wake_up = 1;
+       pvt->dead = should_die;
+       ast_cond_signal(&pvt->cond);
+}
+
+static int default_tps_idle(struct default_taskprocessor_listener_pvt *pvt)
+{
+       SCOPED_MUTEX(lock, &pvt->lock);
+       while (!pvt->wake_up) {
+               ast_cond_wait(&pvt->cond, lock);
+       }
+       pvt->wake_up = 0;
+       return pvt->dead;
+}
+
+/*!
+ * \brief Function that processes tasks in the taskprocessor
+ * \internal
+ */
+static void *tps_processing_function(void *data)
+{
+       struct ast_taskprocessor_listener *listener = data;
+       struct ast_taskprocessor *tps = listener->tps;
+       struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
+       int dead = 0;
+
+       while (!dead) {
+               if (!ast_taskprocessor_execute(tps)) {
+                       dead = default_tps_idle(pvt);
+               }
+       }
+       return NULL;
+}
+
+static int default_listener_start(struct ast_taskprocessor_listener *listener)
+{
+       struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
+
+       if (ast_pthread_create(&pvt->poll_thread, NULL, tps_processing_function, listener)) {
+               return -1;
+       }
+
+       return 0;
+}
+
+static void default_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
+{
+       struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
+
+       if (was_empty) {
+               default_tps_wake_up(pvt, 0);
+       }
+}
+
+static void default_listener_pvt_destroy(struct default_taskprocessor_listener_pvt *pvt)
+{
+       ast_mutex_destroy(&pvt->lock);
+       ast_cond_destroy(&pvt->cond);
+       ast_free(pvt);
+}
+
+static void default_listener_shutdown(struct ast_taskprocessor_listener *listener)
+{
+       struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
+       default_tps_wake_up(pvt, 1);
+       pthread_join(pvt->poll_thread, NULL);
+       pvt->poll_thread = AST_PTHREADT_NULL;
+       default_listener_pvt_destroy(pvt);
+}
+
+static const struct ast_taskprocessor_listener_callbacks default_listener_callbacks = {
+       .start = default_listener_start,
+       .task_pushed = default_task_pushed,
+       .shutdown = default_listener_shutdown,
+};
+
 /*!
  * \internal
  * \brief Clean up resources on Asterisk shutdown
@@ -291,118 +393,67 @@ static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_arg
        return CLI_SUCCESS;
 }
 
-/* this is the task processing worker function */
-static void *tps_processing_function(void *data)
-{
-       struct ast_taskprocessor *i = data;
-       struct tps_task *t;
-       int size;
-
-       if (!i) {
-               ast_log(LOG_ERROR, "cannot start thread_function loop without a ast_taskprocessor structure.\n");
-               return NULL;
-       }
-
-       while (i->poll_thread_run) {
-               ast_mutex_lock(&i->taskprocessor_lock);
-               if (!i->poll_thread_run) {
-                       ast_mutex_unlock(&i->taskprocessor_lock);
-                       break;
-               }
-               if (!(size = tps_taskprocessor_depth(i))) {
-                       ast_cond_wait(&i->poll_cond, &i->taskprocessor_lock);
-                       if (!i->poll_thread_run) {
-                               ast_mutex_unlock(&i->taskprocessor_lock);
-                               break;
-                       }
-               }
-               ast_mutex_unlock(&i->taskprocessor_lock);
-               /* stuff is in the queue */
-               if (!(t = tps_taskprocessor_pop(i))) {
-                       ast_log(LOG_ERROR, "Wtf?? %d tasks in the queue, but we're popping blanks!\n", size);
-                       continue;
-               }
-               if (!t->execute) {
-                       ast_log(LOG_WARNING, "Task is missing a function to execute!\n");
-                       tps_task_free(t);
-                       continue;
-               }
-               t->execute(t->datap);
-
-               ast_mutex_lock(&i->taskprocessor_lock);
-               if (i->stats) {
-                       i->stats->_tasks_processed_count++;
-                       if (size > i->stats->max_qsize) {
-                               i->stats->max_qsize = size;
-                       }
-               }
-               ast_mutex_unlock(&i->taskprocessor_lock);
-
-               tps_task_free(t);
-       }
-       while ((t = tps_taskprocessor_pop(i))) {
-               tps_task_free(t);
-       }
-       return NULL;
-}
-
 /* hash callback for astobj2 */
 static int tps_hash_cb(const void *obj, const int flags)
 {
        const struct ast_taskprocessor *tps = obj;
+       const char *name = flags & OBJ_KEY ? obj : tps->name;
 
-       return ast_str_case_hash(tps->name);
+       return ast_str_case_hash(name);
 }
 
 /* compare callback for astobj2 */
 static int tps_cmp_cb(void *obj, void *arg, int flags)
 {
        struct ast_taskprocessor *lhs = obj, *rhs = arg;
+       const char *rhsname = flags & OBJ_KEY ? arg : rhs->name;
 
-       return !strcasecmp(lhs->name, rhs->name) ? CMP_MATCH | CMP_STOP : 0;
+       return !strcasecmp(lhs->name, rhsname) ? CMP_MATCH | CMP_STOP : 0;
 }
 
 /* destroy the taskprocessor */
 static void tps_taskprocessor_destroy(void *tps)
 {
        struct ast_taskprocessor *t = tps;
+       struct tps_task *task;
 
        if (!tps) {
                ast_log(LOG_ERROR, "missing taskprocessor\n");
                return;
        }
        ast_debug(1, "destroying taskprocessor '%s'\n", t->name);
-       /* kill it */
-       ast_mutex_lock(&t->taskprocessor_lock);
-       t->poll_thread_run = 0;
-       ast_cond_signal(&t->poll_cond);
-       ast_mutex_unlock(&t->taskprocessor_lock);
-       pthread_join(t->poll_thread, NULL);
-       t->poll_thread = AST_PTHREADT_NULL;
-       ast_mutex_destroy(&t->taskprocessor_lock);
-       ast_cond_destroy(&t->poll_cond);
        /* free it */
        if (t->stats) {
                ast_free(t->stats);
                t->stats = NULL;
        }
        ast_free((char *) t->name);
+       if (t->listener) {
+               /* This code should not be reached since the listener
+                * should have been destroyed before the taskprocessor could
+                * be destroyed
+                */
+               ao2_ref(t->listener, -1);
+               t->listener = NULL;
+       }
+       while ((task = AST_LIST_REMOVE_HEAD(&t->tps_queue, list))) {
+               tps_task_free(task);
+       }
 }
 
 /* pop the front task and return it */
 static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps)
 {
        struct tps_task *task;
+       SCOPED_AO2LOCK(lock, tps);
 
-       if (!tps) {
-               ast_log(LOG_ERROR, "missing taskprocessor\n");
+       if (tps->shutting_down) {
                return NULL;
        }
-       ast_mutex_lock(&tps->taskprocessor_lock);
+
        if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) {
                tps->tps_queue_size--;
        }
-       ast_mutex_unlock(&tps->taskprocessor_lock);
        return task;
 }
 
@@ -421,80 +472,169 @@ const char *ast_taskprocessor_name(struct ast_taskprocessor *tps)
        return tps->name;
 }
 
+static void listener_shutdown(struct ast_taskprocessor_listener *listener)
+{
+       listener->callbacks->shutdown(listener);
+       ao2_ref(listener->tps, -1);
+}
+
+struct ast_taskprocessor_listener *ast_taskprocessor_listener_alloc(const struct ast_taskprocessor_listener_callbacks *callbacks, void *user_data)
+{
+       RAII_VAR(struct ast_taskprocessor_listener *, listener,
+                       ao2_alloc(sizeof(*listener), NULL), ao2_cleanup);
+
+       if (!listener) {
+               return NULL;
+       }
+       listener->callbacks = callbacks;
+       listener->user_data = user_data;
+
+       ao2_ref(listener, +1);
+       return listener;
+}
+
+struct ast_taskprocessor *ast_taskprocessor_listener_get_tps(const struct ast_taskprocessor_listener *listener)
+{
+       ao2_ref(listener->tps, +1);
+       return listener->tps;
+}
+
+void *ast_taskprocessor_listener_get_user_data(const struct ast_taskprocessor_listener *listener)
+{
+       return listener->user_data;
+}
+
+static void *default_listener_pvt_alloc(void)
+{
+       struct default_taskprocessor_listener_pvt *pvt;
+
+       pvt = ast_calloc(1, sizeof(*pvt));
+       if (!pvt) {
+               return NULL;
+       }
+       ast_cond_init(&pvt->cond, NULL);
+       ast_mutex_init(&pvt->lock);
+       pvt->poll_thread = AST_PTHREADT_NULL;
+       return pvt;
+}
+
+static struct ast_taskprocessor *__allocate_taskprocessor(const char *name, struct ast_taskprocessor_listener *listener)
+{
+       RAII_VAR(struct ast_taskprocessor *, p,
+                       ao2_alloc(sizeof(*p), tps_taskprocessor_destroy), ao2_cleanup);
+
+       if (!p) {
+               ast_log(LOG_WARNING, "failed to create taskprocessor '%s'\n", name);
+               return NULL;
+       }
+
+       if (!(p->stats = ast_calloc(1, sizeof(*p->stats)))) {
+               ast_log(LOG_WARNING, "failed to create taskprocessor stats for '%s'\n", name);
+               return NULL;
+       }
+       if (!(p->name = ast_strdup(name))) {
+               ao2_ref(p, -1);
+               return NULL;
+       }
+
+       ao2_ref(listener, +1);
+       p->listener = listener;
+
+       ao2_ref(p, +1);
+       listener->tps = p;
+
+       if (!(ao2_link(tps_singletons, p))) {
+               ast_log(LOG_ERROR, "Failed to add taskprocessor '%s' to container\n", p->name);
+               return NULL;
+       }
+
+       if (p->listener->callbacks->start(p->listener)) {
+               ast_log(LOG_ERROR, "Unable to start taskprocessor listener for taskprocessor %s\n", p->name);
+               ast_taskprocessor_unreference(p);
+               return NULL;
+       }
+
+       /* RAII_VAR will decrement the refcount at the end of the function.
+        * Since we want to pass back a reference to p, we bump the refcount
+        */
+       ao2_ref(p, +1);
+       return p;
+
+}
+
 /* Provide a reference to a taskprocessor.  Create the taskprocessor if necessary, but don't
  * create the taskprocessor if we were told via ast_tps_options to return a reference only
  * if it already exists */
 struct ast_taskprocessor *ast_taskprocessor_get(const char *name, enum ast_tps_options create)
 {
-       struct ast_taskprocessor *p, tmp_tps = {
-               .name = name,
-       };
+       struct ast_taskprocessor *p;
+       struct ast_taskprocessor_listener *listener;
+       struct default_taskprocessor_listener_pvt *pvt;
 
        if (ast_strlen_zero(name)) {
                ast_log(LOG_ERROR, "requesting a nameless taskprocessor!!!\n");
                return NULL;
        }
-       ao2_lock(tps_singletons);
-       p = ao2_find(tps_singletons, &tmp_tps, OBJ_POINTER);
+       p = ao2_find(tps_singletons, name, OBJ_KEY);
        if (p) {
-               ao2_unlock(tps_singletons);
                return p;
        }
        if (create & TPS_REF_IF_EXISTS) {
                /* calling function does not want a new taskprocessor to be created if it doesn't already exist */
-               ao2_unlock(tps_singletons);
                return NULL;
        }
-       /* create a new taskprocessor */
-       if (!(p = ao2_alloc(sizeof(*p), tps_taskprocessor_destroy))) {
-               ao2_unlock(tps_singletons);
-               ast_log(LOG_WARNING, "failed to create taskprocessor '%s'\n", name);
+       /* Create a new taskprocessor. Start by creating a default listener */
+       pvt = default_listener_pvt_alloc();
+       if (!pvt) {
                return NULL;
        }
-
-       ast_cond_init(&p->poll_cond, NULL);
-       ast_mutex_init(&p->taskprocessor_lock);
-
-       if (!(p->stats = ast_calloc(1, sizeof(*p->stats)))) {
-               ao2_unlock(tps_singletons);
-               ast_log(LOG_WARNING, "failed to create taskprocessor stats for '%s'\n", name);
-               ao2_ref(p, -1);
+       listener = ast_taskprocessor_listener_alloc(&default_listener_callbacks, pvt);
+       if (!listener) {
+               default_listener_pvt_destroy(pvt);
                return NULL;
        }
-       if (!(p->name = ast_strdup(name))) {
-               ao2_unlock(tps_singletons);
-               ao2_ref(p, -1);
-               return NULL;
-       }
-       p->poll_thread_run = 1;
-       p->poll_thread = AST_PTHREADT_NULL;
-       if (ast_pthread_create(&p->poll_thread, NULL, tps_processing_function, p) < 0) {
-               ao2_unlock(tps_singletons);
-               ast_log(LOG_ERROR, "Taskprocessor '%s' failed to create the processing thread.\n", p->name);
-               ao2_ref(p, -1);
+
+       p = __allocate_taskprocessor(name, listener);
+       if (!p) {
+               default_listener_pvt_destroy(pvt);
+               ao2_ref(listener, -1);
                return NULL;
        }
-       if (!(ao2_link(tps_singletons, p))) {
-               ao2_unlock(tps_singletons);
-               ast_log(LOG_ERROR, "Failed to add taskprocessor '%s' to container\n", p->name);
-               ao2_ref(p, -1);
+
+       /* Unref listener here since the taskprocessor has gained a reference to the listener */
+       ao2_ref(listener, -1);
+       return p;
+
+}
+
+struct ast_taskprocessor *ast_taskprocessor_create_with_listener(const char *name, struct ast_taskprocessor_listener *listener)
+{
+       struct ast_taskprocessor *p = ao2_find(tps_singletons, name, OBJ_KEY);
+
+       if (p) {
+               ast_taskprocessor_unreference(p);
                return NULL;
        }
-       ao2_unlock(tps_singletons);
-       return p;
+       return __allocate_taskprocessor(name, listener);
 }
 
 /* decrement the taskprocessor reference count and unlink from the container if necessary */
 void *ast_taskprocessor_unreference(struct ast_taskprocessor *tps)
 {
-       if (tps) {
-               ao2_lock(tps_singletons);
-               ao2_unlink(tps_singletons, tps);
-               if (ao2_ref(tps, -1) > 1) {
-                       ao2_link(tps_singletons, tps);
-               }
-               ao2_unlock(tps_singletons);
+       if (!tps) {
+               return NULL;
        }
+
+       if (ao2_ref(tps, -1) > 3) {
+               return NULL;
+       }
+       /* If we're down to 3 references, then those must be:
+        * 1. The reference we just got rid of
+        * 2. The container
+        * 3. The listener
+        */
+       ao2_unlink(tps_singletons, tps);
+       listener_shutdown(tps->listener);
        return NULL;
 }
 
@@ -502,6 +642,7 @@ void *ast_taskprocessor_unreference(struct ast_taskprocessor *tps)
 int ast_taskprocessor_push(struct ast_taskprocessor *tps, int (*task_exe)(void *datap), void *datap)
 {
        struct tps_task *t;
+       int previous_size;
 
        if (!tps || !task_exe) {
                ast_log(LOG_ERROR, "%s is missing!!\n", (tps) ? "task callback" : "taskprocessor");
@@ -511,11 +652,40 @@ int ast_taskprocessor_push(struct ast_taskprocessor *tps, int (*task_exe)(void *
                ast_log(LOG_ERROR, "failed to allocate task!  Can't push to '%s'\n", tps->name);
                return -1;
        }
-       ast_mutex_lock(&tps->taskprocessor_lock);
+       ao2_lock(tps);
        AST_LIST_INSERT_TAIL(&tps->tps_queue, t, list);
-       tps->tps_queue_size++;
-       ast_cond_signal(&tps->poll_cond);
-       ast_mutex_unlock(&tps->taskprocessor_lock);
+       previous_size = tps->tps_queue_size++;
+       ao2_unlock(tps);
+       tps->listener->callbacks->task_pushed(tps->listener, previous_size ? 0 : 1);
        return 0;
 }
 
+int ast_taskprocessor_execute(struct ast_taskprocessor *tps)
+{
+       struct tps_task *t;
+       int size;
+
+       if (!(t = tps_taskprocessor_pop(tps))) {
+               return 0;
+       }
+
+       t->execute(t->datap);
+
+       tps_task_free(t);
+
+       ao2_lock(tps);
+       size = tps_taskprocessor_depth(tps);
+       if (tps->stats) {
+               tps->stats->_tasks_processed_count++;
+               if (size > tps->stats->max_qsize) {
+                       tps->stats->max_qsize = size;
+               }
+       }
+       ao2_unlock(tps);
+
+       if (size == 0 && tps->listener->callbacks->emptied) {
+               tps->listener->callbacks->emptied(tps->listener);
+               return 0;
+       }
+       return 1;
+}
diff --git a/main/threadpool.c b/main/threadpool.c
new file mode 100644 (file)
index 0000000..adaf8a5
--- /dev/null
@@ -0,0 +1,1105 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2012-2013, Digium, Inc.
+ *
+ * Mark Michelson <mmmichelson@digium.com>
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE file
+ * at the top of the source tree.
+ */
+
+
+#include "asterisk.h"
+
+#include "asterisk/threadpool.h"
+#include "asterisk/taskprocessor.h"
+#include "asterisk/astobj2.h"
+#include "asterisk/utils.h"
+
+/* Needs to stay prime if increased */
+#define THREAD_BUCKETS 89
+
+/*!
+ * \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;
+       /*! True if the threadpool is in the processof shutting down */
+       int shutting_down;
+       /*! Threadpool-specific options */
+       struct ast_threadpool_options options;
+};
+
+/*!
+ * \brief listener for a threadpool
+ *
+ * The listener is notified of changes in a threadpool. It can
+ * react by doing things like increasing the number of threads
+ * in the pool
+ */
+struct ast_threadpool_listener {
+       /*! Callbacks called by the threadpool */
+       const struct ast_threadpool_listener_callbacks *callbacks;
+       /*! User data for the listener */
+       void *user_data;
+};
+
+/*!
+ * \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,
+};
+
+/*!
+ * 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;
+       /*! Options for this threadpool */
+       struct ast_threadpool_options options;
+};
+
+/* Worker thread forward declarations. See definitions for documentation */
+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_thread_start(struct worker_thread *worker);
+static int worker_idle(struct worker_thread *worker);
+static void worker_set_state(struct worker_thread *worker, enum worker_state state);
+static void worker_shutdown(struct worker_thread *worker);
+
+/*!
+ * \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);
+       int idle_size = ao2_container_count(pool->idle_threads);
+
+       if (pool->listener && pool->listener->callbacks->state_changed) {
+               pool->listener->callbacks->state_changed(pool, pool->listener, active_size, idle_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;
+       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)
+{
+       struct thread_worker_pair *pair = ao2_alloc(sizeof(*pair), thread_worker_pair_destructor);
+       if (!pair) {
+               return NULL;
+       }
+       pair->pool = pool;
+       ao2_ref(worker, +1);
+       pair->worker = worker;
+       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;
+
+       ao2_link(pair->pool->idle_threads, pair->worker);
+       ao2_unlink(pair->pool->active_threads, pair->worker);
+
+       threadpool_send_state_changed(pair->pool);
+
+       ao2_ref(pair, -1);
+       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)
+{
+       struct thread_worker_pair *pair;
+       SCOPED_AO2LOCK(lock, pool);
+       if (pool->shutting_down) {
+               return;
+       }
+       pair = thread_worker_pair_alloc(pool, worker);
+       if (!pair) {
+               return;
+       }
+       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);
+       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)
+{
+       struct thread_worker_pair *pair;
+       SCOPED_AO2LOCK(lock, pool);
+       if (pool->shutting_down) {
+               return;
+       }
+       pair = thread_worker_pair_alloc(pool, worker);
+       if (!pair) {
+               return;
+       }
+       ast_taskprocessor_push(pool->control_tps, queued_zombie_thread_dead, pair);
+}
+
+static int queued_idle_thread_dead(void *data)
+{
+       struct thread_worker_pair *pair = data;
+
+       ao2_unlink(pair->pool->idle_threads, pair->worker);
+       threadpool_send_state_changed(pair->pool);
+
+       ao2_ref(pair, -1);
+       return 0;
+}
+
+static void threadpool_idle_thread_dead(struct ast_threadpool *pool,
+               struct worker_thread *worker)
+{
+       struct thread_worker_pair *pair;
+       SCOPED_AO2LOCK(lock, pool);
+       if (pool->shutting_down) {
+               return;
+       }
+       pair = thread_worker_pair_alloc(pool, worker);
+       if (!pair) {
+               return;
+       }
+       ast_taskprocessor_push(pool->control_tps, queued_idle_thread_dead, pair);
+}
+
+/*!
+ * \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)
+{
+       ao2_lock(pool);
+       if (!pool->shutting_down) {
+               ao2_unlock(pool);
+               return ast_taskprocessor_execute(pool->tps);
+       }
+       ao2_unlock(pool);
+       return 0;
+}
+
+/*!
+ * \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 = obj;
+       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 name The name of the threadpool.
+ * \param options The options the threadpool uses.
+ * \retval NULL Could not initialize threadpool properly
+ * \retval non-NULL The newly-allocated threadpool
+ */
+static void *threadpool_alloc(const char *name, const struct ast_threadpool_options *options)
+{
+       RAII_VAR(struct ast_threadpool *, pool,
+                       ao2_alloc(sizeof(*pool), threadpool_destructor), ao2_cleanup);
+       struct ast_str *control_tps_name = ast_str_create(64);
+
+       if (!control_tps_name) {
+               return NULL;
+       }
+
+       ast_str_set(&control_tps_name, 0, "%s-control", name);
+
+       pool->control_tps = ast_taskprocessor_get(ast_str_buffer(control_tps_name), TPS_REF_DEFAULT);
+       ast_free(control_tps_name);
+       if (!pool->control_tps) {
+               return NULL;
+       }
+       pool->active_threads = ao2_container_alloc(THREAD_BUCKETS, worker_thread_hash, worker_thread_cmp);
+       if (!pool->active_threads) {
+               return NULL;
+       }
+       pool->idle_threads = ao2_container_alloc(THREAD_BUCKETS, worker_thread_hash, worker_thread_cmp);
+       if (!pool->idle_threads) {
+               return NULL;
+       }
+       pool->zombie_threads = ao2_container_alloc(THREAD_BUCKETS, worker_thread_hash, worker_thread_cmp);
+       if (!pool->zombie_threads) {
+               return NULL;
+       }
+       pool->options = *options;
+
+       ao2_ref(pool, +1);
+       return pool;
+}
+
+static int threadpool_tps_start(struct ast_taskprocessor_listener *listener)
+{
+       return 0;
+}
+
+/*!
+ * \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 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)
+{
+       struct task_pushed_data *tpd = ao2_alloc(sizeof(*tpd), NULL);
+
+       if (!tpd) {
+               return NULL;
+       }
+       tpd->pool = pool;
+       tpd->was_empty = was_empty;
+       return tpd;
+}
+
+/*!
+ * \brief Activate idle threads
+ *
+ * This function always returns CMP_MATCH because all workers 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_thread(void *obj, void *arg, int flags)
+{
+       struct worker_thread *worker = obj;
+       struct ast_threadpool *pool = arg;
+
+       if (!ao2_link(pool->active_threads, worker)) {
+               /* If we can't link the idle thread into the active container, then
+                * we'll just leave the thread idle and not wake it up.
+                */
+               ast_log(LOG_WARNING, "Failed to activate thread %d. Remaining idle\n",
+                               worker->id);
+               return 0;
+       }
+       worker_set_state(worker, ALIVE);
+       return CMP_MATCH;
+}
+
+/*!
+ * \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;
+
+       int current_size = ao2_container_count(pool->active_threads) +
+               ao2_container_count(pool->idle_threads);
+
+       if (pool->options.max_size && current_size + delta > pool->options.max_size) {
+               delta = pool->options.max_size - current_size;
+       }
+
+       ast_debug(3, "Increasing threadpool %s's size by %d\n",
+                       ast_taskprocessor_name(pool->tps), delta);
+
+       for (i = 0; i < delta; ++i) {
+               struct worker_thread *worker = worker_thread_alloc(pool);
+               if (!worker) {
+                       return;
+               }
+               if (ao2_link(pool->active_threads, worker)) {
+                       if (worker_thread_start(worker)) {
+                               ast_log(LOG_ERROR, "Unable to start worker thread %d. Destroying.\n", worker->id);
+                               ao2_unlink(pool->active_threads, worker);
+                       }
+               } else {
+                       ast_log(LOG_WARNING, "Failed to activate worker thread %d. Destroying.\n", worker->id);
+               }
+               ao2_ref(worker, -1);
+       }
+}
+
+/*!
+ * \brief Queued 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 queued_task_pushed(void *data)
+{
+       struct task_pushed_data *tpd = data;
+       struct ast_threadpool *pool = tpd->pool;
+       int was_empty = tpd->was_empty;
+       int state_changed;
+
+       if (pool->listener && pool->listener->callbacks->task_pushed) {
+               pool->listener->callbacks->task_pushed(pool, pool->listener, was_empty);
+       }
+       if (ao2_container_count(pool->idle_threads) == 0) {
+               if (pool->options.auto_increment > 0) {
+                       grow(pool, pool->options.auto_increment);
+                       state_changed = 1;
+               }
+       } else {
+               ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA,
+                               activate_thread, pool);
+               state_changed = 1;
+       }
+       if (state_changed) {
+               threadpool_send_state_changed(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)
+{
+       struct ast_threadpool *pool = ast_taskprocessor_listener_get_user_data(listener);
+       struct task_pushed_data *tpd;
+       SCOPED_AO2LOCK(lock, pool);
+
+       if (pool->shutting_down) {
+               return;
+       }
+       tpd = task_pushed_data_alloc(pool, was_empty);
+       if (!tpd) {
+               return;
+       }
+
+       ast_taskprocessor_push(pool->control_tps, queued_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 queued_emptied(void *data)
+{
+       struct ast_threadpool *pool = data;
+
+       /* We already checked for existence of this callback when this was queued */
+       pool->listener->callbacks->emptied(pool, pool->listener);
+       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 = ast_taskprocessor_listener_get_user_data(listener);
+       SCOPED_AO2LOCK(lock, pool);
+
+       if (pool->shutting_down) {
+               return;
+       }
+
+       if (pool->listener && pool->listener->callbacks->emptied) {
+               ast_taskprocessor_push(pool->control_tps, queued_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)
+{
+       struct ast_threadpool *pool = ast_taskprocessor_listener_get_user_data(listener);
+
+       if (pool->listener && pool->listener->callbacks->shutdown) {
+               pool->listener->callbacks->shutdown(pool->listener);
+       }
+       ao2_cleanup(pool->active_threads);
+       ao2_cleanup(pool->idle_threads);
+       ao2_cleanup(pool->zombie_threads);
+       ao2_cleanup(pool);
+}
+
+/*!
+ * \brief Table of taskprocessor listener callbacks for threadpool's main taskprocessor
+ */
+static struct ast_taskprocessor_listener_callbacks threadpool_tps_listener_callbacks = {
+       .start = threadpool_tps_start,
+       .task_pushed = threadpool_tps_task_pushed,
+       .emptied = threadpool_tps_emptied,
+       .shutdown = threadpool_tps_shutdown,
+};
+
+/*!
+ * \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)
+{
+       int *num_to_kill = arg;
+
+       if (*num_to_kill > 0) {
+               --(*num_to_kill);
+               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;
+       struct ast_threadpool *pool = arg;
+       int *num_to_zombify = data;
+
+       if ((*num_to_zombify)-- > 0) {
+               if (!ao2_link(pool->zombie_threads, worker)) {
+                       ast_log(LOG_WARNING, "Failed to zombify active thread %d. Thread will remain active\n", worker->id);
+                       return 0;
+               }
+               worker_set_state(worker, ZOMBIE);
+               return CMP_MATCH;
+       } else {
+               return CMP_STOP;
+       }
+}
+
+/*!
+ * \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)
+{
+       /*
+        * Preference is to kill idle threads, but
+        * we'll move on to deactivating active threads
+        * if we have to
+        */
+       int idle_threads = ao2_container_count(pool->idle_threads);
+       int idle_threads_to_kill = MIN(delta, idle_threads);
+       int active_threads_to_zombify = delta - idle_threads_to_kill;
+
+       ast_debug(3, "Destroying %d idle threads in threadpool %s\n", idle_threads_to_kill,
+                       ast_taskprocessor_name(pool->tps));
+
+       ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
+                       kill_threads, &idle_threads_to_kill);
+
+       ast_debug(3, "Destroying %d active threads in threadpool %s\n", active_threads_to_zombify,
+                       ast_taskprocessor_name(pool->tps));
+
+       ao2_callback_data(pool->active_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
+                       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 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)
+{
+       struct set_size_data *ssd = ao2_alloc(sizeof(*ssd), NULL);
+       if (!ssd) {
+               return NULL;
+       }
+
+       ssd->pool = pool;
+       ssd->size = size;
+       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;
+       struct ast_threadpool *pool = ssd->pool;
+       unsigned int num_threads = ssd->size;
+
+       /* We don't count zombie threads as being "live" when potentially resizing */
+       unsigned int current_size = ao2_container_count(pool->active_threads) +
+               ao2_container_count(pool->idle_threads);
+
+       if (current_size == num_threads) {
+               ast_log(LOG_NOTICE, "Not changing threadpool size since new size %u is the same as current %u\n",
+                               num_threads, current_size);
+               return 0;
+       }
+
+       if (current_size < num_threads) {
+               grow(pool, num_threads - current_size);
+       } else {
+               shrink(pool, current_size - num_threads);
+       }
+
+       threadpool_send_state_changed(pool);
+       ao2_ref(ssd, -1);
+       return 0;
+}
+
+void ast_threadpool_set_size(struct ast_threadpool *pool, unsigned int size)
+{
+       struct set_size_data *ssd;
+       SCOPED_AO2LOCK(lock, pool);
+       if (pool->shutting_down) {
+               return;
+       }
+
+       ssd = set_size_data_alloc(pool, size);
+       if (!ssd) {
+               return;
+       }
+
+       ast_taskprocessor_push(pool->control_tps, queued_set_size, ssd);
+}
+
+struct ast_threadpool_listener *ast_threadpool_listener_alloc(
+               const struct ast_threadpool_listener_callbacks *callbacks, void *user_data)
+{
+       struct ast_threadpool_listener *listener = ao2_alloc(sizeof(*listener), NULL);
+       if (!listener) {
+               return NULL;
+       }
+       listener->callbacks = callbacks;
+       listener->user_data = user_data;
+       return listener;
+}
+
+void *ast_threadpool_listener_get_user_data(const struct ast_threadpool_listener *listener)
+{
+       return listener->user_data;
+}
+
+struct pool_options_pair {
+       struct ast_threadpool *pool;
+       struct ast_threadpool_options options;
+};
+
+struct ast_threadpool *ast_threadpool_create(const char *name,
+               struct ast_threadpool_listener *listener,
+               const struct ast_threadpool_options *options)
+{
+       struct ast_taskprocessor *tps;
+       RAII_VAR(struct ast_taskprocessor_listener *, tps_listener, NULL, ao2_cleanup);
+       RAII_VAR(struct ast_threadpool *, pool, threadpool_alloc(name, options), ao2_cleanup);
+
+       if (!pool) {
+               return NULL;
+       }
+       
+       tps_listener = ast_taskprocessor_listener_alloc(&threadpool_tps_listener_callbacks, pool);
+       if (!tps_listener) {
+               return NULL;
+       }
+
+       if (options->version != AST_THREADPOOL_OPTIONS_VERSION) {
+               ast_log(LOG_WARNING, "Incompatible version of threadpool options in use.\n");
+               return NULL;
+       }
+
+       tps = ast_taskprocessor_create_with_listener(name, tps_listener);
+       if (!tps) {
+               return NULL;
+       }
+
+       pool->tps = tps;
+       if (listener) {
+               ao2_ref(listener, +1);
+               pool->listener = listener;
+       }
+       ast_threadpool_set_size(pool, pool->options.initial_size);
+       ao2_ref(pool, +1);
+       return pool;
+}
+
+int ast_threadpool_push(struct ast_threadpool *pool, int (*task)(void *data), void *data)
+{
+       SCOPED_AO2LOCK(lock, pool);
+       if (!pool->shutting_down) {
+               return ast_taskprocessor_push(pool->tps, task, data);
+       }
+       return 0;
+}
+
+void ast_threadpool_shutdown(struct ast_threadpool *pool)
+{
+       if (!pool) {
+               return;
+       }
+       /* Shut down the taskprocessors and everything else just
+        * takes care of itself via the taskprocessor callbacks
+        */
+       ao2_lock(pool);
+       pool->shutting_down = 1;
+       ao2_unlock(pool);
+       ast_taskprocessor_unreference(pool->control_tps);
+       ast_taskprocessor_unreference(pool->tps);
+}
+
+/*!
+ * 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)
+{
+       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;
+       ast_debug(3, "Destroying worker thread %d\n", worker->id);
+       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;
+       worker->options = pool->options;
+       return worker;
+}
+
+static int worker_thread_start(struct worker_thread *worker)
+{
+       return ast_pthread_create(&worker->thread, NULL, worker_start, 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)
+{
+       struct timeval start = ast_tvnow();
+       struct timespec end = {
+               .tv_sec = start.tv_sec + worker->options.idle_timeout,
+               .tv_nsec = start.tv_usec * 1000,
+       };
+       SCOPED_MUTEX(lock, &worker->lock);
+       if (worker->state != ALIVE) {
+               return 0;
+       }
+       threadpool_active_thread_idle(worker->pool, worker);
+       while (!worker->wake_up) {
+               if (worker->options.idle_timeout <= 0) {
+                       ast_cond_wait(&worker->cond, lock);
+               } else if (ast_cond_timedwait(&worker->cond, lock, &end) == ETIMEDOUT) {
+                       break;
+               }
+       }
+
+       if (!worker->wake_up) {
+               ast_debug(1, "Worker thread idle timeout reached. Dying.\n");
+               threadpool_idle_thread_dead(worker->pool, worker);
+               worker->state = DEAD;
+       }
+       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);
+}
+
diff --git a/tests/test_taskprocessor.c b/tests/test_taskprocessor.c
new file mode 100644 (file)
index 0000000..e370dd7
--- /dev/null
@@ -0,0 +1,469 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2012-2013, Digium, Inc.
+ *
+ * Mark Michelson <mmichelson@digium.com>
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE file
+ * at the top of the source tree.
+ */
+
+/*!
+ * \file
+ * \brief taskprocessor unit tests
+ *
+ * \author Mark Michelson <mmichelson@digium.com>
+ *
+ */
+
+/*** MODULEINFO
+       <depend>TEST_FRAMEWORK</depend>
+       <support_level>core</support_level>
+ ***/
+
+#include "asterisk.h"
+
+#include "asterisk/test.h"
+#include "asterisk/taskprocessor.h"
+#include "asterisk/module.h"
+#include "asterisk/astobj2.h"
+
+/*!
+ * \brief userdata associated with baseline taskprocessor test
+ */
+struct task_data {
+       /* Condition used to signal to queuing thread that task was executed */
+       ast_cond_t cond;
+       /* Lock protecting the condition */
+       ast_mutex_t lock;
+       /*! Boolean indicating that the task was run */
+       int task_complete;
+};
+
+/*!
+ * \brief Queued task for baseline test.
+ *
+ * The task simply sets a boolean to indicate the
+ * task has been run and then signals a condition
+ * saying it's complete
+ */
+static int task(void *data)
+{
+       struct task_data *task_data = data;
+       SCOPED_MUTEX(lock, &task_data->lock);
+       task_data->task_complete = 1;
+       ast_cond_signal(&task_data->cond);
+       return 0;
+}
+
+/*!
+ * \brief Baseline test for default taskprocessor
+ *
+ * This test ensures that when a task is added to a taskprocessor that
+ * has been allocated with a default listener that the task gets executed
+ * as expected
+ */
+AST_TEST_DEFINE(default_taskprocessor)
+{
+       struct ast_taskprocessor *tps;
+       struct task_data task_data;
+       struct timeval start;
+       struct timespec ts;
+       enum ast_test_result_state res = AST_TEST_PASS;
+       int timedwait_res;
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "default_taskprocessor";
+               info->category = "/main/taskprocessor/";
+               info->summary = "Test of default taskproccesor";
+               info->description =
+                       "Ensures that a queued task gets executed.";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tps = ast_taskprocessor_get("test", TPS_REF_DEFAULT);
+
+       if (!tps) {
+               ast_test_status_update(test, "Unable to create test taskprocessor\n");
+               return AST_TEST_FAIL;
+       }
+
+       start = ast_tvnow();
+
+       ts.tv_sec = start.tv_sec + 30;
+       ts.tv_nsec = start.tv_usec * 1000;
+
+       ast_cond_init(&task_data.cond, NULL);
+       ast_mutex_init(&task_data.lock);
+       task_data.task_complete = 0;
+
+       ast_taskprocessor_push(tps, task, &task_data);
+       ast_mutex_lock(&task_data.lock);
+       while (!task_data.task_complete) {
+               timedwait_res = ast_cond_timedwait(&task_data.cond, &task_data.lock, &ts);
+               if (timedwait_res == ETIMEDOUT) {
+                       break;
+               }
+       }
+       ast_mutex_unlock(&task_data.lock);
+
+       if (!task_data.task_complete) {
+               ast_test_status_update(test, "Queued task did not execute!\n");
+               res = AST_TEST_FAIL;
+               goto test_end;
+       }
+
+test_end:
+       tps = ast_taskprocessor_unreference(tps);
+       ast_mutex_destroy(&task_data.lock);
+       ast_cond_destroy(&task_data.cond);
+       return res;
+}
+
+#define NUM_TASKS 20000
+
+/*!
+ * \brief Relevant data associated with taskprocessor load test
+ */
+static struct load_task_data {
+       /*! Condition used to indicate a task has completed executing */
+       ast_cond_t cond;
+       /*! Lock used to protect the condition */
+       ast_mutex_t lock;
+       /*! Counter of the number of completed tasks */
+       int tasks_completed;
+       /*! Storage for task-specific data */
+       int task_rand[NUM_TASKS];
+} load_task_results;
+
+/*!
+ * \brief a queued task to be used in the taskprocessor load test
+ *
+ * The task increments the number of tasks executed and puts the passed-in
+ * data into the next slot in the array of random data.
+ */
+static int load_task(void *data)
+{
+       int *randdata = data;
+       SCOPED_MUTEX(lock, &load_task_results.lock);
+       load_task_results.task_rand[load_task_results.tasks_completed++] = *randdata;
+       ast_cond_signal(&load_task_results.cond);
+       return 0;
+}
+
+/*!
+ * \brief Load test for taskprocessor with default listener
+ *
+ * This test queues a large number of tasks, each with random data associated.
+ * The test ensures that all of the tasks are run and that the tasks are executed
+ * in the same order that they were queued
+ */
+AST_TEST_DEFINE(default_taskprocessor_load)
+{
+       struct ast_taskprocessor *tps;
+       struct timeval start;
+       struct timespec ts;
+       enum ast_test_result_state res = AST_TEST_PASS;
+       int timedwait_res;
+       int i;
+       int rand_data[NUM_TASKS];
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "default_taskprocessor_load";
+               info->category = "/main/taskprocessor/";
+               info->summary = "Load test of default taskproccesor";
+               info->description =
+                       "Ensure that a large number of queued tasks are executed in the proper order.";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tps = ast_taskprocessor_get("test", TPS_REF_DEFAULT);
+
+       if (!tps) {
+               ast_test_status_update(test, "Unable to create test taskprocessor\n");
+               return AST_TEST_FAIL;
+       }
+
+       start = ast_tvnow();
+
+       ts.tv_sec = start.tv_sec + 60;
+       ts.tv_nsec = start.tv_usec * 1000;
+
+       ast_cond_init(&load_task_results.cond, NULL);
+       ast_mutex_init(&load_task_results.lock);
+       load_task_results.tasks_completed = 0;
+
+       for (i = 0; i < NUM_TASKS; ++i) {
+               rand_data[i] = ast_random();
+               ast_taskprocessor_push(tps, load_task, &rand_data[i]);
+       }
+
+       ast_mutex_lock(&load_task_results.lock);
+       while (load_task_results.tasks_completed < NUM_TASKS) {
+               timedwait_res = ast_cond_timedwait(&load_task_results.cond, &load_task_results.lock, &ts);
+               if (timedwait_res == ETIMEDOUT) {
+                       break;
+               }
+       }
+       ast_mutex_unlock(&load_task_results.lock);
+
+       if (load_task_results.tasks_completed != NUM_TASKS) {
+               ast_test_status_update(test, "Unexpected number of tasks executed. Expected %d but got %d\n",
+                               NUM_TASKS, load_task_results.tasks_completed);
+               res = AST_TEST_FAIL;
+               goto test_end;
+       }
+
+       for (i = 0; i < NUM_TASKS; ++i) {
+               if (rand_data[i] != load_task_results.task_rand[i]) {
+                       ast_test_status_update(test, "Queued tasks did not execute in order\n");
+                       res = AST_TEST_FAIL;
+                       goto test_end;
+               }
+       }
+
+test_end:
+       tps = ast_taskprocessor_unreference(tps);
+       ast_mutex_destroy(&load_task_results.lock);
+       ast_cond_destroy(&load_task_results.cond);
+       return res;
+}
+
+/*!
+ * \brief Private data for the test taskprocessor listener
+ */
+struct test_listener_pvt {
+       /* Counter of number of tasks pushed to the queue */
+       int num_pushed;
+       /* Counter of number of times the queue was emptied */
+       int num_emptied;
+       /* Counter of number of times that a pushed task occurred on an empty queue */
+       int num_was_empty;
+       /* Boolean indicating whether the shutdown callback was called */
+       int shutdown;
+};
+
+/*!
+ * \brief test taskprocessor listener's alloc callback
+ */
+static void *test_listener_pvt_alloc(void)
+{
+       struct test_listener_pvt *pvt;
+
+       pvt = ast_calloc(1, sizeof(*pvt));
+       return pvt;
+}
+
+/*!
+ * \brief test taskprocessor listener's start callback
+ */
+static int test_start(struct ast_taskprocessor_listener *listener)
+{
+       return 0;
+}
+
+/*!
+ * \brief test taskprocessor listener's task_pushed callback
+ *
+ * Adjusts private data's stats as indicated by the parameters.
+ */
+static void test_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
+{
+       struct test_listener_pvt *pvt = ast_taskprocessor_listener_get_user_data(listener);
+       ++pvt->num_pushed;
+       if (was_empty) {
+               ++pvt->num_was_empty;
+       }
+}
+
+/*!
+ * \brief test taskprocessor listener's emptied callback.
+ */
+static void test_emptied(struct ast_taskprocessor_listener *listener)
+{
+       struct test_listener_pvt *pvt = ast_taskprocessor_listener_get_user_data(listener);
+       ++pvt->num_emptied;
+}
+
+/*!
+ * \brief test taskprocessor listener's shutdown callback.
+ */
+static void test_shutdown(struct ast_taskprocessor_listener *listener)
+{
+       struct test_listener_pvt *pvt = ast_taskprocessor_listener_get_user_data(listener);
+       pvt->shutdown = 1;
+}
+
+static const struct ast_taskprocessor_listener_callbacks test_callbacks = {
+       .start = test_start,
+       .task_pushed = test_task_pushed,
+       .emptied = test_emptied,
+       .shutdown = test_shutdown,
+};
+
+/*!
+ * \brief Queued task for taskprocessor listener test.
+ *
+ * Does nothing.
+ */
+static int listener_test_task(void *ignore)
+{
+       return 0;
+}
+
+/*!
+ * \brief helper to ensure that statistics the listener is keeping are what we expect
+ *
+ * \param test The currently-running test
+ * \param pvt The private data for the taskprocessor listener
+ * \param num_pushed The expected current number of tasks pushed to the processor
+ * \param num_emptied The expected current number of times the taskprocessor has become empty
+ * \param num_was_empty The expected current number of times that tasks were pushed to an empty taskprocessor
+ * \retval -1 Stats were not as expected
+ * \retval 0 Stats were as expected
+ */
+static int check_stats(struct ast_test *test, const struct test_listener_pvt *pvt, int num_pushed, int num_emptied, int num_was_empty)
+{
+       if (pvt->num_pushed != num_pushed) {
+               ast_test_status_update(test, "Unexpected number of tasks pushed. Expected %d but got %d\n",
+                               num_pushed, pvt->num_pushed);
+               return -1;
+       }
+
+       if (pvt->num_emptied != num_emptied) {
+               ast_test_status_update(test, "Unexpected number of empties. Expected %d but got %d\n",
+                               num_emptied, pvt->num_emptied);
+               return -1;
+       }
+
+       if (pvt->num_was_empty != num_was_empty) {
+               ast_test_status_update(test, "Unexpected number of empties. Expected %d but got %d\n",
+                               num_was_empty, pvt->num_emptied);
+               return -1;
+       }
+
+       return 0;
+}
+
+/*!
+ * \brief Test for a taskprocessor with custom listener.
+ *
+ * This test pushes tasks to a taskprocessor with a custom listener, executes the taskss,
+ * and destroys the taskprocessor.
+ *
+ * The test ensures that the listener's callbacks are called when expected and that the data
+ * being passed in is accurate.
+ */
+AST_TEST_DEFINE(taskprocessor_listener)
+{
+       struct ast_taskprocessor *tps = NULL;
+       struct ast_taskprocessor_listener *listener = NULL;
+       struct test_listener_pvt *pvt = NULL;
+       enum ast_test_result_state res = AST_TEST_PASS;
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "taskprocessor_listener";
+               info->category = "/main/taskprocessor/";
+               info->summary = "Test of taskproccesor listeners";
+               info->description =
+                       "Ensures that listener callbacks are called when expected.";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       pvt = test_listener_pvt_alloc();
+       if (!pvt) {
+               ast_test_status_update(test, "Unable to allocate test taskprocessor listener user data\n");
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_taskprocessor_listener_alloc(&test_callbacks, pvt);
+       if (!listener) {
+               ast_test_status_update(test, "Unable to allocate test taskprocessor listener\n");
+               res = AST_TEST_FAIL;
+               goto test_exit;
+       }
+
+       tps = ast_taskprocessor_create_with_listener("test_listener", listener);
+       if (!tps) {
+               ast_test_status_update(test, "Unable to allocate test taskprocessor\n");
+               res = AST_TEST_FAIL;
+               goto test_exit;
+       }
+
+       ast_taskprocessor_push(tps, listener_test_task, NULL);
+
+       if (check_stats(test, pvt, 1, 0, 1) < 0) {
+               res = AST_TEST_FAIL;
+               goto test_exit;
+       }
+
+       ast_taskprocessor_push(tps, listener_test_task, NULL);
+
+       if (check_stats(test, pvt, 2, 0, 1) < 0) {
+               res = AST_TEST_FAIL;
+               goto test_exit;
+       }
+
+       ast_taskprocessor_execute(tps);
+
+       if (check_stats(test, pvt, 2, 0, 1) < 0) {
+               res = AST_TEST_FAIL;
+               goto test_exit;
+       }
+
+       ast_taskprocessor_execute(tps);
+
+       if (check_stats(test, pvt, 2, 1, 1) < 0) {
+               res = AST_TEST_FAIL;
+               goto test_exit;
+       }
+
+       tps = ast_taskprocessor_unreference(tps);
+
+       if (!pvt->shutdown) {
+               res = AST_TEST_FAIL;
+               goto test_exit;
+       }
+
+test_exit:
+       ao2_cleanup(listener);
+       /* This is safe even if tps is NULL */
+       ast_taskprocessor_unreference(tps);
+       ast_free(pvt);
+       return res;
+}
+
+static int unload_module(void)
+{
+       ast_test_unregister(default_taskprocessor);
+       ast_test_unregister(default_taskprocessor_load);
+       ast_test_unregister(taskprocessor_listener);
+       return 0;
+}
+
+static int load_module(void)
+{
+       ast_test_register(default_taskprocessor);
+       ast_test_register(default_taskprocessor_load);
+       ast_test_register(taskprocessor_listener);
+       return AST_MODULE_LOAD_SUCCESS;
+}
+
+AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "taskprocessor test module");
diff --git a/tests/test_threadpool.c b/tests/test_threadpool.c
new file mode 100644 (file)
index 0000000..712b858
--- /dev/null
@@ -0,0 +1,1430 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2012-2013, Digium, Inc.
+ *
+ * Mark Michelson <mmichelson@digium.com>
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE file
+ * at the top of the source tree.
+ */
+
+/*!
+ * \file
+ * \brief threadpool unit tests
+ *
+ * \author Mark Michelson <mmichelson@digium.com>
+ *
+ */
+
+/*** MODULEINFO
+       <depend>TEST_FRAMEWORK</depend>
+       <support_level>core</support_level>
+ ***/
+
+#include "asterisk.h"
+
+#include "asterisk/test.h"
+#include "asterisk/threadpool.h"
+#include "asterisk/module.h"
+#include "asterisk/lock.h"
+#include "asterisk/astobj2.h"
+#include "asterisk/logger.h"
+
+struct test_listener_data {
+       int num_active;
+       int num_idle;
+       int task_pushed;
+       int num_tasks;
+       int empty_notice;
+       int was_empty;
+       ast_mutex_t lock;
+       ast_cond_t cond;
+};
+
+static struct test_listener_data *test_alloc(void)
+{
+       struct test_listener_data *tld = ast_calloc(1, sizeof(*tld));
+       if (!tld) {
+               return NULL;
+       }
+       ast_mutex_init(&tld->lock);
+       ast_cond_init(&tld->cond, NULL);
+       return tld;
+}
+
+static void test_state_changed(struct ast_threadpool *pool,
+               struct ast_threadpool_listener *listener,
+               int active_threads,
+               int idle_threads)
+{
+       struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
+       SCOPED_MUTEX(lock, &tld->lock);
+       tld->num_active = active_threads;
+       tld->num_idle = idle_threads;
+       ast_log(LOG_NOTICE, "Thread state: %d active, %d idle\n", tld->num_active, tld->num_idle);
+       ast_cond_signal(&tld->cond);
+}
+
+static void test_task_pushed(struct ast_threadpool *pool,
+               struct ast_threadpool_listener *listener,
+               int was_empty)
+{
+       struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
+       SCOPED_MUTEX(lock, &tld->lock);
+       tld->task_pushed = 1;
+       ++tld->num_tasks;
+       tld->was_empty = was_empty;
+       ast_cond_signal(&tld->cond);
+}
+
+static void test_emptied(struct ast_threadpool *pool,
+               struct ast_threadpool_listener *listener)
+{
+       struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
+       SCOPED_MUTEX(lock, &tld->lock);
+       tld->empty_notice = 1;
+       ast_cond_signal(&tld->cond);
+}
+
+static void test_shutdown(struct ast_threadpool_listener *listener)
+{
+       struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
+       ast_cond_destroy(&tld->cond);
+       ast_mutex_destroy(&tld->lock);
+}
+
+static const struct ast_threadpool_listener_callbacks test_callbacks = {
+       .state_changed = test_state_changed,
+       .task_pushed = test_task_pushed,
+       .emptied = test_emptied,
+       .shutdown = test_shutdown,
+};
+
+struct simple_task_data {
+       int task_executed;
+       ast_mutex_t lock;
+       ast_cond_t cond;
+};
+
+static struct simple_task_data *simple_task_data_alloc(void)
+{
+       struct simple_task_data *std = ast_calloc(1, sizeof(*std));
+
+       if (!std) {
+               return NULL;
+       }
+       ast_mutex_init(&std->lock);
+       ast_cond_init(&std->cond, NULL);
+       return std;
+}
+
+static int simple_task(void *data)
+{
+       struct simple_task_data *std = data;
+       SCOPED_MUTEX(lock, &std->lock);
+       std->task_executed = 1;
+       ast_cond_signal(&std->cond);
+       return 0;
+}
+
+static enum ast_test_result_state wait_until_thread_state(struct ast_test *test, struct test_listener_data *tld, int num_active, int num_idle)
+{
+       struct timeval start = ast_tvnow();
+       struct timespec end = {
+               .tv_sec = start.tv_sec + 5,
+               .tv_nsec = start.tv_usec * 1000
+       };
+       enum ast_test_result_state res = AST_TEST_PASS;
+       SCOPED_MUTEX(lock, &tld->lock);
+
+       while (!(tld->num_active == num_active && tld->num_idle == num_idle)) {
+               if (ast_cond_timedwait(&tld->cond, &tld->lock, &end) == ETIMEDOUT) {
+                       break;
+               }
+       }
+
+       if (tld->num_active != num_active && tld->num_idle != num_idle) {
+               ast_test_status_update(test, "Number of active threads and idle threads not what was expected.\n");
+               ast_test_status_update(test, "Expected %d active threads but got %d\n", num_active, tld->num_active);
+               ast_test_status_update(test, "Expected %d idle threads but got %d\n", num_idle, tld->num_idle);
+               res = AST_TEST_FAIL;
+       }
+
+       return res;
+}
+
+static void wait_for_task_pushed(struct ast_threadpool_listener *listener)
+{
+       struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
+       struct timeval start = ast_tvnow();
+       struct timespec end = {
+               .tv_sec = start.tv_sec + 5,
+               .tv_nsec = start.tv_usec * 1000
+       };
+       SCOPED_MUTEX(lock, &tld->lock);
+
+       while (!tld->task_pushed) {
+               if (ast_cond_timedwait(&tld->cond, lock, &end) == ETIMEDOUT) {
+                       break;
+               }
+       }
+}
+
+static enum ast_test_result_state wait_for_completion(struct ast_test *test, struct simple_task_data *std)
+{
+       struct timeval start = ast_tvnow();
+       struct timespec end = {
+               .tv_sec = start.tv_sec + 5,
+               .tv_nsec = start.tv_usec * 1000
+       };
+       enum ast_test_result_state res = AST_TEST_PASS;
+       SCOPED_MUTEX(lock, &std->lock);
+
+       while (!std->task_executed) {
+               if (ast_cond_timedwait(&std->cond, lock, &end) == ETIMEDOUT) {
+                       break;
+               }
+       }
+
+       if (!std->task_executed) {
+               ast_test_status_update(test, "Task execution did not occur\n");
+               res = AST_TEST_FAIL;
+       }
+       return res;
+}
+
+static enum ast_test_result_state wait_for_empty_notice(struct ast_test *test, struct test_listener_data *tld)
+{
+       struct timeval start = ast_tvnow();
+       struct timespec end = {
+               .tv_sec = start.tv_sec + 5,
+               .tv_nsec = start.tv_usec * 1000
+       };
+       enum ast_test_result_state res = AST_TEST_PASS;
+       SCOPED_MUTEX(lock, &tld->lock);
+
+       while (!tld->empty_notice) {
+               if (ast_cond_timedwait(&tld->cond, lock, &end) == ETIMEDOUT) {
+                       break;
+               }
+       }
+
+       if (!tld->empty_notice) {
+               ast_test_status_update(test, "Test listener not notified that threadpool is empty\n");
+               res = AST_TEST_FAIL;
+       }
+
+       return res;
+}
+
+static enum ast_test_result_state listener_check(
+               struct ast_test *test,
+               struct ast_threadpool_listener *listener,
+               int task_pushed,
+               int was_empty,
+               int num_tasks,
+               int num_active,
+               int num_idle,
+               int empty_notice)
+{
+       struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
+       enum ast_test_result_state res = AST_TEST_PASS;
+
+       if (tld->task_pushed != task_pushed) {
+               ast_test_status_update(test, "Expected task %sto be pushed, but it was%s\n",
+                               task_pushed ? "" : "not ", tld->task_pushed ? "" : " not");
+               res = AST_TEST_FAIL;
+       }
+       if (tld->was_empty != was_empty) {
+               ast_test_status_update(test, "Expected %sto be empty, but it was%s\n",
+                               was_empty ? "" : "not ", tld->was_empty ? "" : " not");
+               res = AST_TEST_FAIL;
+       }
+       if (tld->num_tasks!= num_tasks) {
+               ast_test_status_update(test, "Expected %d tasks to be pushed, but got %d\n",
+                               num_tasks, tld->num_tasks);
+               res = AST_TEST_FAIL;
+       }
+       if (tld->num_active != num_active) {
+               ast_test_status_update(test, "Expected %d active threads, but got %d\n",
+                               num_active, tld->num_active);
+               res = AST_TEST_FAIL;
+       }
+       if (tld->num_idle != num_idle) {
+               ast_test_status_update(test, "Expected %d idle threads, but got %d\n",
+                               num_idle, tld->num_idle);
+               res = AST_TEST_FAIL;
+       }
+       if (tld->empty_notice != empty_notice) {
+               ast_test_status_update(test, "Expected %s empty notice, but got %s\n",
+                               was_empty ? "an" : "no", tld->task_pushed ? "one" : "none");
+               res = AST_TEST_FAIL;
+       }
+
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_push)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct simple_task_data *std = NULL;
+       struct test_listener_data *tld = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "push";
+               info->category = "/main/threadpool/";
+               info->summary = "Test task";
+               info->description =
+                       "Basic threadpool test";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       std = simple_task_data_alloc();
+       if (!std) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, simple_task, std);
+
+       wait_for_task_pushed(listener);
+
+       res = listener_check(test, listener, 1, 1, 1, 0, 0, 0);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(std);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_initial_threads)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 3,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "initial_threads";
+               info->category = "/main/threadpool/";
+               info->summary = "Test threadpool initialization state";
+               info->description =
+                       "Ensure that a threadpool created with a specific size contains the\n"
+                       "proper number of idle threads.";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 3);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(tld);
+       return res;
+}
+
+
+AST_TEST_DEFINE(threadpool_thread_creation)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "thread_creation";
+               info->category = "/main/threadpool/";
+               info->summary = "Test threadpool thread creation";
+               info->description =
+                       "Ensure that threads can be added to a threadpool";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       /* Now let's create a thread. It should start active, then go
+        * idle immediately
+        */
+       ast_threadpool_set_size(pool, 1);
+
+       res = wait_until_thread_state(test, tld, 0, 1);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_thread_destruction)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "thread_destruction";
+               info->category = "/main/threadpool/";
+               info->summary = "Test threadpool thread destruction";
+               info->description =
+                       "Ensure that threads are properly destroyed in a threadpool";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       ast_threadpool_set_size(pool, 3);
+
+       res = wait_until_thread_state(test, tld, 0, 3);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 0, 0, 0, 0, 3, 0);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       ast_threadpool_set_size(pool, 2);
+
+       res = wait_until_thread_state(test, tld, 0, 2);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_thread_timeout)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 2,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "thread_timeout";
+               info->category = "/main/threadpool/";
+               info->summary = "Test threadpool thread timeout";
+               info->description =
+                       "Ensure that a thread with a two second timeout dies as expected.";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       ast_threadpool_set_size(pool, 1);
+
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 0, 0, 0, 0, 1, 0);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 0);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 0, 0, 0, 0, 0, 0);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_one_task_one_thread)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct simple_task_data *std = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "one_task_one_thread";
+               info->category = "/main/threadpool/";
+               info->summary = "Test a single task with a single thread";
+               info->description =
+                       "Push a task into an empty threadpool, then add a thread to the pool.";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       std = simple_task_data_alloc();
+       if (!std) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, simple_task, std);
+
+       ast_threadpool_set_size(pool, 1);
+
+       /* Threads added to the pool are active when they start,
+        * so the newly-created thread should immediately execute
+        * the waiting task.
+        */
+       res = wait_for_completion(test, std);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_for_empty_notice(test, tld);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       /* After completing the task, the thread should go idle */
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(std);
+       ast_free(tld);
+       return res;
+
+}
+
+AST_TEST_DEFINE(threadpool_one_thread_one_task)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct simple_task_data *std = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "one_thread_one_task";
+               info->category = "/main/threadpool/";
+               info->summary = "Test a single thread with a single task";
+               info->description =
+                       "Add a thread to the pool and then push a task to it.";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       std = simple_task_data_alloc();
+       if (!std) {
+               goto end;
+       }
+
+       ast_threadpool_set_size(pool, 1);
+
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, simple_task, std);
+
+       res = wait_for_completion(test, std);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_for_empty_notice(test, tld);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       /* After completing the task, the thread should go idle */
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(std);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_one_thread_multiple_tasks)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct simple_task_data *std1 = NULL;
+       struct simple_task_data *std2 = NULL;
+       struct simple_task_data *std3 = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "one_thread_multiple_tasks";
+               info->category = "/main/threadpool/";
+               info->summary = "Test a single thread with multiple tasks";
+               info->description =
+                       "Add a thread to the pool and then push three tasks to it.";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       std1 = simple_task_data_alloc();
+       std2 = simple_task_data_alloc();
+       std3 = simple_task_data_alloc();
+       if (!std1 || !std2 || !std3) {
+               goto end;
+       }
+
+       ast_threadpool_set_size(pool, 1);
+
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, simple_task, std1);
+       ast_threadpool_push(pool, simple_task, std2);
+       ast_threadpool_push(pool, simple_task, std3);
+
+       res = wait_for_completion(test, std1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+       res = wait_for_completion(test, std2);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+       res = wait_for_completion(test, std3);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_for_empty_notice(test, tld);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 0, 3, 0, 1, 1);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(std1);
+       ast_free(std2);
+       ast_free(std3);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_auto_increment)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct simple_task_data *std1 = NULL;
+       struct simple_task_data *std2 = NULL;
+       struct simple_task_data *std3 = NULL;
+       struct simple_task_data *std4 = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 3,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "auto_increment";
+               info->category = "/main/threadpool/";
+               info->summary = "Test that the threadpool grows as tasks are added";
+               info->description =
+                       "Create an empty threadpool and push a task to it. Once the task is\n"
+                       "pushed, the threadpool should add three threads and be able to\n"
+                       "handle the task. The threads should then go idle\n";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       std1 = simple_task_data_alloc();
+       std2 = simple_task_data_alloc();
+       std3 = simple_task_data_alloc();
+       std4 = simple_task_data_alloc();
+       if (!std1 || !std2 || !std3 || !std4) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, simple_task, std1);
+
+       /* Pushing the task should result in the threadpool growing
+        * by three threads. This will allow the task to actually execute
+        */
+       res = wait_for_completion(test, std1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_for_empty_notice(test, tld);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 3);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       /* Now push three tasks into the pool and ensure the pool does not
+        * grow.
+        */
+       ast_threadpool_push(pool, simple_task, std2);
+       ast_threadpool_push(pool, simple_task, std3);
+       ast_threadpool_push(pool, simple_task, std4);
+
+       res = wait_for_completion(test, std2);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+       res = wait_for_completion(test, std3);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+       res = wait_for_completion(test, std4);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_for_empty_notice(test, tld);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 3);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+       res = listener_check(test, listener, 1, 0, 4, 0, 3, 1);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(std1);
+       ast_free(std2);
+       ast_free(std3);
+       ast_free(std4);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_max_size)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct simple_task_data *std = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 3,
+               .initial_size = 0,
+               .max_size = 2,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "max_size";
+               info->category = "/main/threadpool/";
+               info->summary = "Test that the threadpool does not exceed its maximum size restriction";
+               info->description =
+                       "Create an empty threadpool and push a task to it. Once the task is\n"
+                       "pushed, the threadpool should attempt to grow by three threads, but the\n"
+                       "pool's restrictions should only allow two threads to be added.\n";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       std = simple_task_data_alloc();
+       if (!std) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, simple_task, std);
+
+       res = wait_for_completion(test, std);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 2);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 1, 1, 0, 2, 1);
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(std);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_reactivation)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct simple_task_data *std1 = NULL;
+       struct simple_task_data *std2 = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "reactivation";
+               info->category = "/main/threadpool/";
+               info->summary = "Test that a threadpool reactivates when work is added";
+               info->description =
+                       "Push a task into a threadpool. Make sure the task executes and the\n"
+                       "thread goes idle. Then push a second task and ensure that the thread\n"
+                       "awakens and executes the second task.\n";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       std1 = simple_task_data_alloc();
+       std2 = simple_task_data_alloc();
+       if (!std1 || !std2) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, simple_task, std1);
+
+       ast_threadpool_set_size(pool, 1);
+
+       res = wait_for_completion(test, std1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_for_empty_notice(test, tld);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       /* Now make sure the threadpool reactivates when we add a second task */
+       ast_threadpool_push(pool, simple_task, std2);
+
+       res = wait_for_completion(test, std2);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_for_empty_notice(test, tld);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 1, 2, 0, 1, 1);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(std1);
+       ast_free(std2);
+       ast_free(tld);
+       return res;
+
+}
+
+struct complex_task_data {
+       int task_executed;
+       int continue_task;
+       ast_mutex_t lock;
+       ast_cond_t stall_cond;
+       ast_cond_t done_cond;
+};
+
+static struct complex_task_data *complex_task_data_alloc(void)
+{
+       struct complex_task_data *ctd = ast_calloc(1, sizeof(*ctd));
+
+       if (!ctd) {
+               return NULL;
+       }
+       ast_mutex_init(&ctd->lock);
+       ast_cond_init(&ctd->stall_cond, NULL);
+       ast_cond_init(&ctd->done_cond, NULL);
+       return ctd;
+}
+
+static int complex_task(void *data)
+{
+       struct complex_task_data *ctd = data;
+       SCOPED_MUTEX(lock, &ctd->lock);
+       while (!ctd->continue_task) {
+               ast_cond_wait(&ctd->stall_cond, lock);
+       }
+       /* We got poked. Finish up */
+       ctd->task_executed = 1;
+       ast_cond_signal(&ctd->done_cond);
+       return 0;
+}
+
+static void poke_worker(struct complex_task_data *ctd)
+{
+       SCOPED_MUTEX(lock, &ctd->lock);
+       ctd->continue_task = 1;
+       ast_cond_signal(&ctd->stall_cond);
+}
+
+static enum ast_test_result_state wait_for_complex_completion(struct complex_task_data *ctd)
+{
+       struct timeval start = ast_tvnow();
+       struct timespec end = {
+               .tv_sec = start.tv_sec + 5,
+               .tv_nsec = start.tv_usec * 1000
+       };
+       enum ast_test_result_state res = AST_TEST_PASS;
+       SCOPED_MUTEX(lock, &ctd->lock);
+
+       while (!ctd->task_executed) {
+               if (ast_cond_timedwait(&ctd->done_cond, lock, &end) == ETIMEDOUT) {
+                       break;
+               }
+       }
+
+       if (!ctd->task_executed) {
+               res = AST_TEST_FAIL;
+       }
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_task_distribution)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct complex_task_data *ctd1 = NULL;
+       struct complex_task_data *ctd2 = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "task_distribution";
+               info->category = "/main/threadpool/";
+               info->summary = "Test that tasks are evenly distributed to threads";
+               info->description =
+                       "Push two tasks into a threadpool. Ensure that each is handled by\n"
+                       "a separate thread\n";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       ctd1 = complex_task_data_alloc();
+       ctd2 = complex_task_data_alloc();
+       if (!ctd1 || !ctd2) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, complex_task, ctd1);
+       ast_threadpool_push(pool, complex_task, ctd2);
+
+       ast_threadpool_set_size(pool, 2);
+
+       res = wait_until_thread_state(test, tld, 2, 0);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 0, 2, 2, 0, 0);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       /* The tasks are stalled until we poke them */
+       poke_worker(ctd1);
+       poke_worker(ctd2);
+
+       res = wait_for_complex_completion(ctd1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+       res = wait_for_complex_completion(ctd2);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 2);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 0, 2, 0, 2, 1);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(ctd1);
+       ast_free(ctd2);
+       ast_free(tld);
+       return res;
+}
+
+AST_TEST_DEFINE(threadpool_more_destruction)
+{
+       struct ast_threadpool *pool = NULL;
+       struct ast_threadpool_listener *listener = NULL;
+       struct complex_task_data *ctd1 = NULL;
+       struct complex_task_data *ctd2 = NULL;
+       enum ast_test_result_state res = AST_TEST_FAIL;
+       struct test_listener_data *tld = NULL;
+       struct ast_threadpool_options options = {
+               .version = AST_THREADPOOL_OPTIONS_VERSION,
+               .idle_timeout = 0,
+               .auto_increment = 0,
+               .initial_size = 0,
+               .max_size = 0,
+       };
+
+       switch (cmd) {
+       case TEST_INIT:
+               info->name = "more_destruction";
+               info->category = "/main/threadpool/";
+               info->summary = "Test that threads are destroyed as expected";
+               info->description =
+                       "Push two tasks into a threadpool. Set the threadpool size to 4\n"
+                       "Ensure that there are 2 active and 2 idle threads. Then shrink the\n"
+                       "threadpool down to 1 thread. Ensure that the thread leftover is active\n"
+                       "and ensure that both tasks complete.\n";
+               return AST_TEST_NOT_RUN;
+       case TEST_EXECUTE:
+               break;
+       }
+
+       tld = test_alloc();
+       if (!tld) {
+               return AST_TEST_FAIL;
+       }
+
+       listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
+       if (!listener) {
+               goto end;
+       }
+
+       pool = ast_threadpool_create(info->name, listener, &options);
+       if (!pool) {
+               goto end;
+       }
+
+       ctd1 = complex_task_data_alloc();
+       ctd2 = complex_task_data_alloc();
+       if (!ctd1 || !ctd2) {
+               goto end;
+       }
+
+       ast_threadpool_push(pool, complex_task, ctd1);
+       ast_threadpool_push(pool, complex_task, ctd2);
+
+       ast_threadpool_set_size(pool, 4);
+
+       res = wait_until_thread_state(test, tld, 2, 2);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 0, 2, 2, 2, 0);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       ast_threadpool_set_size(pool, 1);
+
+       /* Shrinking the threadpool should kill off the two idle threads
+        * and one of the active threads.
+        */
+       res = wait_until_thread_state(test, tld, 1, 0);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 0, 2, 1, 0, 0);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       /* The tasks are stalled until we poke them */
+       poke_worker(ctd1);
+       poke_worker(ctd2);
+
+       res = wait_for_complex_completion(ctd1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+       res = wait_for_complex_completion(ctd2);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = wait_until_thread_state(test, tld, 0, 1);
+       if (res == AST_TEST_FAIL) {
+               goto end;
+       }
+
+       res = listener_check(test, listener, 1, 0, 2, 0, 1, 1);
+
+end:
+       ast_threadpool_shutdown(pool);
+       ao2_cleanup(listener);
+       ast_free(ctd1);
+       ast_free(ctd2);
+       ast_free(tld);
+       return res;
+}
+
+static int unload_module(void)
+{
+       ast_test_unregister(threadpool_push);
+       ast_test_unregister(threadpool_initial_threads);
+       ast_test_unregister(threadpool_thread_creation);
+       ast_test_unregister(threadpool_thread_destruction);
+       ast_test_unregister(threadpool_thread_timeout);
+       ast_test_unregister(threadpool_one_task_one_thread);
+       ast_test_unregister(threadpool_one_thread_one_task);
+       ast_test_unregister(threadpool_one_thread_multiple_tasks);
+       ast_test_unregister(threadpool_auto_increment);
+       ast_test_unregister(threadpool_max_size);
+       ast_test_unregister(threadpool_reactivation);
+       ast_test_unregister(threadpool_task_distribution);
+       ast_test_unregister(threadpool_more_destruction);
+       return 0;
+}
+
+static int load_module(void)
+{
+       ast_test_register(threadpool_push);
+       ast_test_register(threadpool_initial_threads);
+       ast_test_register(threadpool_thread_creation);
+       ast_test_register(threadpool_thread_destruction);
+       ast_test_register(threadpool_thread_timeout);
+       ast_test_register(threadpool_one_task_one_thread);
+       ast_test_register(threadpool_one_thread_one_task);
+       ast_test_register(threadpool_one_thread_multiple_tasks);
+       ast_test_register(threadpool_auto_increment);
+       ast_test_register(threadpool_max_size);
+       ast_test_register(threadpool_reactivation);
+       ast_test_register(threadpool_task_distribution);
+       ast_test_register(threadpool_more_destruction);
+       return AST_MODULE_LOAD_SUCCESS;
+}
+
+AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "threadpool test module");