Add threadpool support to Asterisk.

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

@@ -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>
  *
@@ -22,22 +22,33 @@
  *
  * \author Dwayne M. Hubbard <dhubbard@digium.com>
  *
- * \note A taskprocessor is a named singleton containing a processing thread and
+ * \note A taskprocessor is a named object containing a task queue that
- * a task queue that serializes tasks pushed into it by [a] module(s) that reference the taskprocessor.  
+ * 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()
+ * A taskprocessor is created the first time its name is requested via the
- * function and destroyed when the taskprocessor reference count reaches zero.
+ * 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
+ * A task is a wrapper around a task-handling function pointer and a data
- * to be processed by the singleton processing thread when the task is popped off the front 
+ * pointer.  A task is pushed into a taskprocessor queue using the
- * 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 
  * 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 after the task handling function returns.  A module releases its
- * taskprocessor using the ast_taskprocessor_unreference() function which may result in the
+ * reference to a taskprocessor using the ast_taskprocessor_unreference() function which
- * destruction of the taskprocessor if the taskprocessor's reference count reaches zero.  Tasks waiting
+ * may result in the destruction of the taskprocessor if the taskprocessor's reference
- * to be processed in the taskprocessor queue when the taskprocessor reference count reaches zero
+ * count reaches zero. When the taskprocessor's reference count reaches zero, its
- * will be purged and released from the taskprocessor queue without being processed.
+ * 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__
@@ -60,6 +71,79 @@ 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
  *
@@ -74,6 +158,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.
  *
@@ -96,6 +196,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

include/asterisk/threadpool.h

@@ -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 */

main/taskprocessor.c

@@ -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) {
+	if (tps->shutting_down) {
-		ast_log(LOG_ERROR, "missing taskprocessor\n");
 		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 = {
+	struct ast_taskprocessor *p;
-		.name = name,
+	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, name, OBJ_KEY);
-	p = ao2_find(tps_singletons, &tmp_tps, OBJ_POINTER);
 	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 */
+	/* Create a new taskprocessor. Start by creating a default listener */
-	if (!(p = ao2_alloc(sizeof(*p), tps_taskprocessor_destroy))) {
+	pvt = default_listener_pvt_alloc();
-		ao2_unlock(tps_singletons);
+	if (!pvt) {
-		ast_log(LOG_WARNING, "failed to create taskprocessor '%s'\n", name);
 		return NULL;
 	}
-
+	listener = ast_taskprocessor_listener_alloc(&default_listener_callbacks, pvt);
-	ast_cond_init(&p->poll_cond, NULL);
+	if (!listener) {
-	ast_mutex_init(&p->taskprocessor_lock);
+		default_listener_pvt_destroy(pvt);
-
-	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);
 		return NULL;
 	}
-	if (!(p->name = ast_strdup(name))) {
+
-		ao2_unlock(tps_singletons);
+	p = __allocate_taskprocessor(name, listener);
-		ao2_ref(p, -1);
+	if (!p) {
-		return NULL;
+		default_listener_pvt_destroy(pvt);
-	}
+		ao2_ref(listener, -1);
-	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);
 		return NULL;
 	}
-	if (!(ao2_link(tps_singletons, p))) {
+
-		ao2_unlock(tps_singletons);
+	/* Unref listener here since the taskprocessor has gained a reference to the listener */
-		ast_log(LOG_ERROR, "Failed to add taskprocessor '%s' to container\n", p->name);
+	ao2_ref(listener, -1);
-		ao2_ref(p, -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 __allocate_taskprocessor(name, listener);
-	return p;
 }
 
 /* decrement the taskprocessor reference count and unlink from the container if necessary */
 void *ast_taskprocessor_unreference(struct ast_taskprocessor *tps)
 {
-	if (tps) {
+	if (!tps) {
-		ao2_lock(tps_singletons);
+		return NULL;
-		ao2_unlink(tps_singletons, tps);
-		if (ao2_ref(tps, -1) > 1) {
-			ao2_link(tps_singletons, tps);
-		}
-		ao2_unlock(tps_singletons);
 	}
+
+	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++;
+	previous_size = tps->tps_queue_size++;
-	ast_cond_signal(&tps->poll_cond);
+	ao2_unlock(tps);
-	ast_mutex_unlock(&tps->taskprocessor_lock);
+	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;
+}

tests/test_taskprocessor.c

@@ -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");