taskprocessor.c File Reference

Maintain a container of uniquely-named taskprocessor threads that can be shared across modules. More...

#include "asterisk.h"
#include "asterisk/_private.h"
#include "asterisk/module.h"
#include "asterisk/time.h"
#include "asterisk/astobj2.h"
#include "asterisk/cli.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/sem.h"

Include dependency graph for taskprocessor.c:

Go to the source code of this file.

Data Structures
struct	ast_taskprocessor
	A ast_taskprocessor structure is a singleton by name. More...
struct	ast_taskprocessor_listener
	A listener for taskprocessors. More...
struct	default_taskprocessor_listener_pvt
struct	subsystem_alert
struct	ast_taskprocessor::tps_queue
	Taskprocessor queue. More...
struct	tps_task
	tps_task structure is queued to a taskprocessor More...
struct	tps_taskprocessor_stats
	tps_taskprocessor_stats maintain statistics for a taskprocessor. More...

Macros
#define	ast_taskprocessor_push_internal(tps, task_exe, datap)    __ast_taskprocessor_push(tps, task_exe, datap, __FILE__, __LINE__, __PRETTY_FUNCTION__)
#define	ast_taskprocessor_push_local_internal(tps, task_exe, datap)    __ast_taskprocessor_push_local(tps, task_exe, datap, __FILE__, __LINE__, __PRETTY_FUNCTION__)
#define	AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT   10
	How many seconds to wait for running taskprocessors to finish on shutdown.
#define	FMT_FIELDS   "%-70s %10lu %10lu %10lu %10lu %10lu %10ld %10ld\n"
#define	FMT_FIELDS_SUBSYSTEM   "%-32s %12u\n"
#define	FMT_HEADERS   "%-70s %10s %10s %10s %10s %10s %10s %10s\n"
#define	FMT_HEADERS_SUBSYSTEM   "%-32s %12s\n"
#define	SEQ_STR_SIZE   (1 + 8 + 1) /* Dash plus 8 hex digits plus null terminator */
#define	TPS_MAX_BUCKETS   1567

Functions
static struct ast_taskprocessor *	__allocate_taskprocessor (const char *name, struct ast_taskprocessor_listener *listener)
int	__ast_taskprocessor_push (struct ast_taskprocessor *tps, int(*task_exe)(void *datap), void *datap, const char *file, int line, const char *function)
	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 *datap), void *datap, const char *file, int line, const char *function)
