threadpool.c File Reference

#include "asterisk.h"
#include "asterisk/threadpool.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/astobj2.h"
#include "asterisk/utils.h"

Include dependency graph for threadpool.c:

Go to the source code of this file.

Data Structures
struct	ast_serializer_shutdown_group
struct	ast_threadpool
	An opaque threadpool structure. More...
struct	ast_threadpool_listener
	listener for a threadpool More...
struct	pool_options_pair
struct	serializer
struct	set_size_data
	Helper struct used for queued operations that change the size of the threadpool. More...
struct	task_pushed_data
	helper used for queued task when tasks are pushed More...
struct	thread_worker_pair
	Struct used for queued operations involving worker state changes. More...
struct	worker_thread

Macros
#define	THREAD_BUCKETS   89

Enumerations
enum	worker_state { ALIVE, ZOMBIE, DEAD }
	states for worker threads More...

Functions
static int	activate_thread (void *obj, void *arg, int flags)
	Activate idle threads. More...
struct ast_serializer_shutdown_group *	ast_serializer_shutdown_group_alloc (void)
	Create a serializer group shutdown control object. More...
int	ast_serializer_shutdown_group_join (struct ast_serializer_shutdown_group *shutdown_group, int timeout)
	Wait for the serializers in the group to shutdown with timeout. More...
struct ast_threadpool *	ast_threadpool_create (const char *name, struct ast_threadpool_listener *listener, const struct ast_threadpool_options *options)
	Create a new threadpool. More...
struct ast_threadpool_listener *	ast_threadpool_listener_alloc (const struct ast_threadpool_listener_callbacks *callbacks, void *user_data)
	Allocate a threadpool listener. More...
void *	ast_threadpool_listener_get_user_data (const struct ast_threadpool_listener *listener)
	Get the threadpool listener's user data. More...
int	ast_threadpool_push (struct ast_threadpool *pool, int(*task)(void *data), void *data)
	Push a task to the threadpool. More...
long	ast_threadpool_queue_size (struct ast_threadpool *pool)
	Return the size of the threadpool's task queue. More...
struct ast_taskprocessor *	ast_threadpool_serializer (const char *name, struct ast_threadpool *pool)
	Serialized execution of tasks within a ast_threadpool. More...
struct ast_taskprocessor *	ast_threadpool_serializer_get_current (void)
	Get the threadpool serializer currently associated with this thread. More...
struct ast_taskprocessor *	ast_threadpool_serializer_group (const char *name, struct ast_threadpool *pool, struct ast_serializer_shutdown_group *shutdown_group)
	Serialized execution of tasks within a ast_threadpool. More...
void	ast_threadpool_set_size (struct ast_threadpool *pool, unsigned int size)
	Set the number of threads for the thread pool. More...
void	ast_threadpool_shutdown (struct ast_threadpool *pool)
	Shut down a threadpool and destroy it. More...
	AST_THREADSTORAGE_RAW (current_serializer)
static int	execute_tasks (void *data)
static void	grow (struct ast_threadpool *pool, int delta)
	Add threads to the threadpool. More...
static int	kill_threads (void *obj, void *arg, int flags)
	ao2 callback to kill a set number of threads. More...
static int	queued_active_thread_idle (void *data)
	Move a worker thread from the active container to the idle container. More...
static int	queued_emptied (void *data)
	Queued task that handles the case where the threadpool's taskprocessor is emptied. More...
static int	queued_idle_thread_dead (void *data)
static int	queued_set_size (void *data)
	Change the size of the threadpool. More...
static int	queued_task_pushed (void *data)
	Queued task called when tasks are pushed into the threadpool. More...
static int	queued_zombie_thread_dead (void *data)
	Kill a zombie thread. More...
static struct serializer *	serializer_create (struct ast_threadpool *pool, struct ast_serializer_shutdown_group *shutdown_group)
static void	serializer_dtor (void *obj)
static void	serializer_shutdown (struct ast_taskprocessor_listener *listener)
static void	serializer_shutdown_group_dec (struct ast_serializer_shutdown_group *shutdown_group)
static void	serializer_shutdown_group_dtor (void *vdoomed)
static void	serializer_shutdown_group_inc (struct ast_serializer_shutdown_group *shutdown_group)
static int	serializer_start (struct ast_taskprocessor_listener *listener)
static void	serializer_task_pushed (struct ast_taskprocessor_listener *listener, int was_empty)
static struct set_size_data *	set_size_data_alloc (struct ast_threadpool *pool, unsigned int size)
	Allocate and initialize a set_size_data. More...
static void	shrink (struct ast_threadpool *pool, int delta)
	Remove threads from the threadpool. More...
static struct task_pushed_data *	task_pushed_data_alloc (struct ast_threadpool *pool, int was_empty)
	Allocate and initialize a task_pushed_data. More...
static struct thread_worker_pair *	thread_worker_pair_alloc (struct ast_threadpool *pool, struct worker_thread *worker)
	Allocate and initialize a thread_worker_pair. More...
static void	thread_worker_pair_free (struct thread_worker_pair *pair)
	Destructor for thread_worker_pair. More...
static void	threadpool_active_thread_idle (struct ast_threadpool *pool, struct worker_thread *worker)
	Queue a task to move a thread from the active list to the idle list. More...
static struct ast_threadpool *	threadpool_alloc (const char *name, const struct ast_threadpool_options *options)
static void	threadpool_destructor (void *obj)
	Destroy a threadpool's components. More...
static int	threadpool_execute (struct ast_threadpool *pool)
	Execute a task in the threadpool. More...
static void	threadpool_idle_thread_dead (struct ast_threadpool *pool, struct worker_thread *worker)
static void	threadpool_send_state_changed (struct ast_threadpool *pool)
	Notify the threadpool listener that the state has changed. More...
static void	threadpool_tps_emptied (struct ast_taskprocessor_listener *listener)
	Taskprocessor listener emptied callback. More...
static void	threadpool_tps_shutdown (struct ast_taskprocessor_listener *listener)
	Taskprocessor listener shutdown callback. More...
static int	threadpool_tps_start (struct ast_taskprocessor_listener *listener)
static void	threadpool_tps_task_pushed (struct ast_taskprocessor_listener *listener, int was_empty)
	Taskprocessor listener callback called when a task is added. More...
static void	threadpool_zombie_thread_dead (struct ast_threadpool *pool, struct worker_thread *worker)
	Queue a task to kill a zombie thread. More...
static void	worker_active (struct worker_thread *worker)
	Active loop for worker threads. More...
static int	worker_idle (struct worker_thread *worker)
	Idle function for worker threads. More...
static int	worker_set_state (struct worker_thread *worker, enum worker_state state)
	Change a worker's state. More...
