taskprocessor.h File Reference

An API for managing task processing threads that can be shared across modules. More...

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

Go to the source code of this file.

Data Structures
struct	ast_taskprocessor_listener_callbacks
struct	ast_taskprocessor_local
	Local data parameter. More...

Macros
#define	AST_TASKPROCESSOR_HIGH_WATER_LEVEL   500
#define	AST_TASKPROCESSOR_MAX_NAME   70
	Suggested maximum taskprocessor name length (less null terminator).
#define	ast_taskprocessor_push(tps, task_exe, datap)    __ast_taskprocessor_push(tps, task_exe, datap, __FILE__, __LINE__, __PRETTY_FUNCTION__)
#define	ast_taskprocessor_push_local(tps, task_exe, datap)    __ast_taskprocessor_push_local(tps, task_exe, datap, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Enumerations
enum	ast_tps_options { TPS_REF_DEFAULT = 0 , TPS_REF_IF_EXISTS = (1 << 0) }
	ast_tps_options for specification of taskprocessor options More...

Functions
int	__ast_taskprocessor_push (struct ast_taskprocessor *tps, int(*task_exe)(void *datap), void *datap, const char *file, int line, const char *function) attribute_warn_unused_result
	Push a task into the specified taskprocessor queue and signal the taskprocessor thread.
int	__ast_taskprocessor_push_local (struct ast_taskprocessor *tps, int(*task_exe)(struct ast_taskprocessor_local *local), void *datap, const char *file, int line, const char *function) attribute_warn_unused_result
	Push a task into the specified taskprocessor queue and signal the taskprocessor thread.
unsigned int	ast_taskprocessor_alert_get (void)
	Get the current taskprocessor high water alert count.
int	ast_taskprocessor_alert_set_levels (struct ast_taskprocessor *tps, long low_water, long high_water)
	Set the high and low alert water marks of the given taskprocessor queue.
void	ast_taskprocessor_build_name (char *buf, unsigned int size, const char *format,...)
	Build a taskprocessor name with a sequence number on the end.
struct ast_taskprocessor *	ast_taskprocessor_create_with_listener (const char *name, struct ast_taskprocessor_listener *listener)
	Create a taskprocessor with a custom listener.
int	ast_taskprocessor_execute (struct ast_taskprocessor *tps)
	Pop a task off the taskprocessor and execute it.
struct ast_taskprocessor *	ast_taskprocessor_get (const char *name, enum ast_tps_options create)
	Get a reference to a taskprocessor with the specified name and create the taskprocessor if necessary.
unsigned int	ast_taskprocessor_get_subsystem_alert (const char *subsystem)
	Get the current taskprocessor high water alert count by subsystem.
int	ast_taskprocessor_is_suspended (struct ast_taskprocessor *tps)
	Get the task processor suspend status.
int	ast_taskprocessor_is_task (struct ast_taskprocessor *tps)
	Am I the given taskprocessor's current task.
struct ast_taskprocessor_listener *	ast_taskprocessor_listener (struct ast_taskprocessor *tps)
	Return the listener associated with the taskprocessor.
struct ast_taskprocessor_listener *	ast_taskprocessor_listener_alloc (const struct ast_taskprocessor_listener_callbacks *callbacks, void *user_data)
	Allocate a taskprocessor listener.
struct ast_taskprocessor *	ast_taskprocessor_listener_get_tps (const struct ast_taskprocessor_listener *listener)
	Get a reference to the listener's taskprocessor.
void *	ast_taskprocessor_listener_get_user_data (const struct ast_taskprocessor_listener *listener)
	Get the user data from the listener.
const char *	ast_taskprocessor_name (struct ast_taskprocessor *tps)
	Return the name of the taskprocessor singleton.
void	ast_taskprocessor_name_append (char *buf, unsigned int size, const char *name)
	Append the next sequence number to the given string, and copy into the buffer.
unsigned int	ast_taskprocessor_seq_num (void)
	Get the next sequence number to create a human friendly taskprocessor name.
void	ast_taskprocessor_set_local (struct ast_taskprocessor *tps, void *local_data)
	Sets the local data associated with a taskprocessor.
long	ast_taskprocessor_size (struct ast_taskprocessor *tps)
	Return the current size of the taskprocessor queue.
int	ast_taskprocessor_suspend (struct ast_taskprocessor *tps)
	Indicate the taskprocessor is suspended.
void *	ast_taskprocessor_unreference (struct ast_taskprocessor *tps)
	Unreference the specified taskprocessor and its reference count will decrement.
int	ast_taskprocessor_unsuspend (struct ast_taskprocessor *tps)
	Indicate the taskprocessor is unsuspended.

Detailed Description

An API for managing task processing threads that can be shared across modules.

Author

Dwayne M. Hubbard dhubb.nosp@m.ard@.nosp@m.digiu.nosp@m.m.co.nosp@m.m

Note

A taskprocessor is a named object containing a task queue that serializes tasks pushed into it by [a] module(s) that reference the taskprocessor. A taskprocessor is created the first time its name is requested via the ast_taskprocessor_get() function or the ast_taskprocessor_create_with_listener() function and destroyed when the taskprocessor reference count reaches zero. A taskprocessor also contains an accompanying listener that is notified when changes in the task queue occur.

