taskpool.c File Reference

#include "asterisk.h"
#include "asterisk/_private.h"
#include "asterisk/taskpool.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/astobj2.h"
#include "asterisk/serializer_shutdown_group.h"
#include "asterisk/utils.h"
#include "asterisk/time.h"
#include "asterisk/sched.h"

Include dependency graph for taskpool.c:

Go to the source code of this file.

Data Structures
struct	ast_taskpool
	An opaque taskpool structure. More...
struct	serializer
struct	taskpool_sync_task
struct	taskpool_taskprocessor
	A taskpool taskprocessor. More...
struct	taskpool_taskprocessors
	A container of taskprocessors. More...

Macros
#define	ast_taskpool_push_internal(pool, task, data)    __ast_taskpool_push(pool, task, data, __FILE__, __LINE__, __PRETTY_FUNCTION__)
#define	TASKPOOL_GROW_THRESHOLD   (AST_TASKPROCESSOR_HIGH_WATER_LEVEL * 5) / 10
	The threshold for a taskprocessor at which we consider the pool needing to grow (50% of high water threshold)
#define	TASKPOOL_QUEUE_SIZE_ADD(tps, size)   (size += ast_taskprocessor_size(tps->taskprocessor))
#define	TASKPROCESSOR_IS_IDLE(tps, timeout)   (ast_tvdiff_ms(ast_tvnow(), tps->last_pushed) > (timeout))

Typedefs
typedef void(*	taskpool_selector) (struct ast_taskpool *pool, struct taskpool_taskprocessors *taskprocessors, struct taskpool_taskprocessor **taskprocessor, unsigned int *growth_threshold_reached)

Enumerations
enum	serializer_suspension { SERIALIZER_UNSUSPENDED = 0 , SERIALIZER_SUSPENDING , SERIALIZER_SUSPENDED }

Functions
int	__ast_taskpool_push (struct ast_taskpool *pool, int(*task)(void *data), void *data, const char *file, int line, const char *function)
	Push a task to the taskpool.
int	__ast_taskpool_push_wait (struct ast_taskpool *pool, int(*task)(void *data), void *data, const char *file, int line, const char *function)
	Push a task to the taskpool, and wait for completion.
int	__ast_taskpool_serializer_push_wait (struct ast_taskprocessor *serializer, int(*task)(void *data), void *data, const char *file, int line, const char *function)
	Push a task to a serializer, and wait for completion.
struct ast_taskpool *	ast_taskpool_create (const char *name, const struct ast_taskpool_options *options)
	Create a new taskpool.
static struct ast_taskpool *	ast_taskpool_get_current (void)
int	ast_taskpool_init (void)
int	ast_taskpool_push (struct ast_taskpool *pool, int(*task)(void *data), void *data)
int	ast_taskpool_push_wait (struct ast_taskpool *pool, int(*task)(void *data), void *data)
long	ast_taskpool_queue_size (struct ast_taskpool *pool)
	Get the current number of queued tasks in the taskpool.
struct ast_taskprocessor *	ast_taskpool_serializer (const char *name, struct ast_taskpool *pool)
	Serialized execution of tasks within a ast_taskpool.
struct ast_taskprocessor *	ast_taskpool_serializer_get_current (void)
	Get the taskpool serializer currently associated with this thread.
struct ast_taskprocessor *	ast_taskpool_serializer_group (const char *name, struct ast_taskpool *pool, struct ast_serializer_shutdown_group *shutdown_group)
	Serialized execution of tasks within a ast_taskpool.
int	ast_taskpool_serializer_push_wait (struct ast_taskprocessor *serializer, int(*task)(void *data), void *data)
int	ast_taskpool_serializer_suspend (struct ast_taskprocessor *serializer)
	Suspend a serializer, causing tasks to be queued until unsuspended.
int	ast_taskpool_serializer_unsuspend (struct ast_taskprocessor *serializer)
	Unsuspend a serializer, causing tasks to be executed.
void	ast_taskpool_shutdown (struct ast_taskpool *pool)
	Shut down a taskpool and remove the underlying taskprocessors.
size_t	ast_taskpool_taskprocessors_count (struct ast_taskpool *pool)
	Get the current number of taskprocessors in the taskpool.
	AST_THREADSTORAGE_RAW (current_taskpool_pool)
	Thread storage for the current taskpool.
	AST_THREADSTORAGE_RAW (current_taskpool_serializer)
static int	execute_tasks (void *data)
static struct serializer *	serializer_create (struct ast_taskpool *pool, struct ast_serializer_shutdown_group *shutdown_group)
static void	serializer_dtor (void *obj)
static void	serializer_shutdown (struct ast_taskprocessor_listener *listener)
static int	serializer_start (struct ast_taskprocessor_listener *listener)
static void	serializer_task_pushed (struct ast_taskprocessor_listener *listener, int was_empty)
static void	taskpool_dynamic_pool_grow (struct ast_taskpool *pool, struct taskpool_taskprocessor **taskprocessor)
static int	taskpool_dynamic_pool_shrink (const void *data)
static void	taskpool_least_full_selector (struct ast_taskpool *pool, struct taskpool_taskprocessors *taskprocessors, struct taskpool_taskprocessor **taskprocessor, unsigned int *growth_threshold_reached)
	Least full taskprocessor selector.
static void	taskpool_sequential_selector (struct ast_taskpool *pool, struct taskpool_taskprocessors *taskprocessors, struct taskpool_taskprocessor **taskprocessor, unsigned int *growth_threshold_reached)
static int	taskpool_serializer_empty_task (void *data)
static int	taskpool_serializer_suspend_task (void *data)
static void	taskpool_shutdown (void)
static int	taskpool_sync_task (void *data)
static void	taskpool_sync_task_cleanup (struct taskpool_sync_task *sync_task)
static int	taskpool_sync_task_init (struct taskpool_sync_task *sync_task, int(*task)(void *), void *data)
static struct taskpool_taskprocessor *	taskpool_taskprocessor_alloc (struct ast_taskpool *pool, char type)
static void	taskpool_taskprocessor_dtor (void *obj)
static int	taskpool_taskprocessor_start (void *data)
static int	taskpool_taskprocessor_stop (void *data)
static void	taskpool_taskprocessors_cleanup (struct taskpool_taskprocessors *taskprocessors)
static int	taskpool_taskprocessors_init (struct taskpool_taskprocessors *taskprocessors, unsigned int size)

Variables
static struct ast_sched_context *	sched
	Scheduler used for dynamic pool shrinking.
static struct ast_taskprocessor_listener_callbacks	serializer_tps_listener_callbacks

Macro Definition Documentation

ast_taskpool_push_internal

#define ast_taskpool_push_internal	(		pool,
			task,
			data
	)		__ast_taskpool_push(pool, task, data, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Definition at line 523 of file taskpool.c.

TASKPOOL_GROW_THRESHOLD

#define TASKPOOL_GROW_THRESHOLD   (AST_TASKPROCESSOR_HIGH_WATER_LEVEL * 5) / 10

The threshold for a taskprocessor at which we consider the pool needing to grow (50% of high water threshold)

Definition at line 80 of file taskpool.c.

TASKPOOL_QUEUE_SIZE_ADD

#define TASKPOOL_QUEUE_SIZE_ADD	(		tps,
			size
	)		(size += ast_taskprocessor_size(tps->taskprocessor))

Definition at line 462 of file taskpool.c.