static void	worker_shutdown (struct worker_thread *worker)
	shut a worker thread down More...
static void *	worker_start (void *arg)
	start point for worker threads More...
static struct worker_thread *	worker_thread_alloc (struct ast_threadpool *pool)
	Allocate and initialize a new worker thread. More...
static int	worker_thread_cmp (void *obj, void *arg, int flags)
static void	worker_thread_destroy (void *obj)
	Worker thread destructor. More...
static int	worker_thread_hash (const void *obj, int flags)
static int	worker_thread_start (struct worker_thread *worker)
static int	zombify_threads (void *obj, void *arg, void *data, int flags)
	ao2 callback to zombify a set number of threads. More...

Variables
static struct ast_taskprocessor_listener_callbacks	serializer_tps_listener_callbacks
static struct ast_taskprocessor_listener_callbacks	threadpool_tps_listener_callbacks
	Table of taskprocessor listener callbacks for threadpool's main taskprocessor. More...
static int	worker_id_counter

Macro Definition Documentation

THREAD_BUCKETS

#define THREAD_BUCKETS   89

Definition at line 28 of file threadpool.c.

Referenced by threadpool_alloc().

Enumeration Type Documentation

worker_state

enum worker_state

states for worker threads

Enumerator
ALIVE	The worker is either active or idle
ZOMBIE	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.
DEAD	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.

Definition at line 120 of file threadpool.c.

                  {
   /*! 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,
};

Function Documentation

activate_thread()

static int activate_thread ( void *  obj, void *  arg, int  flags  )

static

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.

Parameters

obj	The worker to activate
arg	The pool where the worker belongs

Return values

CMP_MATCH

Definition at line 491 of file threadpool.c.

References ast_threadpool::active_threads, ALIVE, ao2_link, ao2_unlink, ast_debug, ast_log, CMP_MATCH, worker_thread::id, LOG_WARNING, pool, and worker_set_state().

Referenced by queued_set_size(), and queued_task_pushed().

{
   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;
   }

   if (worker_set_state(worker, ALIVE)) {
      ast_debug(1, "Failed to activate thread %d. It is dead\n",
            worker->id);
      /* The worker thread will no longer exist in the active threads or
       * idle threads container after this.
       */
      ao2_unlink(pool->active_threads, worker);
   }

   return CMP_MATCH;
}

ast_serializer_shutdown_group_alloc()

struct ast_serializer_shutdown_group* ast_serializer_shutdown_group_alloc

(

void

)

Create a serializer group shutdown control object.

Since

13.5.0

Returns

ao2 object to control shutdown of a serializer group.

Definition at line 1229 of file threadpool.c.

References ao2_alloc, ast_cond_init, ast_serializer_shutdown_group::cond, NULL, serializer_shutdown_group_dtor(), and shutdown_group.

Referenced by ast_serializer_pool_create(), load_module(), and sip_options_init_task().

{
   struct ast_serializer_shutdown_group *shutdown_group;

   shutdown_group = ao2_alloc(sizeof(*shutdown_group), serializer_shutdown_group_dtor);
   if (!shutdown_group) {
      return NULL;
   }
   ast_cond_init(&shutdown_group->cond, NULL);
   return shutdown_group;
}

ast_serializer_shutdown_group_join()

int ast_serializer_shutdown_group_join	(	struct ast_serializer_shutdown_group *	shutdown_group,
		int	timeout
	)

Wait for the serializers in the group to shutdown with timeout.

Since

13.5.0

Parameters

shutdown_group	Group shutdown controller. (Returns 0 immediately if NULL)
timeout	Number of seconds to wait for the serializers in the group to shutdown. Zero if the timeout is disabled.

Returns

Number of seriaizers that did not get shutdown within the timeout.

Definition at line 1241 of file threadpool.c.

References ao2_lock, ao2_object_get_lockaddr(), ao2_unlock, ast_assert, ast_cond_timedwait, ast_cond_wait, ast_tvnow(), ast_serializer_shutdown_group::cond, ast_serializer_shutdown_group::count, lock, NULL, and timeout.

Referenced by ast_res_pjsip_cleanup_options_handling(), ast_serializer_pool_destroy(), and unload_module().

{
   int remaining;
   ast_mutex_t *lock;

   if (!shutdown_group) {
      return 0;
   }

   lock = ao2_object_get_lockaddr(shutdown_group);
   ast_assert(lock != NULL);

   ao2_lock(shutdown_group);
   if (timeout) {
      struct timeval start;
      struct timespec end;

      start = ast_tvnow();
      end.tv_sec = start.tv_sec + timeout;
      end.tv_nsec = start.tv_usec * 1000;
      while (shutdown_group->count) {
         if (ast_cond_timedwait(&shutdown_group->cond, lock, &end)) {
            /* Error or timed out waiting for the count to reach zero. */
            break;
         }
      }
   } else {
      while (shutdown_group->count) {
         if (ast_cond_wait(&shutdown_group->cond, lock)) {
            /* Error */
            break;
         }
      }
   }
   remaining = shutdown_group->count;
   ao2_unlock(shutdown_group);
   return remaining;
}

ast_threadpool_create()

struct ast_threadpool* ast_threadpool_create	(	const char *	name,
		struct ast_threadpool_listener *	listener,
		const struct ast_threadpool_options *	options
	)

Create a new threadpool.

This function creates a threadpool. Tasks may be pushed onto this thread pool 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.

Parameters

name	The unique name for the threadpool
listener	The listener the threadpool will notify of changes. Can be NULL.
options	The behavioral options for this threadpool

Return values

NULL	Failed to create the threadpool
non-NULL	The newly-created threadpool

Definition at line 915 of file threadpool.c.

References ao2_cleanup, ao2_ref, ast_alloca, ast_log, ast_taskprocessor_create_with_listener(), ast_taskprocessor_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_set_size(), ast_threadpool::listener, LOG_WARNING, NULL, pool, RAII_VAR, threadpool_alloc(), ast_threadpool::tps, and ast_threadpool_options::version.

Referenced by ast_sorcery_init(), AST_TEST_DEFINE(), load_module(), and stasis_init().

{
   struct ast_taskprocessor *tps;
   RAII_VAR(struct ast_taskprocessor_listener *, tps_listener, NULL, ao2_cleanup);
   RAII_VAR(struct ast_threadpool *, pool, NULL, ao2_cleanup);
   char *fullname;

   pool = threadpool_alloc(name, options);
   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;
   }

   fullname = ast_alloca(strlen(name) + strlen("/pool") + 1);
   sprintf(fullname, "%s/pool", name); /* Safe */
   tps = ast_taskprocessor_create_with_listener(fullname, 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;
}

ast_threadpool_listener_alloc()