A task is a wrapper around a task-handling function pointer and a data pointer. A task is pushed into a taskprocessor queue using the ast_taskprocessor_push(taskprocessor, taskhandler, taskdata) function and freed by the taskprocessor after the task handling function returns. A module releases its reference to a taskprocessor using the ast_taskprocessor_unreference() function which may result in the destruction of the taskprocessor if the taskprocessor's reference count reaches zero. When the taskprocessor's reference count reaches zero, its listener's shutdown() callback will be called. Any further attempts to execute tasks will be denied.

The taskprocessor listener has the flexibility of doling out tasks to best fit the module's needs. For instance, a taskprocessor listener may have a single dispatch thread that handles all tasks, or it may dispatch tasks to a thread pool.

There is a default taskprocessor listener that will be used if a taskprocessor is created without any explicit listener. This default listener runs tasks sequentially in a single thread. The listener will execute tasks as long as there are tasks to be processed. When the taskprocessor is shut down, the default listener will stop processing tasks and join its execution thread.

Definition in file taskprocessor.h.

Macro Definition Documentation

AST_TASKPROCESSOR_HIGH_WATER_LEVEL

#define AST_TASKPROCESSOR_HIGH_WATER_LEVEL   500

Default taskprocessor high water level alert trigger

Definition at line 64 of file taskprocessor.h.

AST_TASKPROCESSOR_MAX_NAME

#define AST_TASKPROCESSOR_MAX_NAME   70

Suggested maximum taskprocessor name length (less null terminator).

Definition at line 61 of file taskprocessor.h.

ast_taskprocessor_push

#define ast_taskprocessor_push	(		tps,
			task_exe,
			datap
	)		__ast_taskprocessor_push(tps, task_exe, datap, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Definition at line 219 of file taskprocessor.h.

                               {
    /*! Local data, associated with the taskprocessor. */
    void *local_data;
    /*! Data pointer passed with this task. */
    void *data;
};
 
/*!
 * \brief Push a task into the specified taskprocessor queue and signal the
 * taskprocessor thread.
 *
 * The callback receives a \ref ast_taskprocessor_local struct, which contains
 * both the provided \a datap pointer, and any local data set on the
 * taskprocessor with ast_taskprocessor_set_local().
 *
 * \param tps The taskprocessor structure
 * \param task_exe The task handling function to push into the taskprocessor queue
 * \param datap The data to be used by the task handling function
 * \retval 0 success
 * \retval -1 failure
 * \since 12.0.0
 */
int __ast_taskprocessor_push_local(struct ast_taskprocessor *tps,
    int (*task_exe)(struct ast_taskprocessor_local *local), void *datap,
    const char *file, int line, const char *function) attribute_warn_unused_result;
#define ast_taskprocessor_push_local(tps, task_exe, datap) \
    __ast_taskprocessor_push_local(tps, task_exe, datap, __FILE__, __LINE__, __PRETTY_FUNCTION__)
 
/*!
 * \brief Indicate the taskprocessor is suspended.
 *
 * \since 13.12.0
 *
 * \param tps Task processor.
 * \retval 0 success
 * \retval -1 failure
 */
int ast_taskprocessor_suspend(struct ast_taskprocessor *tps);
 
/*!
 * \brief Indicate the taskprocessor is unsuspended.
 *
 * \since 13.12.0
 *
 * \param tps Task processor.
 * \retval 0 success
 * \retval -1 failure
 */
int ast_taskprocessor_unsuspend(struct ast_taskprocessor *tps);
 
/*!
 * \brief Get the task processor suspend status
 *
 * \since 13.12.0
 *
 * \param tps Task processor.
 * \retval non-zero if the task processor is suspended
 */
int ast_taskprocessor_is_suspended(struct ast_taskprocessor *tps);
 
/*!
 * \brief Pop a task off the taskprocessor and execute it.
 *
 * \since 12.0.0
 *
 * \param tps The taskprocessor from which to execute.
 * \retval 0 There is no further work to be done.
 * \retval 1 Tasks still remain in the taskprocessor queue.
 */
int ast_taskprocessor_execute(struct ast_taskprocessor *tps);
 
/*!
 * \brief Am I the given taskprocessor's current task.
 * \since 12.7.0
 *
 * \param tps Taskprocessor to check.
 *
 * \retval non-zero if current thread is the taskprocessor thread.
 */
int ast_taskprocessor_is_task(struct ast_taskprocessor *tps);
 
/*!
 * \brief Get the next sequence number to create a human friendly taskprocessor name.
 * \since 13.8.0
 *
 * \return Sequence number for use in creating human friendly taskprocessor names.
 */
unsigned int ast_taskprocessor_seq_num(void);
 
/*!
 * \brief Append the next sequence number to the given string, and copy into the buffer.
 *
 * \param buf Where to copy the appended taskprocessor name.
 * \param size How large is buf including null terminator.
 * \param name A name to append the sequence number to.
 */
void ast_taskprocessor_name_append(char *buf, unsigned int size, const char *name);
 
/*!
 * \brief Build a taskprocessor name with a sequence number on the end.
 * \since 13.8.0
 *
 * \param buf Where to put the built taskprocessor name.
 * \param size How large is buf including null terminator.
 * \param format printf format to create the non-sequenced part of the name.
 *
 * \note The user supplied part of the taskprocessor name is truncated
 * to allow the full sequence number to be appended within the supplied
 * buffer size.
 */
void __attribute__((format(printf, 3, 4))) ast_taskprocessor_build_name(char *buf, unsigned int size, const char *format, ...);
 
/*!
 * \brief Return the name of the taskprocessor singleton
 * \since 1.6.1
 */
const char *ast_taskprocessor_name(struct ast_taskprocessor *tps);
 
/*!
 * \brief Return the current size of the taskprocessor queue
 * \since 13.7.0
 */
long ast_taskprocessor_size(struct ast_taskprocessor *tps);
 
/*!
 * \brief Return the listener associated with the taskprocessor
 */
struct ast_taskprocessor_listener *ast_taskprocessor_listener(struct ast_taskprocessor *tps);
 
/*!
 * \brief Get the current taskprocessor high water alert count.
 * \since 13.10.0
 *
 * \retval 0 if no taskprocessors are in high water alert.
 * \retval non-zero if some task processors are in high water alert.
 */
unsigned int ast_taskprocessor_alert_get(void);
 
 
/*!
 * \brief Get the current taskprocessor high water alert count by subsystem.
 * \since 13.26.0
 * \since 16.3.0
 *
 * \param subsystem The subsystem name
 *
 * \retval 0 if no taskprocessors are in high water alert.
 * \retval non-zero if some task processors are in high water alert.
 */
unsigned int ast_taskprocessor_get_subsystem_alert(const char *subsystem);
 
/*!
 * \brief Set the high and low alert water marks of the given taskprocessor queue.
 * \since 13.10.0
 *
 * \param tps Taskprocessor to update queue water marks.
 * \param low_water New queue low water mark. (-1 to set as 90% of high_water)
 * \param high_water New queue high water mark.
 *
 * \retval 0 on success.
 * \retval -1 on error (water marks not changed).
 */
int ast_taskprocessor_alert_set_levels(struct ast_taskprocessor *tps, long low_water, long high_water);
 
#endif /* __AST_TASKPROCESSOR_H__ */

ast_taskprocessor_push_local

#define ast_taskprocessor_push_local	(		tps,
			task_exe,
			datap
	)		__ast_taskprocessor_push_local(tps, task_exe, datap, __FILE__, __LINE__, __PRETTY_FUNCTION__)