static struct ast_taskprocessor *	__start_taskprocessor (struct ast_taskprocessor *p)
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.
int	ast_taskprocessor_push (struct ast_taskprocessor *tps, int(*task_exe)(void *datap), void *datap)
int	ast_taskprocessor_push_local (struct ast_taskprocessor *tps, int(*task_exe)(struct ast_taskprocessor_local *datap), void *datap)
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.
int	ast_tps_init (void)
static	AST_VECTOR_RW (subsystem_alert_vector, struct subsystem_alert *)
static char *	cli_subsystem_alert_report (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	cli_tps_ping (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	cli_tps_report (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	cli_tps_reset_stats (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	cli_tps_reset_stats_all (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	cli_tps_show_taskprocessor (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static int	default_listener_die (void *data)
static void *	default_listener_pvt_alloc (void)
static void	default_listener_pvt_destroy (struct default_taskprocessor_listener_pvt *pvt)
static void	default_listener_pvt_dtor (struct ast_taskprocessor_listener *listener)
static void	default_listener_shutdown (struct ast_taskprocessor_listener *listener)
static int	default_listener_start (struct ast_taskprocessor_listener *listener)
static void	default_task_pushed (struct ast_taskprocessor_listener *listener, int was_empty)
static void *	default_tps_processing_function (void *data)
	Function that processes tasks in the taskprocessor.
static void	listener_shutdown (struct ast_taskprocessor_listener *listener)
static void	subsystem_alert_decrement (const char *subsystem)
static void	subsystem_alert_increment (const char *subsystem)
static int	subsystem_cmp (struct subsystem_alert *a, struct subsystem_alert *b)
static void	subsystem_copy (struct subsystem_alert *alert, struct subsystem_alert_vector *vector)
static int	subsystem_match (struct subsystem_alert *alert, const char *subsystem)
static void	taskprocessor_listener_dtor (void *obj)
static int	taskprocessor_push (struct ast_taskprocessor *tps, struct tps_task *t)
static void	tps_alert_add (struct ast_taskprocessor *tps, int delta)
static int	tps_cmp_cb (void *obj, void *arg, int flags)
static int	tps_hash_cb (const void *obj, const int flags)
static int	tps_ping_handler (void *datap)
static int	tps_report_taskprocessor_list (int fd, const char *like)
static void	tps_report_taskprocessor_list_helper (int fd, struct ast_taskprocessor *tps)
static void	tps_reset_stats (struct ast_taskprocessor *tps)
static void	tps_shutdown (void)
static int	tps_sort_cb (const void *obj_left, const void *obj_right, int flags)
static struct tps_task *	tps_task_alloc (int(*task_exe)(void *datap), void *datap, const char *file, int line, const char *function)
static struct tps_task *	tps_task_alloc_local (int(*task_exe)(struct ast_taskprocessor_local *local), void *datap, const char *file, int line, const char *function)
static void *	tps_task_free (struct tps_task *task)
static void	tps_taskprocessor_dtor (void *tps)
static struct tps_task *	tps_taskprocessor_pop (struct ast_taskprocessor *tps)
static char *	tps_taskprocessor_tab_complete (struct ast_cli_args *a)

Variables
static const struct ast_taskprocessor_listener_callbacks	default_listener_callbacks
static unsigned int	tps_alert_count
static ast_rwlock_t	tps_alert_lock = AST_RWLOCK_INIT_VALUE

Detailed Description

Maintain a container of uniquely-named taskprocessor threads that can be shared across modules.

Author

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

Definition in file taskprocessor.c.

Macro Definition Documentation

ast_taskprocessor_push_internal

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

Definition at line 210 of file taskprocessor.c.

ast_taskprocessor_push_local_internal

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

Definition at line 215 of file taskprocessor.c.

AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT

#define AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT   10

How many seconds to wait for running taskprocessors to finish on shutdown.

Definition at line 318 of file taskprocessor.c.

FMT_FIELDS

#define FMT_FIELDS   "%-70s %10lu %10lu %10lu %10lu %10lu %10ld %10ld\n"

Definition at line 615 of file taskprocessor.c.

FMT_FIELDS_SUBSYSTEM

#define FMT_FIELDS_SUBSYSTEM   "%-32s %12u\n"

FMT_HEADERS

#define FMT_HEADERS   "%-70s %10s %10s %10s %10s %10s %10s %10s\n"

Definition at line 614 of file taskprocessor.c.

FMT_HEADERS_SUBSYSTEM

#define FMT_HEADERS_SUBSYSTEM   "%-32s %12s\n"

SEQ_STR_SIZE

#define SEQ_STR_SIZE   (1 + 8 + 1) /* Dash plus 8 hex digits plus null terminator */

Definition at line 1515 of file taskprocessor.c.

TPS_MAX_BUCKETS

#define TPS_MAX_BUCKETS   1567

Function Documentation

__allocate_taskprocessor()

static struct ast_taskprocessor * __allocate_taskprocessor ( const char *  name, struct ast_taskprocessor_listener *  listener  )

static

Definition at line 1174 of file taskprocessor.c.

{
    struct ast_taskprocessor *p;
    char *subsystem_separator;
    size_t subsystem_length = 0;
    size_t name_length;
 
    name_length = strlen(name);
    subsystem_separator = strchr(name, '/');
    if (subsystem_separator) {
        subsystem_length = subsystem_separator - name;
    }
 
    p = ao2_alloc(sizeof(*p) + name_length + subsystem_length + 2, tps_taskprocessor_dtor);
    if (!p) {
        ast_log(LOG_WARNING, "failed to create taskprocessor '%s'\n", name);
        return NULL;
    }
 
    /* Set default congestion water level alert triggers. */
    p->tps_queue_low = (AST_TASKPROCESSOR_HIGH_WATER_LEVEL * 9) / 10;
    p->tps_queue_high = AST_TASKPROCESSOR_HIGH_WATER_LEVEL;
 
    strcpy(p->name, name); /* Safe */
    p->subsystem = p->name + name_length + 1;
    ast_copy_string(p->subsystem, name, subsystem_length + 1);
 
    ao2_ref(listener, +1);
    p->listener = listener;
 
    p->thread = AST_PTHREADT_NULL;
 
    ao2_ref(p, +1);
    listener->tps = p;
 
    if (!(ao2_link_flags(tps_singletons, p, OBJ_NOLOCK))) {
        ast_log(LOG_ERROR, "Failed to add taskprocessor '%s' to container\n", p->name);
        listener->tps = NULL;
        ao2_ref(p, -2);
        return NULL;
    }
 
    return p;
}

References ao2_alloc, ao2_link_flags, ao2_ref, ast_copy_string(), ast_log, AST_PTHREADT_NULL, AST_TASKPROCESSOR_HIGH_WATER_LEVEL, listener(), ast_taskprocessor::listener, LOG_ERROR, LOG_WARNING, ast_taskprocessor::name, name, NULL, OBJ_NOLOCK, ast_taskprocessor::subsystem, ast_taskprocessor::thread, ast_taskprocessor::tps_queue_high, ast_taskprocessor::tps_queue_low, and tps_taskprocessor_dtor().

Referenced by ast_taskprocessor_create_with_listener(), and ast_taskprocessor_get().

__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 *datap)	task_exe,
		void *	datap,
		const char *	file,
		int	line,
		const char *	function
	)

Definition at line 1382 of file taskprocessor.c.

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

References taskprocessor_push(), and tps_task_alloc_local().

Referenced by ast_taskprocessor_push_local().

__start_taskprocessor()

static struct ast_taskprocessor * __start_taskprocessor ( struct ast_taskprocessor *  p )

static

Definition at line 1219 of file taskprocessor.c.

{
    if (p && p->listener->callbacks->start(p->listener)) {
        ast_log(LOG_ERROR, "Failed to start taskprocessor listener for '%s'\n",
            p->name);
        ast_taskprocessor_unreference(p);
 
        return NULL;
    }
 
    return p;
}

References ast_log, ast_taskprocessor_unreference(), ast_taskprocessor_listener::callbacks, ast_taskprocessor::listener, LOG_ERROR, ast_taskprocessor::name, NULL, and ast_taskprocessor_listener_callbacks::start.

Referenced by ast_taskprocessor_create_with_listener(), and ast_taskprocessor_get().

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_push()

int ast_taskprocessor_push	(	struct ast_taskprocessor *	tps,
		int(*)(void *datap)	task_exe,
		void *	datap
	)

Definition at line 1377 of file taskprocessor.c.

{
    return __ast_taskprocessor_push(tps, task_exe, datap, NULL, 0, NULL);
}

References __ast_taskprocessor_push(), and NULL.

ast_taskprocessor_push_local()

int ast_taskprocessor_push_local	(	struct ast_taskprocessor *	tps,
		int(*)(struct ast_taskprocessor_local *datap)	task_exe,
		void *	datap
	)

Definition at line 1388 of file taskprocessor.c.

{
    return __ast_taskprocessor_push_local(tps, task_exe, datap, NULL, 0, NULL);
}

References __ast_taskprocessor_push_local(), and NULL.

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().

ast_tps_init()

int ast_tps_init

(

void

)

Provided by taskprocessor.c

Definition at line 397 of file taskprocessor.c.

{
    tps_singletons = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
        TPS_MAX_BUCKETS, tps_hash_cb, NULL, tps_cmp_cb);
    if (!tps_singletons) {
        ast_log(LOG_ERROR, "Failed to initialize taskprocessor container!\n");
        return -1;
    }
 
    if (AST_VECTOR_RW_INIT(&overloaded_subsystems, 10)) {
        ao2_ref(tps_singletons, -1);
        ast_log(LOG_ERROR, "Failed to initialize taskprocessor subsystems tracking vector!\n");
        return -1;
    }
 
    ast_cond_init(&cli_ping_cond, NULL);
 
    ast_cli_register_multiple(taskprocessor_clis, ARRAY_LEN(taskprocessor_clis));
 
    ast_register_cleanup(tps_shutdown);
 
    return 0;
}

References AO2_ALLOC_OPT_LOCK_MUTEX, ao2_container_alloc_hash, ao2_ref, ARRAY_LEN, ast_cli_register_multiple, ast_cond_init, ast_log, ast_register_cleanup(), AST_VECTOR_RW_INIT, LOG_ERROR, NULL, tps_cmp_cb(), tps_hash_cb(), TPS_MAX_BUCKETS, and tps_shutdown().

Referenced by asterisk_daemon().

AST_VECTOR_RW()

static AST_VECTOR_RW ( subsystem_alert_vector  , struct subsystem_alert *    )

static

Definition at line 141 of file taskprocessor.c.

                                                   {
    AST_CLI_DEFINE(cli_tps_ping, "Ping a named task processor"),
    AST_CLI_DEFINE(cli_tps_report, "List instantiated task processors and statistics"),
    AST_CLI_DEFINE(cli_tps_show_taskprocessor, "Display detailed info about a taskprocessor"),
    AST_CLI_DEFINE(cli_subsystem_alert_report, "List task processor subsystems in alert"),
    AST_CLI_DEFINE(cli_tps_reset_stats, "Reset a named task processor's stats"),
    AST_CLI_DEFINE(cli_tps_reset_stats_all, "Reset all task processors' stats"),
};

cli_subsystem_alert_report()

static char * cli_subsystem_alert_report ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 916 of file taskprocessor.c.

{
    struct subsystem_alert_vector sorted_subsystems;
    int i;
 
#define FMT_HEADERS_SUBSYSTEM       "%-32s %12s\n"
#define FMT_FIELDS_SUBSYSTEM        "%-32s %12u\n"
 
    switch (cmd) {
    case CLI_INIT:
        e->command = "core show taskprocessor alerted subsystems";
        e->usage =
            "Usage: core show taskprocessor alerted subsystems\n"
            "   Shows a list of task processor subsystems that are currently alerted\n";
        return NULL;
    case CLI_GENERATE:
        return NULL;
    }
 
    if (a->argc != e->args) {
        return CLI_SHOWUSAGE;
    }
 
    if (AST_VECTOR_INIT(&sorted_subsystems, AST_VECTOR_SIZE(&overloaded_subsystems))) {
        return CLI_FAILURE;
    }
 
    AST_VECTOR_RW_RDLOCK(&overloaded_subsystems);
    for (i = 0; i < AST_VECTOR_SIZE(&overloaded_subsystems); i++) {
        subsystem_copy(AST_VECTOR_GET(&overloaded_subsystems, i), &sorted_subsystems);
    }
    AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
 
    ast_cli(a->fd, "\n" FMT_HEADERS_SUBSYSTEM, "Subsystem", "Alert Count");
 
    for (i = 0; i < AST_VECTOR_SIZE(&sorted_subsystems); i++) {
        struct subsystem_alert *alert = AST_VECTOR_GET(&sorted_subsystems, i);
        ast_cli(a->fd, FMT_FIELDS_SUBSYSTEM, alert->subsystem, alert->alert_count);
    }
 
    ast_cli(a->fd, "\n%zu subsystems\n\n", AST_VECTOR_SIZE(&sorted_subsystems));
 
    AST_VECTOR_CALLBACK_VOID(&sorted_subsystems, ast_free);
    AST_VECTOR_FREE(&sorted_subsystems);
 
    return CLI_SUCCESS;
}

References a, subsystem_alert::alert_count, ast_cli_entry::args, ast_cli(), ast_free, AST_VECTOR_CALLBACK_VOID, AST_VECTOR_FREE, AST_VECTOR_GET, AST_VECTOR_INIT, AST_VECTOR_RW_RDLOCK, AST_VECTOR_RW_UNLOCK, AST_VECTOR_SIZE, CLI_FAILURE, CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, FMT_FIELDS_SUBSYSTEM, FMT_HEADERS_SUBSYSTEM, NULL, subsystem_alert::subsystem, subsystem_copy(), and ast_cli_entry::usage.

cli_tps_ping()

static char * cli_tps_ping ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 515 of file taskprocessor.c.

{
    struct timeval begin, end, delta;
    const char *name;
    struct timeval when;
    struct timespec ts;
    struct ast_taskprocessor *tps;
 
    switch (cmd) {
    case CLI_INIT:
        e->command = "core ping taskprocessor";
        e->usage =
            "Usage: core ping taskprocessor <taskprocessor>\n"
            "   Displays the time required for a task to be processed\n";
        return NULL;
    case CLI_GENERATE:
        if (a->pos == 3) {
            return tps_taskprocessor_tab_complete(a);
        } else {
            return NULL;
        }
    }
 
    if (a->argc != 4)
        return CLI_SHOWUSAGE;
 
    name = a->argv[3];
    if (!(tps = ast_taskprocessor_get(name, TPS_REF_IF_EXISTS))) {
        ast_cli(a->fd, "\nping failed: %s not found\n\n", name);
        return CLI_SUCCESS;
    }
    ast_cli(a->fd, "\npinging %s ...", name);
 
    /*
     * Wait up to 5 seconds for a ping reply.
     *
     * On a very busy system it could take awhile to get a
     * ping response from some taskprocessors.
     */
    begin = ast_tvnow();
    when = ast_tvadd(begin, ast_samp2tv(5000, 1000));
    ts.tv_sec = when.tv_sec;
    ts.tv_nsec = when.tv_usec * 1000;
 
    ast_mutex_lock(&cli_ping_cond_lock);
    if (ast_taskprocessor_push_internal(tps, tps_ping_handler, 0) < 0) {
        ast_mutex_unlock(&cli_ping_cond_lock);
        ast_cli(a->fd, "\nping failed: could not push task to %s\n\n", name);
        ast_taskprocessor_unreference(tps);
        return CLI_FAILURE;
    }
    ast_cond_timedwait(&cli_ping_cond, &cli_ping_cond_lock, &ts);
    ast_mutex_unlock(&cli_ping_cond_lock);
 
    end = ast_tvnow();
    delta = ast_tvsub(end, begin);
    ast_cli(a->fd, "\n\t%24s ping time: %.1ld.%.6ld sec\n\n", name, (long)delta.tv_sec, (long int)delta.tv_usec);
    ast_taskprocessor_unreference(tps);
    return CLI_SUCCESS;
}

References a, ast_cli(), ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_samp2tv(), ast_taskprocessor_get(), ast_taskprocessor_push_internal, ast_taskprocessor_unreference(), ast_tvadd(), ast_tvnow(), ast_tvsub(), CLI_FAILURE, CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, end, name, NULL, tps_ping_handler(), TPS_REF_IF_EXISTS, tps_taskprocessor_tab_complete(), and ast_cli_entry::usage.

cli_tps_report()

static char * cli_tps_report ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 681 of file taskprocessor.c.

{
    const char *like;
 
    switch (cmd) {
    case CLI_INIT:
        e->command = "core show taskprocessors [like]";
        e->usage =
            "Usage: core show taskprocessors [like keyword]\n"
            "   Shows a list of instantiated task processors and their statistics\n";
        return NULL;
    case CLI_GENERATE:
        if (a->pos == e->args) {
            return tps_taskprocessor_tab_complete(a);
        } else {
            return NULL;
        }
    }
 
    if (a->argc == e->args - 1) {
        like = "";
    } else if (a->argc == e->args + 1 && !strcasecmp(a->argv[e->args-1], "like")) {
        like = a->argv[e->args];
    } else {
        return CLI_SHOWUSAGE;
    }
 
    ast_cli(a->fd, "\n" FMT_HEADERS, "Processor", "Processed", "In Queue", "Max Depth", "Low water", "High water", "Low time(us)", "High time(us)");
    ast_cli(a->fd, "\n%d taskprocessors\n\n", tps_report_taskprocessor_list(a->fd, like));
 
    return CLI_SUCCESS;
}

References a, ast_cli_entry::args, ast_cli(), CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, FMT_HEADERS, NULL, tps_report_taskprocessor_list(), tps_taskprocessor_tab_complete(), and ast_cli_entry::usage.

cli_tps_reset_stats()

static char * cli_tps_reset_stats ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 1566 of file taskprocessor.c.

{
    const char *name;
    struct ast_taskprocessor *tps;
 
    switch (cmd) {
    case CLI_INIT:
        e->command = "core reset taskprocessor";
        e->usage =
            "Usage: core reset taskprocessor <taskprocessor>\n"
            "    Resets stats for the specified taskprocessor\n";
        return NULL;
    case CLI_GENERATE:
        return tps_taskprocessor_tab_complete(a);
    }
 
    if (a->argc != 4) {
        return CLI_SHOWUSAGE;
    }
 
    name = a->argv[3];
    if (!(tps = ast_taskprocessor_get(name, TPS_REF_IF_EXISTS))) {
        ast_cli(a->fd, "\nReset failed: %s not found\n\n", name);
        return CLI_SUCCESS;
    }
    ast_cli(a->fd, "\nResetting %s\n\n", name);
 
    tps_reset_stats(tps);
 
    ast_taskprocessor_unreference(tps);
 
    return CLI_SUCCESS;
}

References a, ast_cli(), ast_taskprocessor_get(), ast_taskprocessor_unreference(), CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, name, NULL, TPS_REF_IF_EXISTS, tps_reset_stats(), tps_taskprocessor_tab_complete(), and ast_cli_entry::usage.

cli_tps_reset_stats_all()

static char * cli_tps_reset_stats_all ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 1600 of file taskprocessor.c.

{
    struct ast_taskprocessor *tps;
    struct ao2_iterator iter;
 
    switch (cmd) {
    case CLI_INIT:
        e->command = "core reset taskprocessors";
        e->usage =
            "Usage: core reset taskprocessors\n"
            "    Resets stats for all taskprocessors\n";
        return NULL;
    case CLI_GENERATE:
        return NULL;
    }
 
    if (a->argc != e->args) {
        return CLI_SHOWUSAGE;
    }
 
    ast_cli(a->fd, "\nResetting stats for all taskprocessors\n\n");
 
    iter = ao2_iterator_init(tps_singletons, 0);
    while ((tps = ao2_iterator_next(&iter))) {
        tps_reset_stats(tps);
        ast_taskprocessor_unreference(tps);
    }
    ao2_iterator_destroy(&iter);
 
    return CLI_SUCCESS;
}

References a, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ast_cli_entry::args, ast_cli(), ast_taskprocessor_unreference(), CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, NULL, tps_reset_stats(), and ast_cli_entry::usage.

cli_tps_show_taskprocessor()

static char * cli_tps_show_taskprocessor ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 714 of file taskprocessor.c.

{
    const char *name;
    struct ast_taskprocessor *tps;
    struct tps_task *task;
    int task_count = 0;
 
    switch (cmd) {
    case CLI_INIT:
        e->command = "core show taskprocessor name";
        e->usage =
            "Usage: core show taskprocessor name <taskprocessor>\n"
            "    Displays detailed information about a specific taskprocessor,\n"
            "    including all queued tasks and their origins (DEVMODE only).\n";
        return NULL;
    case CLI_GENERATE:
        if (a->pos == 4) {
            return tps_taskprocessor_tab_complete(a);
        }
        return NULL;
    }
 
    if (a->argc != 5) {
        return CLI_SHOWUSAGE;
    }
 
    name = a->argv[4];
    tps = ast_taskprocessor_get(name, TPS_REF_IF_EXISTS);
    if (!tps) {
        ast_cli(a->fd, "\nTaskprocessor '%s' not found\n\n", name);
        return CLI_SUCCESS;
    }
 
    ao2_lock(tps);
 
    ast_cli(a->fd, "\nTaskprocessor: %s\n", tps->name);
    ast_cli(a->fd, "===========================================\n");
    ast_cli(a->fd, "Subsystem:            %s\n", tps->subsystem[0] ? tps->subsystem : "(none)");
    ast_cli(a->fd, "Tasks processed:      %lu\n", tps->stats._tasks_processed_count);
    ast_cli(a->fd, "Current queue size:   %ld\n", tps->tps_queue_size);
    ast_cli(a->fd, "Max queue depth:      %lu\n", tps->stats.max_qsize);
    ast_cli(a->fd, "Low water mark:       %ld\n", tps->tps_queue_low);
    ast_cli(a->fd, "High water mark:      %ld\n", tps->tps_queue_high);
    ast_cli(a->fd, "High water alert:     %s\n", tps->high_water_alert ? "Yes" : "No");
    ast_cli(a->fd, "Suspended:            %s\n", tps->suspended ? "Yes" : "No");
    ast_cli(a->fd, "Currently executing:  %s\n", tps->executing ? "Yes" : "No");
    ast_cli(a->fd, "Highest time (us):    %ld\n", tps->stats.highest_time_processed);
    if (tps->stats.highest_time_task_file) {
        ast_cli(a->fd, "  Highest task origin: %s:%d (%s)\n",
            tps->stats.highest_time_task_file,
            tps->stats.highest_time_task_line,
            tps->stats.highest_time_task_function);
    }
    ast_cli(a->fd, "Lowest time (us):     %ld\n", tps->stats.lowest_time_processed);
 
    if (tps->tps_queue_size > 0) {
        ast_cli(a->fd, "\nQueued Tasks:\n");
        ast_cli(a->fd, "-------------------------------------------\n");
 
        AST_LIST_TRAVERSE(&tps->tps_queue, task, list) {
            task_count++;
            if (task->file) {
                ast_cli(a->fd, "  Task #%d:\n", task_count);
                ast_cli(a->fd, "    Origin:   %s:%d\n", task->file, task->line);
                ast_cli(a->fd, "    Function: %s\n", task->function);
                ast_cli(a->fd, "    Type:     %s\n", task->wants_local ? "Local" : "Standard");
            } else {
                ast_cli(a->fd, "  Task #%d: (origin not available)\n", task_count);
            }
        }
        ast_cli(a->fd, "\nTotal queued tasks: %d\n", task_count);
    } else {
        ast_cli(a->fd, "\nNo tasks currently queued.\n");
    }
 
    ao2_unlock(tps);
    ast_taskprocessor_unreference(tps);
 
    ast_cli(a->fd, "\n");
    return CLI_SUCCESS;
}

References tps_taskprocessor_stats::_tasks_processed_count, a, ao2_lock, ao2_unlock, ast_cli(), AST_LIST_TRAVERSE, ast_taskprocessor_get(), ast_taskprocessor_unreference(), CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, ast_taskprocessor::executing, ast_taskprocessor::high_water_alert, 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::list, tps_taskprocessor_stats::lowest_time_processed, tps_taskprocessor_stats::max_qsize, ast_taskprocessor::name, name, NULL, ast_taskprocessor::stats, ast_taskprocessor::subsystem, ast_taskprocessor::suspended, task(), task_count, ast_taskprocessor::tps_queue, ast_taskprocessor::tps_queue_high, ast_taskprocessor::tps_queue_low, ast_taskprocessor::tps_queue_size, TPS_REF_IF_EXISTS, tps_taskprocessor_tab_complete(), and ast_cli_entry::usage.

default_listener_die()

static int default_listener_die ( void *  data )

static

Definition at line 273 of file taskprocessor.c.

{
    struct default_taskprocessor_listener_pvt *pvt = data;
    pvt->dead = 1;
    return 0;
}

References default_taskprocessor_listener_pvt::dead.

Referenced by default_listener_shutdown().

default_listener_pvt_alloc()

static void * default_listener_pvt_alloc ( void  )

static

Definition at line 1145 of file taskprocessor.c.

{
    struct default_taskprocessor_listener_pvt *pvt;
 
    pvt = ast_calloc(1, sizeof(*pvt));
    if (!pvt) {
        ast_log(LOG_ERROR, "Failed to allocate memory for taskprocessor listener\n");
        return NULL;
    }
    pvt->poll_thread = AST_PTHREADT_NULL;
    if (ast_sem_init(&pvt->sem, 0, 0) != 0) {
        ast_log(LOG_ERROR, "Failed to initialize taskprocessor semaphore: %s\n", strerror(errno));
        ast_free(pvt);
        return NULL;
    }
    return pvt;
}

References ast_calloc, ast_free, ast_log, AST_PTHREADT_NULL, ast_sem_init(), errno, LOG_ERROR, NULL, default_taskprocessor_listener_pvt::poll_thread, and default_taskprocessor_listener_pvt::sem.

Referenced by ast_taskprocessor_get().

default_listener_pvt_destroy()

static void default_listener_pvt_destroy ( struct default_taskprocessor_listener_pvt *  pvt )

static

Definition at line 192 of file taskprocessor.c.

{
    ast_assert(pvt->dead);
    ast_sem_destroy(&pvt->sem);
    ast_free(pvt);
}

References ast_assert, ast_free, ast_sem_destroy(), default_taskprocessor_listener_pvt::dead, and default_taskprocessor_listener_pvt::sem.

Referenced by ast_taskprocessor_get(), and default_listener_pvt_dtor().

default_listener_pvt_dtor()

static void default_listener_pvt_dtor ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 199 of file taskprocessor.c.

{
    struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
 
    default_listener_pvt_destroy(pvt);
 
    listener->user_data = NULL;
}

References default_listener_pvt_destroy(), listener(), and NULL.

default_listener_shutdown()

static void default_listener_shutdown ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 280 of file taskprocessor.c.

{
    struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
    int res;
 
    /* Hold a reference during shutdown */
    ao2_t_ref(listener->tps, +1, "tps-shutdown");
 
    if (ast_taskprocessor_push_internal(listener->tps, default_listener_die, pvt)) {
        /* This will cause the thread to exit early without completing tasks already
         * in the queue.  This is probably the least bad option in this situation. */
        default_listener_die(pvt);
    }
 
    ast_assert(pvt->poll_thread != AST_PTHREADT_NULL);
 
    if (pthread_equal(pthread_self(), pvt->poll_thread)) {
        res = pthread_detach(pvt->poll_thread);
        if (res != 0) {
            ast_log(LOG_ERROR, "pthread_detach(): %s\n", strerror(errno));
        }
    } else {
        res = pthread_join(pvt->poll_thread, NULL);
        if (res != 0) {
            ast_log(LOG_ERROR, "pthread_join(): %s\n", strerror(errno));
        }
    }
    pvt->poll_thread = AST_PTHREADT_NULL;
}

References ao2_t_ref, ast_assert, ast_log, AST_PTHREADT_NULL, ast_taskprocessor_push_internal, default_listener_die(), errno, listener(), LOG_ERROR, NULL, and default_taskprocessor_listener_pvt::poll_thread.

default_listener_start()

static int default_listener_start ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 252 of file taskprocessor.c.

{
    struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
 
    if (ast_pthread_create(&pvt->poll_thread, NULL, default_tps_processing_function, listener)) {
        return -1;
    }
 
    return 0;
}

References ast_pthread_create, default_tps_processing_function(), listener(), NULL, and default_taskprocessor_listener_pvt::poll_thread.

default_task_pushed()

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

static

Definition at line 263 of file taskprocessor.c.

{
    struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
 
    if (ast_sem_post(&pvt->sem) != 0) {
        ast_log(LOG_ERROR, "Taskprocessor '%s': Failed to signal task enqueue: %s\n",
            listener->tps->name, strerror(errno));
    }
}

References ast_log, ast_sem_post(), errno, listener(), LOG_ERROR, and default_taskprocessor_listener_pvt::sem.

default_tps_processing_function()

static void * default_tps_processing_function ( void *  data )

static

Function that processes tasks in the taskprocessor.

Definition at line 223 of file taskprocessor.c.

{
    struct ast_taskprocessor_listener *listener = data;
    struct ast_taskprocessor *tps = listener->tps;
    struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
    int sem_value;
    int res;
 
    while (!pvt->dead) {
        res = ast_sem_wait(&pvt->sem);
        if (res != 0 && errno != EINTR) {
            ast_log(LOG_ERROR, "Taskprocessor '%s': Semaphore wait failed: %s\n",
                tps->name, strerror(errno));
            /* Just give up */
            break;
        }
        ast_taskprocessor_execute(tps);
    }
 
    /* No posting to a dead taskprocessor! */
    res = ast_sem_getvalue(&pvt->sem, &sem_value);
    ast_assert(res == 0 && sem_value == 0);
 
    /* Free the shutdown reference (see default_listener_shutdown) */
    ao2_t_ref(listener->tps, -1, "tps-shutdown");
 
    return NULL;
}

References ao2_t_ref, ast_assert, ast_log, ast_sem_getvalue(), ast_sem_wait(), ast_taskprocessor_execute(), default_taskprocessor_listener_pvt::dead, errno, listener(), LOG_ERROR, ast_taskprocessor::name, NULL, and default_taskprocessor_listener_pvt::sem.

Referenced by default_listener_start().

listener_shutdown()

static void listener_shutdown ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 1105 of file taskprocessor.c.

{
    listener->callbacks->shutdown(listener);
    ao2_ref(listener->tps, -1);
}

References ao2_ref, and listener().

Referenced by ast_taskprocessor_unreference().

subsystem_alert_decrement()

static void subsystem_alert_decrement ( const char *  subsystem )

static

Definition at line 873 of file taskprocessor.c.

{
    struct subsystem_alert *alert;
    int idx;
 
    if (ast_strlen_zero(subsystem)) {
        return;
    }
 
    AST_VECTOR_RW_WRLOCK(&overloaded_subsystems);
    idx = AST_VECTOR_GET_INDEX(&overloaded_subsystems, subsystem, subsystem_match);
    if (idx < 0) {
        ast_log(LOG_ERROR,
            "Can't decrement alert count for subsystem '%s' as it wasn't in alert\n", subsystem);
        AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
        return;
    }
    alert = AST_VECTOR_GET(&overloaded_subsystems, idx);
 
    alert->alert_count--;
    if (alert->alert_count <= 0) {
        AST_VECTOR_REMOVE(&overloaded_subsystems, idx, 0);
        ast_free(alert);
    }
 
    AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
}

References subsystem_alert::alert_count, ast_free, ast_log, ast_strlen_zero(), AST_VECTOR_GET, AST_VECTOR_GET_INDEX, AST_VECTOR_REMOVE, AST_VECTOR_RW_UNLOCK, AST_VECTOR_RW_WRLOCK, LOG_ERROR, subsystem_alert::subsystem, and subsystem_match().

Referenced by tps_alert_add().

subsystem_alert_increment()

static void subsystem_alert_increment ( const char *  subsystem )

static

Definition at line 841 of file taskprocessor.c.

{
    struct subsystem_alert *alert;
    int idx;
 
    if (ast_strlen_zero(subsystem)) {
        return;
    }
 
    AST_VECTOR_RW_WRLOCK(&overloaded_subsystems);
    idx = AST_VECTOR_GET_INDEX(&overloaded_subsystems, subsystem, subsystem_match);
    if (idx >= 0) {
        alert = AST_VECTOR_GET(&overloaded_subsystems, idx);
        alert->alert_count++;
        AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
        return;
    }
 
    alert = ast_malloc(sizeof(*alert) + strlen(subsystem) + 1);
    if (!alert) {
        AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
        return;
    }
    alert->alert_count = 1;
    strcpy(alert->subsystem, subsystem); /* Safe */
 
    if (AST_VECTOR_APPEND(&overloaded_subsystems, alert)) {
        ast_free(alert);
    }
    AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
}

References subsystem_alert::alert_count, ast_free, ast_malloc, ast_strlen_zero(), AST_VECTOR_APPEND, AST_VECTOR_GET, AST_VECTOR_GET_INDEX, AST_VECTOR_RW_UNLOCK, AST_VECTOR_RW_WRLOCK, subsystem_alert::subsystem, and subsystem_match().

Referenced by tps_alert_add().

subsystem_cmp()

static int subsystem_cmp ( struct subsystem_alert *  a, struct subsystem_alert *  b  )

static

Definition at line 819 of file taskprocessor.c.

{
    return strcmp(a->subsystem, b->subsystem);
}

References a, and b.

Referenced by subsystem_copy().

subsystem_copy()

static void subsystem_copy ( struct subsystem_alert *  alert, struct subsystem_alert_vector *  vector  )

static

Definition at line 901 of file taskprocessor.c.

{
    struct subsystem_alert *alert_copy;
    alert_copy = ast_malloc(sizeof(*alert_copy) + strlen(alert->subsystem) + 1);
    if (!alert_copy) {
        return;
    }
    alert_copy->alert_count = alert->alert_count;
    strcpy(alert_copy->subsystem, alert->subsystem); /* Safe */
    if (AST_VECTOR_ADD_SORTED(vector, alert_copy, subsystem_cmp)) {
        ast_free(alert_copy);
    }
}

References subsystem_alert::alert_count, ast_free, ast_malloc, AST_VECTOR_ADD_SORTED, subsystem_alert::subsystem, and subsystem_cmp().

Referenced by cli_subsystem_alert_report().

subsystem_match()

static int subsystem_match ( struct subsystem_alert *  alert, const char *  subsystem  )

static

Definition at line 814 of file taskprocessor.c.

{
    return !strcmp(alert->subsystem, subsystem);
}

References ast_taskprocessor::subsystem, and subsystem_alert::subsystem.

Referenced by ast_taskprocessor_get_subsystem_alert(), subsystem_alert_decrement(), and subsystem_alert_increment().

taskprocessor_listener_dtor()

static void taskprocessor_listener_dtor ( void *  obj )

static

Definition at line 1111 of file taskprocessor.c.

{
    struct ast_taskprocessor_listener *listener = obj;
 
    if (listener->callbacks->dtor) {
        listener->callbacks->dtor(listener);
    }
}

References ast_taskprocessor_listener::callbacks, ast_taskprocessor_listener_callbacks::dtor, and listener().

Referenced by ast_taskprocessor_listener_alloc().

taskprocessor_push()

static int taskprocessor_push ( struct ast_taskprocessor *  tps, struct tps_task *  t  )

static

Definition at line 1330 of file taskprocessor.c.

{
    int previous_size;
    int was_empty;
 
    if (!tps) {
        ast_log(LOG_ERROR, "Taskprocessor is NULL!\n");
        return -1;
    }
 
    if (!t) {
        ast_log(LOG_ERROR, "Task is NULL!\n");
        return -1;
    }
 
    if (t->file) {
        ast_debug(3, "Taskprocessor '%s': Task pushed from %s:%d (%s)\n",
            tps->name, t->file, t->line, t->function);
    }
 
    ao2_lock(tps);
    AST_LIST_INSERT_TAIL(&tps->tps_queue, t, list);
    previous_size = tps->tps_queue_size++;
 
    if (tps->tps_queue_high <= tps->tps_queue_size) {
        if (!tps->high_water_alert) {
            ast_log(LOG_WARNING, "Taskprocessor '%s' queue reached %ld scheduled tasks (high water mark: %ld)%s.\n",
                tps->name, tps->tps_queue_size, tps->tps_queue_high, tps->high_water_warned ? " again" : "");
            tps->high_water_warned = 1;
            tps->high_water_alert = 1;
            tps_alert_add(tps, +1);
        }
    }
 
    /* The currently executing task counts as still in queue */
    was_empty = tps->executing ? 0 : previous_size == 0;
    ao2_unlock(tps);
    tps->listener->callbacks->task_pushed(tps->listener, was_empty);
    return 0;
}

References ao2_lock, ao2_unlock, ast_debug, AST_LIST_INSERT_TAIL, ast_log, ast_taskprocessor_listener::callbacks, ast_taskprocessor::executing, tps_task::file, tps_task::function, ast_taskprocessor::high_water_alert, ast_taskprocessor::high_water_warned, tps_task::line, ast_taskprocessor::listener, LOG_ERROR, LOG_WARNING, ast_taskprocessor::name, ast_taskprocessor_listener_callbacks::task_pushed, tps_alert_add(), ast_taskprocessor::tps_queue, ast_taskprocessor::tps_queue_high, and ast_taskprocessor::tps_queue_size.

Referenced by __ast_taskprocessor_push(), and __ast_taskprocessor_push_local().

tps_alert_add()

static void tps_alert_add ( struct ast_taskprocessor *  tps, int  delta  )

static

Definition at line 979 of file taskprocessor.c.

{
    unsigned int old;
 
    ast_rwlock_wrlock(&tps_alert_lock);
    old = tps_alert_count;
    tps_alert_count += delta;
    if (DEBUG_ATLEAST(3)
        /* and tps_alert_count becomes zero or non-zero */
        && !old != !tps_alert_count) {
        ast_log(LOG_DEBUG, "Taskprocessor '%s' %s the high water alert (total alerts: %u).\n",
            tps->name, tps_alert_count ? "triggered" : "cleared", tps_alert_count);
    }
 
    if (tps->subsystem[0] != '\0') {
        if (delta > 0) {
            subsystem_alert_increment(tps->subsystem);
        } else {
            subsystem_alert_decrement(tps->subsystem);
        }
    }
 
    ast_rwlock_unlock(&tps_alert_lock);
}

References ast_log, ast_rwlock_unlock, ast_rwlock_wrlock, DEBUG_ATLEAST, LOG_DEBUG, ast_taskprocessor::name, ast_taskprocessor::subsystem, subsystem_alert_decrement(), subsystem_alert_increment(), tps_alert_count, and tps_alert_lock.

Referenced by ast_taskprocessor_alert_set_levels(), taskprocessor_push(), tps_taskprocessor_dtor(), and tps_taskprocessor_pop().

tps_cmp_cb()

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

static

Definition at line 806 of file taskprocessor.c.

{
    struct ast_taskprocessor *lhs = obj, *rhs = arg;
    const char *rhsname = flags & OBJ_KEY ? arg : rhs->name;
 
    return !strcasecmp(lhs->name, rhsname) ? CMP_MATCH | CMP_STOP : 0;
}

References CMP_MATCH, CMP_STOP, ast_taskprocessor::name, and OBJ_KEY.

Referenced by ast_tps_init().

tps_hash_cb()

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

static

Definition at line 797 of file taskprocessor.c.

{
    const struct ast_taskprocessor *tps = obj;
    const char *name = flags & OBJ_KEY ? obj : tps->name;
 
    return ast_str_case_hash(name);
}

References ast_str_case_hash(), ast_taskprocessor::name, name, and OBJ_KEY.

Referenced by ast_tps_init().

tps_ping_handler()

static int tps_ping_handler ( void *  datap )

static

Definition at line 506 of file taskprocessor.c.

{
    ast_mutex_lock(&cli_ping_cond_lock);
    ast_cond_signal(&cli_ping_cond);
    ast_mutex_unlock(&cli_ping_cond_lock);
    return 0;
}

References ast_cond_signal, ast_mutex_lock, and ast_mutex_unlock.

Referenced by cli_tps_ping().

tps_report_taskprocessor_list()

static int tps_report_taskprocessor_list ( int  fd, const char *  like  )

static

Definition at line 644 of file taskprocessor.c.

{
    int tps_count = 0;
    int word_len;
    struct ao2_container *sorted_tps;
    struct ast_taskprocessor *tps;
    struct ao2_iterator iter;
 
    sorted_tps = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, tps_sort_cb,
        NULL);
    if (!sorted_tps
        || ao2_container_dup(sorted_tps, tps_singletons, 0)) {
        ast_debug(1, "Failed to retrieve sorted taskprocessors\n");
        ao2_cleanup(sorted_tps);
        return 0;
    }
 
    word_len = strlen(like);
    iter = ao2_iterator_init(sorted_tps, AO2_ITERATOR_UNLINK);
    while ((tps = ao2_iterator_next(&iter))) {
        if (like) {
            if (!strncasecmp(like, tps->name, word_len)) {
                tps_report_taskprocessor_list_helper(fd, tps);
                tps_count++;
            }
        } else {
            tps_report_taskprocessor_list_helper(fd, tps);
            tps_count++;
        }
        ast_taskprocessor_unreference(tps);
    }
    ao2_iterator_destroy(&iter);
    ao2_ref(sorted_tps, -1);
 
    return tps_count;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_cleanup, ao2_container_alloc_rbtree, ao2_container_dup(), ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, AO2_ITERATOR_UNLINK, ao2_ref, ast_debug, ast_taskprocessor_unreference(), ast_taskprocessor::name, NULL, tps_report_taskprocessor_list_helper(), and tps_sort_cb().

Referenced by cli_tps_report().

tps_report_taskprocessor_list_helper()

static void tps_report_taskprocessor_list_helper ( int  fd, struct ast_taskprocessor *  tps  )

static

Definition at line 625 of file taskprocessor.c.

{
    ast_cli(fd, FMT_FIELDS, tps->name, tps->stats._tasks_processed_count,
        tps->tps_queue_size, tps->stats.max_qsize, tps->tps_queue_low,
        tps->tps_queue_high, tps->stats.lowest_time_processed,
        tps->stats.highest_time_processed);
}

References tps_taskprocessor_stats::_tasks_processed_count, ast_cli(), FMT_FIELDS, tps_taskprocessor_stats::highest_time_processed, tps_taskprocessor_stats::lowest_time_processed, tps_taskprocessor_stats::max_qsize, ast_taskprocessor::name, ast_taskprocessor::stats, ast_taskprocessor::tps_queue_high, ast_taskprocessor::tps_queue_low, and ast_taskprocessor::tps_queue_size.

Referenced by tps_report_taskprocessor_list().

tps_reset_stats()

static void tps_reset_stats ( struct ast_taskprocessor *  tps )

static

Definition at line 1553 of file taskprocessor.c.

{
    ao2_lock(tps);
    tps->stats._tasks_processed_count = 0;
    tps->stats.max_qsize = 0;
    tps->stats.highest_time_processed = 0;
    tps->stats.lowest_time_processed = 0;
    tps->stats.highest_time_task_file = NULL;
    tps->stats.highest_time_task_line = 0;
    tps->stats.highest_time_task_function = NULL;
    ao2_unlock(tps);
}

References tps_taskprocessor_stats::_tasks_processed_count, ao2_lock, ao2_unlock, 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_taskprocessor_stats::lowest_time_processed, tps_taskprocessor_stats::max_qsize, NULL, and ast_taskprocessor::stats.

Referenced by cli_tps_reset_stats(), and cli_tps_reset_stats_all().

tps_shutdown()

static void tps_shutdown ( void  )

static

Definition at line 324 of file taskprocessor.c.

{
    int objcount;
    int tries;
    struct ao2_container *sorted_tps;
    struct ast_taskprocessor *tps;
    struct ao2_iterator iter;
    struct timespec delay = {1, 0};
 
    /* During shutdown there may still be taskprocessor threads running and those
     * tasprocessors reference tps_singletons.  When those taskprocessors finish
     * they will call ast_taskprocessor_unreference, creating a race condition which
     * can result in tps_singletons being referenced after being deleted. To try and
     * avoid this we check the container count and if greater than zero, give the
     * running taskprocessors a chance to finish */
    objcount = ao2_container_count(tps_singletons);
    if (objcount > 0) {
        ast_log(LOG_DEBUG,
            "Taskprocessor shutdown: Waiting for %d taskprocessor(s) to complete.\n",
            objcount);
 
        /* give the running taskprocessors a chance to finish, up to
         * AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT seconds */
        for (tries = 0; tries < AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT; tries++) {
            while (nanosleep(&delay, &delay));
            objcount = ao2_container_count(tps_singletons);
            /* if count is 0, we are done waiting */
            if (objcount == 0) {
                break;
            }
            delay.tv_sec = 1;
            delay.tv_nsec = 0;
            ast_log(LOG_DEBUG,
                "Taskprocessor shutdown: Still waiting for %d taskprocessor(s) after %d second(s).\n",
                objcount, tries + 1);
        }
    }
 
    /* rather than try forever, risk an assertion on shutdown.  This probably indicates
     * a taskprocessor was not cleaned up somewhere */
    if (objcount > 0) {
        ast_log(LOG_ERROR,
            "Taskprocessor shutdown: %d taskprocessor(s) still running after %d seconds. Assertion may occur:\n",
            objcount, AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT);
 
        sorted_tps = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, tps_sort_cb,
            NULL);
        if (!sorted_tps || ao2_container_dup(sorted_tps, tps_singletons, 0)) {
            ast_log(LOG_ERROR, "Unable to get sorted list of taskprocessors for shutdown report\n");
        }
        else {
            iter = ao2_iterator_init(sorted_tps, AO2_ITERATOR_UNLINK);
            while ((tps = ao2_iterator_next(&iter))) {
                ast_log(LOG_ERROR, "  - Taskprocessor '%s' (queue size: %ld)\n",
                    tps->name, tps->tps_queue_size);
            }
        }
 
        ao2_cleanup(sorted_tps);
    }
    else {
        ast_log(LOG_DEBUG,
                "Taskprocessor shutdown: All taskprocessors completed successfully.\n");
    }
 
    ast_cli_unregister_multiple(taskprocessor_clis, ARRAY_LEN(taskprocessor_clis));
    AST_VECTOR_CALLBACK_VOID(&overloaded_subsystems, ast_free);
    AST_VECTOR_RW_FREE(&overloaded_subsystems);
    ao2_t_ref(tps_singletons, -1, "Unref tps_singletons in shutdown");
    tps_singletons = NULL;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_cleanup, ao2_container_alloc_rbtree, ao2_container_count(), ao2_container_dup(), ao2_iterator_init(), ao2_iterator_next, AO2_ITERATOR_UNLINK, ao2_t_ref, ARRAY_LEN, ast_cli_unregister_multiple(), ast_free, ast_log, AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT, AST_VECTOR_CALLBACK_VOID, AST_VECTOR_RW_FREE, LOG_DEBUG, LOG_ERROR, ast_taskprocessor::name, NULL, ast_taskprocessor_listener_callbacks::start, ast_taskprocessor::tps_queue_size, and tps_sort_cb().

Referenced by ast_tps_init().

tps_sort_cb()

static int tps_sort_cb ( const void *  obj_left, const void *  obj_right, int  flags  )

static

Definition at line 592 of file taskprocessor.c.

{
    const struct ast_taskprocessor *tps_left = obj_left;
    const struct ast_taskprocessor *tps_right = obj_right;
    const char *right_key = obj_right;
    int cmp;
 
    switch (flags & OBJ_SEARCH_MASK) {
    default:
    case OBJ_SEARCH_OBJECT:
        right_key = tps_right->name;
        /* Fall through */
    case OBJ_SEARCH_KEY:
        cmp = strcasecmp(tps_left->name, right_key);
        break;
    case OBJ_SEARCH_PARTIAL_KEY:
        cmp = strncasecmp(tps_left->name, right_key, strlen(right_key));
        break;
    }
    return cmp;
}

References ast_taskprocessor::name, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, and OBJ_SEARCH_PARTIAL_KEY.

Referenced by tps_report_taskprocessor_list(), and tps_shutdown().

tps_task_alloc()

static struct tps_task * tps_task_alloc ( int(*)(void *datap)  task_exe, void *  datap, const char *  file, int  line, const char *  function  )

static

Definition at line 422 of file taskprocessor.c.

{
    struct tps_task *t;
    if (!task_exe) {
        ast_log(LOG_ERROR, "Task callback function is NULL!\n");
        return NULL;
    }
 
    t = ast_calloc(1, sizeof(*t));
    if (!t) {
        ast_log(LOG_ERROR, "Failed to allocate memory for task!\n");
        return NULL;
    }
 
    t->callback.execute = task_exe;
    t->datap = datap;
    t->file = file;
    t->line = line;
    t->function = function;
 
    return t;
}

References ast_calloc, ast_log, tps_task::callback, tps_task::datap, tps_task::execute, tps_task::file, tps_task::function, tps_task::line, LOG_ERROR, and NULL.

Referenced by __ast_taskprocessor_push().

tps_task_alloc_local()

static struct tps_task * tps_task_alloc_local ( int(*)(struct ast_taskprocessor_local *local)  task_exe, void *  datap, const char *  file, int  line, const char *  function  )

static

Definition at line 446 of file taskprocessor.c.

{
    struct tps_task *t;
    if (!task_exe) {
        ast_log(LOG_ERROR, "Task callback function is NULL!\n");
        return NULL;
    }
 
    t = ast_calloc(1, sizeof(*t));
    if (!t) {
        ast_log(LOG_ERROR, "Failed to allocate memory for task!\n");
        return NULL;
    }
 
    t->callback.execute_local = task_exe;
    t->datap = datap;
    t->wants_local = 1;
    t->file = file;
    t->line = line;
    t->function = function;
 
    return t;
}

References ast_calloc, ast_log, tps_task::callback, tps_task::datap, tps_task::execute_local, tps_task::file, tps_task::function, tps_task::line, LOG_ERROR, NULL, and tps_task::wants_local.

Referenced by __ast_taskprocessor_push_local().

tps_task_free()

static void * tps_task_free ( struct tps_task *  task )

static

Definition at line 472 of file taskprocessor.c.

{
    ast_free(task);
    return NULL;
}

References ast_free, NULL, and task().

Referenced by ast_taskprocessor_execute(), and tps_taskprocessor_dtor().

tps_taskprocessor_dtor()

static void tps_taskprocessor_dtor ( void *  tps )

static

Definition at line 1051 of file taskprocessor.c.

{
    struct ast_taskprocessor *t = tps;
    struct tps_task *task;
 
    while ((task = AST_LIST_REMOVE_HEAD(&t->tps_queue, list))) {
        tps_task_free(task);
    }
    t->tps_queue_size = 0;
 
    if (t->high_water_alert) {
        t->high_water_alert = 0;
        tps_alert_add(t, -1);
    }
 
    ao2_cleanup(t->listener);
    t->listener = NULL;
}

References ao2_cleanup, AST_LIST_REMOVE_HEAD, ast_taskprocessor::high_water_alert, tps_task::list, ast_taskprocessor::listener, NULL, task(), tps_alert_add(), ast_taskprocessor::tps_queue, ast_taskprocessor::tps_queue_size, and tps_task_free().

Referenced by __allocate_taskprocessor().

tps_taskprocessor_pop()

static struct tps_task * tps_taskprocessor_pop ( struct ast_taskprocessor *  tps )

static

Definition at line 1071 of file taskprocessor.c.

{
    struct tps_task *task;
 
    if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) {
        --tps->tps_queue_size;
        if (tps->high_water_alert && tps->tps_queue_size <= tps->tps_queue_low) {
            tps->high_water_alert = 0;
            tps_alert_add(tps, -1);
        }
    }
    return task;
}

References AST_LIST_REMOVE_HEAD, ast_taskprocessor::high_water_alert, tps_task::list, task(), tps_alert_add(), ast_taskprocessor::tps_queue, ast_taskprocessor::tps_queue_low, and ast_taskprocessor::tps_queue_size.

Referenced by ast_taskprocessor_execute().

tps_taskprocessor_tab_complete()

static char * tps_taskprocessor_tab_complete ( struct ast_cli_args *  a )

static

Definition at line 483 of file taskprocessor.c.

{
    int tklen;
    struct ast_taskprocessor *p;
    struct ao2_iterator i;
 
    tklen = strlen(a->word);
    i = ao2_iterator_init(tps_singletons, 0);
    while ((p = ao2_iterator_next(&i))) {
        if (!strncasecmp(a->word, p->name, tklen)) {
            if (ast_cli_completion_add(ast_strdup(p->name))) {
                ast_taskprocessor_unreference(p);
                break;
            }
        }
        ast_taskprocessor_unreference(p);
    }
    ao2_iterator_destroy(&i);
 
    return NULL;
}

References a, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ast_cli_completion_add(), ast_strdup, ast_taskprocessor_unreference(), ast_taskprocessor::name, and NULL.

Referenced by cli_tps_ping(), cli_tps_report(), cli_tps_reset_stats(), and cli_tps_show_taskprocessor().

Variable Documentation

default_listener_callbacks

const struct ast_taskprocessor_listener_callbacks default_listener_callbacks

static

Definition at line 310 of file taskprocessor.c.

                                                                                      {
    .start = default_listener_start,
    .task_pushed = default_task_pushed,
    .shutdown = default_listener_shutdown,
    .dtor = default_listener_pvt_dtor,
};

Referenced by ast_taskprocessor_get().

tps_alert_count

unsigned int tps_alert_count

static

Count of the number of taskprocessors in high water alert.

Definition at line 966 of file taskprocessor.c.

Referenced by ast_taskprocessor_alert_get(), and tps_alert_add().

tps_alert_lock

ast_rwlock_t tps_alert_lock = AST_RWLOCK_INIT_VALUE

static

Access protection for tps_alert_count

Definition at line 969 of file taskprocessor.c.

Referenced by ast_taskprocessor_alert_get(), and tps_alert_add().