struct ast_threadpool_listener* ast_threadpool_listener_alloc	(	const struct ast_threadpool_listener_callbacks *	callbacks,
		void *	user_data
	)

Allocate a threadpool listener.

This function will call back into the alloc callback for the listener.

Parameters

callbacks	Listener callbacks to assign to the listener
user_data	User data to be stored in the threadpool listener

Return values

NULL	Failed to allocate the listener
non-NULL	The newly-created threadpool listener

Definition at line 893 of file threadpool.c.

References ao2_alloc, ast_threadpool_listener::callbacks, ast_threadpool::listener, NULL, and ast_threadpool_listener::user_data.

Referenced by AST_TEST_DEFINE().

{
   struct ast_threadpool_listener *listener = ao2_alloc(sizeof(*listener), NULL);
   if (!listener) {
      return NULL;
   }
   listener->callbacks = callbacks;
   listener->user_data = user_data;
   return listener;
}

ast_threadpool_listener_get_user_data()

void* ast_threadpool_listener_get_user_data

(

const struct ast_threadpool_listener *

listener

)

Get the threadpool listener's user data.

Parameters

listener

The threadpool listener

Returns

The user data

Definition at line 905 of file threadpool.c.

References ast_threadpool_listener::user_data.

Referenced by listener_check(), test_emptied(), test_shutdown(), test_state_changed(), test_task_pushed(), and wait_for_task_pushed().

{
   return listener->user_data;
}

ast_threadpool_push()

int ast_threadpool_push	(	struct ast_threadpool *	pool,
		int(*)(void *data)	task,
		void *	data
	)

Push a task to the threadpool.

Tasks pushed into the threadpool will be automatically taken by one of the threads within

Parameters

pool	The threadpool to add the task to
task	The task to add
data	The parameter for the task

Return values

0	success
-1	failure

Definition at line 956 of file threadpool.c.

References ast_taskprocessor_push(), lock, SCOPED_AO2LOCK, ast_threadpool::shutting_down, task(), and ast_threadpool::tps.

Referenced by AST_TEST_DEFINE(), and serializer_task_pushed().

{
   SCOPED_AO2LOCK(lock, pool);
   if (!pool->shutting_down) {
      return ast_taskprocessor_push(pool->tps, task, data);
   }
   return -1;
}

ast_threadpool_queue_size()

long ast_threadpool_queue_size

(

struct ast_threadpool *

pool

)

Return the size of the threadpool's task queue.

Since

13.7.0

Definition at line 1437 of file threadpool.c.

References ast_taskprocessor_size(), and ast_threadpool::tps.

Referenced by ast_sip_threadpool_queue_size().

{
   return ast_taskprocessor_size(pool->tps);
}

ast_threadpool_serializer()

struct ast_taskprocessor* ast_threadpool_serializer	(	const char *	name,
		struct ast_threadpool *	pool
	)

Serialized execution of tasks within a ast_threadpool.

Since

12.0.0

A ast_taskprocessor with the same contract as a default taskprocessor (tasks execute serially) except instead of executing out of a dedicated thread, execution occurs in a thread from a ast_threadpool. Think of it as a lightweight thread.

While it guarantees that each task will complete before executing the next, there is no guarantee as to which thread from the pool individual tasks will execute. This normally only matters if your code relys on thread specific information, such as thread locals.

Use ast_taskprocessor_unreference() to dispose of the returned ast_taskprocessor.

Only a single taskprocessor with a given name may exist. This function will fail if a taskprocessor with the given name already exists.

Parameters

name	Name of the serializer. (must be unique)
pool	ast_threadpool for execution.

Returns

ast_taskprocessor for enqueuing work.

NULL on error.

Definition at line 1432 of file threadpool.c.

References ast_threadpool_serializer_group(), and NULL.

Referenced by AST_TEST_DEFINE(), internal_stasis_subscribe(), and sorcery_object_type_alloc().

{
   return ast_threadpool_serializer_group(name, pool, NULL);
}

ast_threadpool_serializer_get_current()

struct ast_taskprocessor* ast_threadpool_serializer_get_current

(

void

)

Get the threadpool serializer currently associated with this thread.

Since

14.0.0

Note

The returned pointer is valid while the serializer thread is running.

Use ao2_ref() on serializer if you are going to keep it for another thread. To unref it you must then use ast_taskprocessor_unreference().

Return values

serializer	on success.
NULL	on error or no serializer associated with the thread.

Definition at line 1397 of file threadpool.c.

References ast_threadstorage_get_ptr().

Referenced by record_serializer(), rfc3326_outgoing_request(), rfc3326_outgoing_response(), and sip_resolve().

{
   return ast_threadstorage_get_ptr(&current_serializer);
}

ast_threadpool_serializer_group()

struct ast_taskprocessor* ast_threadpool_serializer_group	(	const char *	name,
		struct ast_threadpool *	pool,
		struct ast_serializer_shutdown_group *	shutdown_group
	)

Serialized execution of tasks within a ast_threadpool.

Since

13.5.0

A ast_taskprocessor with the same contract as a default taskprocessor (tasks execute serially) except instead of executing out of a dedicated thread, execution occurs in a thread from a ast_threadpool. Think of it as a lightweight thread.

While it guarantees that each task will complete before executing the next, there is no guarantee as to which thread from the pool individual tasks will execute. This normally only matters if your code relys on thread specific information, such as thread locals.

Use ast_taskprocessor_unreference() to dispose of the returned ast_taskprocessor.

Only a single taskprocessor with a given name may exist. This function will fail if a taskprocessor with the given name already exists.

Parameters

name	Name of the serializer. (must be unique)
pool	ast_threadpool for execution.
shutdown_group	Group shutdown controller. (NULL if no group association)

Returns

ast_taskprocessor for enqueuing work.

NULL on error.

Definition at line 1402 of file threadpool.c.

References ao2_ref, ast_taskprocessor_create_with_listener(), ast_taskprocessor_listener_alloc(), ast_threadpool::listener, NULL, serializer_create(), serializer_shutdown_group_inc(), and ast_threadpool::tps.

Referenced by ast_serializer_pool_create(), ast_sip_create_serializer_group(), and ast_threadpool_serializer().

{
   struct serializer *ser;
   struct ast_taskprocessor_listener *listener;
   struct ast_taskprocessor *tps;

   ser = serializer_create(pool, shutdown_group);
   if (!ser) {
      return NULL;
   }

   listener = ast_taskprocessor_listener_alloc(&serializer_tps_listener_callbacks, ser);
   if (!listener) {
      ao2_ref(ser, -1);
      return NULL;
   }

   tps = ast_taskprocessor_create_with_listener(name, listener);
   if (!tps) {
      /* ser ref transferred to listener but not cleaned without tps */
      ao2_ref(ser, -1);
   } else if (shutdown_group) {
      serializer_shutdown_group_inc(shutdown_group);
   }

   ao2_ref(listener, -1);
   return tps;
}