Definition at line 248 of file taskprocessor.h.

Enumeration Type Documentation

ast_tps_options

enum ast_tps_options

ast_tps_options for specification of taskprocessor options

Specify whether a taskprocessor should be created via ast_taskprocessor_get() if the taskprocessor does not already exist. The default behavior is to create a taskprocessor if it does not already exist and provide its reference to the calling function. To only return a reference to a taskprocessor if and only if it exists, use the TPS_REF_IF_EXISTS option in ast_taskprocessor_get().

Enumerator
TPS_REF_DEFAULT	return a reference to a taskprocessor, create one if it does not exist
TPS_REF_IF_EXISTS	return a reference to a taskprocessor ONLY if it already exists

Definition at line 74 of file taskprocessor.h.

                     {
    /*! \brief return a reference to a taskprocessor, create one if it does not exist */
    TPS_REF_DEFAULT = 0,
    /*! \brief return a reference to a taskprocessor ONLY if it already exists */
    TPS_REF_IF_EXISTS = (1 << 0),
};

Function Documentation

__ast_taskprocessor_push()

int __ast_taskprocessor_push	(	struct ast_taskprocessor *	tps,
		int(*)(void *datap)	task_exe,
		void *	datap,
		const char *	file,
		int	line,
		const char *	function
	)

Push a task into the specified taskprocessor queue and signal the taskprocessor thread.

Parameters

tps	The taskprocessor structure
task_exe	The task handling function to push into the taskprocessor queue
datap	The data to be used by the task handling function

Return values

0	success
-1	failure

Since

1.6.1

Definition at line 1371 of file taskprocessor.c.

{
    return taskprocessor_push(tps, tps_task_alloc(task_exe, datap, file, line, function));
}

References taskprocessor_push(), and tps_task_alloc().

Referenced by __ast_sip_push_task(), __ast_taskpool_push(), __ast_threadpool_push(), and ast_taskprocessor_push().

__ast_taskprocessor_push_local()

int __ast_taskprocessor_push_local	(	struct ast_taskprocessor *	tps,
		int(*)(struct ast_taskprocessor_local *local)	task_exe,
		void *	datap,
		const char *	file,
		int	line,
		const char *	function
	)

Push a task into the specified taskprocessor queue and signal the taskprocessor thread.

The callback receives a ast_taskprocessor_local struct, which contains both the provided datap pointer, and any local data set on the taskprocessor with ast_taskprocessor_set_local().

Parameters

tps	The taskprocessor structure
task_exe	The task handling function to push into the taskprocessor queue
datap	The data to be used by the task handling function

