taskprocessor.h File Reference

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

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

Go to the source code of this file.

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

Macros
#define	AST_TASKPROCESSOR_HIGH_WATER_LEVEL   500
#define	AST_TASKPROCESSOR_MAX_NAME   70
	Suggested maximum taskprocessor name length (less null terminator). More...

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

Functions
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) attribute_warn_unused_result
	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 *local), void *datap) attribute_warn_unused_result
	Push a task into the specified taskprocessor queue and signal the taskprocessor thread. More...
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...

Detailed Description

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

Author

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

Note

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

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

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

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

Definition in file taskprocessor.h.

Macro Definition Documentation

AST_TASKPROCESSOR_HIGH_WATER_LEVEL

#define AST_TASKPROCESSOR_HIGH_WATER_LEVEL   500

Default taskprocessor high water level alert trigger

Definition at line 63 of file taskprocessor.h.

Referenced by __allocate_taskprocessor(), actual_load_config(), ast_res_pjsip_init_options_handling(), ast_serializer_pool_set_alerts(), ast_sip_initialize_sorcery_location(), create_routes(), load_module(), and manager_subscriptions_init().

AST_TASKPROCESSOR_MAX_NAME

#define AST_TASKPROCESSOR_MAX_NAME   70

Suggested maximum taskprocessor name length (less null terminator).

Definition at line 60 of file taskprocessor.h.

Referenced by alloc_playback_chan(), allocate_subscription_tree(), ast_serializer_pool_create(), 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().

Enumeration Type Documentation

ast_tps_options

enum ast_tps_options

ast_tps_options for specification of taskprocessor options

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

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

Definition at line 73 of file taskprocessor.h.

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

Function Documentation

ast_taskprocessor_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 *local)	task_exe,
		void *	datap
	)

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

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

Parameters

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

Return values

0	success
-1	failure

Since

12.0.0

Referenced by AST_TEST_DEFINE(), and dispatch_message().

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