ast_threadpool_set_size()

void ast_threadpool_set_size	(	struct ast_threadpool *	threadpool,
		unsigned int	size
	)

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.

Parameters

threadpool	The threadpool to adjust
size	The new desired size of the threadpool

Definition at line 874 of file threadpool.c.

References ast_free, ast_taskprocessor_push(), ast_threadpool::control_tps, lock, queued_set_size(), SCOPED_AO2LOCK, set_size_data_alloc(), and ast_threadpool::shutting_down.

Referenced by AST_TEST_DEFINE(), and ast_threadpool_create().

{
   struct set_size_data *ssd;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   ssd = set_size_data_alloc(pool, size);
   if (!ssd) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_set_size, ssd)) {
      ast_free(ssd);
   }
}

ast_threadpool_shutdown()

void ast_threadpool_shutdown

(

struct ast_threadpool *

pool

)

Shut down a threadpool and destroy it.

Parameters

pool	The pool to shut down

Definition at line 965 of file threadpool.c.

References ao2_lock, ao2_unlock, ast_taskprocessor_unreference(), ast_threadpool::control_tps, ast_threadpool::shutting_down, and ast_threadpool::tps.

Referenced by AST_TEST_DEFINE(), load_module(), sorcery_cleanup(), stasis_cleanup(), and unload_module().

{
   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);
}

AST_THREADSTORAGE_RAW()

AST_THREADSTORAGE_RAW

(

current_serializer

)

Referenced by serializer_create().

execute_tasks()

static int execute_tasks ( void *  data )

static

Definition at line 1349 of file threadpool.c.

References ast_taskprocessor_execute(), ast_taskprocessor_unreference(), ast_threadstorage_set_ptr(), NULL, and ast_threadpool::tps.

Referenced by serializer_task_pushed().

{
   struct ast_taskprocessor *tps = data;

   ast_threadstorage_set_ptr(&current_serializer, tps);
   while (ast_taskprocessor_execute(tps)) {
      /* No-op */
   }
   ast_threadstorage_set_ptr(&current_serializer, NULL);

   ast_taskprocessor_unreference(tps);
   return 0;
}

grow()

static void grow ( struct ast_threadpool *  pool, int  delta  )

static

Add threads to the threadpool.

This function is called from the threadpool's control taskprocessor thread.

Parameters

pool	The pool that is expanding The number of threads to add to the pool

Definition at line 524 of file threadpool.c.

References ast_threadpool::active_threads, ao2_container_count(), ao2_link, ao2_ref, ao2_unlink, ast_debug, ast_log, ast_taskprocessor_name(), worker_thread::id, ast_threadpool::idle_threads, LOG_ERROR, LOG_WARNING, ast_threadpool_options::max_size, ast_threadpool::options, ast_threadpool::tps, worker_thread_alloc(), and worker_thread_start().

Referenced by __ast_string_field_ptr_build_va(), __ast_string_field_ptr_grow(), queued_set_size(), and queued_task_pushed().

{
   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->idle_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);
   }
}

kill_threads()

static int kill_threads ( void *  obj, void *  arg, int  flags  )

static

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.

Parameters

obj	The worker thread up for possible destruction
arg	The counter
flags	Unused

Return values

CMP_MATCH	The counter has not reached zero, so this flag should be removed.
CMP_STOP	The counter has reached zero so no more threads should be removed.

Definition at line 714 of file threadpool.c.

References CMP_MATCH, and CMP_STOP.

Referenced by shrink().

{
   int *num_to_kill = arg;

   if (*num_to_kill > 0) {
      --(*num_to_kill);
      return CMP_MATCH;
   } else {
      return CMP_STOP;
   }
}

queued_active_thread_idle()

static int queued_active_thread_idle ( void *  data )

static

Move a worker thread from the active container to the idle container.

This function is called from the threadpool's control taskprocessor thread.

Parameters

data	A thread_worker_pair containing the threadpool and the worker to move.

Returns

0

Definition at line 235 of file threadpool.c.

References ast_threadpool::active_threads, ao2_link, ao2_unlink, ast_threadpool::idle_threads, thread_worker_pair::pool, thread_worker_pair_free(), threadpool_send_state_changed(), and thread_worker_pair::worker.

Referenced by threadpool_active_thread_idle().

{
   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);

   thread_worker_pair_free(pair);
   return 0;
}

queued_emptied()

static int queued_emptied ( void *  data )

static

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

Parameters

data	The pool that has become empty

Returns

0

Definition at line 638 of file threadpool.c.

References ast_threadpool_listener::callbacks, ast_threadpool_listener_callbacks::emptied, ast_threadpool::listener, and pool.

Referenced by threadpool_tps_emptied().

{
   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;
}

queued_idle_thread_dead()

static int queued_idle_thread_dead ( void *  data )

static

Definition at line 321 of file threadpool.c.

References ao2_unlink, ast_threadpool::idle_threads, thread_worker_pair::pool, thread_worker_pair_free(), threadpool_send_state_changed(), and thread_worker_pair::worker.

Referenced by threadpool_idle_thread_dead().

{
   struct thread_worker_pair *pair = data;

   ao2_unlink(pair->pool->idle_threads, pair->worker);
   threadpool_send_state_changed(pair->pool);

   thread_worker_pair_free(pair);
   return 0;
}

queued_set_size()

static int queued_set_size ( void *  data )

static

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

Parameters

data	A set_size_data used for determining how to act

Returns

0

Definition at line 838 of file threadpool.c.

References activate_thread(), ast_threadpool::active_threads, ao2_callback, ao2_container_count(), ast_debug, ast_free, grow(), ast_threadpool::idle_threads, OBJ_MULTIPLE, OBJ_NODATA, OBJ_NOLOCK, OBJ_UNLINK, pool, set_size_data::pool, shrink(), set_size_data::size, and threadpool_send_state_changed().

Referenced by ast_threadpool_set_size().

{
   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);

   ast_free(ssd);

   if (current_size == num_threads) {
      ast_debug(3, "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) {
      ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
            activate_thread, pool);

      /* As the above may have altered the number of current threads update it */
      current_size = ao2_container_count(pool->active_threads) +
            ao2_container_count(pool->idle_threads);
      grow(pool, num_threads - current_size);
      ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA | OBJ_MULTIPLE,
            activate_thread, pool);
   } else {
      shrink(pool, current_size - num_threads);
   }

   threadpool_send_state_changed(pool);
   return 0;
}

queued_task_pushed()

static int queued_task_pushed ( void *  data )

static

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.

Parameters

data	A task_pushed_data

Returns

0

Definition at line 564 of file threadpool.c.

