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	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 sybsystem. 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_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

FMT_FIELDS

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

Definition at line 511 of file taskprocessor.c.

Referenced by tps_report_taskprocessor_list_helper().

FMT_FIELDS_SUBSYSTEM

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

Referenced by cli_subsystem_alert_report().

FMT_HEADERS

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

Definition at line 510 of file taskprocessor.c.

Referenced by cli_tps_report().

FMT_HEADERS_SUBSYSTEM

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

Referenced by cli_subsystem_alert_report().

SEQ_STR_SIZE

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

Definition at line 1283 of file taskprocessor.c.

Referenced by ast_taskprocessor_build_name(), and ast_taskprocessor_name_append().

TPS_MAX_BUCKETS

#define TPS_MAX_BUCKETS   1567

Referenced by ast_tps_init().

Function Documentation

__allocate_taskprocessor()

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

static

Definition at line 983 of file taskprocessor.c.

References ao2_alloc, ao2_link_flags, ao2_ref, ast_copy_string(), ast_log, AST_PTHREADT_NULL, AST_TASKPROCESSOR_HIGH_WATER_LEVEL, ast_taskprocessor::listener, listener(), LOG_ERROR, LOG_WARNING, name, ast_taskprocessor::name, NULL, OBJ_NOLOCK, ast_taskprocessor::subsystem, ast_taskprocessor::thread, ast_taskprocessor_listener::tps, 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().

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

__start_taskprocessor()

static struct ast_taskprocessor* __start_taskprocessor ( struct ast_taskprocessor *  p )

static

Definition at line 1028 of file taskprocessor.c.

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

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

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

References ast_rwlock_rdlock, ast_rwlock_unlock, tps_alert_count, and tps_alert_lock.

Referenced by AST_TEST_DEFINE(), and distributor().

{
   unsigned int count;

   ast_rwlock_rdlock(&tps_alert_lock);
   count = tps_alert_count;
   ast_rwlock_unlock(&tps_alert_lock);

   return count;
}

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

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

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

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.

Returns

Nothing

Definition at line 1295 of file taskprocessor.c.

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

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

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

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

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

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

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

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

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_data, ast_taskprocessor_local::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_TEST_DEFINE(), default_tps_processing_function(), execute_tasks(), and threadpool_execute().

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

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

References __allocate_taskprocessor(), __start_taskprocessor(), ao2_find, ao2_lock, ao2_ref, ao2_unlock, ast_log, ast_strlen_zero, ast_taskprocessor_listener_alloc(), default_listener_pvt_alloc(), default_listener_pvt_destroy(), listener(), LOG_ERROR, 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(), and threadpool_alloc().

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

ast_taskprocessor_get_subsystem_alert()

unsigned int ast_taskprocessor_get_subsystem_alert

(

const char *

subsystem

)

Get the current taskprocessor high water alert count by sybsystem.

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

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

Referenced by AST_TEST_DEFINE(), and distributor().

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

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

References ast_taskprocessor::suspended.

Referenced by ast_sip_session_suspend().

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

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

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