Return values

0	success
-1	failure

Since

12.0.0

ast_taskprocessor_alert_get()

unsigned int ast_taskprocessor_alert_get

(

void

)

Get the current taskprocessor high water alert count.

Since

13.10.0

Return values

0	if no taskprocessors are in high water alert.
non-zero	if some task processors are in high water alert.

Definition at line 1004 of file taskprocessor.c.

{
    unsigned int count;
 
    ast_rwlock_rdlock(&tps_alert_lock);
    count = tps_alert_count;
    ast_rwlock_unlock(&tps_alert_lock);
 
    return count;
}

References ast_rwlock_rdlock, ast_rwlock_unlock, tps_alert_count, and tps_alert_lock.

Referenced by AST_TEST_DEFINE(), and distributor().

ast_taskprocessor_alert_set_levels()

int ast_taskprocessor_alert_set_levels	(	struct ast_taskprocessor *	tps,
		long	low_water,
		long	high_water
	)

Set the high and low alert water marks of the given taskprocessor queue.

Since

13.10.0

Parameters

tps	Taskprocessor to update queue water marks.
low_water	New queue low water mark. (-1 to set as 90% of high_water)
high_water	New queue high water mark.

Return values

0	on success.
-1	on error (water marks not changed).

Definition at line 1015 of file taskprocessor.c.

{
    if (!tps || high_water < 0 || high_water < low_water) {
        return -1;
    }
 
    if (low_water < 0) {
        /* Set low water level to 90% of high water level */
        low_water = (high_water * 9) / 10;
    }
 
    ao2_lock(tps);
 
    tps->tps_queue_low = low_water;
    tps->tps_queue_high = high_water;
 
    if (tps->high_water_alert) {
        if (!tps->tps_queue_size || tps->tps_queue_size < low_water) {
            /* Update water mark alert immediately */
            tps->high_water_alert = 0;
            tps_alert_add(tps, -1);
        }
    } else {
        if (high_water < tps->tps_queue_size) {
            /* Update water mark alert immediately */
            tps->high_water_alert = 1;
            tps_alert_add(tps, +1);
        }
    }
 
    ao2_unlock(tps);
 
    return 0;
}

References ao2_lock, ao2_unlock, ast_taskprocessor::high_water_alert, tps_alert_add(), ast_taskprocessor::tps_queue_high, ast_taskprocessor::tps_queue_low, and ast_taskprocessor::tps_queue_size.

Referenced by actual_load_config(), ast_res_pjsip_init_options_handling(), ast_serializer_pool_set_alerts(), ast_sorcery_object_set_congestion_levels(), AST_TEST_DEFINE(), and stasis_subscription_set_congestion_limits().

ast_taskprocessor_build_name()

void ast_taskprocessor_build_name	(	char *	buf,
		unsigned int	size,
		const char *	format,
			...
	)

Build a taskprocessor name with a sequence number on the end.

Since

13.8.0

Parameters

buf	Where to put the built taskprocessor name.
size	How large is buf including null terminator.
format	printf format to create the non-sequenced part of the name.

Note

The user supplied part of the taskprocessor name is truncated to allow the full sequence number to be appended within the supplied buffer size.

Definition at line 1527 of file taskprocessor.c.

{
    va_list ap;
    int user_size;
 
    ast_assert(buf != NULL);
    ast_assert(SEQ_STR_SIZE <= size);
 
    va_start(ap, format);
    user_size = vsnprintf(buf, size - (SEQ_STR_SIZE - 1), format, ap);
    va_end(ap);
    if (user_size < 0) {
        /*
         * Wow!  We got an output error to a memory buffer.
         * Assume no user part of name written.
         */
        user_size = 0;
    } else if (size < user_size + SEQ_STR_SIZE) {
        /* Truncate user part of name to make sequence number fit. */
        user_size = size - SEQ_STR_SIZE;
    }
 
    /* Append sequence number to end of user name. */
    snprintf(buf + user_size, SEQ_STR_SIZE, "-%08x", ast_taskprocessor_seq_num());
}

References ast_assert, ast_taskprocessor_seq_num(), buf, NULL, and SEQ_STR_SIZE.

Referenced by alloc_playback_chan(), allocate_subscription_tree(), ast_sip_session_alloc(), create_websocket_serializer(), distributor_pool_setup(), handle_cli_taskpool_push_serializer_efficiency(), handle_cli_threadpool_push_serializer_efficiency(), internal_stasis_subscribe(), refer_progress_alloc(), sip_options_aor_alloc(), sip_outbound_publisher_alloc(), sip_outbound_registration_state_alloc(), sorcery_object_type_alloc(), and taskpool_taskprocessor_alloc().

ast_taskprocessor_create_with_listener()

struct ast_taskprocessor * ast_taskprocessor_create_with_listener	(	const char *	name,
		struct ast_taskprocessor_listener *	listener
	)

Create a taskprocessor with a custom listener.

Since

12.0.0

Note that when a taskprocessor is created in this way, it does not create any threads to execute the tasks. This job is left up to the listener. The listener's start() callback will be called during this function.

Parameters

