Asterisk - The Open Source Telephony Project GIT-master-0644429
|
An API for managing task processing threads that can be shared across modules. More...
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 subsystem. More... | |
int | ast_taskprocessor_is_suspended (struct ast_taskprocessor *tps) |
Get the task processor suspend status. More... | |
int | ast_taskprocessor_is_task (struct ast_taskprocessor *tps) |
Am I the given taskprocessor's current task. More... | |
struct ast_taskprocessor_listener * | ast_taskprocessor_listener_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... | |
An API for managing task processing threads that can be shared across modules.
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.
#define AST_TASKPROCESSOR_HIGH_WATER_LEVEL 500 |
Default taskprocessor high water level alert trigger
Definition at line 64 of file taskprocessor.h.
#define AST_TASKPROCESSOR_MAX_NAME 70 |
Suggested maximum taskprocessor name length (less null terminator).
Definition at line 61 of file taskprocessor.h.
enum ast_tps_options |
ast_tps_options for specification of taskprocessor options
Specify whether a taskprocessor should be created via ast_taskprocessor_get() if the taskprocessor does not already exist. The default behavior is to create a taskprocessor if it does not already exist and provide its reference to the calling function. To only return a reference to a taskprocessor if and only if it exists, use the TPS_REF_IF_EXISTS option in ast_taskprocessor_get().
Enumerator | |
---|---|
TPS_REF_DEFAULT | return a reference to a taskprocessor, create one if it does not exist |
TPS_REF_IF_EXISTS | return a reference to a taskprocessor ONLY if it already exists |
Definition at line 74 of file taskprocessor.h.
unsigned int ast_taskprocessor_alert_get | ( | void | ) |
Get the current taskprocessor high water alert count.
0 | if no taskprocessors are in high water alert. |
non-zero | if some task processors are in high water alert. |
Definition at line 884 of file taskprocessor.c.
References ast_rwlock_rdlock, ast_rwlock_unlock, tps_alert_count, and tps_alert_lock.
Referenced by AST_TEST_DEFINE(), and distributor().
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.
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. |
0 | on success. |
-1 | on error (water marks not changed). |
Definition at line 895 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().
void ast_taskprocessor_build_name | ( | char * | buf, |
unsigned int | size, | ||
const char * | format, | ||
... | |||
) |
Build a taskprocessor name with a sequence number on the end.
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. |
Definition at line 1360 of file taskprocessor.c.
References ast_assert, ast_taskprocessor_seq_num(), buf, NULL, and SEQ_STR_SIZE.
Referenced by alloc_playback_chan(), allocate_subscription_tree(), ast_sip_session_alloc(), create_websocket_serializer(), distributor_pool_setup(), internal_stasis_subscribe(), refer_progress_alloc(), sip_options_aor_alloc(), sip_outbound_publisher_alloc(), sip_outbound_registration_state_alloc(), and sorcery_object_type_alloc().
struct ast_taskprocessor * ast_taskprocessor_create_with_listener | ( | const char * | name, |
struct ast_taskprocessor_listener * | listener | ||
) |
Create a taskprocessor with a custom listener.
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.
name | The name of the taskprocessor to create |
listener | The listener for operations on this taskprocessor |
NULL | Failure |
non-NULL | success |
Definition at line 1148 of file taskprocessor.c.
References __allocate_taskprocessor(), __start_taskprocessor(), ao2_find, ao2_lock, ao2_unlock, ast_taskprocessor_unreference(), listener(), name, NULL, OBJ_KEY, and OBJ_NOLOCK.
Referenced by AST_TEST_DEFINE(), ast_threadpool_create(), and ast_threadpool_serializer_group().
int ast_taskprocessor_execute | ( | struct ast_taskprocessor * | tps | ) |
Pop a task off the taskprocessor and execute it.
tps | The taskprocessor from which to execute. |
0 | There is no further work to be done. |
1 | Tasks still remain in the taskprocessor queue. |
Definition at line 1277 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::local_data, ast_taskprocessor::local_data, tps_taskprocessor_stats::max_qsize, ast_taskprocessor::stats, ast_taskprocessor::thread, tps_task_free(), tps_taskprocessor_pop(), and tps_task::wants_local.
Referenced by AST_TEST_DEFINE(), default_tps_processing_function(), execute_tasks(), and threadpool_execute().
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().
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 |
Definition at line 1109 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_callbacks, default_listener_pvt_alloc(), default_listener_pvt_destroy(), listener(), LOG_ERROR, name, NULL, OBJ_KEY, OBJ_NOLOCK, and TPS_REF_IF_EXISTS.
Referenced by alloc_playback_chan(), ast_dns_system_resolver_init(), ast_msg_init(), AST_TEST_DEFINE(), cli_tps_ping(), cli_tps_reset_stats(), find_request_serializer(), internal_stasis_subscribe(), load_module(), load_objects(), and threadpool_alloc().
unsigned int ast_taskprocessor_get_subsystem_alert | ( | const char * | subsystem | ) |
Get the current taskprocessor high water alert count by subsystem.
subsystem | The subsystem name |
0 | if no taskprocessors are in high water alert. |
non-zero | if some task processors are in high water alert. |
Definition at line 704 of file taskprocessor.c.
References subsystem_alert::alert_count, AST_VECTOR_GET, AST_VECTOR_GET_INDEX, AST_VECTOR_RW_RDLOCK, AST_VECTOR_RW_UNLOCK, subsystem_alert::subsystem, and subsystem_match().
Referenced by AST_TEST_DEFINE(), and distributor().
int ast_taskprocessor_is_suspended | ( | struct ast_taskprocessor * | tps | ) |
Get the task processor suspend status.
tps | Task processor. |
non-zero | if the task processor is suspended |
Definition at line 1272 of file taskprocessor.c.
References ast_taskprocessor::suspended.
Referenced by ast_sip_session_suspend().
int ast_taskprocessor_is_task | ( | struct ast_taskprocessor * | tps | ) |
Am I the given taskprocessor's current task.
tps | Taskprocessor to check. |
non-zero | if current thread is the taskprocessor thread. |
Definition at line 1331 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().
struct ast_taskprocessor_listener * ast_taskprocessor_listener_alloc | ( | const struct ast_taskprocessor_listener_callbacks * | callbacks, |
void * | user_data | ||
) |
Allocate a taskprocessor listener.
This will result in the listener being allocated with the specified callbacks.
callbacks | The callbacks to assign to the listener |
user_data | The user data for the listener |
NULL | Failure |
non-NULL | The newly allocated taskprocessor listener |
Definition at line 995 of file taskprocessor.c.
References ao2_alloc, 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 * 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()
listener | The listener that has the taskprocessor |
Definition at line 1009 of file taskprocessor.c.
References ao2_ref, and listener().
Referenced by serializer_task_pushed().
void * ast_taskprocessor_listener_get_user_data | ( | const struct ast_taskprocessor_listener * | listener | ) |
Get the user data from the listener.
listener | The taskprocessor listener |
Definition at line 1015 of file taskprocessor.c.
References listener().
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().
const char * ast_taskprocessor_name | ( | struct ast_taskprocessor * | tps | ) |
Return the name of the taskprocessor singleton.
Definition at line 971 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().
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.
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 1350 of file taskprocessor.c.
References ast_assert, ast_taskprocessor_seq_num(), buf, name, NULL, and SEQ_STR_SIZE.
Referenced by ast_serializer_pool_create().
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.
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 |
0 | success |
-1 | failure |
Definition at line 1240 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().
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().
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 |
0 | success |
-1 | failure |
Referenced by AST_TEST_DEFINE(), and dispatch_message().
unsigned int ast_taskprocessor_seq_num | ( | void | ) |
Get the next sequence number to create a human friendly taskprocessor name.
Definition at line 1341 of file taskprocessor.c.
References ast_atomic_fetchadd_int().
Referenced by ast_taskprocessor_build_name(), and ast_taskprocessor_name_append().
void ast_taskprocessor_set_local | ( | struct ast_taskprocessor * | tps, |
void * | local_data | ||
) |
Sets the local data associated with a taskprocessor.
See ast_taskprocessor_push_local().
tps | Task processor. |
local_data | Local data to associate with tps. |
Definition at line 1166 of file taskprocessor.c.
References ast_taskprocessor::local_data, lock, and SCOPED_AO2LOCK.
Referenced by AST_TEST_DEFINE(), and internal_stasis_subscribe().
long ast_taskprocessor_size | ( | struct ast_taskprocessor * | tps | ) |
Return the current size of the taskprocessor queue.
Definition at line 965 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().
int ast_taskprocessor_suspend | ( | struct ast_taskprocessor * | tps | ) |
Indicate the taskprocessor is suspended.
tps | Task processor. |
0 | success |
-1 | failure |
Definition at line 1250 of file taskprocessor.c.
References ao2_lock, ao2_unlock, and ast_taskprocessor::suspended.
Referenced by ast_sip_session_suspend(), and AST_TEST_DEFINE().
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.
tps | taskprocessor to unreference |
Definition at line 1174 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(), 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().
int ast_taskprocessor_unsuspend | ( | struct ast_taskprocessor * | tps | ) |
Indicate the taskprocessor is unsuspended.
tps | Task processor. |
0 | success |
-1 | failure |
Definition at line 1261 of file taskprocessor.c.
References ao2_lock, ao2_unlock, and ast_taskprocessor::suspended.
Referenced by ast_sip_session_unsuspend(), and AST_TEST_DEFINE().