References activate_thread(), ast_threadpool::active_threads, ao2_callback, ao2_container_count(), ast_free, ast_threadpool_options::auto_increment, ast_threadpool_listener::callbacks, grow(), ast_threadpool::idle_threads, ast_threadpool::listener, OBJ_NODATA, OBJ_NOLOCK, OBJ_UNLINK, ast_threadpool::options, pool, task_pushed_data::pool, ast_threadpool_listener_callbacks::task_pushed, threadpool_send_state_changed(), and task_pushed_data::was_empty.

Referenced by threadpool_tps_task_pushed().

{
   struct task_pushed_data *tpd = data;
   struct ast_threadpool *pool = tpd->pool;
   int was_empty = tpd->was_empty;
   unsigned int existing_active;

   ast_free(tpd);

   if (pool->listener && pool->listener->callbacks->task_pushed) {
      pool->listener->callbacks->task_pushed(pool, pool->listener, was_empty);
   }

   existing_active = ao2_container_count(pool->active_threads);

   /* The first pass transitions any existing idle threads to be active, and
    * will also remove any worker threads that have recently entered the dead
    * state.
    */
   ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA,
         activate_thread, pool);

   /* If no idle threads could be transitioned to active grow the pool as permitted. */
   if (ao2_container_count(pool->active_threads) == existing_active) {
      if (!pool->options.auto_increment) {
         return 0;
      }
      grow(pool, pool->options.auto_increment);
      /* An optional second pass transitions any newly added threads. */
      ao2_callback(pool->idle_threads, OBJ_UNLINK | OBJ_NOLOCK | OBJ_NODATA,
            activate_thread, pool);
   }

   threadpool_send_state_changed(pool);
   return 0;
}

queued_zombie_thread_dead()

static int queued_zombie_thread_dead ( void *  data )

static

Kill a zombie thread.

This runs from the threadpool's control taskprocessor thread.

Parameters

data	A thread_worker_pair containing the threadpool and the zombie thread

Returns

0

Definition at line 284 of file threadpool.c.

References ao2_unlink, thread_worker_pair::pool, thread_worker_pair_free(), threadpool_send_state_changed(), thread_worker_pair::worker, and ast_threadpool::zombie_threads.

Referenced by threadpool_zombie_thread_dead().

{
   struct thread_worker_pair *pair = data;

   ao2_unlink(pair->pool->zombie_threads, pair->worker);
   threadpool_send_state_changed(pair->pool);

   thread_worker_pair_free(pair);
   return 0;
}

serializer_create()

static struct serializer* serializer_create ( struct ast_threadpool *  pool, struct ast_serializer_shutdown_group *  shutdown_group  )

static

Definition at line 1332 of file threadpool.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_bump, ao2_ref, AST_THREADSTORAGE_RAW(), NULL, pool, serializer::pool, serializer_dtor(), and serializer::shutdown_group.

Referenced by ast_threadpool_serializer_group().

{
   struct serializer *ser;

   ser = ao2_alloc_options(sizeof(*ser), serializer_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
   if (!ser) {
      return NULL;
   }
   ao2_ref(pool, +1);
   ser->pool = pool;
   ser->shutdown_group = ao2_bump(shutdown_group);
   return ser;
}

serializer_dtor()

static void serializer_dtor ( void *  obj )

static

Definition at line 1322 of file threadpool.c.

References ao2_cleanup, NULL, serializer::pool, and serializer::shutdown_group.

Referenced by serializer_create().

{
   struct serializer *ser = obj;

   ao2_cleanup(ser->pool);
   ser->pool = NULL;
   ao2_cleanup(ser->shutdown_group);
   ser->shutdown_group = NULL;
}

serializer_shutdown()

static void serializer_shutdown ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 1381 of file threadpool.c.

References ao2_cleanup, ast_taskprocessor_listener_get_user_data(), serializer_shutdown_group_dec(), and serializer::shutdown_group.

{
   struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);

   if (ser->shutdown_group) {
      serializer_shutdown_group_dec(ser->shutdown_group);
   }
   ao2_cleanup(ser);
}

serializer_shutdown_group_dec()

static void serializer_shutdown_group_dec ( struct ast_serializer_shutdown_group *  shutdown_group )

static

Definition at line 1305 of file threadpool.c.

References ao2_lock, ao2_unlock, ast_cond_signal, ast_serializer_shutdown_group::cond, and ast_serializer_shutdown_group::count.

Referenced by serializer_shutdown().

{
   ao2_lock(shutdown_group);
   --shutdown_group->count;
   if (!shutdown_group->count) {
      ast_cond_signal(&shutdown_group->cond);
   }
   ao2_unlock(shutdown_group);
}

serializer_shutdown_group_dtor()

static void serializer_shutdown_group_dtor ( void *  vdoomed )

static

Definition at line 1222 of file threadpool.c.

References ast_cond_destroy, and ast_serializer_shutdown_group::cond.

Referenced by ast_serializer_shutdown_group_alloc().

{
   struct ast_serializer_shutdown_group *doomed = vdoomed;

   ast_cond_destroy(&doomed->cond);
}

serializer_shutdown_group_inc()

static void serializer_shutdown_group_inc ( struct ast_serializer_shutdown_group *  shutdown_group )

static

Definition at line 1289 of file threadpool.c.

References ao2_lock, ao2_unlock, and ast_serializer_shutdown_group::count.

Referenced by ast_threadpool_serializer_group().

{
   ao2_lock(shutdown_group);
   ++shutdown_group->count;
   ao2_unlock(shutdown_group);
}

serializer_start()

static int serializer_start ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 1375 of file threadpool.c.

{
   /* No-op */
   return 0;
}

serializer_task_pushed()

static void serializer_task_pushed ( struct ast_taskprocessor_listener *  listener, int  was_empty  )

static

Definition at line 1363 of file threadpool.c.

References ast_taskprocessor_listener_get_tps(), ast_taskprocessor_listener_get_user_data(), ast_taskprocessor_unreference(), ast_threadpool_push(), execute_tasks(), serializer::pool, and ast_threadpool::tps.

{
   if (was_empty) {
      struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
      struct ast_taskprocessor *tps = ast_taskprocessor_listener_get_tps(listener);

      if (ast_threadpool_push(ser->pool, execute_tasks, tps)) {
         ast_taskprocessor_unreference(tps);
      }
   }
}

set_size_data_alloc()

static struct set_size_data* set_size_data_alloc ( struct ast_threadpool *  pool, unsigned int  size  )

static

Allocate and initialize a set_size_data.

Parameters

pool	The pool for the set_size_data
size	The size to store in the set_size_data

Definition at line 814 of file threadpool.c.

References ast_malloc, NULL, pool, set_size_data::pool, and set_size_data::size.

