[asterisk-commits] mmichelson: trunk r379432 - in /trunk: ./ include/asterisk/ main/ tests/
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Fri Jan 18 09:42:15 CST 2013
Author: mmichelson
Date: Fri Jan 18 09:42:10 2013
New Revision: 379432
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=379432
Log:
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
Added:
trunk/include/asterisk/threadpool.h (with props)
trunk/main/threadpool.c (with props)
trunk/tests/test_taskprocessor.c (with props)
trunk/tests/test_threadpool.c (with props)
Modified:
trunk/ (props changed)
trunk/include/asterisk/taskprocessor.h
trunk/main/taskprocessor.c
Propchange: trunk/
------------------------------------------------------------------------------
svn:mergeinfo = /team/mmichelson/threadpool:376024-379375
Modified: trunk/include/asterisk/taskprocessor.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/taskprocessor.h?view=diff&rev=379432&r1=379431&r2=379432
==============================================================================
--- trunk/include/asterisk/taskprocessor.h (original)
+++ trunk/include/asterisk/taskprocessor.h Fri Jan 18 09:42:10 2013
@@ -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 at digium.com>
*
@@ -22,22 +22,33 @@
*
* \author Dwayne M. Hubbard <dhubbard at 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.
- *
- * 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
+ * \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.
+ *
+ * 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 @@
/*!
* \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,19 +71,108 @@
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
* \since 1.6.1
*/
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.
@@ -97,6 +197,17 @@
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
*/
Added: trunk/include/asterisk/threadpool.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/threadpool.h?view=auto&rev=379432
==============================================================================
--- trunk/include/asterisk/threadpool.h (added)
+++ trunk/include/asterisk/threadpool.h Fri Jan 18 09:42:10 2013
@@ -1,0 +1,180 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2012-2013, Digium, Inc.
+ *
+ * Mark Michelson <mmmichelson at 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 */
Propchange: trunk/include/asterisk/threadpool.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: trunk/include/asterisk/threadpool.h
------------------------------------------------------------------------------
svn:keywords = Author Date Id Revision
Propchange: trunk/include/asterisk/threadpool.h
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: trunk/main/taskprocessor.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/taskprocessor.c?view=diff&rev=379432&r1=379431&r2=379432
==============================================================================
--- trunk/main/taskprocessor.c (original)
+++ trunk/main/taskprocessor.c Fri Jan 18 09:42:10 2013
@@ -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 at digium.com>
*
@@ -37,7 +37,6 @@
#include "asterisk/astobj2.h"
#include "asterisk/cli.h"
#include "asterisk/taskprocessor.h"
-
/*!
* \brief tps_task structure is queued to a taskprocessor
@@ -67,14 +66,6 @@
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 @@
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;
@@ -120,6 +134,94 @@
static struct ast_cli_entry taskprocessor_clis[] = {
AST_CLI_DEFINE(cli_tps_ping, "Ping a named task processor"),
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,
};
/*!
@@ -291,118 +393,67 @@
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;
-
- return ast_str_case_hash(tps->name);
+ const char *name = flags & OBJ_KEY ? obj : 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;
-
- return !strcasecmp(lhs->name, rhs->name) ? CMP_MATCH | CMP_STOP : 0;
+ const char *rhsname = flags & OBJ_KEY ? arg : rhs->name;
+
+ 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;
-
- if (!tps) {
- ast_log(LOG_ERROR, "missing taskprocessor\n");
- return NULL;
- }
- ast_mutex_lock(&tps->taskprocessor_lock);
+ SCOPED_AO2LOCK(lock, tps);
+
+ if (tps->shutting_down) {
+ return NULL;
+ }
+
if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) {
tps->tps_queue_size--;
}
- ast_mutex_unlock(&tps->taskprocessor_lock);
return task;
}
@@ -419,6 +470,96 @@
return NULL;
}
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
@@ -426,75 +567,74 @@
* 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);
- 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);
- 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);
- 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);
- return NULL;
- }
- ao2_unlock(tps_singletons);
+ return NULL;
+ }
+ /* Create a new taskprocessor. Start by creating a default listener */
+ pvt = default_listener_pvt_alloc();
+ if (!pvt) {
+ return NULL;
+ }
+ listener = ast_taskprocessor_listener_alloc(&default_listener_callbacks, pvt);
+ if (!listener) {
+ default_listener_pvt_destroy(pvt);
+ return NULL;
+ }
+
+ p = __allocate_taskprocessor(name, listener);
+ if (!p) {
+ default_listener_pvt_destroy(pvt);
+ ao2_ref(listener, -1);
+ return NULL;
+ }
+
+ /* 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;
+ }
+ 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 @@
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 @@
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;
+}
Added: trunk/main/threadpool.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/threadpool.c?view=auto&rev=379432
==============================================================================
--- trunk/main/threadpool.c (added)
+++ trunk/main/threadpool.c Fri Jan 18 09:42:10 2013
@@ -1,0 +1,1105 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2012-2013, Digium, Inc.
+ *
+ * Mark Michelson <mmmichelson at 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);
[... 2888 lines stripped ...]
More information about the asterisk-commits
mailing list