diff --git a/include/asterisk/res_pjsip.h b/include/asterisk/res_pjsip.h index f38e0b39a6..b26aba90d0 100644 --- a/include/asterisk/res_pjsip.h +++ b/include/asterisk/res_pjsip.h @@ -2980,6 +2980,11 @@ const char *ast_sip_get_host_ip_string(int af); */ long ast_sip_threadpool_queue_size(void); +/*! + * \brief Retrieve the SIP threadpool object + */ +struct ast_threadpool *ast_sip_threadpool(void); + /*! * \brief Retrieve transport state * \since 13.7.1 diff --git a/res/res_pjsip.c b/res/res_pjsip.c index c6594708bb..0dcbcea83b 100644 --- a/res/res_pjsip.c +++ b/res/res_pjsip.c @@ -34,6 +34,7 @@ #include "asterisk/utils.h" #include "asterisk/astobj2.h" #include "asterisk/module.h" +#include "asterisk/serializer.h" #include "asterisk/threadpool.h" #include "asterisk/taskprocessor.h" #include "asterisk/uuid.h" @@ -2840,7 +2841,7 @@ #define SERIALIZER_POOL_SIZE 8 /*! Pool of serializers to use if not supplied. */ -static struct ast_taskprocessor *serializer_pool[SERIALIZER_POOL_SIZE]; +static struct ast_serializer_pool *sip_serializer_pool; static pjsip_endpoint *ast_pjsip_endpoint; @@ -4626,71 +4627,10 @@ struct ast_taskprocessor *ast_sip_create_serializer(const char *name) return ast_sip_create_serializer_group(name, NULL); } -/*! - * \internal - * \brief Shutdown the serializers in the default pool. - * \since 14.0.0 - * - * \return Nothing - */ -static void serializer_pool_shutdown(void) -{ - int idx; - - for (idx = 0; idx < SERIALIZER_POOL_SIZE; ++idx) { - ast_taskprocessor_unreference(serializer_pool[idx]); - serializer_pool[idx] = NULL; - } -} - -/*! - * \internal - * \brief Setup the serializers in the default pool. - * \since 14.0.0 - * - * \retval 0 on success. - * \retval -1 on error. - */ -static int serializer_pool_setup(void) -{ - char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1]; - int idx; - - for (idx = 0; idx < SERIALIZER_POOL_SIZE; ++idx) { - /* Create name with seq number appended. */ - ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/default"); - - serializer_pool[idx] = ast_sip_create_serializer(tps_name); - if (!serializer_pool[idx]) { - serializer_pool_shutdown(); - return -1; - } - } - return 0; -} - -static struct ast_taskprocessor *serializer_pool_pick(void) -{ - int idx; - int pos = 0; - - if (!serializer_pool[0]) { - return NULL; - } - - for (idx = 1; idx < SERIALIZER_POOL_SIZE; ++idx) { - if (ast_taskprocessor_size(serializer_pool[idx]) < ast_taskprocessor_size(serializer_pool[pos])) { - pos = idx; - } - } - - return serializer_pool[pos]; -} - int ast_sip_push_task(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data) { if (!serializer) { - serializer = serializer_pool_pick(); + serializer = ast_serializer_pool_get(sip_serializer_pool); } return ast_taskprocessor_push(serializer, sip_task, task_data); @@ -4771,7 +4711,7 @@ int ast_sip_push_task_wait_serializer(struct ast_taskprocessor *serializer, int { if (!serializer) { /* Caller doesn't care which PJSIP serializer the task executes under. */ - serializer = serializer_pool_pick(); + serializer = ast_serializer_pool_get(sip_serializer_pool); if (!serializer) { /* No serializer picked to execute the task */ return -1; @@ -5133,6 +5073,11 @@ long ast_sip_threadpool_queue_size(void) return ast_threadpool_queue_size(sip_threadpool); } +struct ast_threadpool *ast_sip_threadpool(void) +{ + return sip_threadpool; +} + #ifdef TEST_FRAMEWORK AST_TEST_DEFINE(xml_sanitization_end_null) { @@ -5204,7 +5149,7 @@ static int unload_pjsip(void *data) * These calls need the pjsip endpoint and serializer to clean up. * If they're not set, then there's nothing to clean up anyway. */ - if (ast_pjsip_endpoint && serializer_pool[0]) { + if (ast_pjsip_endpoint && sip_serializer_pool) { ast_res_pjsip_cleanup_options_handling(); ast_res_pjsip_cleanup_message_filter(); ast_sip_destroy_distributor(); @@ -5340,7 +5285,9 @@ static int load_module(void) goto error; } - if (serializer_pool_setup()) { + sip_serializer_pool = ast_serializer_pool_create( + "pjsip/default", SERIALIZER_POOL_SIZE, sip_threadpool, -1); + if (!sip_serializer_pool) { ast_log(LOG_ERROR, "Failed to create SIP serializer pool. Aborting load\n"); goto error; } @@ -5413,7 +5360,7 @@ error: /* These functions all check for NULLs and are safe to call at any time */ ast_sip_destroy_scheduler(); - serializer_pool_shutdown(); + ast_serializer_pool_destroy(sip_serializer_pool); ast_threadpool_shutdown(sip_threadpool); return AST_MODULE_LOAD_DECLINE; @@ -5444,7 +5391,7 @@ static int unload_module(void) */ ast_sip_push_task_wait_servant(NULL, unload_pjsip, NULL); ast_sip_destroy_scheduler(); - serializer_pool_shutdown(); + ast_serializer_pool_destroy(sip_serializer_pool); ast_threadpool_shutdown(sip_threadpool); return 0; diff --git a/res/res_pjsip_mwi.c b/res/res_pjsip_mwi.c index c89d383855..49b5dc56c4 100644 --- a/res/res_pjsip_mwi.c +++ b/res/res_pjsip_mwi.c @@ -36,6 +36,7 @@ #include "asterisk/logger.h" #include "asterisk/astobj2.h" #include "asterisk/taskprocessor.h" +#include "asterisk/serializer.h" #include "asterisk/sorcery.h" #include "asterisk/stasis.h" #include "asterisk/mwi.h" @@ -57,8 +58,11 @@ static char *default_voicemail_extension; /*! Number of serializers in pool if one not supplied. */ #define MWI_SERIALIZER_POOL_SIZE 8 +/*! Max timeout for all threads to join during an unload. */ +#define MAX_UNLOAD_TIMEOUT_TIME 10 /* Seconds */ + /*! Pool of serializers to use if not supplied. */ -static struct ast_taskprocessor *mwi_serializer_pool[MWI_SERIALIZER_POOL_SIZE]; +static struct ast_serializer_pool *mwi_serializer_pool; static void mwi_subscription_shutdown(struct ast_sip_subscription *sub); static void mwi_to_ami(struct ast_sip_subscription *sub, struct ast_str **buf); @@ -129,117 +133,6 @@ struct mwi_subscription { char id[1]; }; -/*! - * \internal - * \brief Shutdown the serializers in the mwi pool. - * \since 13.12.0 - * - * \return Nothing - */ -static void mwi_serializer_pool_shutdown(void) -{ - int idx; - - for (idx = 0; idx < MWI_SERIALIZER_POOL_SIZE; ++idx) { - ast_taskprocessor_unreference(mwi_serializer_pool[idx]); - mwi_serializer_pool[idx] = NULL; - } -} - -/*! - * \internal - * \brief Setup the serializers in the mwi pool. - * \since 13.12.0 - * - * \retval 0 on success. - * \retval -1 on error. - */ -static int mwi_serializer_pool_setup(void) -{ - char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1]; - int idx; - - for (idx = 0; idx < MWI_SERIALIZER_POOL_SIZE; ++idx) { - /* Create name with seq number appended. */ - ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/mwi"); - - mwi_serializer_pool[idx] = ast_sip_create_serializer(tps_name); - if (!mwi_serializer_pool[idx]) { - mwi_serializer_pool_shutdown(); - return -1; - } - } - return 0; -} - -/*! - * \internal - * \brief Pick a mwi serializer from the pool. - * \since 13.12.0 - * - * \retval least queue size task processor. - */ -static struct ast_taskprocessor *get_mwi_serializer(void) -{ - int idx; - int pos; - - if (!mwi_serializer_pool[0]) { - return NULL; - } - - for (pos = idx = 0; idx < MWI_SERIALIZER_POOL_SIZE; ++idx) { - if (ast_taskprocessor_size(mwi_serializer_pool[idx]) < ast_taskprocessor_size(mwi_serializer_pool[pos])) { - pos = idx; - } - } - - return mwi_serializer_pool[pos]; -} - -/*! - * \internal - * \brief Set taskprocessor alert levels for the serializers in the mwi pool. - * \since 13.12.0 - * - * \retval 0 on success. - * \retval -1 on error. - */ -static int mwi_serializer_set_alert_levels(void) -{ - int idx; - long tps_queue_high; - long tps_queue_low; - - if (!mwi_serializer_pool[0]) { - return -1; - } - - tps_queue_high = ast_sip_get_mwi_tps_queue_high(); - if (tps_queue_high <= 0) { - ast_log(AST_LOG_WARNING, "Invalid taskprocessor high water alert trigger level '%ld'\n", - tps_queue_high); - tps_queue_high = AST_TASKPROCESSOR_HIGH_WATER_LEVEL; - } - - tps_queue_low = ast_sip_get_mwi_tps_queue_low(); - if (tps_queue_low < -1 || tps_queue_high < tps_queue_low) { - ast_log(AST_LOG_WARNING, "Invalid taskprocessor low water clear alert level '%ld'\n", - tps_queue_low); - tps_queue_low = -1; - } - - for (idx = 0; idx < MWI_SERIALIZER_POOL_SIZE; ++idx) { - if (ast_taskprocessor_alert_set_levels(mwi_serializer_pool[idx], tps_queue_low, tps_queue_high)) { - ast_log(AST_LOG_WARNING, "Failed to set alert levels for MWI serializer pool #%d.\n", - idx); - } - } - - return 0; -} - - static void mwi_stasis_cb(void *userdata, struct stasis_subscription *sub, struct stasis_message *msg); @@ -1213,7 +1106,7 @@ static int send_notify(void *obj, void *arg, int flags) struct mwi_subscription *mwi_sub = obj; struct ast_taskprocessor *serializer = mwi_sub->is_solicited ? ast_sip_subscription_get_serializer(mwi_sub->sip_sub) - : get_mwi_serializer(); + : ast_serializer_pool_get(mwi_serializer_pool); if (ast_sip_push_task(serializer, serialized_notify, ao2_bump(mwi_sub))) { ao2_ref(mwi_sub, -1); @@ -1228,7 +1121,8 @@ static void mwi_stasis_cb(void *userdata, struct stasis_subscription *sub, struct mwi_subscription *mwi_sub = userdata; if (stasis_subscription_final_message(sub, msg)) { - if (ast_sip_push_task(NULL, serialized_cleanup, ao2_bump(mwi_sub))) { + if (ast_sip_push_task(ast_serializer_pool_get(mwi_serializer_pool), + serialized_cleanup, ao2_bump(mwi_sub))) { ao2_ref(mwi_sub, -1); } return; @@ -1391,7 +1285,8 @@ static int send_contact_notify(void *obj, void *arg, int flags) return 0; } - if (ast_sip_push_task(get_mwi_serializer(), serialized_notify, ao2_bump(mwi_sub))) { + if (ast_sip_push_task(ast_serializer_pool_get(mwi_serializer_pool), + serialized_notify, ao2_bump(mwi_sub))) { ao2_ref(mwi_sub, -1); } @@ -1513,7 +1408,8 @@ static void mwi_startup_event_cb(void *data, struct stasis_subscription *sub, st return; } - ast_sip_push_task(NULL, send_initial_notify_all, NULL); + ast_sip_push_task(ast_serializer_pool_get(mwi_serializer_pool), + send_initial_notify_all, NULL); stasis_unsubscribe(sub); } @@ -1522,7 +1418,8 @@ static void global_loaded(const char *object_type) { ast_free(default_voicemail_extension); default_voicemail_extension = ast_sip_get_default_voicemail_extension(); - mwi_serializer_set_alert_levels(); + ast_serializer_pool_set_alerts(mwi_serializer_pool, + ast_sip_get_mwi_tps_queue_high(), ast_sip_get_mwi_tps_queue_low()); } static struct ast_sorcery_observer global_observer = { @@ -1543,14 +1440,16 @@ static int load_module(void) return AST_MODULE_LOAD_DECLINE; } - if (mwi_serializer_pool_setup()) { + mwi_serializer_pool = ast_serializer_pool_create("pjsip/mwi", + MWI_SERIALIZER_POOL_SIZE, ast_sip_threadpool(), MAX_UNLOAD_TIMEOUT_TIME); + if (!mwi_serializer_pool) { ast_log(AST_LOG_WARNING, "Failed to create MWI serializer pool. The default SIP pool will be used for MWI\n"); } solicited_mwi = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0, MWI_BUCKETS, mwi_sub_hash, NULL, mwi_sub_cmp); if (!solicited_mwi) { - mwi_serializer_pool_shutdown(); + ast_serializer_pool_destroy(mwi_serializer_pool); ast_sip_unregister_subscription_handler(&mwi_handler); return AST_MODULE_LOAD_DECLINE; } @@ -1558,7 +1457,7 @@ static int load_module(void) unsolicited_mwi = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0, MWI_BUCKETS, mwi_sub_hash, NULL, mwi_sub_cmp); if (!unsolicited_mwi) { - mwi_serializer_pool_shutdown(); + ast_serializer_pool_destroy(mwi_serializer_pool); ast_sip_unregister_subscription_handler(&mwi_handler); ao2_ref(solicited_mwi, -1); return AST_MODULE_LOAD_DECLINE; @@ -1571,7 +1470,8 @@ static int load_module(void) if (!ast_sip_get_mwi_disable_initial_unsolicited()) { create_mwi_subscriptions(); if (ast_test_flag(&ast_options, AST_OPT_FLAG_FULLY_BOOTED)) { - ast_sip_push_task(NULL, send_initial_notify_all, NULL); + ast_sip_push_task(ast_serializer_pool_get(mwi_serializer_pool), + send_initial_notify_all, NULL); } else { struct stasis_subscription *sub; @@ -1581,6 +1481,15 @@ static int load_module(void) } } + if (!mwi_serializer_pool) { + /* + * If the mwi serializer pool was unable to be established then the module will + * use the default serializer pool. If this happens prevent manual unloading + * since there would now exist the potential for a crash on unload. + */ + ast_module_shutdown_ref(ast_module_info->self); + } + return AST_MODULE_LOAD_SUCCESS; } @@ -1589,13 +1498,22 @@ static int unload_module(void) ast_sorcery_observer_remove(ast_sip_get_sorcery(), "global", &global_observer); ast_sorcery_observer_remove(ast_sip_get_sorcery(), "contact", &mwi_contact_observer); - ao2_callback(unsolicited_mwi, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE, unsubscribe, NULL); - ao2_ref(unsolicited_mwi, -1); - unsolicited_mwi = NULL; + if (unsolicited_mwi) { + ao2_callback(unsolicited_mwi, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE, unsubscribe, NULL); + ao2_ref(unsolicited_mwi, -1); + unsolicited_mwi = NULL; + } - ao2_cleanup(solicited_mwi); + if (solicited_mwi) { + ao2_ref(solicited_mwi, -1); + solicited_mwi = NULL; + } - mwi_serializer_pool_shutdown(); + if (ast_serializer_pool_destroy(mwi_serializer_pool)) { + ast_log(LOG_WARNING, "Unload incomplete. Try again later\n"); + return -1; + } + mwi_serializer_pool = NULL; ast_sip_unregister_subscription_handler(&mwi_handler);