TASKPROCESSOR_IS_IDLE

#define TASKPROCESSOR_IS_IDLE	(		tps,
			timeout
	)		(ast_tvdiff_ms(ast_tvnow(), tps->last_pushed) > (timeout))

Definition at line 221 of file taskpool.c.

Typedef Documentation

taskpool_selector

typedef void(* taskpool_selector) (struct ast_taskpool *pool, struct taskpool_taskprocessors *taskprocessors, struct taskpool_taskprocessor **taskprocessor, unsigned int *growth_threshold_reached)

Definition at line 51 of file taskpool.c.

Enumeration Type Documentation

serializer_suspension

enum serializer_suspension

Enumerator
SERIALIZER_UNSUSPENDED
SERIALIZER_SUSPENDING
SERIALIZER_SUSPENDED

Definition at line 696 of file taskpool.c.

                           {
    SERIALIZER_UNSUSPENDED = 0, /* The serializer is unsuspended */
    SERIALIZER_SUSPENDING,      /* The serializer is pending suspension */
    SERIALIZER_SUSPENDED,       /* The serializer is suspended */
};

Function Documentation

__ast_taskpool_push()

int __ast_taskpool_push	(	struct ast_taskpool *	pool,
		int(*)(void *data)	task,
		void *	data,
		const char *	file,
		int	line,
		const char *	function
	)

Push a task to the taskpool.

Since

23.1.0

22.7.0

20.17.0

Tasks pushed into the taskpool will be automatically taken by one of the taskprocessors within

Parameters

pool	The taskpool 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 527 of file taskpool.c.

{
    RAII_VAR(struct taskpool_taskprocessor *, taskprocessor, NULL, ao2_cleanup);
 
    /* Select the taskprocessor in the pool to use for pushing this task */
    ao2_lock(pool);
    if (!pool->shutting_down) {
        unsigned int growth_threshold_reached = 0;
 
        /* A selector doesn't set taskprocessor to NULL, it will only change the value if a better
         * taskprocessor is found. This means that even if the selector for a dynamic taskprocessor
         * fails for some reason, it will still fall back to the initially found static one if
         * it is present.
         */
        pool->selector(pool, &pool->static_taskprocessors, &taskprocessor, &growth_threshold_reached);
        if (pool->options.auto_increment && growth_threshold_reached) {
            /* If we need to grow then try dynamic taskprocessors */
            pool->selector(pool, &pool->dynamic_taskprocessors, &taskprocessor, &growth_threshold_reached);
            if (growth_threshold_reached) {
                /* If we STILL need to grow then grow the dynamic taskprocessor pool if allowed */
                taskpool_dynamic_pool_grow(pool, &taskprocessor);
            }
 
            /* If a dynamic taskprocessor was used update its last push time */
            if (taskprocessor) {
                taskprocessor->last_pushed = ast_tvnow();
            }
        }
        ao2_bump(taskprocessor);
    }
    ao2_unlock(pool);
 
    if (!taskprocessor) {
        return -1;
    }
 
    if (__ast_taskprocessor_push(taskprocessor->taskprocessor, task, data, file, line, function)) {
        return -1;
    }
 
    return 0;
}

References __ast_taskprocessor_push(), ao2_bump, ao2_cleanup, ao2_lock, ao2_unlock, ast_tvnow(), ast_taskpool_options::auto_increment, ast_taskpool::dynamic_taskprocessors, NULL, ast_taskpool::options, RAII_VAR, ast_taskpool::selector, ast_taskpool::shutting_down, ast_taskpool::static_taskprocessors, task(), taskpool_dynamic_pool_grow(), and taskpool_taskprocessor::taskprocessor.

Referenced by __ast_sip_push_task(), __ast_taskpool_push_wait(), and ast_taskpool_push().

__ast_taskpool_push_wait()

int __ast_taskpool_push_wait	(	struct ast_taskpool *	pool,
		int(*)(void *data)	task,
		void *	data,
		const char *	file,
		int	line,
		const char *	function
	)

Push a task to the taskpool, and wait for completion.

Since

23.1.0

22.7.0

20.17.0

Tasks pushed into the taskpool will be automatically taken by one of the taskprocessors within

Parameters

pool	The taskpool 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 635 of file taskpool.c.

{
    struct taskpool_sync_task sync_task;
 
    /* If we are already executing within a taskpool taskprocessor then
     * don't bother pushing a new task, just directly execute the task.
     */
    if (ast_taskpool_get_current()) {
        return task(data);
    }
 
    if (taskpool_sync_task_init(&sync_task, task, data)) {
        return -1;
    }
 
    if (__ast_taskpool_push(pool, taskpool_sync_task, &sync_task, file, line, function)) {
        taskpool_sync_task_cleanup(&sync_task);
        return -1;
    }
 
    ast_mutex_lock(&sync_task.lock);
    while (!sync_task.complete) {
        ast_cond_wait(&sync_task.cond, &sync_task.lock);
    }
    ast_mutex_unlock(&sync_task.lock);
 
    taskpool_sync_task_cleanup(&sync_task);
    return sync_task.fail;
}

References __ast_taskpool_push(), ast_cond_wait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_get_current(), taskpool_sync_task::complete, taskpool_sync_task::cond, taskpool_sync_task::fail, taskpool_sync_task::lock, task(), taskpool_sync_task_cleanup(), and taskpool_sync_task_init().

Referenced by __ast_sip_push_task_wait(), __ast_sip_push_task_wait_serializer(), and ast_taskpool_push_wait().

__ast_taskpool_serializer_push_wait()

int __ast_taskpool_serializer_push_wait	(	struct ast_taskprocessor *	serializer,
		int(*)(void *data)	task,
		void *	data,
		const char *	file,
		int	line,
		const char *	function
	)

Push a task to a serializer, and wait for completion.

Since

23.1.0

22.7.0

20.17.0

Parameters

serializer	The serializer 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 882 of file taskpool.c.