Referenced by ast_threadpool_set_size().

{
   struct set_size_data *ssd = ast_malloc(sizeof(*ssd));
   if (!ssd) {
      return NULL;
   }

   ssd->pool = pool;
   ssd->size = size;
   return ssd;
}

shrink()

static void shrink ( struct ast_threadpool *  pool, int  delta  )

static

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.

Parameters

pool	The threadpool to remove threads from
delta	The number of threads to remove

Definition at line 775 of file threadpool.c.

References ast_threadpool::active_threads, ao2_callback, ao2_callback_data, ao2_container_count(), ast_debug, ast_taskprocessor_name(), ast_threadpool::idle_threads, kill_threads(), MIN, OBJ_MULTIPLE, OBJ_NODATA, OBJ_NOLOCK, OBJ_UNLINK, ast_threadpool::tps, and zombify_threads().

Referenced by queued_set_size().

{
   /*
    * 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);
}

task_pushed_data_alloc()

static struct task_pushed_data* task_pushed_data_alloc ( struct ast_threadpool *  pool, int  was_empty  )

static

Allocate and initialize a task_pushed_data.

Parameters

pool	The threadpool to set in the task_pushed_data
was_empty	The was_empty value to set in the task_pushed_data

Return values

NULL	Unable to allocate task_pushed_data
non-NULL	The newly-allocated task_pushed_data

Definition at line 466 of file threadpool.c.

References ast_malloc, NULL, pool, task_pushed_data::pool, and task_pushed_data::was_empty.

Referenced by threadpool_tps_task_pushed().

{
   struct task_pushed_data *tpd = ast_malloc(sizeof(*tpd));

   if (!tpd) {
      return NULL;
   }
   tpd->pool = pool;
   tpd->was_empty = was_empty;
   return tpd;
}

thread_worker_pair_alloc()

static struct thread_worker_pair* thread_worker_pair_alloc ( struct ast_threadpool *  pool, struct worker_thread *  worker  )

static

Allocate and initialize a thread_worker_pair.

Parameters

pool	Threadpool to assign to the thread_worker_pair
worker	Worker thread to assign to the thread_worker_pair

Definition at line 214 of file threadpool.c.

References ao2_ref, ast_malloc, NULL, thread_worker_pair::pool, pool, and thread_worker_pair::worker.

Referenced by threadpool_active_thread_idle(), threadpool_idle_thread_dead(), and threadpool_zombie_thread_dead().

{
   struct thread_worker_pair *pair = ast_malloc(sizeof(*pair));
   if (!pair) {
      return NULL;
   }
   pair->pool = pool;
   ao2_ref(worker, +1);
   pair->worker = worker;

   return pair;
}

thread_worker_pair_free()

static void thread_worker_pair_free ( struct thread_worker_pair *  pair )

static

Destructor for thread_worker_pair.

Definition at line 203 of file threadpool.c.

References ao2_ref, ast_free, and thread_worker_pair::worker.

Referenced by queued_active_thread_idle(), queued_idle_thread_dead(), queued_zombie_thread_dead(), threadpool_active_thread_idle(), threadpool_idle_thread_dead(), and threadpool_zombie_thread_dead().

{
   ao2_ref(pair->worker, -1);
   ast_free(pair);
}

threadpool_active_thread_idle()

static void threadpool_active_thread_idle ( struct ast_threadpool *  pool, struct worker_thread *  worker  )

static

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.

Parameters

pool	The threadpool to which the worker belongs
worker	The worker thread that has gone idle

Definition at line 256 of file threadpool.c.

References ast_taskprocessor_push(), ast_threadpool::control_tps, lock, queued_active_thread_idle(), SCOPED_AO2LOCK, ast_threadpool::shutting_down, thread_worker_pair_alloc(), and thread_worker_pair_free().

Referenced by worker_start().

{
   struct thread_worker_pair *pair;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   pair = thread_worker_pair_alloc(pool, worker);
   if (!pair) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_active_thread_idle, pair)) {
      thread_worker_pair_free(pair);
   }
}

threadpool_alloc()

static struct ast_threadpool* threadpool_alloc ( const char *  name, const struct ast_threadpool_options *  options  )

static

Definition at line 404 of file threadpool.c.

References ao2_alloc, AO2_ALLOC_OPT_LOCK_MUTEX, ao2_cleanup, ao2_container_alloc_hash, ao2_ref, ast_free, ast_str_buffer(), ast_str_create, ast_str_set(), ast_taskprocessor_get(), NULL, ast_threadpool::options, pool, RAII_VAR, THREAD_BUCKETS, threadpool_destructor(), TPS_REF_DEFAULT, worker_thread_cmp(), and worker_thread_hash().

Referenced by ast_threadpool_create().

{
   RAII_VAR(struct ast_threadpool *, pool, NULL, ao2_cleanup);
   struct ast_str *control_tps_name;

   pool = ao2_alloc(sizeof(*pool), threadpool_destructor);
   control_tps_name = ast_str_create(64);
   if (!pool || !control_tps_name) {
      ast_free(control_tps_name);
      return NULL;
   }

   ast_str_set(&control_tps_name, 0, "%s/pool-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_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
      THREAD_BUCKETS, worker_thread_hash, NULL, worker_thread_cmp);
   if (!pool->active_threads) {
      return NULL;
   }
   pool->idle_threads = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
      THREAD_BUCKETS, worker_thread_hash, NULL, worker_thread_cmp);
   if (!pool->idle_threads) {
      return NULL;
   }
   pool->zombie_threads = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
      THREAD_BUCKETS, worker_thread_hash, NULL, worker_thread_cmp);
   if (!pool->zombie_threads) {
      return NULL;
   }
   pool->options = *options;

   ao2_ref(pool, +1);
   return pool;
}

threadpool_destructor()

static void threadpool_destructor ( void *  obj )

static

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.

Parameters

obj	The pool to destroy

Definition at line 386 of file threadpool.c.

References ao2_cleanup, ast_threadpool::listener, and pool.

Referenced by threadpool_alloc().

{
   struct ast_threadpool *pool = obj;
   ao2_cleanup(pool->listener);
}

threadpool_execute()

static int threadpool_execute ( struct ast_threadpool *  pool )

static

Execute a task in the threadpool.

This is the function that worker threads call in order to execute tasks in the threadpool

Parameters

pool	The pool to which the tasks belong.

Return values

0	Either the pool has been shut down or there are no tasks.
1	There are still tasks remaining in the pool.

Definition at line 362 of file threadpool.c.

References ao2_lock, ao2_unlock, ast_taskprocessor_execute(), ast_threadpool::shutting_down, and ast_threadpool::tps.

Referenced by worker_active().

{
   ao2_lock(pool);
   if (!pool->shutting_down) {
      ao2_unlock(pool);
      return ast_taskprocessor_execute(pool->tps);
   }
   ao2_unlock(pool);
   return 0;
}

threadpool_idle_thread_dead()

static void threadpool_idle_thread_dead ( struct ast_threadpool *  pool, struct worker_thread *  worker  )

static

Definition at line 332 of file threadpool.c.

References ast_taskprocessor_push(), ast_threadpool::control_tps, lock, queued_idle_thread_dead(), SCOPED_AO2LOCK, ast_threadpool::shutting_down, thread_worker_pair_alloc(), and thread_worker_pair_free().

Referenced by worker_idle().

{
   struct thread_worker_pair *pair;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   pair = thread_worker_pair_alloc(pool, worker);
   if (!pair) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_idle_thread_dead, pair)) {
      thread_worker_pair_free(pair);
   }
}

threadpool_send_state_changed()

static void threadpool_send_state_changed ( struct ast_threadpool *  pool )

static

Notify the threadpool listener that the state has changed.

This notifies the threadpool listener via its state_changed callback.

Parameters

pool	The threadpool whose state has changed

Definition at line 180 of file threadpool.c.

References ast_threadpool::active_threads, ao2_container_count(), ast_threadpool_listener::callbacks, ast_threadpool::idle_threads, ast_threadpool::listener, and ast_threadpool_listener_callbacks::state_changed.

Referenced by queued_active_thread_idle(), queued_idle_thread_dead(), queued_set_size(), queued_task_pushed(), and queued_zombie_thread_dead().

{
   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);
   }
}

threadpool_tps_emptied()

static void threadpool_tps_emptied ( struct ast_taskprocessor_listener *  listener )

static

Taskprocessor listener emptied callback.

The threadpool queues a task to let the threadpool listener know that the threadpool no longer contains any tasks.

Parameters

listener

The taskprocessor listener. The threadpool is the listener's private data.

Definition at line 654 of file threadpool.c.

References ast_taskprocessor_listener_get_user_data(), ast_taskprocessor_push(), ast_threadpool_listener::callbacks, ast_threadpool::control_tps, ast_threadpool_listener_callbacks::emptied, ast_threadpool::listener, lock, pool, queued_emptied(), SCOPED_AO2LOCK, and ast_threadpool::shutting_down.

{
   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) {
      if (ast_taskprocessor_push(pool->control_tps, queued_emptied, pool)) {
         /* Nothing to do here but we need the check to keep the compiler happy. */
      }
   }
}