{
   int is_task;

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

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

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

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

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

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

References ao2_ref, and ast_taskprocessor_listener::tps.

Referenced by serializer_task_pushed().

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

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

References ast_taskprocessor_listener::user_data.

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

{
   return listener->user_data;
}

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

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

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

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

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

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

Referenced by ast_serializer_pool_create().

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

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

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_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(), hepv3_send_packet(), iax2_transmit(), mwi_handle_subscribe(), mwi_handle_unsubscribe(), play_sound_helper(), sorcery_object_load(), stasis_unsubscribe(), threadpool_active_thread_idle(), threadpool_idle_thread_dead(), threadpool_tps_emptied(), threadpool_tps_task_pushed(), and threadpool_zombie_thread_dead().

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

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

References taskprocessor_push(), and tps_task_alloc_local().

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

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

References ast_atomic_fetchadd_int().

Referenced by ast_taskprocessor_build_name(), and ast_taskprocessor_name_append().

{
   static int seq_num;

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

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

References ast_taskprocessor::local_data, lock, and SCOPED_AO2LOCK.

Referenced by AST_TEST_DEFINE(), and internal_stasis_subscribe().

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

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

References ast_taskprocessor::tps_queue_size.

Referenced by ast_serializer_pool_get(), ast_taskprocessor_execute(), AST_TEST_DEFINE(), and ast_threadpool_queue_size().

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

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

References ao2_lock, ao2_unlock, and ast_taskprocessor::suspended.

Referenced by ast_sip_session_suspend(), and AST_TEST_DEFINE().

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

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

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(), 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(), sip_resolve_destroy(), sorcery_object_type_destructor(), subscription_dtor(), subscription_persistence_recreate(), subscription_tree_destructor(), tps_report_taskprocessor_list(), tps_shutdown_thread(), tps_taskprocessor_tab_complete(), unload_module(), and websocket_cb().

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

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

References ao2_lock, ao2_unlock, and ast_taskprocessor::suspended.

Referenced by ast_sip_session_unsuspend(), and AST_TEST_DEFINE().

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

ast_tps_init()

int ast_tps_init

(

void

)

Provided by taskprocessor.c

Definition at line 301 of file taskprocessor.c.

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

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

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

References subsystem_alert::alert_count, ast_cli_args::argc, 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, ast_cli_args::fd, FMT_FIELDS_SUBSYSTEM, FMT_HEADERS_SUBSYSTEM, NULL, subsystem_alert::subsystem, subsystem_copy(), and ast_cli_entry::usage.

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

cli_tps_ping()

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

static

Definition at line 411 of file taskprocessor.c.

References ast_cli_args::argc, ast_cli_args::argv, 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, ast_cli_args::fd, name, NULL, ast_cli_args::pos, tps_ping_handler(), TPS_REF_IF_EXISTS, tps_taskprocessor_tab_complete(), and ast_cli_entry::usage.

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

cli_tps_report()

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

static

Definition at line 576 of file taskprocessor.c.

References ast_cli_args::argc, ast_cli_entry::args, ast_cli_args::argv, ast_cli(), CLI_GENERATE, CLI_INIT, CLI_SHOWUSAGE, CLI_SUCCESS, ast_cli_entry::command, ast_cli_args::fd, FMT_HEADERS, NULL, ast_cli_args::pos, tps_report_taskprocessor_list(), tps_taskprocessor_tab_complete(), and ast_cli_entry::usage.

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

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

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

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

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

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

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

default_listener_die()

static int default_listener_die ( void *  data )

static

Definition at line 243 of file taskprocessor.c.

References default_taskprocessor_listener_pvt::dead.

Referenced by default_listener_shutdown().

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

default_listener_pvt_alloc()

static void* default_listener_pvt_alloc ( void  )

static

Definition at line 955 of file taskprocessor.c.

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

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

default_listener_pvt_destroy()

static void default_listener_pvt_destroy ( struct default_taskprocessor_listener_pvt *  pvt )

static

Definition at line 173 of file taskprocessor.c.

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

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

default_listener_pvt_dtor()

static void default_listener_pvt_dtor ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 180 of file taskprocessor.c.

References default_listener_pvt_destroy(), NULL, and ast_taskprocessor_listener::user_data.

{
   struct default_taskprocessor_listener_pvt *pvt = listener->user_data;

   default_listener_pvt_destroy(pvt);

   listener->user_data = NULL;
}

default_listener_shutdown()

static void default_listener_shutdown ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 250 of file taskprocessor.c.

References ao2_t_ref, ast_assert, ast_log, AST_PTHREADT_NULL, ast_taskprocessor_push(), default_listener_die(), errno, LOG_ERROR, NULL, default_taskprocessor_listener_pvt::poll_thread, ast_taskprocessor_listener::tps, and ast_taskprocessor_listener::user_data.

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

default_listener_start()

static int default_listener_start ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 222 of file taskprocessor.c.

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

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

default_task_pushed()

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

static

Definition at line 233 of file taskprocessor.c.

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

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

default_tps_processing_function()

static void* default_tps_processing_function ( void *  data )

static

Function that processes tasks in the taskprocessor.

Definition at line 193 of file taskprocessor.c.

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, default_taskprocessor_listener_pvt::sem, ast_taskprocessor_listener::tps, and ast_taskprocessor_listener::user_data.

Referenced by default_listener_start().

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

listener_shutdown()

static void listener_shutdown ( struct ast_taskprocessor_listener *  listener )

static

Definition at line 915 of file taskprocessor.c.

References ao2_ref, ast_taskprocessor_listener::callbacks, ast_taskprocessor_listener_callbacks::shutdown, and ast_taskprocessor_listener::tps.

Referenced by ast_taskprocessor_unreference().

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

subsystem_alert_decrement()

static void subsystem_alert_decrement ( const char *  subsystem )

static

Definition at line 686 of file taskprocessor.c.

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, and subsystem_match().

Referenced by tps_alert_add().

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

subsystem_alert_increment()

static void subsystem_alert_increment ( const char *  subsystem )

static

Definition at line 654 of file taskprocessor.c.

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

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

subsystem_cmp()

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

static

Definition at line 632 of file taskprocessor.c.

References subsystem_alert::subsystem.

Referenced by subsystem_copy().

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

subsystem_copy()

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

static

Definition at line 714 of file taskprocessor.c.

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

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

subsystem_match()

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

static

Definition at line 627 of file taskprocessor.c.

References subsystem_alert::subsystem.

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

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

taskprocessor_listener_dtor()

static void taskprocessor_listener_dtor ( void *  obj )

static

Definition at line 921 of file taskprocessor.c.

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

Referenced by ast_taskprocessor_listener_alloc().

{
   struct ast_taskprocessor_listener *listener = obj;

   if (listener->callbacks->dtor) {
      listener->callbacks->dtor(listener);
   }
}

taskprocessor_push()

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

static

Definition at line 1139 of file taskprocessor.c.

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, tps_task::list, 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().

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

tps_alert_add()

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

static

Definition at line 794 of file taskprocessor.c.

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

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

tps_cmp_cb()

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

static

The astobj2 compare callback for taskprocessors.

Definition at line 619 of file taskprocessor.c.

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

Referenced by ast_tps_init().

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

tps_hash_cb()

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

static

The astobj2 hash callback for taskprocessors.

Definition at line 610 of file taskprocessor.c.

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

Referenced by ast_tps_init().

{
   const struct ast_taskprocessor *tps = obj;
   const char *name = flags & OBJ_KEY ? obj : tps->name;

   return ast_str_case_hash(name);
}

tps_ping_handler()

static int tps_ping_handler ( void *  datap )

static

CLI taskprocessor ping <blah>handler function.

Definition at line 402 of file taskprocessor.c.

References ast_cond_signal, ast_mutex_lock, and ast_mutex_unlock.

Referenced by cli_tps_ping().

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

tps_report_taskprocessor_list()

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

static

Definition at line 539 of file taskprocessor.c.

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

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

tps_report_taskprocessor_list_helper()

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

static

Definition at line 521 of file taskprocessor.c.

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

{
   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_reset_stats()

static void tps_reset_stats ( struct ast_taskprocessor *  tps )

static

Definition at line 1321 of file taskprocessor.c.

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

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

tps_shutdown()

static void tps_shutdown ( void  )

static

Definition at line 291 of file taskprocessor.c.

References ao2_t_ref, ARRAY_LEN, ast_cli_unregister_multiple(), ast_free, AST_VECTOR_CALLBACK_VOID, AST_VECTOR_RW_FREE, and NULL.

Referenced by ast_tps_init().

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

tps_sort_cb()

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

static

Definition at line 488 of file taskprocessor.c.

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

Referenced by tps_report_taskprocessor_list().

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

tps_task_alloc()

static struct tps_task* tps_task_alloc ( int(*)(void *datap)  task_exe, void *  datap  )

static

Definition at line 326 of file taskprocessor.c.

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

Referenced by ast_taskprocessor_push().

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

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

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

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

tps_task_free()

static void* tps_task_free ( struct tps_task *  task )

static

Definition at line 368 of file taskprocessor.c.

References ast_free, and NULL.

Referenced by ast_taskprocessor_execute(), and tps_taskprocessor_dtor().

{
   ast_free(task);
   return NULL;
}

tps_taskprocessor_dtor()

static void tps_taskprocessor_dtor ( void *  tps )

static

Definition at line 866 of file taskprocessor.c.

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

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

tps_taskprocessor_pop()

static struct tps_task* tps_taskprocessor_pop ( struct ast_taskprocessor *  tps )

static

Definition at line 886 of file taskprocessor.c.

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

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

tps_taskprocessor_tab_complete()

static char* tps_taskprocessor_tab_complete ( struct ast_cli_args *  a )

static

Definition at line 379 of file taskprocessor.c.

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

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

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

Variable Documentation

default_listener_callbacks

const struct ast_taskprocessor_listener_callbacks default_listener_callbacks

static

Definition at line 280 of file taskprocessor.c.

taskprocessor_clis

struct ast_cli_entry taskprocessor_clis[]

static

Definition at line 159 of file taskprocessor.c.

tps_alert_count

unsigned int tps_alert_count

static

Count of the number of taskprocessors in high water alert.

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

Referenced by ast_taskprocessor_alert_get(), and tps_alert_add().