{
    struct ast_taskprocessor_listener *listener = ast_taskprocessor_listener(serializer);
    struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
    struct ast_taskprocessor *prior_serializer;
    struct taskpool_sync_task sync_task;
 
    /* If not in a taskpool taskprocessor we can just queue the task like normal and
     * wait. */
    if (!ast_taskpool_get_current()) {
        if (taskpool_sync_task_init(&sync_task, task, data)) {
            return -1;
        }
 
        if (__ast_taskprocessor_push(serializer, taskpool_sync_task, &sync_task, file, line, function)) {
            taskpool_sync_task_cleanup(&sync_task);
            return -1;
        }
 
        ast_mutex_lock(&sync_task.lock);
        while (!sync_task.complete) {
            ast_cond_wait(&sync_task.cond, &sync_task.lock);
        }
        ast_mutex_unlock(&sync_task.lock);
 
        taskpool_sync_task_cleanup(&sync_task);
        return sync_task.fail;
    }
 
    /* It is possible that we are already executing within a serializer, so stash the existing
     * away so we can restore it.
     */
    prior_serializer = ast_taskpool_serializer_get_current();
 
    ao2_lock(ser);
 
    /* There are two cases where we can or have to directly execute this task:
     * 1. There are no other tasks in the serializer
     * 2. We are already in the serializer
     * In the second case if we don't execute the task now, we will deadlock waiting
     * on it as it will never occur.
     */
    if (!ast_taskprocessor_size(serializer) || prior_serializer == serializer) {
        ast_threadstorage_set_ptr(&current_taskpool_serializer, serializer);
        sync_task.fail = task(data);
        ao2_unlock(ser);
        ast_threadstorage_set_ptr(&current_taskpool_serializer, prior_serializer);
        return sync_task.fail;
    }
 
    if (taskpool_sync_task_init(&sync_task, task, data)) {
        ao2_unlock(ser);
        return -1;
    }
 
    /* First we queue the serialized task */
    if (__ast_taskprocessor_push(serializer, taskpool_sync_task, &sync_task, file, line, function)) {
        taskpool_sync_task_cleanup(&sync_task);
        ao2_unlock(ser);
        return -1;
    }
 
    /* Next we queue the empty task to ensure the serializer doesn't reach empty, this
     * stops two tasks from being queued for the same serializer at the same time.
     */
    if (ast_taskprocessor_push(serializer, taskpool_serializer_empty_task, NULL)) {
        taskpool_sync_task_cleanup(&sync_task);
        ao2_unlock(ser);
        return -1;
    }
 
    /* Now we execute the tasks on the serializer until our sync task is complete */
    ast_threadstorage_set_ptr(&current_taskpool_serializer, serializer);
    while (!sync_task.complete) {
        /* If the serializer is suspended wait until it unsuspends */
        while (ser->suspended == SERIALIZER_SUSPENDED) {
            ast_cond_wait(&ser->cond,  ao2_object_get_lockaddr(ser));
        }
 
        /* The sync task is guaranteed to be executed, so doing a while loop on the complete
         * flag is safe.
         */
        ast_taskprocessor_execute(serializer);
    }
    taskpool_sync_task_cleanup(&sync_task);
    ao2_unlock(ser);
 
    ast_threadstorage_set_ptr(&current_taskpool_serializer, prior_serializer);
 
    return sync_task.fail;
}

References __ast_taskprocessor_push(), ao2_lock, ao2_object_get_lockaddr(), ao2_unlock, ast_cond_wait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_get_current(), ast_taskpool_serializer_get_current(), ast_taskprocessor_execute(), ast_taskprocessor_listener_get_user_data(), ast_taskprocessor_push, ast_taskprocessor_size(), ast_threadstorage_set_ptr(), taskpool_sync_task::complete, taskpool_sync_task::cond, serializer::cond, taskpool_sync_task::fail, listener(), taskpool_sync_task::lock, NULL, SERIALIZER_SUSPENDED, serializer::suspended, task(), taskpool_serializer_empty_task(), taskpool_sync_task_cleanup(), and taskpool_sync_task_init().

Referenced by __ast_sip_push_task_wait(), __ast_sip_push_task_wait_serializer(), and ast_taskpool_serializer_push_wait().

ast_taskpool_create()

struct ast_taskpool * ast_taskpool_create	(	const char *	name,
		const struct ast_taskpool_options *	options
	)

Create a new taskpool.

Since

23.1.0

22.7.0

20.17.0

This function creates a taskpool. Tasks may be pushed onto this task pool and will be automatically acted upon by taskprocessors within the pool.

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

Parameters

name	The unique name for the taskpool
options	The behavioral options for this taskpool

Return values

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

Note

The ast_taskpool_shutdown function must be called to shut down the taskpool and clean up underlying resources fully.

Definition at line 324 of file taskpool.c.

{
    struct ast_taskpool *pool;
 
    /* Enforce versioning on the passed-in options */
    if (options->version != AST_TASKPOOL_OPTIONS_VERSION) {
        return NULL;
    }
 
    pool = ao2_alloc(sizeof(*pool) + strlen(name) + 1, NULL);
    if (!pool) {
        return NULL;
    }
 
    strcpy(pool->name, name); /* Safe */
    memcpy(&pool->options, options, sizeof(pool->options));
    pool->shrink_sched_id = -1;
 
    /* Verify the passed-in options are valid, and adjust if needed */
    if (options->initial_size < options->minimum_size) {
        pool->options.initial_size = options->minimum_size;
        ast_log(LOG_WARNING, "Taskpool '%s' has an initial size of %d, which is less than the minimum size of %d. Adjusting to %d.\n",
            name, options->initial_size, options->minimum_size, options->minimum_size);
    }
 
    if (options->max_size && pool->options.initial_size > options->max_size) {
        pool->options.max_size = pool->options.initial_size;
        ast_log(LOG_WARNING, "Taskpool '%s' has a max size of %d, which is less than the initial size of %d. Adjusting to %d.\n",
            name, options->max_size, pool->options.initial_size, pool->options.initial_size);
    }
 
    if (!options->auto_increment) {
        if (!pool->options.minimum_size) {
            pool->options.minimum_size = 1;
            ast_log(LOG_WARNING, "Taskpool '%s' has a minimum size of 0, which is not valid without auto increment. Adjusting to 1.\n", name);
        }
        if (!pool->options.max_size) {
            pool->options.max_size = pool->options.minimum_size;
            ast_log(LOG_WARNING, "Taskpool '%s' has a max size of 0, which is not valid without auto increment. Adjusting to %d.\n", name, pool->options.minimum_size);
        }
        if (pool->options.minimum_size != pool->options.max_size) {
            pool->options.minimum_size = pool->options.max_size;
            pool->options.initial_size = pool->options.max_size;
            ast_log(LOG_WARNING, "Taskpool '%s' has a minimum size of %d, while max size is %d. Adjusting all sizes to %d due to lack of auto increment.\n",
                name, options->minimum_size, pool->options.max_size, pool->options.max_size);
        }
    } else if (!options->growth_threshold) {
        pool->options.growth_threshold = TASKPOOL_GROW_THRESHOLD;
    }
 
    if (options->selector == AST_TASKPOOL_SELECTOR_DEFAULT || options->selector == AST_TASKPOOL_SELECTOR_LEAST_FULL) {
        pool->selector = taskpool_least_full_selector;
    } else if (options->selector == AST_TASKPOOL_SELECTOR_SEQUENTIAL) {
        pool->selector = taskpool_sequential_selector;
    } else {
        ast_log(LOG_WARNING, "Taskpool '%s' has an invalid selector of %d. Adjusting to default selector.\n",
            name, options->selector);
        pool->selector = taskpool_least_full_selector;
    }
 
    if (taskpool_taskprocessors_init(&pool->static_taskprocessors, pool->options.minimum_size)) {
        ao2_ref(pool, -1);
        return NULL;
    }
 
    /* Create the static taskprocessors based on the passed-in options */
    for (int i = 0; i < pool->options.minimum_size; i++) {
        struct taskpool_taskprocessor *taskprocessor;
 
        taskprocessor = taskpool_taskprocessor_alloc(pool, 's');
        if (!taskprocessor) {
            /* The reference to pool is passed to ast_taskpool_shutdown */
            ast_taskpool_shutdown(pool);
            return NULL;
        }
 
        if (AST_VECTOR_APPEND(&pool->static_taskprocessors.taskprocessors, taskprocessor)) {
            ao2_ref(taskprocessor, -1);
            /* The reference to pool is passed to ast_taskpool_shutdown */
            ast_taskpool_shutdown(pool);
            return NULL;
        }
    }
 
    if (taskpool_taskprocessors_init(&pool->dynamic_taskprocessors,
        pool->options.initial_size - pool->options.minimum_size)) {
        ast_taskpool_shutdown(pool);
        return NULL;
    }
 
    /* Create the dynamic taskprocessor based on the passed-in options */
    for (int i = 0; i < (pool->options.initial_size - pool->options.minimum_size); i++) {
        struct taskpool_taskprocessor *taskprocessor;
 
        taskprocessor = taskpool_taskprocessor_alloc(pool, 'd');
        if (!taskprocessor) {
            /* The reference to pool is passed to ast_taskpool_shutdown */
            ast_taskpool_shutdown(pool);
            return NULL;
        }
 
        if (AST_VECTOR_APPEND(&pool->dynamic_taskprocessors.taskprocessors, taskprocessor)) {
            ao2_ref(taskprocessor, -1);
            /* The reference to pool is passed to ast_taskpool_shutdown */
            ast_taskpool_shutdown(pool);
            return NULL;
        }
    }
 
    /* If idle timeout support is enabled kick off a scheduled task to shrink the dynamic pool periodically, we do
     * this no matter if there are dynamic taskprocessor present to reduce the work needed within the push function
     * and to reduce complexity.
     */
    if (options->idle_timeout && options->auto_increment) {
        pool->shrink_sched_id = ast_sched_add(sched, options->idle_timeout * 1000, taskpool_dynamic_pool_shrink, ao2_bump(pool));
        if (pool->shrink_sched_id < 0) {
            ao2_ref(pool, -1);
            /* The second reference to pool is passed to ast_taskpool_shutdown */
            ast_taskpool_shutdown(pool);
            return NULL;
        }
    }
 
    return pool;
}