name	The name of the taskprocessor to create
listener	The listener for operations on this taskprocessor

Return values

NULL	Failure
non-NULL	success

Definition at line 1274 of file taskprocessor.c.

{
    struct ast_taskprocessor *p;
 
    ao2_lock(tps_singletons);
    p = ao2_find(tps_singletons, name, OBJ_KEY | OBJ_NOLOCK);
    if (p) {
        ao2_unlock(tps_singletons);
        ast_taskprocessor_unreference(p);
        return NULL;
    }
 
    p = __allocate_taskprocessor(name, listener);
    ao2_unlock(tps_singletons);
 
    return __start_taskprocessor(p);
}

References __allocate_taskprocessor(), __start_taskprocessor(), ao2_find, ao2_lock, ao2_unlock, ast_taskprocessor_unreference(), listener(), name, NULL, OBJ_KEY, and OBJ_NOLOCK.

Referenced by ast_taskpool_serializer_group(), AST_TEST_DEFINE(), ast_threadpool_create(), and ast_threadpool_serializer_group().

ast_taskprocessor_execute()

int ast_taskprocessor_execute

(

struct ast_taskprocessor *

tps

)

Pop a task off the taskprocessor and execute it.

Since

12.0.0

Parameters

tps	The taskprocessor from which to execute.

Return values

0	There is no further work to be done.
1	Tasks still remain in the taskprocessor queue.

Definition at line 1420 of file taskprocessor.c.

{
    struct ast_taskprocessor_local local;
    struct tps_task *t;
    long size;
    struct timeval start;
    long elapsed;
    const char *task_file = NULL;
    int task_line = 0;
    const char *task_function = NULL;
 
    ao2_lock(tps);
    t = tps_taskprocessor_pop(tps);
    if (!t) {
        ao2_unlock(tps);
        return 0;
    }
 
    tps->thread = pthread_self();
    tps->executing = 1;
 
    if (t->wants_local) {
        local.local_data = tps->local_data;
        local.data = t->datap;
    }
 
    /* Save task origin info before we free the task */
    task_file = t->file;
    task_line = t->line;
    task_function = t->function;
    ao2_unlock(tps);
 
    start = ast_tvnow();
 
    if (t->wants_local) {
        t->callback.execute_local(&local);
    } else {
        t->callback.execute(t->datap);
    }
    tps_task_free(t);
 
    ao2_lock(tps);
    tps->thread = AST_PTHREADT_NULL;
    /* We need to check size in the same critical section where we reset the
     * executing bit. Avoids a race condition where a task is pushed right
     * after we pop an empty stack.
     */
    tps->executing = 0;
    size = ast_taskprocessor_size(tps);
 
    /* Update the stats */
    ++tps->stats._tasks_processed_count;
 
    /* Include the task we just executed as part of the queue size. */
    if (size >= tps->stats.max_qsize) {
        tps->stats.max_qsize = size + 1;
    }
 
    elapsed = ast_tvdiff_us(ast_tvnow(), start);
    if (elapsed > tps->stats.highest_time_processed) {
        tps->stats.highest_time_processed = elapsed;
        tps->stats.highest_time_task_file = task_file;
        tps->stats.highest_time_task_line = task_line;
        tps->stats.highest_time_task_function = task_function;
    }
    if (elapsed < tps->stats.lowest_time_processed) {
        tps->stats.lowest_time_processed = elapsed;
    }
 
    ao2_unlock(tps);
 
    /* If we executed a task, check for the transition to empty */
    if (size == 0 && tps->listener->callbacks->emptied) {
        tps->listener->callbacks->emptied(tps->listener);
    }
    return size > 0;
}

References tps_taskprocessor_stats::_tasks_processed_count, ao2_lock, ao2_unlock, AST_PTHREADT_NULL, ast_taskprocessor_size(), ast_tvdiff_us(), ast_tvnow(), tps_task::callback, ast_taskprocessor_listener::callbacks, ast_taskprocessor_local::data, tps_task::datap, ast_taskprocessor_listener_callbacks::emptied, tps_task::execute, tps_task::execute_local, ast_taskprocessor::executing, tps_task::file, tps_task::function, tps_taskprocessor_stats::highest_time_processed, tps_taskprocessor_stats::highest_time_task_file, tps_taskprocessor_stats::highest_time_task_function, tps_taskprocessor_stats::highest_time_task_line, tps_task::line, ast_taskprocessor::listener, ast_taskprocessor_local::local_data, ast_taskprocessor::local_data, tps_taskprocessor_stats::lowest_time_processed, tps_taskprocessor_stats::max_qsize, NULL, ast_taskprocessor::stats, ast_taskprocessor::thread, tps_task_free(), tps_taskprocessor_pop(), and tps_task::wants_local.

Referenced by ast_taskpool_serializer_push_wait(), AST_TEST_DEFINE(), default_tps_processing_function(), execute_tasks(), execute_tasks(), and threadpool_execute().

ast_taskprocessor_get()

struct ast_taskprocessor * ast_taskprocessor_get	(	const char *	name,
		enum ast_tps_options	create
	)

Get a reference to a taskprocessor with the specified name and create the taskprocessor if necessary.

