threadpool.h File Reference

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	ast_threadpool_listener_callbacks
struct	ast_threadpool_options

Macros
#define	AST_THREADPOOL_OPTIONS_VERSION   1

Functions
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) attribute_warn_unused_result
	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 *threadpool, 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...

Macro Definition Documentation

AST_THREADPOOL_OPTIONS_VERSION

#define AST_THREADPOOL_OPTIONS_VERSION   1

Definition at line 71 of file threadpool.h.

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

Function Documentation

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