References ao2_alloc, ao2_bump, ao2_ref, ast_log, ast_sched_add(), AST_TASKPOOL_OPTIONS_VERSION, AST_TASKPOOL_SELECTOR_DEFAULT, AST_TASKPOOL_SELECTOR_LEAST_FULL, AST_TASKPOOL_SELECTOR_SEQUENTIAL, ast_taskpool_shutdown(), AST_VECTOR_APPEND, ast_taskpool::dynamic_taskprocessors, ast_taskpool_options::growth_threshold, ast_taskpool_options::initial_size, LOG_WARNING, ast_taskpool_options::max_size, ast_taskpool_options::minimum_size, ast_taskpool::name, name, NULL, ast_taskpool::options, options, ast_taskpool::selector, ast_taskpool::shrink_sched_id, ast_taskpool::static_taskprocessors, taskpool_dynamic_pool_shrink(), TASKPOOL_GROW_THRESHOLD, taskpool_least_full_selector(), taskpool_sequential_selector(), taskpool_taskprocessor_alloc(), taskpool_taskprocessors_init(), taskpool_taskprocessor::taskprocessor, and taskpool_taskprocessors::taskprocessors.

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

ast_taskpool_get_current()

static struct ast_taskpool * ast_taskpool_get_current ( void  )

static

Definition at line 92 of file taskpool.c.

{
    return ast_threadstorage_get_ptr(&current_taskpool_pool);
}

References ast_threadstorage_get_ptr().

Referenced by __ast_taskpool_push_wait(), __ast_taskpool_serializer_push_wait(), ast_taskpool_serializer_suspend(), ast_taskpool_serializer_unsuspend(), execute_tasks(), and taskpool_taskprocessor_stop().

ast_taskpool_init()

int ast_taskpool_init

(

void

)

Provided by taskpool.c

Definition at line 1093 of file taskpool.c.

{
    sched = ast_sched_context_create();
    if (!sched) {
        return -1;
    }
 
    if (ast_sched_start_thread(sched)) {
        return -1;
    }
 
    ast_register_cleanup(taskpool_shutdown);
 
    return 0;
}

References ast_register_cleanup(), ast_sched_context_create(), ast_sched_start_thread(), and taskpool_shutdown().

Referenced by asterisk_daemon().

ast_taskpool_push()

int ast_taskpool_push	(	struct ast_taskpool *	pool,
		int(*)(void *data)	task,
		void *	data
	)

Definition at line 572 of file taskpool.c.

{
    return __ast_taskpool_push(pool, task, data, NULL, 0, NULL);
}

References __ast_taskpool_push(), NULL, and task().

ast_taskpool_push_wait()

int ast_taskpool_push_wait	(	struct ast_taskpool *	pool,
		int(*)(void *data)	task,
		void *	data
	)

Definition at line 670 of file taskpool.c.

{
    return __ast_taskpool_push_wait(pool, task, data, NULL, 0, NULL);
}

References __ast_taskpool_push_wait(), NULL, and task().

ast_taskpool_queue_size()

long ast_taskpool_queue_size

(

struct ast_taskpool *

pool

)

Get the current number of queued tasks in the taskpool.

Since

23.1.0

22.7.0

20.17.0

Parameters

pool	The taskpool to query

Return values

The	number of queued tasks in the taskpool

Definition at line 464 of file taskpool.c.

{
    long queue_size = 0;
 
    ao2_lock(pool);
    AST_VECTOR_CALLBACK_VOID(&pool->static_taskprocessors.taskprocessors, TASKPOOL_QUEUE_SIZE_ADD, queue_size);
    AST_VECTOR_CALLBACK_VOID(&pool->dynamic_taskprocessors.taskprocessors, TASKPOOL_QUEUE_SIZE_ADD, queue_size);
    ao2_unlock(pool);
 
    return queue_size;
}

References ao2_lock, ao2_unlock, AST_VECTOR_CALLBACK_VOID, ast_taskpool::dynamic_taskprocessors, ast_taskpool::static_taskprocessors, TASKPOOL_QUEUE_SIZE_ADD, and taskpool_taskprocessors::taskprocessors.

Referenced by ast_sip_taskpool_queue_size().

ast_taskpool_serializer()

struct ast_taskprocessor * ast_taskpool_serializer	(	const char *	name,
		struct ast_taskpool *	pool
	)

Serialized execution of tasks within a ast_taskpool.

Since

23.1.0

22.7.0

20.17.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 taskprocessor from a ast_taskpool.

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 relies 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_taskpool for execution.

Returns

ast_taskprocessor for enqueuing work.

Return values

NULL

on error.

Definition at line 860 of file taskpool.c.

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

References ast_taskpool_serializer_group(), name, and NULL.

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

ast_taskpool_serializer_get_current()

struct ast_taskprocessor * ast_taskpool_serializer_get_current

(

void

)

Get the taskpool serializer currently associated with this thread.

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 825 of file taskpool.c.

{
    return ast_threadstorage_get_ptr(&current_taskpool_serializer);
}

References ast_threadstorage_get_ptr().

Referenced by __ast_taskpool_serializer_push_wait(), record_serializer(), requeue_task(), rfc3326_outgoing_request(), rfc3326_outgoing_response(), serializer_efficiency_task(), and simple_task().

ast_taskpool_serializer_group()

struct ast_taskprocessor * ast_taskpool_serializer_group	(	const char *	name,
		struct ast_taskpool *	pool,
		struct ast_serializer_shutdown_group *	shutdown_group
	)

Serialized execution of tasks within a ast_taskpool.

Since

23.1.0

22.7.0

20.17.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 taskprocessor from a ast_taskpool.

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 relies 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_taskpool for execution.
shutdown_group	Group shutdown controller. (NULL if no group association)

Returns

ast_taskprocessor for enqueuing work.

Return values

NULL

on error.

Definition at line 830 of file taskpool.c.

{
    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) {
        ast_serializer_shutdown_group_inc(shutdown_group);
    }
 
    ao2_ref(listener, -1);
    return tps;
}