The default behavior of instantiating a taskprocessor if one does not already exist can be disabled by specifying the TPS_REF_IF_EXISTS ast_tps_options as the second argument to ast_taskprocessor_get().

Parameters

name	The name of the taskprocessor
create	Use 0 by default or specify TPS_REF_IF_EXISTS to return NULL if the taskprocessor does not already exist return A pointer to a reference counted taskprocessor under normal conditions, or NULL if the TPS_REF_IF_EXISTS reference type is specified and the taskprocessor does not exist

Since

1.6.1

Definition at line 1235 of file taskprocessor.c.

{
    struct ast_taskprocessor *p;
    struct ast_taskprocessor_listener *listener;
    struct default_taskprocessor_listener_pvt *pvt;
 
    if (ast_strlen_zero(name)) {
        ast_log(LOG_ERROR, "Cannot get taskprocessor with empty name!\n");
        return NULL;
    }
    ao2_lock(tps_singletons);
    p = ao2_find(tps_singletons, name, OBJ_KEY | OBJ_NOLOCK);
    if (p || (create & TPS_REF_IF_EXISTS)) {
        /* calling function does not want a new taskprocessor to be created if it doesn't already exist */
        ao2_unlock(tps_singletons);
        return p;
    }
 
    /* Create a new taskprocessor. Start by creating a default listener */
    pvt = default_listener_pvt_alloc();
    if (!pvt) {
        ao2_unlock(tps_singletons);
        return NULL;
    }
    listener = ast_taskprocessor_listener_alloc(&default_listener_callbacks, pvt);
    if (!listener) {
        ao2_unlock(tps_singletons);
        default_listener_pvt_destroy(pvt);
        return NULL;
    }
 
    p = __allocate_taskprocessor(name, listener);
    ao2_unlock(tps_singletons);
    p = __start_taskprocessor(p);
    ao2_ref(listener, -1);
 
    return p;
}

References __allocate_taskprocessor(), __start_taskprocessor(), ao2_find, ao2_lock, ao2_ref, ao2_unlock, ast_log, ast_strlen_zero(), ast_taskprocessor_listener_alloc(), default_listener_callbacks, default_listener_pvt_alloc(), default_listener_pvt_destroy(), listener(), LOG_ERROR, name, NULL, OBJ_KEY, OBJ_NOLOCK, and TPS_REF_IF_EXISTS.

Referenced by alloc_playback_chan(), ast_dns_system_resolver_init(), ast_msg_init(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), cli_tps_ping(), cli_tps_reset_stats(), cli_tps_show_taskprocessor(), find_request_serializer(), internal_stasis_subscribe(), load_module(), load_module(), load_module(), load_objects(), taskpool_taskprocessor_alloc(), and threadpool_alloc().

ast_taskprocessor_get_subsystem_alert()

unsigned int ast_taskprocessor_get_subsystem_alert

(

const char *

subsystem

)

Get the current taskprocessor high water alert count by subsystem.

Since

13.26.0

16.3.0

Parameters

subsystem

The subsystem name

Return values

0	if no taskprocessors are in high water alert.
non-zero	if some task processors are in high water alert.

Definition at line 824 of file taskprocessor.c.

{
    struct subsystem_alert *alert;
    unsigned int count = 0;
    int idx;
 
    AST_VECTOR_RW_RDLOCK(&overloaded_subsystems);
    idx = AST_VECTOR_GET_INDEX(&overloaded_subsystems, subsystem, subsystem_match);
    if (idx >= 0) {
        alert = AST_VECTOR_GET(&overloaded_subsystems, idx);
        count = alert->alert_count;
    }
    AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
 
    return count;
}

References subsystem_alert::alert_count, AST_VECTOR_GET, AST_VECTOR_GET_INDEX, AST_VECTOR_RW_RDLOCK, AST_VECTOR_RW_UNLOCK, subsystem_alert::subsystem, and subsystem_match().

Referenced by AST_TEST_DEFINE(), and distributor().

ast_taskprocessor_is_suspended()

int ast_taskprocessor_is_suspended

(

struct ast_taskprocessor *

tps

)

Get the task processor suspend status.

Since

13.12.0

Parameters

tps	Task processor.

Return values

non-zero

if the task processor is suspended

Definition at line 1415 of file taskprocessor.c.

{
    return tps ? tps->suspended : -1;
}

References ast_taskprocessor::suspended.

Referenced by ast_sip_session_suspend().

ast_taskprocessor_is_task()

int ast_taskprocessor_is_task

(

struct ast_taskprocessor *

tps

)

Am I the given taskprocessor's current task.

Since

12.7.0

Parameters

tps	Taskprocessor to check.

Return values

non-zero

if current thread is the taskprocessor thread.

Definition at line 1498 of file taskprocessor.c.

{
    int is_task;
 
    ao2_lock(tps);
    is_task = pthread_equal(tps->thread, pthread_self());
    ao2_unlock(tps);
    return is_task;
}

References ao2_lock, ao2_unlock, and ast_taskprocessor::thread.

Referenced by __ast_sip_push_task_wait_serializer(), ast_sip_session_suspend(), and handle_new_invite_request().