threadpool_tps_shutdown()

static void threadpool_tps_shutdown ( struct ast_taskprocessor_listener *  listener )

static

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.

Parameters

listener

The taskprocessor listener. The threadpool is the listener's private data.

Definition at line 679 of file threadpool.c.

References ast_threadpool::active_threads, ao2_cleanup, ast_taskprocessor_listener_get_user_data(), ast_threadpool_listener::callbacks, ast_threadpool::idle_threads, ast_threadpool::listener, pool, ast_threadpool_listener_callbacks::shutdown, and ast_threadpool::zombie_threads.

{
   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);
}

threadpool_tps_start()

static int threadpool_tps_start ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 444 of file threadpool.c.

{
   return 0;
}

threadpool_tps_task_pushed()

static void threadpool_tps_task_pushed ( struct ast_taskprocessor_listener *  listener, int  was_empty  )

static

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.

Parameters

listener	The taskprocessor listener. The threadpool is the listener's private data
was_empty	True if the taskprocessor was empty prior to the task being pushed

Definition at line 610 of file threadpool.c.

References ast_free, ast_taskprocessor_listener_get_user_data(), ast_taskprocessor_push(), ast_threadpool::control_tps, lock, pool, queued_task_pushed(), SCOPED_AO2LOCK, ast_threadpool::shutting_down, and task_pushed_data_alloc().

{
   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;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_task_pushed, tpd)) {
      ast_free(tpd);
   }
}

threadpool_zombie_thread_dead()

static void threadpool_zombie_thread_dead ( struct ast_threadpool *  pool, struct worker_thread *  worker  )

static

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.

Definition at line 301 of file threadpool.c.

References ast_taskprocessor_push(), ast_threadpool::control_tps, lock, queued_zombie_thread_dead(), SCOPED_AO2LOCK, ast_threadpool::shutting_down, thread_worker_pair_alloc(), and thread_worker_pair_free().

Referenced by worker_start().

{
   struct thread_worker_pair *pair;
   SCOPED_AO2LOCK(lock, pool);

   if (pool->shutting_down) {
      return;
   }

   pair = thread_worker_pair_alloc(pool, worker);
   if (!pair) {
      return;
   }

   if (ast_taskprocessor_push(pool->control_tps, queued_zombie_thread_dead, pair)) {
      thread_worker_pair_free(pair);
   }
}

worker_active()

static void worker_active ( struct worker_thread *  worker )

static

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.

Parameters

worker

The worker thread executing tasks.

Definition at line 1124 of file threadpool.c.

References worker_thread::pool, and threadpool_execute().

Referenced by worker_start().

{
   int alive;

   /* The following is equivalent to
    *
    * while (threadpool_execute(worker->pool));
    *
    * However, reviewers have suggested in the past
    * doing that can cause optimizers to (wrongly)
    * optimize the code away.
    */
   do {
      alive = threadpool_execute(worker->pool);
   } while (alive);
}

worker_idle()

static int worker_idle ( struct worker_thread *  worker )

static

Idle function for worker threads.

The worker waits here until it gets told by the threadpool to wake up.

worker is locked before entering this function.

Parameters

worker

The idle worker

Return values

0	The thread is being woken up so that it can conclude.
non-zero	The thread is being woken up to do more work.

Definition at line 1153 of file threadpool.c.

References ALIVE, ast_cond_timedwait, ast_cond_wait, ast_debug, ast_tvnow(), worker_thread::cond, DEAD, ast_threadpool_options::idle_timeout, worker_thread::lock, worker_thread::options, worker_thread::pool, worker_thread::state, threadpool_idle_thread_dead(), and worker_thread::wake_up.

Referenced by worker_start().