References ao2_ref, ast_serializer_shutdown_group_inc(), ast_taskprocessor_create_with_listener(), ast_taskprocessor_listener_alloc(), listener(), name, NULL, serializer_create(), serializer_tps_listener_callbacks, and shutdown_group.

Referenced by ast_serializer_taskpool_create(), ast_sip_create_serializer_group(), and ast_taskpool_serializer().

ast_taskpool_serializer_push_wait()

int ast_taskpool_serializer_push_wait	(	struct ast_taskprocessor *	serializer,
		int(*)(void *data)	task,
		void *	data
	)

Definition at line 877 of file taskpool.c.

{
    return __ast_taskpool_serializer_push_wait(serializer, task, data, NULL, 0, NULL);
}

References __ast_taskpool_serializer_push_wait(), NULL, and task().

ast_taskpool_serializer_suspend()

int ast_taskpool_serializer_suspend

(

struct ast_taskprocessor *

serializer

)

Suspend a serializer, causing tasks to be queued until unsuspended.

Since

23.2.0

22.8.0

20.18.0

Parameters

serializer

The serializer to suspend

Return values

0	success
-1	failure

Note

May only be invoked from outside of the taskpool

Definition at line 1001 of file taskpool.c.

{
    struct ast_taskprocessor_listener *listener = ast_taskprocessor_listener(serializer);
    struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
 
    /* This suspension process works by inserting a checkpoint into the queue of the
     * serializer. Once this checkpoint is reached the taskpool taskprocessor handling
     * the queue stops prematurely and does not get requeued. For the case where a
     * synchronous task wait is in progress it is instead paused temporarily. Once
     * the serializer is unsuspended a new execution task is queued into the taskpool
     * to resume execution and any paused synchronous task waits are awoken to resume
     * their own execution as well. This approach minimizes the number of threads that
     * are paused waiting, nominally to 0.
     */
 
    if (ast_taskpool_get_current()) {
        return -1;
    }
 
    ao2_lock(ser);
 
    /* If the serializer is already suspending or suspended, just return immediately.
     * This mirrors the original behavior from PJSIP.
     */
    if (ser->suspended != SERIALIZER_UNSUSPENDED) {
        ao2_unlock(ser);
        return 0;
    }
 
    ser->suspended = SERIALIZER_SUSPENDING;
 
    ao2_unlock(ser);
 
    /* Once this returns successfully there is no thread executing the tasks on the serializer,
     * so they will accumulate until the serializer is unsuspended.
     */
    if (ast_taskpool_serializer_push_wait(serializer, taskpool_serializer_suspend_task, serializer)) {
        /* Suspension failed, so unsuspend as doing otherwise would leave the serializer in a stuck
         * state.
         */
        ao2_lock(ser);
        ser->suspended = SERIALIZER_UNSUSPENDED;
        ao2_unlock(ser);
        return -1;
    }
 
    return 0;
}

References ao2_lock, ao2_unlock, ast_taskpool_get_current(), ast_taskpool_serializer_push_wait, ast_taskprocessor_listener_get_user_data(), listener(), SERIALIZER_SUSPENDING, SERIALIZER_UNSUSPENDED, serializer::suspended, and taskpool_serializer_suspend_task().

