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_SHUTDOWN_MAX_WAIT   10
	How many seconds to wait for running taskprocessors to finish on shutdown. More...
#define	FMT_FIELDS   "%-70s %10lu %10lu %10lu %10lu %10lu\n"
#define	FMT_FIELDS_SUBSYSTEM   "%-32s %12u\n"
#define	FMT_HEADERS   "%-70s %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)
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. More...
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. More...
void	ast_taskprocessor_build_name (char *buf, unsigned int size, const char *format,...)
	Build a taskprocessor name with a sequence number on the end. More...
struct ast_taskprocessor *	ast_taskprocessor_create_with_listener (const char *name, struct ast_taskprocessor_listener *listener)
	Create a taskprocessor with a custom listener. More...
int	ast_taskprocessor_execute (struct ast_taskprocessor *tps)
	Pop a task off the taskprocessor and execute it. More...
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. More...
unsigned int	ast_taskprocessor_get_subsystem_alert (const char *subsystem)
	Get the current taskprocessor high water alert count by subsystem. More...
int	ast_taskprocessor_is_suspended (struct ast_taskprocessor *tps)
	Get the task processor suspend status. More...
int	ast_taskprocessor_is_task (struct ast_taskprocessor *tps)
	Am I the given taskprocessor's current task. More...
struct ast_taskprocessor_listener *	ast_taskprocessor_listener (struct ast_taskprocessor *tps)
	Return the listener associated with the taskprocessor. More...
struct ast_taskprocessor_listener *	ast_taskprocessor_listener_alloc (const struct ast_taskprocessor_listener_callbacks *callbacks, void *user_data)
	Allocate a taskprocessor listener. More...
struct ast_taskprocessor *	ast_taskprocessor_listener_get_tps (const struct ast_taskprocessor_listener *listener)
	Get a reference to the listener's taskprocessor. More...
void *	ast_taskprocessor_listener_get_user_data (const struct ast_taskprocessor_listener *listener)
	Get the user data from the listener. More...
const char *	ast_taskprocessor_name (struct ast_taskprocessor *tps)
	Return the name of the taskprocessor singleton. More...
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. More...
int	ast_taskprocessor_push (struct ast_taskprocessor *tps, int(*task_exe)(void *datap), void *datap)
	Push a task into the specified taskprocessor queue and signal the taskprocessor thread. More...
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. More...
void	ast_taskprocessor_set_local (struct ast_taskprocessor *tps, void *local_data)
	Sets the local data associated with a taskprocessor. More...
long	ast_taskprocessor_size (struct ast_taskprocessor *tps)
	Return the current size of the taskprocessor queue. More...
int	ast_taskprocessor_suspend (struct ast_taskprocessor *tps)
	Indicate the taskprocessor is suspended. More...
void *	ast_taskprocessor_unreference (struct ast_taskprocessor *tps)
	Unreference the specified taskprocessor and its reference count will decrement. More...
int	ast_taskprocessor_unsuspend (struct ast_taskprocessor *tps)
	Indicate the taskprocessor is unsuspended. More...
int	ast_tps_init (void)
static	AST_VECTOR_RW (subsystem_alert_vector, struct subsystem_alert *)
	CLI taskprocessor ping <blah>operation requires a ping condition lock. More...
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 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. More...
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)
	The astobj2 compare callback for taskprocessors. More...
static int	tps_hash_cb (const void *obj, const int flags)
	The astobj2 hash callback for taskprocessors. More...
static int	tps_ping_handler (void *datap)
	CLI taskprocessor ping <blah>handler function. More...
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)
static struct tps_task *	tps_task_alloc_local (int(*task_exe)(struct ast_taskprocessor_local *local), void *datap)
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 struct ast_cli_entry	taskprocessor_clis []
static unsigned int	tps_alert_count
static ast_rwlock_t	tps_alert_lock = { PTHREAD_RWLOCK_INITIALIZER , NULL, {1, 0} }

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_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 291 of file taskprocessor.c.

FMT_FIELDS

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

Definition at line 578 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\n"

Definition at line 577 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 1353 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 1053 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().

__start_taskprocessor()

static struct ast_taskprocessor * __start_taskprocessor ( struct ast_taskprocessor *  p )

static

Definition at line 1098 of file taskprocessor.c.