{
   struct timeval start = ast_tvnow();
   struct timespec end = {
      .tv_sec = start.tv_sec + worker->options.idle_timeout,
      .tv_nsec = start.tv_usec * 1000,
   };
   while (!worker->wake_up) {
      if (worker->options.idle_timeout <= 0) {
         ast_cond_wait(&worker->cond, &worker->lock);
      } else if (ast_cond_timedwait(&worker->cond, &worker->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;
}

worker_set_state()

static int worker_set_state ( struct worker_thread *  worker, enum worker_state  state  )

static

Change a worker's state.

The threadpool calls into this function in order to let a worker know how it should proceed.

Return values

-1	failure (state transition not permitted)
0	success

Definition at line 1186 of file threadpool.c.

References ALIVE, ast_assert, ast_cond_signal, worker_thread::cond, DEAD, worker_thread::lock, lock, SCOPED_MUTEX, worker_thread::state, state, worker_thread::wake_up, and ZOMBIE.

Referenced by activate_thread(), worker_shutdown(), and zombify_threads().

{
   SCOPED_MUTEX(lock, &worker->lock);

   switch (state) {
   case ALIVE:
      /* This can occur due to a race condition between being told to go active
       * and an idle timeout happening.
       */
      if (worker->state == DEAD) {
         return -1;
      }
      ast_assert(worker->state != ZOMBIE);
      break;
   case DEAD:
      break;
   case ZOMBIE:
      ast_assert(worker->state != DEAD);
      break;
   }

   worker->state = state;
   worker->wake_up = 1;
   ast_cond_signal(&worker->cond);

   return 0;
}

worker_shutdown()

static void worker_shutdown ( struct worker_thread *  worker )

static

shut a worker thread down

Set the worker dead and then wait for its thread to finish executing.

Parameters

worker

The worker thread to shut down

Definition at line 1009 of file threadpool.c.

References AST_PTHREADT_NULL, DEAD, NULL, worker_thread::thread, and worker_set_state().

Referenced by worker_thread_destroy().

{
   worker_set_state(worker, DEAD);
   if (worker->thread != AST_PTHREADT_NULL) {
      pthread_join(worker->thread, NULL);
      worker->thread = AST_PTHREADT_NULL;
   }
}

worker_start()

static void * worker_start ( void *  arg )

static

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

Parameters

arg	The worker thread

Return values

NULL

Definition at line 1044 of file threadpool.c.

References ALIVE, ast_mutex_lock, ast_mutex_unlock, worker_thread::lock, NULL, worker_thread::options, worker_thread::pool, worker_thread::state, ast_threadpool_options::thread_end, ast_threadpool_options::thread_start, threadpool_active_thread_idle(), threadpool_zombie_thread_dead(), worker_active(), worker_idle(), and ZOMBIE.

Referenced by worker_thread_start().

{
   struct worker_thread *worker = arg;
   enum worker_state saved_state;

   if (worker->options.thread_start) {
      worker->options.thread_start();
   }

   ast_mutex_lock(&worker->lock);
   while (worker_idle(worker)) {
      ast_mutex_unlock(&worker->lock);
      worker_active(worker);
      ast_mutex_lock(&worker->lock);
      if (worker->state != ALIVE) {
         break;
      }
      threadpool_active_thread_idle(worker->pool, worker);
   }
   saved_state = worker->state;
   ast_mutex_unlock(&worker->lock);

   /* 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 (saved_state == ZOMBIE) {
      threadpool_zombie_thread_dead(worker->pool, worker);
   }

   if (worker->options.thread_end) {
      worker->options.thread_end();
   }
   return NULL;
}

worker_thread_alloc()

static struct worker_thread * worker_thread_alloc ( struct ast_threadpool *  pool )

static

Allocate and initialize a new worker thread.

This will create, initialize, and start the thread.

Parameters

pool	The threadpool to which the worker will be added

Return values

NULL	Failed to allocate or start the worker thread
non-NULL	The newly-created worker thread

Definition at line 1093 of file threadpool.c.

References ALIVE, ao2_alloc, ast_atomic_fetchadd_int(), ast_cond_init, ast_mutex_init, AST_PTHREADT_NULL, worker_thread::cond, worker_thread::id, worker_thread::lock, NULL, ast_threadpool::options, worker_thread::options, worker_thread::pool, pool, worker_thread::state, worker_thread::thread, and worker_thread_destroy().

Referenced by grow().

{
   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;
}

worker_thread_cmp()

static int worker_thread_cmp ( void *  obj, void *  arg, int  flags  )

static

Definition at line 993 of file threadpool.c.

References CMP_MATCH, and worker_thread::id.

Referenced by threadpool_alloc().

{
   struct worker_thread *worker1 = obj;
   struct worker_thread *worker2 = arg;

   return worker1->id == worker2->id ? CMP_MATCH : 0;
}

worker_thread_destroy()

static void worker_thread_destroy ( void *  obj )

static

Worker thread destructor.

Called automatically when refcount reaches 0. Shuts down the worker thread and destroys its component parts

Definition at line 1025 of file threadpool.c.

References ast_cond_destroy, ast_debug, ast_mutex_destroy, worker_thread::cond, worker_thread::id, worker_thread::lock, and worker_shutdown().

Referenced by worker_thread_alloc().

{
   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);
}

worker_thread_hash()

static int worker_thread_hash ( const void *  obj, int  flags  )

static

Definition at line 986 of file threadpool.c.

References worker_thread::id.

Referenced by threadpool_alloc().

{
   const struct worker_thread *worker = obj;

   return worker->id;
}

worker_thread_start()

static int worker_thread_start ( struct worker_thread *  worker )

static

Definition at line 1109 of file threadpool.c.

References ast_pthread_create, NULL, worker_thread::thread, and worker_start().

Referenced by grow().

{
   return ast_pthread_create(&worker->thread, NULL, worker_start, worker);
}

zombify_threads()

static int zombify_threads ( void *  obj, void *  arg, void *  data, int  flags  )

static

ao2 callback to zombify a set number of threads.

Threads will be zombified as long 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.

Parameters

obj	The worker thread that may be zombified
arg	The pool to which the worker belongs
data	The counter
flags	Unused

Return values

CMP_MATCH	The zombified thread should be removed from its current container
CMP_STOP	Stop attempting to zombify threads

Definition at line 745 of file threadpool.c.

References ao2_link, ast_log, CMP_MATCH, CMP_STOP, worker_thread::id, LOG_WARNING, pool, worker_set_state(), ZOMBIE, and ast_threadpool::zombie_threads.

Referenced by shrink().

{
   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;
   }
}

Variable Documentation

serializer_tps_listener_callbacks

struct ast_taskprocessor_listener_callbacks serializer_tps_listener_callbacks

static

Initial value:

= {
   .task_pushed = serializer_task_pushed,
   .start = serializer_start,
   .shutdown = serializer_shutdown,
}

Definition at line 1391 of file threadpool.c.

threadpool_tps_listener_callbacks

struct ast_taskprocessor_listener_callbacks threadpool_tps_listener_callbacks

static

Table of taskprocessor listener callbacks for threadpool's main taskprocessor.

Definition at line 695 of file threadpool.c.

worker_id_counter

int worker_id_counter

static

A monotonically increasing integer used for worker thread identification.

Definition at line 984 of file threadpool.c.