Referenced by ast_sip_session_suspend(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

ast_taskpool_serializer_unsuspend()

int ast_taskpool_serializer_unsuspend

(

struct ast_taskprocessor *

serializer

)

Unsuspend a serializer, causing tasks to be executed.

Since

23.2.0

22.8.0

20.18.0

Parameters

serializer

The serializer to unsuspend

Return values

0	success
-1	failure

Note

May only be invoked from outside of the taskpool

Definition at line 1050 of file taskpool.c.

{
    struct ast_taskprocessor_listener *listener = ast_taskprocessor_listener(serializer);
    struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
 
    if (ast_taskpool_get_current()) {
        return -1;
    }
 
    ao2_lock(ser);
 
    if (ser->suspended != SERIALIZER_SUSPENDED) {
        ao2_unlock(ser);
        return 0;
    }
 
    ser->suspended = SERIALIZER_UNSUSPENDED;
 
    /* Notify any other interested threads that this one has awoken */
    ast_cond_broadcast(&ser->cond);
 
    /* And now we kick off handling of the queued tasks once again */
    if (ast_taskpool_push(ser->pool, execute_tasks, ao2_bump(serializer))) {
        ast_taskprocessor_unreference(serializer);
    }
 
    ao2_unlock(ser);
 
    return 0;
}

References ao2_bump, ao2_lock, ao2_unlock, ast_cond_broadcast, ast_taskpool_get_current(), ast_taskpool_push, ast_taskprocessor_listener_get_user_data(), ast_taskprocessor_unreference(), serializer::cond, execute_tasks(), listener(), serializer::pool, SERIALIZER_SUSPENDED, SERIALIZER_UNSUSPENDED, and serializer::suspended.

Referenced by ast_sip_session_unsuspend(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), and AST_TEST_DEFINE().

ast_taskpool_shutdown()

void ast_taskpool_shutdown

(

struct ast_taskpool *

pool

)

Shut down a taskpool and remove the underlying taskprocessors.

Since

23.1.0

22.7.0

20.17.0

Parameters

pool	The pool to shut down

Note

This will decrement the reference to the pool

Definition at line 675 of file taskpool.c.

{
    if (!pool) {
        return;
    }
 
    /* Mark this pool as shutting down so nothing new is pushed */
    ao2_lock(pool);
    pool->shutting_down = 1;
    ao2_unlock(pool);
 
    /* Stop the shrink scheduled item if present */
    AST_SCHED_DEL_UNREF(sched, pool->shrink_sched_id, ao2_ref(pool, -1));
 
    /* Clean up all the taskprocessors */
    taskpool_taskprocessors_cleanup(&pool->static_taskprocessors);
    taskpool_taskprocessors_cleanup(&pool->dynamic_taskprocessors);
 
    ao2_ref(pool, -1);
}

References ao2_lock, ao2_ref, ao2_unlock, AST_SCHED_DEL_UNREF, ast_taskpool::dynamic_taskprocessors, ast_taskpool::shrink_sched_id, ast_taskpool::shutting_down, ast_taskpool::static_taskprocessors, and taskpool_taskprocessors_cleanup().

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

ast_taskpool_taskprocessors_count()

size_t ast_taskpool_taskprocessors_count

(

struct ast_taskpool *

pool

)

Get the current number of taskprocessors in the taskpool.

Since

23.1.0

22.7.0

20.17.0

Parameters

pool	The taskpool to query

Return values

The	number of taskprocessors in the taskpool

Definition at line 451 of file taskpool.c.

{
    size_t count;
 
    ao2_lock(pool);
    count = AST_VECTOR_SIZE(&pool->static_taskprocessors.taskprocessors) + AST_VECTOR_SIZE(&pool->dynamic_taskprocessors.taskprocessors);
    ao2_unlock(pool);
 
    return count;
}

References ao2_lock, ao2_unlock, AST_VECTOR_SIZE, ast_taskpool::dynamic_taskprocessors, ast_taskpool::static_taskprocessors, and taskpool_taskprocessors::taskprocessors.

Referenced by AST_TEST_DEFINE(), and AST_TEST_DEFINE().

AST_THREADSTORAGE_RAW() [1/2]

AST_THREADSTORAGE_RAW

(

current_taskpool_pool

)

Thread storage for the current taskpool.

AST_THREADSTORAGE_RAW() [2/2]

AST_THREADSTORAGE_RAW

(

current_taskpool_serializer

)

execute_tasks()

static int execute_tasks ( void *  data )

static

Definition at line 742 of file taskpool.c.

{
    struct ast_taskpool *pool = ast_taskpool_get_current();
    struct ast_taskprocessor *tps = data;
    struct ast_taskprocessor_listener *listener = ast_taskprocessor_listener(tps);
    struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
    size_t remaining, requeue = 0;
 
    /* In a normal scenario this lock will not be in contention with
     * anything else. It is only if a synchronous task is pushed to
     * the serializer that it may be blocked on the synchronous
     * task thread. This is done to ensure that only one thread is executing
     * tasks from the serializer at a given time, and not out of order
     * either.
     */
    ao2_lock(ser);
 
    ast_threadstorage_set_ptr(&current_taskpool_serializer, tps);
    for (remaining = ast_taskprocessor_size(tps); remaining > 0; remaining--) {
        requeue = ast_taskprocessor_execute(tps);
 
        /* If the serializer is suspended we will not execute any more tasks and
         * we will not requeue the taskpool task. Instead it will be requeued when
         * the serializer is unsuspended.
         */
        if (ser->suspended == SERIALIZER_SUSPENDED) {
            requeue = 0;
            break;
        }
    }
    ast_threadstorage_set_ptr(&current_taskpool_serializer, NULL);
 
    ao2_unlock(ser);
 
    /* If there are remaining tasks we requeue, this way the serializer
     * does not hold exclusivity of the taskpool taskprocessor
     */
    if (requeue) {
        /* Ownership passes to the new task */
        if (ast_taskpool_push(pool, execute_tasks, tps)) {
            ast_taskprocessor_unreference(tps);
        }
    } else {
        ast_taskprocessor_unreference(tps);
    }
 
    return 0;
}

References ao2_lock, ao2_unlock, ast_taskpool_get_current(), ast_taskpool_push, ast_taskprocessor_execute(), ast_taskprocessor_listener_get_user_data(), ast_taskprocessor_size(), ast_taskprocessor_unreference(), ast_threadstorage_set_ptr(), execute_tasks(), listener(), NULL, serializer::pool, SERIALIZER_SUSPENDED, serializer::suspended, and ast_taskprocessor_listener::tps.

Referenced by ast_taskpool_serializer_unsuspend(), execute_tasks(), and serializer_task_pushed().

serializer_create()

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

static

Definition at line 722 of file taskpool.c.

{
    struct serializer *ser;
 
    /* This object has a lock so it can be used to ensure exclusive access
     * to the execution of tasks within the serializer.
     */
    ser = ao2_alloc(sizeof(*ser), serializer_dtor);
    if (!ser) {
        return NULL;
    }
    ser->pool = ao2_bump(pool);
    ser->shutdown_group = ao2_bump(shutdown_group);
    ast_cond_init(&ser->cond, NULL);
    return ser;
}

References ao2_alloc, ao2_bump, ast_cond_init, serializer::cond, NULL, serializer::pool, serializer_dtor(), serializer::shutdown_group, and shutdown_group.

Referenced by ast_taskpool_serializer_group().

serializer_dtor()

static void serializer_dtor ( void *  obj )

static

Definition at line 713 of file taskpool.c.

{
    struct serializer *ser = obj;
 
    ao2_cleanup(ser->pool);
    ao2_cleanup(ser->shutdown_group);
    ast_cond_destroy(&ser->cond);
}

References ao2_cleanup, ast_cond_destroy, serializer::cond, serializer::pool, and serializer::shutdown_group.

Referenced by serializer_create().

serializer_shutdown()

static void serializer_shutdown ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 809 of file taskpool.c.

{
    struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
 
    if (ser->shutdown_group) {
        ast_serializer_shutdown_group_dec(ser->shutdown_group);
    }
    ao2_cleanup(ser);
}

References ao2_cleanup, ast_serializer_shutdown_group_dec(), ast_taskprocessor_listener_get_user_data(), listener(), and serializer::shutdown_group.

serializer_start()

static int serializer_start ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 803 of file taskpool.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 791 of file taskpool.c.

{
    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_taskpool_push(ser->pool, execute_tasks, tps)) {
            ast_taskprocessor_unreference(tps);
        }
    }
}

References ast_taskpool_push, ast_taskprocessor_listener_get_tps(), ast_taskprocessor_listener_get_user_data(), ast_taskprocessor_unreference(), execute_tasks(), listener(), and serializer::pool.

taskpool_dynamic_pool_grow()

static void taskpool_dynamic_pool_grow ( struct ast_taskpool *  pool, struct taskpool_taskprocessor **  taskprocessor  )

static

Definition at line 479 of file taskpool.c.

{
    unsigned int num_to_add = pool->options.auto_increment;
    int i;
 
    if (!num_to_add) {
        return;
    }
 
    /* If a maximum size is enforced, then determine if we have to limit how many taskprocessors we add */
    if (pool->options.max_size) {
        unsigned int current_size = AST_VECTOR_SIZE(&pool->dynamic_taskprocessors.taskprocessors) + AST_VECTOR_SIZE(&pool->static_taskprocessors.taskprocessors);
 
        if (current_size + num_to_add > pool->options.max_size) {
            num_to_add = pool->options.max_size - current_size;
        }
    }
 
    for (i = 0; i < num_to_add; i++) {
        struct taskpool_taskprocessor *new_taskprocessor;
 
        new_taskprocessor = taskpool_taskprocessor_alloc(pool, 'd');
        if (!new_taskprocessor) {
            return;
        }
 
        if (AST_VECTOR_APPEND(&pool->dynamic_taskprocessors.taskprocessors, new_taskprocessor)) {
            ao2_ref(new_taskprocessor, -1);
            return;
        }
 
        if (i == 0) {
            /* On the first iteration we return the taskprocessor we just added */
            *taskprocessor = new_taskprocessor;
            /* We assume we will be going back to the first taskprocessor, since we are at the end of the vector */
            pool->dynamic_taskprocessors.taskprocessor_num = 0;
        } else if (i == 1) {
            /* On the second iteration we update the next taskprocessor to use to be this one */
            pool->dynamic_taskprocessors.taskprocessor_num = AST_VECTOR_SIZE(&pool->dynamic_taskprocessors.taskprocessors) - 1;
        }
    }
}

References ao2_ref, AST_VECTOR_APPEND, AST_VECTOR_SIZE, ast_taskpool_options::auto_increment, ast_taskpool::dynamic_taskprocessors, ast_taskpool_options::max_size, ast_taskpool::options, ast_taskpool::static_taskprocessors, taskpool_taskprocessor_alloc(), taskpool_taskprocessor::taskprocessor, taskpool_taskprocessors::taskprocessor_num, and taskpool_taskprocessors::taskprocessors.

Referenced by __ast_taskpool_push().

taskpool_dynamic_pool_shrink()

static int taskpool_dynamic_pool_shrink ( const void *  data )

static

Definition at line 226 of file taskpool.c.