{
    if (p && p->listener->callbacks->start(p->listener)) {
        ast_log(LOG_ERROR, "Unable to start taskprocessor listener for taskprocessor %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 884 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 895 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 1365 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 1153 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 1282 of file taskprocessor.c.

{
    struct ast_taskprocessor_local local;
    struct tps_task *t;
    long size;
 
    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;
    }
    ao2_unlock(tps);
 
    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;
    }
    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(), 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, ast_taskprocessor::listener, ast_taskprocessor_local::local_data, ast_taskprocessor::local_data, tps_taskprocessor_stats::max_qsize, 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(), 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 1114 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, "requesting a nameless taskprocessor!!!\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(), cli_tps_ping(), cli_tps_reset_stats(), find_request_serializer(), internal_stasis_subscribe(), 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 704 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 1277 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 1336 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 970 of file taskprocessor.c.

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

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

Referenced by ast_taskpool_serializer_push_wait(), and execute_tasks().

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 1000 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 1014 of file taskprocessor.c.

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

References ao2_ref, and listener().

Referenced by 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 1020 of file taskprocessor.c.

{
    return listener->user_data;
}

References listener().

Referenced by ast_taskpool_serializer_push_wait(), execute_tasks(), serializer_shutdown(), 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 976 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 1355 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
	)

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 1245 of file taskprocessor.c.

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

References taskprocessor_push(), and tps_task_alloc().

Referenced by ast_cc_agent_status_response(), ast_cc_monitor_failed(), ast_cc_monitor_party_b_free(), ast_cc_monitor_status_request(), ast_cc_monitor_stop_ringing(), ast_msg_queue(), ast_sip_push_task(), ast_sorcery_create(), ast_sorcery_delete(), ast_sorcery_update(), ast_taskpool_push(), ast_taskpool_serializer_push_wait(), AST_TEST_DEFINE(), ast_threadpool_push(), ast_threadpool_set_size(), async_delete_name_rec(), async_play_sound_helper(), cc_request_state_change(), cli_tps_ping(), default_listener_shutdown(), destroy_conference_bridge(), dns_system_resolver_resolve(), generic_monitor_devstate_cb(), handle_cc_status(), handle_cli_taskpool_push_serializer_efficiency(), handle_cli_threadpool_push_serializer_efficiency(), hepv3_send_packet(), iax2_transmit(), mwi_handle_subscribe(), mwi_handle_unsubscribe(), play_sound_helper(), serializer_efficiency_task(), sorcery_object_load(), stasis_unsubscribe(), taskpool_taskprocessor_alloc(), taskpool_taskprocessor_dtor(), threadpool_active_thread_idle(), threadpool_idle_thread_dead(), threadpool_tps_emptied(), threadpool_tps_task_pushed(), and threadpool_zombie_thread_dead().

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 1250 of file taskprocessor.c.

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

References taskprocessor_push(), and tps_task_alloc_local().

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 1346 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 1171 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 965 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 1255 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 1179 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_threadpool_shutdown(), cli_tps_ping(), cli_tps_reset_stats(), cli_tps_reset_stats_all(), destroy_conference_bridge(), distributor(), distributor_pool_shutdown(), dns_system_resolver_destroy(), 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(), 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(), 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 1266 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 368 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, "taskprocessor container failed to initialize!\n");
        return -1;
    }
 
    if (AST_VECTOR_RW_INIT(&overloaded_subsystems, 10)) {
        ao2_ref(tps_singletons, -1);
        ast_log(LOG_ERROR, "taskprocessor subsystems vector failed to initialize!\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, taskprocessor_clis, 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

CLI taskprocessor ping <blah>operation requires a ping condition lock.

Definition at line 127 of file taskprocessor.c.

                                                   {

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 796 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 478 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(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(), 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 643 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");
    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 1399 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 1433 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.

default_listener_die()

static int default_listener_die ( void *  data )

static

Definition at line 246 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 1025 of file taskprocessor.c.

{
    struct default_taskprocessor_listener_pvt *pvt;
 
    pvt = ast_calloc(1, sizeof(*pvt));
    if (!pvt) {
        return NULL;
    }
    pvt->poll_thread = AST_PTHREADT_NULL;
    if (ast_sem_init(&pvt->sem, 0, 0) != 0) {
        ast_log(LOG_ERROR, "ast_sem_init(): %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 176 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 183 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 253 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(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(), 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 225 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 236 of file taskprocessor.c.

{
    struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
 
    if (ast_sem_post(&pvt->sem) != 0) {
        ast_log(LOG_ERROR, "Failed to notify of enqueued task: %s\n",
            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 196 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, "ast_sem_wait(): %s\n",
                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, 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 985 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 753 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 721 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 699 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 781 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 694 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 991 of file taskprocessor.c.

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

References 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 1209 of file taskprocessor.c.

{
    int previous_size;
    int was_empty;
 
    if (!tps) {
        ast_log(LOG_ERROR, "tps is NULL!\n");
        return -1;
    }
 
    if (!t) {
        ast_log(LOG_ERROR, "t is NULL!\n");
        return -1;
    }
 
    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, "The '%s' task processor queue reached %ld scheduled tasks%s.\n",
                tps->name, tps->tps_queue_size, 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_LIST_INSERT_TAIL, ast_log, ast_taskprocessor_listener::callbacks, ast_taskprocessor::executing, ast_taskprocessor::high_water_alert, ast_taskprocessor::high_water_warned, 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 859 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.\n",
            tps->name, tps_alert_count ? "triggered" : "cleared");
    }
 
    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

The astobj2 compare callback for taskprocessors.

Definition at line 686 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

The astobj2 hash callback for taskprocessors.

Definition at line 677 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

CLI taskprocessor ping <blah>handler function.

Definition at line 469 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 606 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 588 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);
}

References tps_taskprocessor_stats::_tasks_processed_count, ast_cli(), FMT_FIELDS, 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 1391 of file taskprocessor.c.

{
    ao2_lock(tps);
    tps->stats._tasks_processed_count = 0;
    tps->stats.max_qsize = 0;
    ao2_unlock(tps);
}

References tps_taskprocessor_stats::_tasks_processed_count, ao2_lock, ao2_unlock, tps_taskprocessor_stats::max_qsize, 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 297 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,
            "waiting for taskprocessor shutdown, %d tps object(s) still allocated.\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,
                "waiting for taskprocessor shutdown, %d tps object(s) still allocated.\n",
                objcount);
        }
    }
 
    /* 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,
            "Assertion may occur, the following taskprocessors are still running:\n");
 
        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");
        }
        else {
            iter = ao2_iterator_init(sorted_tps, AO2_ITERATOR_UNLINK);
            while ((tps = ao2_iterator_next(&iter))) {
                ast_log(LOG_ERROR, "taskprocessor '%s'\n", tps->name);
            }
        }
 
        ao2_cleanup(sorted_tps);
    }
    else {
        ast_log(LOG_DEBUG,
                "All waiting taskprocessors cleared!\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, taskprocessor_clis, 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 555 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  )

static

Definition at line 393 of file taskprocessor.c.

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

References ast_calloc, ast_log, tps_task::callback, tps_task::datap, tps_task::execute, 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  )

static

Definition at line 413 of file taskprocessor.c.

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

References ast_calloc, ast_log, tps_task::callback, tps_task::datap, tps_task::execute_local, 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 435 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 931 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 951 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 446 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(), and cli_tps_reset_stats().

Variable Documentation

default_listener_callbacks

const struct ast_taskprocessor_listener_callbacks default_listener_callbacks

static

Definition at line 283 of file taskprocessor.c.

Referenced by ast_taskprocessor_get().

taskprocessor_clis

struct ast_cli_entry taskprocessor_clis[]

static

Definition at line 162 of file taskprocessor.c.

Referenced by ast_tps_init(), and tps_shutdown().

tps_alert_count

unsigned int tps_alert_count

static

Count of the number of taskprocessors in high water alert.

Definition at line 846 of file taskprocessor.c.

Referenced by ast_taskprocessor_alert_get(), and tps_alert_add().

tps_alert_lock

ast_rwlock_t tps_alert_lock = { PTHREAD_RWLOCK_INITIALIZER , NULL, {1, 0} }

static

Access protection for tps_alert_count

Definition at line 849 of file taskprocessor.c.

Referenced by ast_taskprocessor_alert_get(), and tps_alert_add().