From 6fa3ed067957e580e208ead7a62a095ac26f5fa9 Mon Sep 17 00:00:00 2001 From: Alexei Gradinari Date: Wed, 22 Jun 2016 15:25:23 -0400 Subject: [PATCH] res_pjsip: improve realtime performance #2 The patch removes updating all Endpoints' status on startup. Instead, only non-qualified aors with static contact and non-qualified non-expired contacts are retrieved from the realtime to update the endpoint status to ONLINE. The endpoint name was added to the contact object to simply find the endpoint that created this contact. The status of endpoints with qualified aors will be updated by 'qualify' functions. ASTERISK-26061 #close Change-Id: Id324c1776fa55d3741e0c5457ecac0304cb1a0df --- ...64_ps_contacts_add_endpoint_and_modify_.py | 33 +++++ include/asterisk/res_pjsip.h | 12 ++ res/res_pjsip.c | 6 + res/res_pjsip/location.c | 6 + res/res_pjsip/pjsip_configuration.c | 126 +++++++++++------ res/res_pjsip/pjsip_options.c | 128 +++++++++++++++++- 6 files changed, 265 insertions(+), 46 deletions(-) create mode 100644 contrib/ast-db-manage/config/versions/ef7efc2d3964_ps_contacts_add_endpoint_and_modify_.py diff --git a/contrib/ast-db-manage/config/versions/ef7efc2d3964_ps_contacts_add_endpoint_and_modify_.py b/contrib/ast-db-manage/config/versions/ef7efc2d3964_ps_contacts_add_endpoint_and_modify_.py new file mode 100644 index 0000000000..ff18343290 --- /dev/null +++ b/contrib/ast-db-manage/config/versions/ef7efc2d3964_ps_contacts_add_endpoint_and_modify_.py @@ -0,0 +1,33 @@ +"""ps_contacts add endpoint and modify expiration_time to bigint + +Revision ID: ef7efc2d3964 +Revises: a845e4d8ade8 +Create Date: 2016-06-02 18:18:46.231920 + +""" + +# revision identifiers, used by Alembic. +revision = 'ef7efc2d3964' +down_revision = 'a845e4d8ade8' + +from alembic import op +import sqlalchemy as sa + + +def upgrade(): + context = op.get_context() + + op.add_column('ps_contacts', sa.Column('endpoint', sa.String(40))) + + if context.bind.dialect.name != 'postgresql': + op.alter_column('ps_contacts', 'expiration_time', type_=sa.BigInteger) + else: + op.execute('ALTER TABLE ps_contacts ALTER COLUMN expiration_time TYPE BIGINT USING expiration_time::bigint') + + op.create_index('ps_contacts_qualifyfreq_exptime', 'ps_contacts', ['qualify_frequency', 'expiration_time']) + op.create_index('ps_aors_qualifyfreq_contact', 'ps_aors', ['qualify_frequency', 'contact']) +def downgrade(): + op.drop_index('ps_aors_qualifyfreq_contact') + op.drop_index('ps_contacts_qualifyfreq_exptime') + op.drop_column('ps_contacts', 'endpoint') + op.alter_column('ps_contacts', 'expiration_time', type_=sa.String(40)) diff --git a/include/asterisk/res_pjsip.h b/include/asterisk/res_pjsip.h index d1f0c9825b..f9f9e2037e 100644 --- a/include/asterisk/res_pjsip.h +++ b/include/asterisk/res_pjsip.h @@ -256,6 +256,8 @@ struct ast_sip_contact { int via_port; /*! Content of the Call-ID header in REGISTER request */ AST_STRING_FIELD_EXTENDED(call_id); + /*! The name of the endpoint that added the contact */ + AST_STRING_FIELD_EXTENDED(endpoint_name); }; #define CONTACT_STATUS "contact_status" @@ -977,6 +979,16 @@ void ast_sip_unregister_endpoint_identifier(struct ast_sip_endpoint_identifier * */ void *ast_sip_endpoint_alloc(const char *name); +/*! + * \brief Change state of a persistent endpoint. + * + * \param endpoint The SIP endpoint name to change state. + * \param state The new state + * \retval 0 Success + * \retval -1 Endpoint not found + */ +int ast_sip_persistent_endpoint_update_state(const char *endpoint_name, enum ast_endpoint_state state); + /*! * \brief Get a pointer to the PJSIP endpoint. * diff --git a/res/res_pjsip.c b/res/res_pjsip.c index 9944702c35..12bb930b56 100644 --- a/res/res_pjsip.c +++ b/res/res_pjsip.c @@ -1169,6 +1169,12 @@ REGISTER requests and is not intended to be configured manually. + + Endpoint name + + The name of the endpoint this contact belongs to + + Asterisk Server name diff --git a/res/res_pjsip/location.c b/res/res_pjsip/location.c index 8f8c030c35..aad76a6947 100644 --- a/res/res_pjsip/location.c +++ b/res/res_pjsip/location.c @@ -121,6 +121,7 @@ static void *contact_alloc(const char *name) return NULL; } + ast_string_field_init_extended(contact, endpoint_name); ast_string_field_init_extended(contact, reg_server); ast_string_field_init_extended(contact, via_addr); ast_string_field_init_extended(contact, call_id); @@ -352,6 +353,10 @@ int ast_sip_location_add_contact_nolock(struct ast_sip_aor *aor, const char *uri contact->endpoint = ao2_bump(endpoint); + if (endpoint) { + ast_string_field_set(contact, endpoint_name, ast_sorcery_object_get_id(endpoint)); + } + return ast_sorcery_create(ast_sip_get_sorcery(), contact); } @@ -1136,6 +1141,7 @@ int ast_sip_initialize_sorcery_location(void) ast_sorcery_object_field_register(sorcery, "contact", "authenticate_qualify", "no", OPT_BOOL_T, 1, FLDSET(struct ast_sip_contact, authenticate_qualify)); ast_sorcery_object_field_register(sorcery, "contact", "outbound_proxy", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_contact, outbound_proxy)); ast_sorcery_object_field_register(sorcery, "contact", "user_agent", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_contact, user_agent)); + ast_sorcery_object_field_register(sorcery, "contact", "endpoint", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_contact, endpoint_name)); ast_sorcery_object_field_register(sorcery, "contact", "reg_server", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_contact, reg_server)); ast_sorcery_object_field_register(sorcery, "contact", "via_addr", "", OPT_STRINGFIELD_T, 0, STRFLDSET(struct ast_sip_contact, via_addr)); ast_sorcery_object_field_register(sorcery, "contact", "via_port", "0", OPT_UINT_T, 0, FLDSET(struct ast_sip_contact, via_port)); diff --git a/res/res_pjsip/pjsip_configuration.c b/res/res_pjsip/pjsip_configuration.c index 3c4949573a..8a5ff416ad 100644 --- a/res/res_pjsip/pjsip_configuration.c +++ b/res/res_pjsip/pjsip_configuration.c @@ -58,6 +58,53 @@ static int persistent_endpoint_cmp(void *obj, void *arg, int flags) return !strcmp(ast_endpoint_get_resource(persistent1->endpoint), id) ? CMP_MATCH | CMP_STOP : 0; } +/*! \brief Internal function for changing the state of an endpoint */ +static void endpoint_update_state(struct ast_endpoint *endpoint, enum ast_endpoint_state state) +{ + struct ast_json *blob; + char *regcontext; + + /* If there was no state change, don't publish anything. */ + if (ast_endpoint_get_state(endpoint) == state) { + return; + } + + regcontext = ast_sip_get_regcontext(); + + if (state == AST_ENDPOINT_ONLINE) { + ast_endpoint_set_state(endpoint, AST_ENDPOINT_ONLINE); + blob = ast_json_pack("{s: s}", "peer_status", "Reachable"); + + if (!ast_strlen_zero(regcontext)) { + if (!ast_exists_extension(NULL, regcontext, ast_endpoint_get_resource(endpoint), 1, NULL)) { + ast_add_extension(regcontext, 1, ast_endpoint_get_resource(endpoint), 1, NULL, NULL, + "Noop", ast_strdup(ast_endpoint_get_resource(endpoint)), ast_free_ptr, "SIP"); + } + } + + ast_verb(2, "Endpoint %s is now Reachable\n", ast_endpoint_get_resource(endpoint)); + } else { + ast_endpoint_set_state(endpoint, AST_ENDPOINT_OFFLINE); + blob = ast_json_pack("{s: s}", "peer_status", "Unreachable"); + + if (!ast_strlen_zero(regcontext)) { + struct pbx_find_info q = { .stacklen = 0 }; + + if (pbx_find_extension(NULL, NULL, &q, regcontext, ast_endpoint_get_resource(endpoint), 1, NULL, "", E_MATCH)) { + ast_context_remove_extension(regcontext, ast_endpoint_get_resource(endpoint), 1, NULL); + } + } + + ast_verb(2, "Endpoint %s is now Unreachable\n", ast_endpoint_get_resource(endpoint)); + } + + ast_free(regcontext); + + ast_endpoint_blob_publish(endpoint, ast_endpoint_state_type(), blob); + ast_json_unref(blob); + ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "PJSIP/%s", ast_endpoint_get_resource(endpoint)); +} + /*! \brief Callback function for changing the state of an endpoint */ static int persistent_endpoint_update_state(void *obj, void *arg, int flags) { @@ -69,7 +116,6 @@ static int persistent_endpoint_update_state(void *obj, void *arg, int flags) struct ao2_iterator i; struct ast_sip_contact *contact; enum ast_endpoint_state state = AST_ENDPOINT_OFFLINE; - char *regcontext; if (status) { char rtt[32]; @@ -113,45 +159,7 @@ static int persistent_endpoint_update_state(void *obj, void *arg, int flags) ao2_ref(contacts, -1); } - /* If there was no state change, don't publish anything. */ - if (ast_endpoint_get_state(endpoint) == state) { - return 0; - } - - regcontext = ast_sip_get_regcontext(); - - if (state == AST_ENDPOINT_ONLINE) { - ast_endpoint_set_state(endpoint, AST_ENDPOINT_ONLINE); - blob = ast_json_pack("{s: s}", "peer_status", "Reachable"); - - if (!ast_strlen_zero(regcontext)) { - if (!ast_exists_extension(NULL, regcontext, ast_endpoint_get_resource(endpoint), 1, NULL)) { - ast_add_extension(regcontext, 1, ast_endpoint_get_resource(endpoint), 1, NULL, NULL, - "Noop", ast_strdup(ast_endpoint_get_resource(endpoint)), ast_free_ptr, "SIP"); - } - } - - ast_verb(2, "Endpoint %s is now Reachable\n", ast_endpoint_get_resource(endpoint)); - } else { - ast_endpoint_set_state(endpoint, AST_ENDPOINT_OFFLINE); - blob = ast_json_pack("{s: s}", "peer_status", "Unreachable"); - - if (!ast_strlen_zero(regcontext)) { - struct pbx_find_info q = { .stacklen = 0 }; - - if (pbx_find_extension(NULL, NULL, &q, regcontext, ast_endpoint_get_resource(endpoint), 1, NULL, "", E_MATCH)) { - ast_context_remove_extension(regcontext, ast_endpoint_get_resource(endpoint), 1, NULL); - } - } - - ast_verb(2, "Endpoint %s is now Unreachable\n", ast_endpoint_get_resource(endpoint)); - } - - ast_free(regcontext); - - ast_endpoint_blob_publish(endpoint, ast_endpoint_state_type(), blob); - ast_json_unref(blob); - ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "PJSIP/%s", ast_endpoint_get_resource(endpoint)); + endpoint_update_state(endpoint,state); return 0; } @@ -1181,6 +1189,20 @@ static void persistent_endpoint_destroy(void *obj) ast_free(persistent->aors); } +int ast_sip_persistent_endpoint_update_state(const char *endpoint_name, enum ast_endpoint_state state) +{ + RAII_VAR(struct sip_persistent_endpoint *, persistent, NULL, ao2_cleanup); + SCOPED_AO2LOCK(lock, persistent_endpoints); + + if (!(persistent = ao2_find(persistent_endpoints, endpoint_name, OBJ_KEY | OBJ_NOLOCK))) { + return -1; + } + + endpoint_update_state(persistent->endpoint, state); + + return 0; +} + /*! \brief Internal function which finds (or creates) persistent endpoint information */ static struct ast_endpoint *persistent_endpoint_find_or_create(const struct ast_sip_endpoint *endpoint) { @@ -1198,11 +1220,7 @@ static struct ast_endpoint *persistent_endpoint_find_or_create(const struct ast_ persistent->aors = ast_strdup(endpoint->aors); - if (ast_strlen_zero(persistent->aors)) { - ast_endpoint_set_state(persistent->endpoint, AST_ENDPOINT_UNKNOWN); - } else { - persistent_endpoint_update_state(persistent, NULL, 0); - } + ast_endpoint_set_state(persistent->endpoint, AST_ENDPOINT_UNKNOWN); ao2_link_flags(persistent_endpoints, persistent, OBJ_NOLOCK); } @@ -1683,6 +1701,22 @@ static struct ast_cli_entry cli_commands[] = { struct ast_sip_cli_formatter_entry *channel_formatter; struct ast_sip_cli_formatter_entry *endpoint_formatter; +static int on_load_endpoint(void *obj, void *arg, int flags) +{ + return sip_endpoint_apply_handler(sip_sorcery, obj); +} + +static void load_all_endpoints(void) +{ + struct ao2_container *endpoints; + + endpoints = ast_sorcery_retrieve_by_fields(sip_sorcery, "endpoint", AST_RETRIEVE_FLAG_MULTIPLE | AST_RETRIEVE_FLAG_ALL, NULL); + if (endpoints) { + ao2_callback(endpoints, OBJ_NODATA, on_load_endpoint, NULL); + ao2_ref(endpoints, -1); + } +} + int ast_res_pjsip_initialize_configuration(void) { if (ast_manager_register_xml(AMI_SHOW_ENDPOINTS, EVENT_FLAG_SYSTEM, ami_show_endpoints) || @@ -1885,6 +1919,8 @@ int ast_res_pjsip_initialize_configuration(void) ast_sorcery_load(sip_sorcery); + load_all_endpoints(); + return 0; } diff --git a/res/res_pjsip/pjsip_options.c b/res/res_pjsip/pjsip_options.c index 381de37545..dc3bce06f1 100644 --- a/res/res_pjsip/pjsip_options.c +++ b/res/res_pjsip/pjsip_options.c @@ -330,7 +330,12 @@ static int qualify_contact(struct ast_sip_endpoint *endpoint, struct ast_sip_con if (endpoint) { endpoint_local = ao2_bump(endpoint); } else { - endpoint_local = find_an_endpoint(contact); + if (!ast_strlen_zero(contact->endpoint_name)) { + endpoint_local = ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(), "endpoint", contact->endpoint_name); + } + if (!endpoint_local) { + endpoint_local = find_an_endpoint(contact); + } if (!endpoint_local) { ast_log(LOG_ERROR, "Unable to find an endpoint to qualify contact %s\n", contact->uri); @@ -1240,6 +1245,126 @@ static const struct ast_sorcery_observer observer_callbacks_options = { .deleted = aor_observer_deleted }; +static int aor_update_endpoint_state(void *obj, void *arg, int flags) +{ + struct ast_sip_endpoint *endpoint = obj; + const char *endpoint_name = ast_sorcery_object_get_id(endpoint); + char *aor = arg; + char *endpoint_aor; + char *endpoint_aors; + + if (ast_strlen_zero(aor) || ast_strlen_zero(endpoint->aors)) { + return 0; + } + + endpoint_aors = ast_strdupa(endpoint->aors); + while ((endpoint_aor = ast_strip(strsep(&endpoint_aors, ",")))) { + if (!strcmp(aor, endpoint_aor)) { + if (ast_sip_persistent_endpoint_update_state(endpoint_name, AST_ENDPOINT_ONLINE) == -1) { + ast_log(LOG_WARNING, "Unable to find persistent endpoint '%s' for aor '%s'\n", + endpoint_name, aor); + } + } + } + + return 0; +} + +static int on_aor_update_endpoint_state(void *obj, void *arg, int flags) +{ + struct ast_sip_aor *aor = obj; + struct ao2_container *endpoints; + RAII_VAR(struct ast_variable *, var, NULL, ast_variables_destroy); + const char *aor_name = ast_sorcery_object_get_id(aor); + char *aor_like; + + if (ast_strlen_zero(aor_name)) { + return -1; + } + + if (aor->permanent_contacts && ((int)(aor->qualify_frequency * 1000)) <= 0) { + aor_like = ast_alloca(strlen(aor_name) + 3); + sprintf(aor_like, "%%%s%%", aor_name); + var = ast_variable_new("aors LIKE", aor_like, ""); + if (!var) { + return -1; + } + endpoints = ast_sorcery_retrieve_by_fields(ast_sip_get_sorcery(), + "endpoint", AST_RETRIEVE_FLAG_MULTIPLE, var); + + if (endpoints) { + /* + * Because aors are a string list, we have to use a pattern match but since a simple + * pattern match could return an endpoint that has an aor of "aaabccc" when searching + * for "abc", we still have to iterate over them to find an exact aor match. + */ + ao2_callback(endpoints, 0, aor_update_endpoint_state, (char *)aor_name); + ao2_ref(endpoints, -1); + } + } + + return 0; +} + +static int contact_update_endpoint_state(void *obj, void *arg, int flags) +{ + const struct ast_sip_contact *contact = obj; + struct timeval tv = ast_tvnow(); + + if (!ast_strlen_zero(contact->endpoint_name) && ((int)(contact->qualify_frequency * 1000)) <= 0 && + contact->expiration_time.tv_sec > tv.tv_sec) { + + if (ast_sip_persistent_endpoint_update_state(contact->endpoint_name, AST_ENDPOINT_ONLINE) == -1) { + ast_log(LOG_WARNING, "Unable to find persistent endpoint '%s' for contact '%s/%s'\n", + contact->endpoint_name, contact->aor, contact->uri); + return -1; + } + } + + return 0; +} + +static void update_all_unqualified_endpoints(void) +{ + struct ao2_container *aors; + struct ao2_container *contacts; + RAII_VAR(struct ast_variable *, var_aor, NULL, ast_variables_destroy); + RAII_VAR(struct ast_variable *, var_contact, NULL, ast_variables_destroy); + RAII_VAR(char *, time_now, NULL, ast_free); + struct timeval tv = ast_tvnow(); + + if (!(var_aor = ast_variable_new("contact !=", "", ""))) { + return; + } + if (!(var_aor->next = ast_variable_new("qualify_frequency <=", "0", ""))) { + return; + } + + if (ast_asprintf(&time_now, "%ld", tv.tv_sec) == -1) { + return; + } + if (!(var_contact = ast_variable_new("expiration_time >", time_now, ""))) { + return; + } + if (!(var_contact->next = ast_variable_new("qualify_frequency <=", "0", ""))) { + return; + } + + aors = ast_sorcery_retrieve_by_fields(ast_sip_get_sorcery(), + "aor", AST_RETRIEVE_FLAG_MULTIPLE, var_aor); + if (aors) { + ao2_callback(aors, OBJ_NODATA, on_aor_update_endpoint_state, NULL); + ao2_ref(aors, -1); + } + + contacts = ast_sorcery_retrieve_by_fields(ast_sip_get_sorcery(), + "contact", AST_RETRIEVE_FLAG_MULTIPLE, var_contact); + if (contacts) { + ao2_callback(contacts, OBJ_NODATA, contact_update_endpoint_state, NULL); + ao2_ref(contacts, -1); + } +} + int ast_res_pjsip_init_options_handling(int reload) { static const pj_str_t STR_OPTIONS = { "OPTIONS", 7 }; @@ -1281,6 +1406,7 @@ int ast_res_pjsip_init_options_handling(int reload) ast_manager_register_xml("PJSIPQualify", EVENT_FLAG_SYSTEM | EVENT_FLAG_REPORTING, ami_sip_qualify); ast_cli_register_multiple(cli_options, ARRAY_LEN(cli_options)); + update_all_unqualified_endpoints(); qualify_and_schedule_all(); return 0;