{
    struct ast_taskpool *pool = (struct ast_taskpool *)data;
    int num_removed;
 
    ao2_lock(pool);
 
    /* If the pool is shutting down, do nothing and don't reschedule */
    if (pool->shutting_down) {
        ao2_unlock(pool);
        ao2_ref(pool, -1);
        return 0;
    }
 
    /* Go through the dynamic taskprocessors and find any which have been idle long enough and remove them */
    num_removed = AST_VECTOR_REMOVE_ALL_CMP_UNORDERED(&pool->dynamic_taskprocessors.taskprocessors, pool->options.idle_timeout * 1000,
        TASKPROCESSOR_IS_IDLE, ao2_cleanup);
    if (num_removed) {
        /* If we've removed any taskprocessors the taskprocessor_num may no longer be valid, so update it */
        if (pool->dynamic_taskprocessors.taskprocessor_num >= AST_VECTOR_SIZE(&pool->dynamic_taskprocessors.taskprocessors)) {
            pool->dynamic_taskprocessors.taskprocessor_num = 0;
        }
    }
 
    ao2_unlock(pool);
 
    /* It is possible for the pool to have been shut down between unlocking and returning, this is
     * inherently a race condition we can't eliminate so we will catch it on the next iteration.
     */
    return pool->options.idle_timeout * 1000;
}

References ao2_cleanup, ao2_lock, ao2_ref, ao2_unlock, AST_VECTOR_REMOVE_ALL_CMP_UNORDERED, AST_VECTOR_SIZE, ast_taskpool::dynamic_taskprocessors, ast_taskpool_options::idle_timeout, ast_taskpool::options, ast_taskpool::shutting_down, TASKPROCESSOR_IS_IDLE, taskpool_taskprocessors::taskprocessor_num, and taskpool_taskprocessors::taskprocessors.

Referenced by ast_taskpool_create().

taskpool_least_full_selector()

static void taskpool_least_full_selector ( struct ast_taskpool *  pool, struct taskpool_taskprocessors *  taskprocessors, struct taskpool_taskprocessor **  taskprocessor, unsigned int *  growth_threshold_reached  )

static

Least full taskprocessor selector.

\interal

Definition at line 287 of file taskpool.c.

{
    struct taskpool_taskprocessor *least_full = NULL;
    unsigned int i;
 
    if (!AST_VECTOR_SIZE(&taskprocessors->taskprocessors)) {
        *growth_threshold_reached = 1;
        return;
    }
 
    /* We assume that the growth threshold has not yet been reached, until proven otherwise */
    *growth_threshold_reached = 0;
 
    for (i = 0; i < AST_VECTOR_SIZE(&taskprocessors->taskprocessors); i++) {
        struct taskpool_taskprocessor *tp = AST_VECTOR_GET(&taskprocessors->taskprocessors, i);
 
        /* If this taskprocessor has no outstanding tasks, it is the best choice */
        if (!ast_taskprocessor_size(tp->taskprocessor)) {
            *taskprocessor = tp;
            return;
        }
 
        /* If any of the taskprocessors have reached the growth threshold then we should grow the pool */
        if (ast_taskprocessor_size(tp->taskprocessor) >= pool->options.growth_threshold) {
            *growth_threshold_reached = 1;
        }
 
        /* The taskprocessor with the fewest tasks should be used */
        if (!least_full || ast_taskprocessor_size(tp->taskprocessor) < ast_taskprocessor_size(least_full->taskprocessor)) {
            least_full = tp;
        }
    }
 
    *taskprocessor = least_full;
}

References ast_taskprocessor_size(), AST_VECTOR_GET, AST_VECTOR_SIZE, ast_taskpool_options::growth_threshold, NULL, ast_taskpool::options, taskpool_taskprocessor::taskprocessor, and taskpool_taskprocessors::taskprocessors.

Referenced by ast_taskpool_create().

taskpool_sequential_selector()

static void taskpool_sequential_selector ( struct ast_taskpool *  pool, struct taskpool_taskprocessors *  taskprocessors, struct taskpool_taskprocessor **  taskprocessor, unsigned int *  growth_threshold_reached  )

static

Definition at line 262 of file taskpool.c.

{
    unsigned int taskprocessor_num = taskprocessors->taskprocessor_num;
 
    if (!AST_VECTOR_SIZE(&taskprocessors->taskprocessors)) {
        *growth_threshold_reached = 1;
        return;
    }
 
    taskprocessors->taskprocessor_num++;
    if (taskprocessors->taskprocessor_num == AST_VECTOR_SIZE(&taskprocessors->taskprocessors)) {
        taskprocessors->taskprocessor_num = 0;
    }
 
    *taskprocessor = AST_VECTOR_GET(&taskprocessors->taskprocessors, taskprocessor_num);
 
    /* Check to see if this has reached the growth threshold */
    *growth_threshold_reached = (ast_taskprocessor_size((*taskprocessor)->taskprocessor) >= pool->options.growth_threshold) ? 1 : 0;
}

References ast_taskprocessor_size(), AST_VECTOR_GET, AST_VECTOR_SIZE, ast_taskpool_options::growth_threshold, ast_taskpool::options, taskpool_taskprocessors::taskprocessor_num, and taskpool_taskprocessors::taskprocessors.

Referenced by ast_taskpool_create().

taskpool_serializer_empty_task()

static int taskpool_serializer_empty_task ( void *  data )

static

Definition at line 868 of file taskpool.c.

{
    return 0;
}

Referenced by __ast_taskpool_serializer_push_wait(), and taskpool_serializer_suspend_task().

taskpool_serializer_suspend_task()

static int taskpool_serializer_suspend_task ( void *  data )

static

Definition at line 978 of file taskpool.c.

{
    struct ast_taskprocessor *serializer = data;
    struct ast_taskprocessor_listener *listener = ast_taskprocessor_listener(serializer);
    struct serializer *ser = ast_taskprocessor_listener_get_user_data(listener);
 
    /* First we queue the empty task to ensure the serializer doesn't reach empty, this
     * prevents any threads from queueing up a taskpool task that executes the serializer
     * while it is suspended, allowing us to queue it ourselves when the serializer is
     * unsuspended.
     */
    if (ast_taskprocessor_push(serializer, taskpool_serializer_empty_task, NULL)) {
        return -1;
    }
 
    /* Next we suspend the serializer so that the execute_tasks currently executing stops
     * and doesn't requeue.
     */
    ser->suspended = SERIALIZER_SUSPENDED;
 
    return 0;
}

References ast_taskprocessor_listener_get_user_data(), ast_taskprocessor_push, listener(), NULL, SERIALIZER_SUSPENDED, serializer::suspended, and taskpool_serializer_empty_task().

Referenced by ast_taskpool_serializer_suspend().

taskpool_shutdown()

static void taskpool_shutdown ( void  )

static

Definition at line 1085 of file taskpool.c.

{
    if (sched) {
        ast_sched_context_destroy(sched);
        sched = NULL;
    }
}

References ast_sched_context_destroy(), and NULL.

Referenced by ast_taskpool_init().

taskpool_sync_task()

static int taskpool_sync_task ( void *  data )

static

Definition at line 615 of file taskpool.c.

{
    struct taskpool_sync_task *sync_task = data;
    int ret;
 
    sync_task->fail = sync_task->task(sync_task->task_data);
 
    /*
     * Once we unlock sync_task->lock after signaling, we cannot access
     * sync_task again.  The thread waiting within ast_taskpool_push_wait()
     * is free to continue and release its local variable (sync_task).
     */
    ast_mutex_lock(&sync_task->lock);
    sync_task->complete = 1;
    ast_cond_signal(&sync_task->cond);
    ret = sync_task->fail;
    ast_mutex_unlock(&sync_task->lock);
    return ret;
}