ast_taskprocessor_listener()

struct ast_taskprocessor_listener * ast_taskprocessor_listener

(

struct ast_taskprocessor *

tps

)

Return the listener associated with the taskprocessor.

Definition at line 1090 of file taskprocessor.c.

{
    return tps ? tps->listener : NULL;
}

References ast_taskprocessor::listener, NULL, and ast_taskprocessor_listener::tps.

ast_taskprocessor_listener_alloc()

struct ast_taskprocessor_listener * ast_taskprocessor_listener_alloc	(	const struct ast_taskprocessor_listener_callbacks *	callbacks,
		void *	user_data
	)

Allocate a taskprocessor listener.

Since

12.0.0

This will result in the listener being allocated with the specified callbacks.

Parameters

callbacks	The callbacks to assign to the listener
user_data	The user data for the listener

Return values

NULL	Failure
non-NULL	The newly allocated taskprocessor listener

Definition at line 1120 of file taskprocessor.c.

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

References ao2_alloc, callbacks, listener(), NULL, taskprocessor_listener_dtor(), and ast_taskprocessor_listener::user_data.

Referenced by ast_taskpool_serializer_group(), ast_taskprocessor_get(), AST_TEST_DEFINE(), ast_threadpool_create(), and ast_threadpool_serializer_group().

ast_taskprocessor_listener_get_tps()

struct ast_taskprocessor * ast_taskprocessor_listener_get_tps

(

const struct ast_taskprocessor_listener *

listener

)

Get a reference to the listener's taskprocessor.

This will return the taskprocessor with its reference count increased. Release the reference to this object by using ast_taskprocessor_unreference()

Parameters

listener

The listener that has the taskprocessor

Returns

The taskprocessor

Definition at line 1134 of file taskprocessor.c.

{
    ao2_ref(listener->tps, +1);
    return listener->tps;
}

References ao2_ref, and listener().

Referenced by serializer_task_pushed(), and serializer_task_pushed().

ast_taskprocessor_listener_get_user_data()

void * ast_taskprocessor_listener_get_user_data

(

const struct ast_taskprocessor_listener *

listener

)

Get the user data from the listener.

Parameters

listener

The taskprocessor listener

Returns

The listener's user data

Definition at line 1140 of file taskprocessor.c.

{
    return listener->user_data;
}

References listener().

Referenced by ast_taskpool_serializer_push_wait(), execute_tasks(), serializer_shutdown(), serializer_shutdown(), serializer_task_pushed(), serializer_task_pushed(), test_emptied(), test_shutdown(), test_task_pushed(), threadpool_tps_emptied(), threadpool_tps_shutdown(), and threadpool_tps_task_pushed().

ast_taskprocessor_name()

const char * ast_taskprocessor_name

(

struct ast_taskprocessor *

tps

)

Return the name of the taskprocessor singleton.

Since

1.6.1

Definition at line 1096 of file taskprocessor.c.

{
    if (!tps) {
        ast_log(LOG_ERROR, "no taskprocessor specified!\n");
        return NULL;
    }
    return tps->name;
}

References ast_log, LOG_ERROR, ast_taskprocessor::name, NULL, and ast_taskprocessor_listener::tps.

Referenced by ast_serializer_pool_set_alerts(), ast_sip_get_distributor_serializer(), distributor(), grow(), record_serializer(), shrink(), sip_options_apply_aor_configuration(), and sip_options_contact_add_task().

ast_taskprocessor_name_append()

void ast_taskprocessor_name_append	(	char *	buf,
		unsigned int	size,
		const char *	name
	)

Append the next sequence number to the given string, and copy into the buffer.

Parameters

buf	Where to copy the appended taskprocessor name.
size	How large is buf including null terminator.
name	A name to append the sequence number to.

Definition at line 1517 of file taskprocessor.c.

{
    int final_size = strlen(name) + SEQ_STR_SIZE;
 
    ast_assert(buf != NULL && name != NULL);
    ast_assert(final_size <= size);
 
    snprintf(buf, final_size, "%s-%08x", name, ast_taskprocessor_seq_num());
}

References ast_assert, ast_taskprocessor_seq_num(), buf, name, NULL, and SEQ_STR_SIZE.

Referenced by ast_serializer_pool_create(), and ast_serializer_taskpool_create().

ast_taskprocessor_seq_num()

unsigned int ast_taskprocessor_seq_num

(

void

)

Get the next sequence number to create a human friendly taskprocessor name.

Since

13.8.0

Returns

Sequence number for use in creating human friendly taskprocessor names.

Definition at line 1508 of file taskprocessor.c.

{
    static int seq_num;
 
    return (unsigned int) ast_atomic_fetchadd_int(&seq_num, +1);
}

References ast_atomic_fetchadd_int().

Referenced by ast_taskprocessor_build_name(), and ast_taskprocessor_name_append().

ast_taskprocessor_set_local()

void ast_taskprocessor_set_local	(	struct ast_taskprocessor *	tps,
		void *	local_data
	)

Sets the local data associated with a taskprocessor.

Since

12.0.0

See ast_taskprocessor_push_local().