References ast_cond_signal, ast_mutex_lock, ast_mutex_unlock, taskpool_sync_task::complete, taskpool_sync_task::cond, taskpool_sync_task::fail, taskpool_sync_task::lock, taskpool_sync_task::task, and taskpool_sync_task::task_data.

taskpool_sync_task_cleanup()

static void taskpool_sync_task_cleanup ( struct taskpool_sync_task *  sync_task )

static

Definition at line 606 of file taskpool.c.

{
    ast_mutex_destroy(&sync_task->lock);
    ast_cond_destroy(&sync_task->cond);
}

References ast_cond_destroy, ast_mutex_destroy, taskpool_sync_task::cond, and taskpool_sync_task::lock.

Referenced by __ast_taskpool_push_wait(), and __ast_taskpool_serializer_push_wait().

taskpool_sync_task_init()

static int taskpool_sync_task_init ( struct taskpool_sync_task *  sync_task, int(*)(void *)  task, void *  data  )

static

Definition at line 592 of file taskpool.c.

{
    ast_mutex_init(&sync_task->lock);
    ast_cond_init(&sync_task->cond, NULL);
    sync_task->complete = 0;
    sync_task->fail = 0;
    sync_task->task = task;
    sync_task->task_data = data;
    return 0;
}

References ast_cond_init, ast_mutex_init, taskpool_sync_task::complete, taskpool_sync_task::cond, taskpool_sync_task::fail, taskpool_sync_task::lock, NULL, task(), taskpool_sync_task::task, and taskpool_sync_task::task_data.

Referenced by __ast_taskpool_push_wait(), and __ast_taskpool_serializer_push_wait().

taskpool_taskprocessor_alloc()

static struct taskpool_taskprocessor * taskpool_taskprocessor_alloc ( struct ast_taskpool *  pool, char  type  )

static

Definition at line 152 of file taskpool.c.

{
    struct taskpool_taskprocessor *taskprocessor;
    char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1];
 
    /* We don't actually need locking for each pool taskprocessor, as the only thing
     * mutable is the underlying taskprocessor which has its own internal locking.
     */
    taskprocessor = ao2_alloc_options(sizeof(*taskprocessor), taskpool_taskprocessor_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (!taskprocessor) {
        return NULL;
    }
 
    /* Create name with seq number appended. */
    ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "taskpool/%c:%s", type, pool->name);
 
    taskprocessor->taskprocessor = ast_taskprocessor_get(tps_name, TPS_REF_DEFAULT);
    if (!taskprocessor->taskprocessor) {
        ao2_ref(taskprocessor, -1);
        return NULL;
    }
 
    taskprocessor->last_pushed = ast_tvnow();
 
    if (ast_taskprocessor_push(taskprocessor->taskprocessor, taskpool_taskprocessor_start, ao2_bump(pool))) {
        ao2_ref(pool, -1);
        /* Prevent the taskprocessor from queueing the stop task by explicitly unreferencing and setting it to
         * NULL here.
         */
        ast_taskprocessor_unreference(taskprocessor->taskprocessor);
        taskprocessor->taskprocessor = NULL;
        return NULL;
    }
 
    return taskprocessor;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_bump, ao2_ref, ast_taskprocessor_build_name(), ast_taskprocessor_get(), AST_TASKPROCESSOR_MAX_NAME, ast_taskprocessor_push, ast_taskprocessor_unreference(), ast_tvnow(), ast_taskpool::name, NULL, taskpool_taskprocessor_dtor(), taskpool_taskprocessor_start(), taskpool_taskprocessor::taskprocessor, TPS_REF_DEFAULT, and type.

Referenced by ast_taskpool_create(), and taskpool_dynamic_pool_grow().

taskpool_taskprocessor_dtor()

static void taskpool_taskprocessor_dtor ( void *  obj )

static

Definition at line 116 of file taskpool.c.

{
    struct taskpool_taskprocessor *taskprocessor = obj;
 
    if (taskprocessor->taskprocessor && ast_taskprocessor_push(taskprocessor->taskprocessor, taskpool_taskprocessor_stop, NULL)) {
        /* We can't actually do anything if this fails, so just accept reality */
    }
 
    ast_taskprocessor_unreference(taskprocessor->taskprocessor);
}

References ast_taskprocessor_push, ast_taskprocessor_unreference(), NULL, taskpool_taskprocessor_stop(), and taskpool_taskprocessor::taskprocessor.

Referenced by taskpool_taskprocessor_alloc().

taskpool_taskprocessor_start()

static int taskpool_taskprocessor_start ( void *  data )

static

Definition at line 131 of file taskpool.c.

{
    struct ast_taskpool *pool = data;
 
    /* Set the pool on the thread for this taskprocessor, inheriting the
     * reference passed to the task itself.
     */
    ast_threadstorage_set_ptr(&current_taskpool_pool, pool);
 
    /* If a thread start callback is set on the options, call it */
    if (pool->options.thread_start) {
        pool->options.thread_start();
    }
 
    return 0;
}

References ast_threadstorage_set_ptr(), ast_taskpool::options, and ast_taskpool_options::thread_start.

Referenced by taskpool_taskprocessor_alloc().

taskpool_taskprocessor_stop()

static int taskpool_taskprocessor_stop ( void *  data )

static

Definition at line 101 of file taskpool.c.

 {
    struct ast_taskpool *pool = ast_taskpool_get_current();
 
    /* If a thread stop callback is set on the options, call it */
    if (pool->options.thread_end) {
        pool->options.thread_end();
    }
 
    ao2_cleanup(pool);
 
     return 0;
 }

References ao2_cleanup, ast_taskpool_get_current(), ast_taskpool::options, and ast_taskpool_options::thread_end.

Referenced by taskpool_taskprocessor_dtor().

taskpool_taskprocessors_cleanup()

static void taskpool_taskprocessors_cleanup ( struct taskpool_taskprocessors *  taskprocessors )

static

Definition at line 206 of file taskpool.c.

{
    /* Access/manipulation of taskprocessors is done with the lock held, and
     * with a check of the shutdown flag done. This means that outside of holding
     * the lock we can safely muck with it. Pushing to the taskprocessor is done
     * outside of the lock, but with a reference to the taskprocessor held.
     */
    AST_VECTOR_CALLBACK_VOID(&taskprocessors->taskprocessors, ao2_cleanup);
    AST_VECTOR_FREE(&taskprocessors->taskprocessors);
}

References ao2_cleanup, AST_VECTOR_CALLBACK_VOID, AST_VECTOR_FREE, and taskpool_taskprocessors::taskprocessors.

Referenced by ast_taskpool_shutdown().

taskpool_taskprocessors_init()

static int taskpool_taskprocessors_init ( struct taskpool_taskprocessors *  taskprocessors, unsigned int  size  )

static

Definition at line 193 of file taskpool.c.

{
    if (AST_VECTOR_INIT(&taskprocessors->taskprocessors, size)) {
        return -1;
    }
 
    return 0;
}

References AST_VECTOR_INIT, and taskpool_taskprocessors::taskprocessors.

Referenced by ast_taskpool_create().

Variable Documentation

sched

struct ast_sched_context* sched

static

Scheduler used for dynamic pool shrinking.

Definition at line 83 of file taskpool.c.

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 819 of file taskpool.c.

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

Referenced by ast_taskpool_serializer_group().