Parameters

tps	Task processor.
local_data	Local data to associate with tps.

Definition at line 1292 of file taskprocessor.c.

{
    SCOPED_AO2LOCK(lock, tps);
    tps->local_data = local_data;
}

References ast_taskprocessor::local_data, lock, and SCOPED_AO2LOCK.

Referenced by AST_TEST_DEFINE(), and internal_stasis_subscribe().

ast_taskprocessor_size()

long ast_taskprocessor_size

(

struct ast_taskprocessor *

tps

)

Return the current size of the taskprocessor queue.

Since

13.7.0

Definition at line 1085 of file taskprocessor.c.

{
    return (tps) ? tps->tps_queue_size : -1;
}

References ast_taskprocessor::tps_queue_size.

Referenced by ast_serializer_pool_get(), ast_taskpool_serializer_push_wait(), ast_taskprocessor_execute(), AST_TEST_DEFINE(), ast_threadpool_queue_size(), execute_tasks(), taskpool_least_full_selector(), and taskpool_sequential_selector().

ast_taskprocessor_suspend()

int ast_taskprocessor_suspend

(

struct ast_taskprocessor *

tps

)

Indicate the taskprocessor is suspended.

Since

13.12.0

Parameters

tps	Task processor.

Return values

0	success
-1	failure

Definition at line 1393 of file taskprocessor.c.

{
    if (tps) {
        ao2_lock(tps);
        tps->suspended = 1;
        ao2_unlock(tps);
        return 0;
    }
    return -1;
}

References ao2_lock, ao2_unlock, and ast_taskprocessor::suspended.

Referenced by ast_sip_session_suspend(), and AST_TEST_DEFINE().

ast_taskprocessor_unreference()

void * ast_taskprocessor_unreference

(

struct ast_taskprocessor *

tps

)

Unreference the specified taskprocessor and its reference count will decrement.

Taskprocessors use astobj2 and will unlink from the taskprocessor singleton container and destroy themself when the taskprocessor reference count reaches zero.

Parameters

tps	taskprocessor to unreference

Returns

NULL

Since

1.6.1

Definition at line 1300 of file taskprocessor.c.

{
    if (!tps) {
        return NULL;
    }
 
    /* To prevent another thread from finding and getting a reference to this
     * taskprocessor we hold the singletons lock. If we didn't do this then
     * they may acquire it and find that the listener has been shut down.
     */
    ao2_lock(tps_singletons);
 
    if (ao2_ref(tps, -1) > 3) {
        ao2_unlock(tps_singletons);
        return NULL;
    }
 
    /* If we're down to 3 references, then those must be:
     * 1. The reference we just got rid of
     * 2. The container
     * 3. The listener
     */
    ao2_unlink_flags(tps_singletons, tps, OBJ_NOLOCK);
    ao2_unlock(tps_singletons);
 
    listener_shutdown(tps->listener);
    return NULL;
}

References ao2_lock, ao2_ref, ao2_unlink_flags, ao2_unlock, ast_taskprocessor::listener, listener_shutdown(), NULL, and OBJ_NOLOCK.

Referenced by __start_taskprocessor(), __unload_module(), ast_msg_shutdown(), ast_res_pjsip_cleanup_options_handling(), ast_serializer_pool_destroy(), ast_taskprocessor_create_with_listener(), 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(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), ast_threadpool_shutdown(), cli_tps_ping(), cli_tps_reset_stats(), cli_tps_reset_stats_all(), cli_tps_show_taskprocessor(), destroy_conference_bridge(), distributor(), distributor_pool_shutdown(), dns_system_resolver_destroy(), execute_tasks(), execute_tasks(), exten_state_subscription_destructor(), handle_cli_taskpool_push_serializer_efficiency(), handle_cli_threadpool_push_serializer_efficiency(), refer_progress_destroy(), scheduler(), schtd_dtor(), serializer_task_pushed(), serializer_task_pushed(), session_destructor(), sip_options_aor_dtor(), sip_outbound_publisher_destroy(), sip_outbound_registration_client_state_destroy(), sorcery_object_type_destructor(), subscription_dtor(), subscription_persistence_recreate(), subscription_tree_destructor(), taskpool_taskprocessor_alloc(), taskpool_taskprocessor_dtor(), tps_report_taskprocessor_list(), tps_shutdown_thread(), tps_taskprocessor_tab_complete(), unload_module(), unload_module(), unload_module(), unload_module(), unload_module(), unload_module(), and websocket_cb().

ast_taskprocessor_unsuspend()

int ast_taskprocessor_unsuspend

(

struct ast_taskprocessor *

tps

)

Indicate the taskprocessor is unsuspended.

Since

13.12.0

Parameters

tps	Task processor.

Return values

0	success
-1	failure

Definition at line 1404 of file taskprocessor.c.

{
    if (tps) {
        ao2_lock(tps);
        tps->suspended = 0;
        ao2_unlock(tps);
        return 0;
    }
    return -1;
}

References ao2_lock, ao2_unlock, and ast_taskprocessor::suspended.

Referenced by ast_sip_session_unsuspend(), and AST_TEST_DEFINE().