diff options
44 files changed, 1048 insertions, 455 deletions
@@ -52,6 +52,9 @@ res_pjsip into the "reg_server" field in the ps_contacts table to facilitate multi-server setups. + * When starting Asterisk, received traffic will now be ignored until Asterisk + has loaded all modules and is fully booted. + res_hep ------------------ * Added a new option, 'uuid_type', that sets the preferred source of the Homer @@ -68,6 +71,12 @@ res_pjsip_info_empty Some SBCs will terminate a call if their empty INFO packets are not responded to within a predefined time. +res_odbc +------------------ + * A new option has been added, 'max_connections', which sets the maximum number + of concurrent connections to the database. This option defaults to 1 which + returns the behavior to that of Asterisk 13.7 and prior. + app_confbridge ------------------ * Added a bridge profile option called regcontext that allows you to @@ -76,6 +85,18 @@ app_confbridge server installations via alternate means (DUNDI for example). By default this feature is not used. +Codecs +------------------ + * Added the associated format name to 'core show codecs'. + +res_ari_channels +------------------ + * Added 'formats' to channel create/originate to allow setting the allowed + formats for a channel when no originator channel is available. Especially + useful for Local channel creation where no other format information is + available. 'core show codecs' can now be used to look up suitable format + names. + ------------------------------------------------------------------------------ --- Functionality changes from Asterisk 13.8.0 to Asterisk 13.9.0 ------------ ------------------------------------------------------------------------------ diff --git a/addons/ooh323c/src/Makefile.in b/addons/ooh323c/src/Makefile.in index d3a96024b..15b14f7df 100644 --- a/addons/ooh323c/src/Makefile.in +++ b/addons/ooh323c/src/Makefile.in @@ -104,7 +104,7 @@ CONFIG_HEADER = $(top_builddir)/config.h CONFIG_CLEAN_FILES = LIBRARIES = $(noinst_LIBRARIES) -libooh323c_a_AR = $(AR) cru +libooh323c_a_AR = $(AR) cr libooh323c_a_LIBADD = am_libooh323c_a_OBJECTS = ooLogChan.$(OBJEXT) ooUtils.$(OBJEXT) \ ooGkClient.$(OBJEXT) context.$(OBJEXT) ooDateTime.$(OBJEXT) \ diff --git a/channels/chan_pjsip.c b/channels/chan_pjsip.c index f081bd899..970fef496 100644 --- a/channels/chan_pjsip.c +++ b/channels/chan_pjsip.c @@ -269,6 +269,9 @@ static int direct_media_mitigate_glare(struct ast_sip_session *session) return 0; } +/*! + * \pre chan is locked + */ static int check_for_rtp_changes(struct ast_channel *chan, struct ast_rtp_instance *rtp, struct ast_sip_session_media *media, int rtcp_fd) { @@ -338,6 +341,11 @@ static int send_direct_media_request(void *data) int changed = 0; int res = 0; + /* The channel needs to be locked when checking for RTP changes. + * Otherwise, we could end up destroying an underlying RTCP structure + * at the same time that the channel thread is attempting to read RTCP + */ + ast_channel_lock(cdata->chan); if (pvt->media[SIP_MEDIA_AUDIO]) { changed |= check_for_rtp_changes( cdata->chan, cdata->rtp, pvt->media[SIP_MEDIA_AUDIO], 1); @@ -346,6 +354,7 @@ static int send_direct_media_request(void *data) changed |= check_for_rtp_changes( cdata->chan, cdata->vrtp, pvt->media[SIP_MEDIA_VIDEO], 3); } + ast_channel_unlock(cdata->chan); if (direct_media_mitigate_glare(cdata->session)) { ast_debug(4, "Disregarding setting RTP on %s: mitigating re-INVITE glare\n", ast_channel_name(cdata->chan)); diff --git a/channels/chan_sip.c b/channels/chan_sip.c index f64845472..77f438262 100644 --- a/channels/chan_sip.c +++ b/channels/chan_sip.c @@ -15790,11 +15790,12 @@ static void start_register_timeout(struct sip_registry *reg) static const char *sip_sanitized_host(const char *host) { - struct ast_sockaddr addr = { { 0, 0, }, }; + struct ast_sockaddr addr; /* peer/sip_pvt->tohost and sip_registry->hostname should never have a port * in them, so we use PARSE_PORT_FORBID here. If this lookup fails, we return * the original host which is most likely a host name and not an IP. */ + memset(&addr, 0, sizeof(addr)); if (!ast_sockaddr_parse(&addr, host, PARSE_PORT_FORBID)) { return host; } diff --git a/configs/samples/res_odbc.conf.sample b/configs/samples/res_odbc.conf.sample index 66659ae42..a21e96d07 100644 --- a/configs/samples/res_odbc.conf.sample +++ b/configs/samples/res_odbc.conf.sample @@ -51,6 +51,11 @@ pre-connect => yes ; that we should attempt? ;limit => 5 ; +; The maximum number of connections to have open at any given time. +; This defaults to 1 and it is highly recommended to only set this higher +; if using a version of UnixODBC greater than 2.3.1. +;max_connections => 20 +; ; When the channel is destroyed, should any uncommitted open transactions ; automatically be committed? ;forcecommit => no diff --git a/include/asterisk/codec.h b/include/asterisk/codec.h index 28befec50..fb2b7da38 100644 --- a/include/asterisk/codec.h +++ b/include/asterisk/codec.h @@ -77,6 +77,8 @@ struct ast_codec { unsigned int smooth; /*! \brief The module that registered this codec */ struct ast_module *mod; + /*! \brief A format name for a default sane format using this codec */ + const char *format_name; }; /*! diff --git a/include/asterisk/compat.h b/include/asterisk/compat.h index c9c99c150..c7bc5b093 100644 --- a/include/asterisk/compat.h +++ b/include/asterisk/compat.h @@ -68,7 +68,7 @@ #endif #ifndef AST_POLL_COMPAT -#include <sys/poll.h> +#include <poll.h> #else #include "asterisk/poll-compat.h" #endif diff --git a/include/asterisk/poll-compat.h b/include/asterisk/poll-compat.h index cbb610925..72ac2c3e2 100644 --- a/include/asterisk/poll-compat.h +++ b/include/asterisk/poll-compat.h @@ -83,7 +83,7 @@ #ifndef AST_POLL_COMPAT -#include <sys/poll.h> +#include <poll.h> #define ast_poll(a, b, c) poll(a, b, c) diff --git a/include/asterisk/res_hep.h b/include/asterisk/res_hep.h index bd0129eea..cfd213ad7 100644 --- a/include/asterisk/res_hep.h +++ b/include/asterisk/res_hep.h @@ -118,6 +118,14 @@ int hepv3_send_packet(struct hepv3_capture_info *capture_info); */ enum hep_uuid_type hepv3_get_uuid_type(void); +/*! + * \brief Return whether or not we're currently loaded and active + * + * \retval 0 The module is not loaded + * \retval 1 The module is loaded + */ +int hepv3_is_loaded(void); + #if defined(__cplusplus) || defined(c_plusplus) } #endif diff --git a/include/asterisk/res_pjsip.h b/include/asterisk/res_pjsip.h index b64ad624b..e3eab7cb4 100644 --- a/include/asterisk/res_pjsip.h +++ b/include/asterisk/res_pjsip.h @@ -1336,6 +1336,17 @@ struct ast_taskprocessor *ast_sip_create_serializer_group(struct ast_serializer_ struct ast_taskprocessor *ast_sip_create_serializer_group_named(const char *name, struct ast_serializer_shutdown_group *shutdown_group); /*! + * \brief Determine the distributor serializer for the SIP message. + * \since 13.10.0 + * + * \param rdata The incoming message. + * + * \retval Calculated distributor serializer on success. + * \retval NULL on error. + */ +struct ast_taskprocessor *ast_sip_get_distributor_serializer(pjsip_rx_data *rdata); + +/*! * \brief Set a serializer on a SIP dialog so requests and responses are automatically serialized * * Passing a NULL serializer is a way to remove a serializer from a dialog. diff --git a/include/asterisk/res_pjsip_session.h b/include/asterisk/res_pjsip_session.h index 75d37ac3d..e4c54a173 100644 --- a/include/asterisk/res_pjsip_session.h +++ b/include/asterisk/res_pjsip_session.h @@ -406,9 +406,10 @@ struct ast_sip_channel_pvt *ast_sip_channel_pvt_alloc(void *pvt, struct ast_sip_ * \param endpoint The endpoint that this session communicates with * \param contact The contact associated with this session * \param inv_session The PJSIP INVITE session data + * \param rdata INVITE request received (NULL if for outgoing allocation) */ struct ast_sip_session *ast_sip_session_alloc(struct ast_sip_endpoint *endpoint, - struct ast_sip_contact *contact, pjsip_inv_session *inv); + struct ast_sip_contact *contact, pjsip_inv_session *inv, pjsip_rx_data *rdata); /*! * \brief Request and wait for the session serializer to be suspended. diff --git a/include/asterisk/sorcery.h b/include/asterisk/sorcery.h index 5e947257b..0cb434766 100644 --- a/include/asterisk/sorcery.h +++ b/include/asterisk/sorcery.h @@ -692,6 +692,20 @@ int __ast_sorcery_object_register(struct ast_sorcery *sorcery, const char *type, __ast_sorcery_object_register((sorcery), (type), 1, 1, (alloc), (transform), (apply)) /*! + * \brief Set the high and low alert water marks of the sorcery object type. + * \since 13.10.0 + * + * \param sorcery Pointer to a sorcery structure + * \param type Type of object + * \param low_water New queue low water mark. (-1 to set as 90% of high_water) + * \param high_water New queue high water mark. + * + * \retval 0 on success. + * \retval -1 on error (water marks not changed). + */ +int ast_sorcery_object_set_congestion_levels(struct ast_sorcery *sorcery, const char *type, long low_water, long high_water); + +/*! * \brief Set the copy handler for an object type * * \param sorcery Pointer to a sorcery structure diff --git a/include/asterisk/stasis.h b/include/asterisk/stasis.h index 4fc295bc4..de44206b6 100644 --- a/include/asterisk/stasis.h +++ b/include/asterisk/stasis.h @@ -601,6 +601,20 @@ struct stasis_subscription *stasis_unsubscribe( struct stasis_subscription *subscription); /*! + * \brief Set the high and low alert water marks of the stasis subscription. + * \since 13.10.0 + * + * \param subscription Pointer to a stasis subscription + * \param low_water New queue low water mark. (-1 to set as 90% of high_water) + * \param high_water New queue high water mark. + * + * \retval 0 on success. + * \retval -1 on error (water marks not changed). + */ +int stasis_subscription_set_congestion_limits(struct stasis_subscription *subscription, + long low_water, long high_water); + +/*! * \brief Block until the last message is processed on a subscription. * * This function will not return until the \a subscription's callback for the diff --git a/include/asterisk/stasis_message_router.h b/include/asterisk/stasis_message_router.h index 89657a5ee..50270a788 100644 --- a/include/asterisk/stasis_message_router.h +++ b/include/asterisk/stasis_message_router.h @@ -127,6 +127,20 @@ void stasis_message_router_publish_sync(struct stasis_message_router *router, struct stasis_message *message); /*! + * \brief Set the high and low alert water marks of the stasis message router. + * \since 13.10.0 + * + * \param router Pointer to a stasis message router + * \param low_water New queue low water mark. (-1 to set as 90% of high_water) + * \param high_water New queue high water mark. + * + * \retval 0 on success. + * \retval -1 on error (water marks not changed). + */ +int stasis_message_router_set_congestion_limits(struct stasis_message_router *router, + long low_water, long high_water); + +/*! * \brief Add a route to a message router. * * A particular \a message_type may have at most one route per \a router. If diff --git a/include/asterisk/taskprocessor.h b/include/asterisk/taskprocessor.h index af3ce747f..e51122269 100644 --- a/include/asterisk/taskprocessor.h +++ b/include/asterisk/taskprocessor.h @@ -59,6 +59,7 @@ struct ast_taskprocessor; /*! \brief Suggested maximum taskprocessor name length (less null terminator). */ #define AST_TASKPROCESSOR_MAX_NAME 45 +/*! Default taskprocessor high water level alert trigger */ #define AST_TASKPROCESSOR_HIGH_WATER_LEVEL 500 /*! @@ -297,4 +298,26 @@ const char *ast_taskprocessor_name(struct ast_taskprocessor *tps); */ long ast_taskprocessor_size(struct ast_taskprocessor *tps); +/*! + * \brief Get the current taskprocessor high water alert count. + * \since 13.10.0 + * + * \retval 0 if no taskprocessors are in high water alert. + * \retval non-zero if some task processors are in high water alert. + */ +unsigned int ast_taskprocessor_alert_get(void); + +/*! + * \brief Set the high and low alert water marks of the given taskprocessor queue. + * \since 13.10.0 + * + * \param tps Taskprocessor to update queue water marks. + * \param low_water New queue low water mark. (-1 to set as 90% of high_water) + * \param high_water New queue high water mark. + * + * \retval 0 on success. + * \retval -1 on error (water marks not changed). + */ +int ast_taskprocessor_alert_set_levels(struct ast_taskprocessor *tps, long low_water, long high_water); + #endif /* __AST_TASKPROCESSOR_H__ */ diff --git a/main/ast_expr2.c b/main/ast_expr2.c index 798e3d3ce..a9e4eff44 100644 --- a/main/ast_expr2.c +++ b/main/ast_expr2.c @@ -93,6 +93,7 @@ #include "asterisk.h" +#include <sys/cdefs.h> #include <sys/types.h> #include <stdio.h> diff --git a/main/ast_expr2.y b/main/ast_expr2.y index 83d3effe3..869dfe9ea 100644 --- a/main/ast_expr2.y +++ b/main/ast_expr2.y @@ -14,6 +14,7 @@ #include "asterisk.h" +#include <sys/cdefs.h> #include <sys/types.h> #include <stdio.h> diff --git a/main/astfd.c b/main/astfd.c index d2cb73a6b..a96471d60 100644 --- a/main/astfd.c +++ b/main/astfd.c @@ -271,7 +271,7 @@ static char *handle_show_fd(struct ast_cli_entry *e, int cmd, struct ast_cli_arg case CLI_GENERATE: return NULL; } - getrlimit(RLIMIT_FSIZE, &rl); + getrlimit(RLIMIT_NOFILE, &rl); if (rl.rlim_cur == RLIM_INFINITY || rl.rlim_max == RLIM_INFINITY) { ast_copy_string(line, "unlimited", sizeof(line)); } else { diff --git a/main/cdr.c b/main/cdr.c index 7795a65fd..ab6530ed3 100644 --- a/main/cdr.c +++ b/main/cdr.c @@ -71,6 +71,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "asterisk/stasis_bridges.h" #include "asterisk/stasis_message_router.h" #include "asterisk/astobj2.h" +#include "asterisk/taskprocessor.h" /*** DOCUMENTATION <configInfo name="cdr" language="en_US"> @@ -4184,6 +4185,8 @@ int ast_cdr_engine_init(void) if (!stasis_router) { return -1; } + stasis_message_router_set_congestion_limits(stasis_router, -1, + 10 * AST_TASKPROCESSOR_HIGH_WATER_LEVEL); if (STASIS_MESSAGE_TYPE_INIT(cdr_sync_message_type)) { return -1; diff --git a/main/cel.c b/main/cel.c index 114e77a46..4abaac7c8 100644 --- a/main/cel.c +++ b/main/cel.c @@ -59,6 +59,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "asterisk/parking.h" #include "asterisk/pickup.h" #include "asterisk/core_local.h" +#include "asterisk/taskprocessor.h" /*** DOCUMENTATION <configInfo name="cel" language="en_US"> @@ -1593,6 +1594,8 @@ static int create_routes(void) if (!cel_state_router) { return -1; } + stasis_message_router_set_congestion_limits(cel_state_router, -1, + 6 * AST_TASKPROCESSOR_HIGH_WATER_LEVEL); ret |= stasis_message_router_add(cel_state_router, stasis_cache_update_type(), diff --git a/main/codec.c b/main/codec.c index 543d4d0bd..c8644fd34 100644 --- a/main/codec.c +++ b/main/codec.c @@ -135,8 +135,8 @@ static char *show_codecs(struct ast_cli_entry *e, int cmd, struct ast_cli_args * "\tIt does not indicate anything about your configuration.\n"); } - ast_cli(a->fd, "%8s %5s %8s %s\n","ID","TYPE","NAME","DESCRIPTION"); - ast_cli(a->fd, "-----------------------------------------------------------------------------------\n"); + ast_cli(a->fd, "%8s %-5s %-12s %-16s %s\n","ID","TYPE","NAME","FORMAT","DESCRIPTION"); + ast_cli(a->fd, "------------------------------------------------------------------------------------------------\n"); ao2_rdlock(codecs); i = ao2_iterator_init(codecs, AO2_ITERATOR_DONTLOCK); @@ -164,10 +164,11 @@ static char *show_codecs(struct ast_cli_entry *e, int cmd, struct ast_cli_args * } } - ast_cli(a->fd, "%8u %5s %8s (%s)\n", + ast_cli(a->fd, "%8u %-5s %-12s %-16s (%s)\n", codec->id, ast_codec_media_type2str(codec->type), codec->name, + S_OR(codec->format_name, "no cached format"), codec->description); } @@ -216,7 +217,8 @@ static char *show_codec(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a return CLI_SUCCESS; } - ast_cli(a->fd, "%11u %s\n", (unsigned int) codec->id, codec->description); + ast_cli(a->fd, "%11u %s (%s)\n", (unsigned int) codec->id, codec->description, + S_OR(codec->format_name, "no format")); ao2_ref(codec, -1); diff --git a/main/codec_builtin.c b/main/codec_builtin.c index 346b47b87..d7d253ab8 100644 --- a/main/codec_builtin.c +++ b/main/codec_builtin.c @@ -774,6 +774,7 @@ static struct ast_codec t140 = { int __res_ ## __LINE__ = 0; \ struct ast_format *__fmt_ ## __LINE__; \ struct ast_codec *__codec_ ## __LINE__; \ + codec.format_name = (codec).name; \ res |= __ast_codec_register(&(codec), NULL); \ __codec_ ## __LINE__ = ast_codec_get((codec).name, (codec).type, (codec).sample_rate); \ __fmt_ ## __LINE__ = __codec_ ## __LINE__ ? ast_format_create(__codec_ ## __LINE__) : NULL; \ @@ -783,14 +784,15 @@ static struct ast_codec t140 = { __res_ ## __LINE__; \ }) -#define CODEC_REGISTER_AND_CACHE_NAMED(format_name, codec) \ +#define CODEC_REGISTER_AND_CACHE_NAMED(fmt_name, codec) \ ({ \ int __res_ ## __LINE__ = 0; \ struct ast_format *__fmt_ ## __LINE__; \ struct ast_codec *__codec_ ## __LINE__; \ + codec.format_name = fmt_name; \ res |= __ast_codec_register(&(codec), NULL); \ __codec_ ## __LINE__ = ast_codec_get((codec).name, (codec).type, (codec).sample_rate); \ - __fmt_ ## __LINE__ = ast_format_create_named((format_name), __codec_ ## __LINE__); \ + __fmt_ ## __LINE__ = ast_format_create_named((fmt_name), __codec_ ## __LINE__); \ res |= ast_format_cache_set(__fmt_ ## __LINE__); \ ao2_ref(__fmt_ ## __LINE__, -1); \ ao2_ref(__codec_ ## __LINE__, -1); \ diff --git a/main/editline/Makefile.in b/main/editline/Makefile.in index 112b68b64..2be4333d5 100644 --- a/main/editline/Makefile.in +++ b/main/editline/Makefile.in @@ -187,7 +187,7 @@ distclean : clean # $(LIB_A) : $(BGCSRCS:.c=.o_a) $(CCSRCS:.c=.o_a) - $(AR) cru $@ $? + $(AR) cr $@ $? $(RANLIB) $@ $(LIB_S) : $(BGCSRCS:.c=.o_s) $(CCSRCS:.c=.o_s) diff --git a/main/manager.c b/main/manager.c index ba261e8e9..bc4804d89 100644 --- a/main/manager.c +++ b/main/manager.c @@ -100,6 +100,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "asterisk/rtp_engine.h" #include "asterisk/format_cache.h" #include "asterisk/translate.h" +#include "asterisk/taskprocessor.h" /*** DOCUMENTATION <manager name="Ping" language="en_US"> @@ -8650,6 +8651,8 @@ static int manager_subscriptions_init(void) if (!stasis_router) { return -1; } + stasis_message_router_set_congestion_limits(stasis_router, -1, + 6 * AST_TASKPROCESSOR_HIGH_WATER_LEVEL); res |= stasis_message_router_set_default(stasis_router, manager_default_msg_cb, NULL); diff --git a/main/sorcery.c b/main/sorcery.c index 3a29cfa58..4dedc4d9b 100644 --- a/main/sorcery.c +++ b/main/sorcery.c @@ -1161,6 +1161,20 @@ int __ast_sorcery_object_register(struct ast_sorcery *sorcery, const char *type, return 0; } +int ast_sorcery_object_set_congestion_levels(struct ast_sorcery *sorcery, const char *type, long low_water, long high_water) +{ + struct ast_sorcery_object_type *object_type; + int res = -1; + + object_type = ao2_find(sorcery->types, type, OBJ_SEARCH_KEY); + if (object_type) { + res = ast_taskprocessor_alert_set_levels(object_type->serializer, + low_water, high_water); + ao2_ref(object_type, -1); + } + return res; +} + void ast_sorcery_object_set_copy_handler(struct ast_sorcery *sorcery, const char *type, sorcery_copy_handler copy) { RAII_VAR(struct ast_sorcery_object_type *, object_type, ao2_find(sorcery->types, type, OBJ_KEY), ao2_cleanup); diff --git a/main/stasis.c b/main/stasis.c index 4fb69033f..bbafb69e1 100644 --- a/main/stasis.c +++ b/main/stasis.c @@ -564,6 +564,18 @@ struct stasis_subscription *stasis_unsubscribe(struct stasis_subscription *sub) return NULL; } +int stasis_subscription_set_congestion_limits(struct stasis_subscription *subscription, + long low_water, long high_water) +{ + int res = -1; + + if (subscription) { + res = ast_taskprocessor_alert_set_levels(subscription->mailbox, + low_water, high_water); + } + return res; +} + void stasis_subscription_join(struct stasis_subscription *subscription) { if (subscription) { diff --git a/main/stasis_message_router.c b/main/stasis_message_router.c index 26df76c53..cf0ac787e 100644 --- a/main/stasis_message_router.c +++ b/main/stasis_message_router.c @@ -289,6 +289,18 @@ void stasis_message_router_publish_sync(struct stasis_message_router *router, ao2_cleanup(router); } +int stasis_message_router_set_congestion_limits(struct stasis_message_router *router, + long low_water, long high_water) +{ + int res = -1; + + if (router) { + res = stasis_subscription_set_congestion_limits(router->subscription, + low_water, high_water); + } + return res; +} + int stasis_message_router_add(struct stasis_message_router *router, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data) diff --git a/main/taskprocessor.c b/main/taskprocessor.c index 1ba0c8a2f..7ce3e4f16 100644 --- a/main/taskprocessor.c +++ b/main/taskprocessor.c @@ -76,6 +76,10 @@ struct ast_taskprocessor { void *local_data; /*! \brief Taskprocessor current queue size */ long tps_queue_size; + /*! \brief Taskprocessor low water clear alert level */ + long tps_queue_low; + /*! \brief Taskprocessor high water alert trigger level */ + long tps_queue_high; /*! \brief Taskprocessor queue */ AST_LIST_HEAD_NOLOCK(tps_queue, tps_task) tps_queue; struct ast_taskprocessor_listener *listener; @@ -85,6 +89,8 @@ struct ast_taskprocessor { unsigned int executing:1; /*! Indicates that a high water warning has been issued on this task processor */ unsigned int high_water_warned:1; + /*! Indicates that a high water alert is active on this taskprocessor */ + unsigned int high_water_alert:1; }; /*! @@ -121,15 +127,9 @@ static int tps_hash_cb(const void *obj, const int flags); /*! \brief The astobj2 compare callback for taskprocessors */ static int tps_cmp_cb(void *obj, void *arg, int flags); -/*! \brief Destroy the taskprocessor when its refcount reaches zero */ -static void tps_taskprocessor_destroy(void *tps); - /*! \brief CLI <example>taskprocessor ping <blah></example> handler function */ static int tps_ping_handler(void *datap); -/*! \brief Remove the front task off the taskprocessor queue */ -static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps); - 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); @@ -472,8 +472,8 @@ static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_arg struct ao2_container *sorted_tps; struct ast_taskprocessor *tps; struct ao2_iterator iter; -#define FMT_HEADERS "%-45s %10s %10s %10s\n" -#define FMT_FIELDS "%-45s %10lu %10lu %10lu\n" +#define FMT_HEADERS "%-45s %10s %10s %10s %10s %10s\n" +#define FMT_FIELDS "%-45s %10lu %10lu %10lu %10lu %10lu\n" switch (cmd) { case CLI_INIT: @@ -498,7 +498,7 @@ static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_arg return CLI_FAILURE; } - ast_cli(a->fd, "\n" FMT_HEADERS, "Processor", "Processed", "In Queue", "Max Depth"); + ast_cli(a->fd, "\n" FMT_HEADERS, "Processor", "Processed", "In Queue", "Max Depth", "Low water", "High water"); tcount = 0; iter = ao2_iterator_init(sorted_tps, AO2_ITERATOR_UNLINK); while ((tps = ao2_iterator_next(&iter))) { @@ -511,7 +511,8 @@ static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_arg maxqsize = 0; processed = 0; } - ast_cli(a->fd, FMT_FIELDS, name, processed, qsize, maxqsize); + ast_cli(a->fd, FMT_FIELDS, name, processed, qsize, maxqsize, + tps->tps_queue_low, tps->tps_queue_high); ast_taskprocessor_unreference(tps); ++tcount; } @@ -539,28 +540,106 @@ static int tps_cmp_cb(void *obj, void *arg, int flags) return !strcasecmp(lhs->name, rhsname) ? CMP_MATCH | CMP_STOP : 0; } +/*! Count of the number of taskprocessors in high water alert. */ +static unsigned int tps_alert_count; + +/*! Access protection for tps_alert_count */ +AST_RWLOCK_DEFINE_STATIC(tps_alert_lock); + +/*! + * \internal + * \brief Add a delta to tps_alert_count with protection. + * \since 13.10.0 + * + * \param tps Taskprocessor updating queue water mark alert trigger. + * \param delta The amount to add to tps_alert_count. + * + * \return Nothing + */ +static void tps_alert_add(struct ast_taskprocessor *tps, int delta) +{ + 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"); + } + ast_rwlock_unlock(&tps_alert_lock); +} + +unsigned int ast_taskprocessor_alert_get(void) +{ + unsigned int count; + + ast_rwlock_rdlock(&tps_alert_lock); + count = tps_alert_count; + ast_rwlock_unlock(&tps_alert_lock); + + return count; +} + +int ast_taskprocessor_alert_set_levels(struct ast_taskprocessor *tps, long low_water, long high_water) +{ + 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; +} + /* destroy the taskprocessor */ -static void tps_taskprocessor_destroy(void *tps) +static void tps_taskprocessor_dtor(void *tps) { struct ast_taskprocessor *t = tps; struct tps_task *task; - if (!tps) { - ast_log(LOG_ERROR, "missing taskprocessor\n"); - return; + while ((task = AST_LIST_REMOVE_HEAD(&t->tps_queue, list))) { + tps_task_free(task); } - ast_debug(1, "destroying taskprocessor '%s'\n", t->name); - /* free it */ + t->tps_queue_size = 0; + + if (t->high_water_alert) { + t->high_water_alert = 0; + tps_alert_add(t, -1); + } + ast_free(t->stats); t->stats = NULL; ast_free((char *) t->name); - if (t->listener) { - ao2_ref(t->listener, -1); - t->listener = NULL; - } - while ((task = AST_LIST_REMOVE_HEAD(&t->tps_queue, list))) { - tps_task_free(task); - } + t->name = NULL; + ao2_cleanup(t->listener); + t->listener = NULL; } /* pop the front task and return it */ @@ -569,7 +648,11 @@ static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps) struct tps_task *task; if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) { - tps->tps_queue_size--; + --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; } @@ -648,19 +731,22 @@ static void *default_listener_pvt_alloc(void) static struct ast_taskprocessor *__allocate_taskprocessor(const char *name, struct ast_taskprocessor_listener *listener) { - RAII_VAR(struct ast_taskprocessor *, p, - ao2_alloc(sizeof(*p), tps_taskprocessor_destroy), ao2_cleanup); + struct ast_taskprocessor *p; + p = ao2_alloc(sizeof(*p), tps_taskprocessor_dtor); if (!p) { ast_log(LOG_WARNING, "failed to create taskprocessor '%s'\n", name); return NULL; } - if (!(p->stats = ast_calloc(1, sizeof(*p->stats)))) { - ast_log(LOG_WARNING, "failed to create taskprocessor stats for '%s'\n", name); - return NULL; - } - if (!(p->name = ast_strdup(name))) { + /* 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; + + p->stats = ast_calloc(1, sizeof(*p->stats)); + p->name = ast_strdup(name); + if (!p->stats || !p->name) { + ao2_ref(p, -1); return NULL; } @@ -675,22 +761,18 @@ static struct ast_taskprocessor *__allocate_taskprocessor(const char *name, stru if (!(ao2_link(tps_singletons, p))) { ast_log(LOG_ERROR, "Failed to add taskprocessor '%s' to container\n", p->name); listener->tps = NULL; - ao2_ref(p, -1); + ao2_ref(p, -2); return NULL; } if (p->listener->callbacks->start(p->listener)) { - ast_log(LOG_ERROR, "Unable to start taskprocessor listener for taskprocessor %s\n", p->name); + ast_log(LOG_ERROR, "Unable to start taskprocessor listener for taskprocessor %s\n", + p->name); ast_taskprocessor_unreference(p); return NULL; } - /* RAII_VAR will decrement the refcount at the end of the function. - * Since we want to pass back a reference to p, we bump the refcount - */ - ao2_ref(p, +1); return p; - } /* Provide a reference to a taskprocessor. Create the taskprocessor if necessary, but don't @@ -799,10 +881,16 @@ static int taskprocessor_push(struct ast_taskprocessor *tps, struct tps_task *t) AST_LIST_INSERT_TAIL(&tps->tps_queue, t, list); previous_size = tps->tps_queue_size++; - if (previous_size >= AST_TASKPROCESSOR_HIGH_WATER_LEVEL && !tps->high_water_warned) { - ast_log(LOG_WARNING, "The '%s' task processor queue reached %d scheduled tasks.\n", - tps->name, previous_size); - tps->high_water_warned = 1; + if (previous_size >= tps->tps_queue_high) { + if (!tps->high_water_warned) { + tps->high_water_warned = 1; + ast_log(LOG_WARNING, "The '%s' task processor queue reached %d scheduled tasks.\n", + tps->name, previous_size); + } + if (!tps->high_water_alert) { + tps->high_water_alert = 1; + tps_alert_add(tps, +1); + } } /* The currently executing task counts as still in queue */ diff --git a/res/ari/resource_channels.c b/res/ari/resource_channels.c index 9e2db9de6..6baac7a4e 100644 --- a/res/ari/resource_channels.c +++ b/res/ari/resource_channels.c @@ -912,6 +912,7 @@ static void ari_channels_handle_originate_with_id(const char *args_endpoint, const char *args_channel_id, const char *args_other_channel_id, const char *args_originator, + const char *args_formats, struct ast_ari_response *response) { char *dialtech; @@ -930,6 +931,7 @@ static void ari_channels_handle_originate_with_id(const char *args_endpoint, }; struct ari_origination *origination; pthread_t thread; + struct ast_format_cap *format_cap = NULL; if ((assignedids.uniqueid && AST_MAX_PUBLIC_UNIQUEID < strlen(assignedids.uniqueid)) || (assignedids.uniqueid2 && AST_MAX_PUBLIC_UNIQUEID < strlen(assignedids.uniqueid2))) { @@ -944,6 +946,12 @@ static void ari_channels_handle_originate_with_id(const char *args_endpoint, return; } + if (!ast_strlen_zero(args_originator) && !ast_strlen_zero(args_formats)) { + ast_ari_response_error(response, 400, "Bad Request", + "Originator and formats can't both be specified"); + return; + } + dialtech = ast_strdupa(args_endpoint); if ((stuff = strchr(dialtech, '/'))) { *stuff++ = '\0'; @@ -1066,7 +1074,41 @@ static void ari_channels_handle_originate_with_id(const char *args_endpoint, } } - if (ast_dial_prerun(dial, other, NULL)) { + if (!ast_strlen_zero(args_formats)) { + char *format_name; + char *formats_copy = ast_strdupa(args_formats); + + if (!(format_cap = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT))) { + ast_ari_response_alloc_failed(response); + ast_dial_destroy(dial); + ast_free(origination); + ast_channel_cleanup(other); + return; + } + + while ((format_name = ast_strip(strsep(&formats_copy, ",")))) { + struct ast_format *fmt = ast_format_cache_get(format_name); + + if (!fmt || ast_format_cap_append(format_cap, fmt, 0)) { + if (!fmt) { + ast_ari_response_error( + response, 400, "Bad Request", + "Provided format (%s) was not found", format_name); + } else { + ast_ari_response_alloc_failed(response); + } + ast_dial_destroy(dial); + ast_free(origination); + ast_channel_cleanup(other); + ao2_ref(format_cap, -1); + ao2_cleanup(fmt); + return; + } + ao2_ref(fmt, -1); + } + } + + if (ast_dial_prerun(dial, other, format_cap)) { ast_ari_response_alloc_failed(response); ast_dial_destroy(dial); ast_free(origination); @@ -1075,6 +1117,7 @@ static void ari_channels_handle_originate_with_id(const char *args_endpoint, } ast_channel_cleanup(other); + ao2_cleanup(format_cap); chan = ast_dial_get_channel(dial, 0); if (!chan) { @@ -1215,6 +1258,7 @@ void ast_ari_channels_originate_with_id(struct ast_variable *headers, args->channel_id, args->other_channel_id, args->originator, + args->formats, response); ast_variables_destroy(variables); } @@ -1251,6 +1295,7 @@ void ast_ari_channels_originate(struct ast_variable *headers, args->channel_id, args->other_channel_id, args->originator, + args->formats, response); ast_variables_destroy(variables); } diff --git a/res/ari/resource_channels.h b/res/ari/resource_channels.h index 4d3ad5f8b..5bb6f7f1e 100644 --- a/res/ari/resource_channels.h +++ b/res/ari/resource_channels.h @@ -78,6 +78,8 @@ struct ast_ari_channels_originate_args { const char *other_channel_id; /*! The unique id of the channel which is originating this one. */ const char *originator; + /*! The format name capability list to use if originator is not specified. Ex. "ulaw,slin16". Format names an be found with "core show codecs". */ + const char *formats; }; /*! * \brief Body parsing function for /channels. @@ -141,6 +143,8 @@ struct ast_ari_channels_originate_with_id_args { const char *other_channel_id; /*! The unique id of the channel which is originating this one. */ const char *originator; + /*! The format name capability list to use if originator is not specified. Ex. "ulaw,slin16". Format names an be found with "core show codecs". */ + const char *formats; }; /*! * \brief Body parsing function for /channels/{channelId}. diff --git a/res/res_ari_channels.c b/res/res_ari_channels.c index d1ae80196..a14a9c8cb 100644 --- a/res/res_ari_channels.c +++ b/res/res_ari_channels.c @@ -157,6 +157,10 @@ int ast_ari_channels_originate_parse_body( if (field) { args->originator = ast_json_string_get(field); } + field = ast_json_object_get(body, "formats"); + if (field) { + args->formats = ast_json_string_get(field); + } return 0; } @@ -217,6 +221,9 @@ static void ast_ari_channels_originate_cb( if (strcmp(i->name, "originator") == 0) { args.originator = (i->value); } else + if (strcmp(i->name, "formats") == 0) { + args.formats = (i->value); + } else {} } /* Look for a JSON request entity */ @@ -377,6 +384,10 @@ int ast_ari_channels_originate_with_id_parse_body( if (field) { args->originator = ast_json_string_get(field); } + field = ast_json_object_get(body, "formats"); + if (field) { + args->formats = ast_json_string_get(field); + } return 0; } @@ -434,6 +445,9 @@ static void ast_ari_channels_originate_with_id_cb( if (strcmp(i->name, "originator") == 0) { args.originator = (i->value); } else + if (strcmp(i->name, "formats") == 0) { + args.formats = (i->value); + } else {} } for (i = path_vars; i; i = i->next) { diff --git a/res/res_hep.c b/res/res_hep.c index 723b27df8..15e779012 100644 --- a/res/res_hep.c +++ b/res/res_hep.c @@ -409,9 +409,21 @@ enum hep_uuid_type hepv3_get_uuid_type(void) { RAII_VAR(struct module_config *, config, ao2_global_obj_ref(global_config), ao2_cleanup); + if (!config) { + /* Well, that's unfortunate. Return something. */ + return HEP_UUID_TYPE_CALL_ID; + } + return config->general->uuid_type; } +int hepv3_is_loaded(void) +{ + RAII_VAR(struct module_config *, config, ao2_global_obj_ref(global_config), ao2_cleanup); + + return (config != NULL) ? 1 : 0; +} + struct hepv3_capture_info *hepv3_create_capture_info(const void *payload, size_t len) { struct hepv3_capture_info *info; diff --git a/res/res_hep.exports.in b/res/res_hep.exports.in index df0f2b4f7..e318ac97f 100644 --- a/res/res_hep.exports.in +++ b/res/res_hep.exports.in @@ -3,6 +3,7 @@ LINKER_SYMBOL_PREFIX*hepv3_send_packet; LINKER_SYMBOL_PREFIX*hepv3_create_capture_info; LINKER_SYMBOL_PREFIX*hepv3_get_uuid_type; + LINKER_SYMBOL_PREFIX*hepv3_is_loaded; local: *; }; diff --git a/res/res_hep_pjsip.c b/res/res_hep_pjsip.c index 936db9300..8f5baa2cb 100644 --- a/res/res_hep_pjsip.c +++ b/res/res_hep_pjsip.c @@ -210,6 +210,11 @@ static int load_module(void) { CHECK_PJSIP_MODULE_LOADED(); + if (!ast_module_check("res_hep.so") || !hepv3_is_loaded()) { + ast_log(AST_LOG_WARNING, "res_hep is not loaded or running; declining module load\n"); + return AST_MODULE_LOAD_DECLINE; + } + ast_sip_register_service(&logging_module); return AST_MODULE_LOAD_SUCCESS; } diff --git a/res/res_hep_rtcp.c b/res/res_hep_rtcp.c index 49a92539f..d77b19c92 100644 --- a/res/res_hep_rtcp.c +++ b/res/res_hep_rtcp.c @@ -149,6 +149,10 @@ static void rtp_topic_handler(void *data, struct stasis_subscription *sub, struc static int load_module(void) { + if (!ast_module_check("res_hep.so") || !hepv3_is_loaded()) { + ast_log(AST_LOG_WARNING, "res_hep is not loaded or running; declining module load\n"); + return AST_MODULE_LOAD_DECLINE; + } stasis_rtp_subscription = stasis_subscribe(ast_rtp_topic(), rtp_topic_handler, NULL); diff --git a/res/res_odbc.c b/res/res_odbc.c index 17b7a76c8..b2204ff09 100644 --- a/res/res_odbc.c +++ b/res/res_odbc.c @@ -78,10 +78,19 @@ struct odbc_class unsigned int forcecommit:1; /*!< Should uncommitted transactions be auto-committed on handle release? */ unsigned int isolation; /*!< Flags for how the DB should deal with data in other, uncommitted transactions */ unsigned int conntimeout; /*!< Maximum time the connection process should take */ + unsigned int maxconnections; /*!< Maximum number of allowed connections */ /*! When a connection fails, cache that failure for how long? */ struct timeval negative_connection_cache; /*! When a connection fails, when did that last occur? */ struct timeval last_negative_connect; + /*! A pool of available connections */ + AST_LIST_HEAD_NOLOCK(, odbc_obj) connections; + /*! Lock to protect the connections */ + ast_mutex_t lock; + /*! Condition to notify any pending connection requesters */ + ast_cond_t cond; + /*! The total number of current connections */ + size_t connection_cnt; }; static struct ao2_container *class_container; @@ -90,7 +99,7 @@ static AST_RWLIST_HEAD_STATIC(odbc_tables, odbc_cache_tables); static odbc_status odbc_obj_connect(struct odbc_obj *obj); static odbc_status odbc_obj_disconnect(struct odbc_obj *obj); -static int odbc_register_class(struct odbc_class *class, int connect); +static void odbc_register_class(struct odbc_class *class, int connect); AST_THREADSTORAGE(errors_buf); @@ -157,6 +166,8 @@ int ast_odbc_text2isolation(const char *txt) static void odbc_class_destructor(void *data) { struct odbc_class *class = data; + struct odbc_obj *obj; + /* Due to refcounts, we can safely assume that any objects with a reference * to us will prevent our destruction, so we don't need to worry about them. */ @@ -169,7 +180,14 @@ static void odbc_class_destructor(void *data) if (class->sanitysql) { ast_free(class->sanitysql); } + + while ((obj = AST_LIST_REMOVE_HEAD(&class->connections, list))) { + ao2_ref(obj, -1); + } + SQLFreeHandle(SQL_HANDLE_ENV, class->env); + ast_mutex_destroy(&class->lock); + ast_cond_destroy(&class->cond); } static int null_hash_fn(const void *obj, const int flags) @@ -180,21 +198,23 @@ static int null_hash_fn(const void *obj, const int flags) static void odbc_obj_destructor(void *data) { struct odbc_obj *obj = data; - struct odbc_class *class = obj->parent; - obj->parent = NULL; + odbc_obj_disconnect(obj); - ao2_ref(class, -1); } -static void destroy_table_cache(struct odbc_cache_tables *table) { +static void destroy_table_cache(struct odbc_cache_tables *table) +{ struct odbc_cache_columns *col; + ast_debug(1, "Destroying table cache for %s\n", table->table); + AST_RWLIST_WRLOCK(&table->columns); while ((col = AST_RWLIST_REMOVE_HEAD(&table->columns, list))) { ast_free(col); } AST_RWLIST_UNLOCK(&table->columns); AST_RWLIST_HEAD_DESTROY(&table->columns); + ast_free(table); } @@ -370,18 +390,19 @@ SQLHSTMT ast_odbc_prepare_and_execute(struct odbc_obj *obj, SQLHSTMT (*prepare_c * We must therefore redo everything when we establish a new * connection. */ stmt = prepare_cb(obj, data); + if (!stmt) { + return NULL; + } - if (stmt) { - res = SQLExecute(stmt); - if ((res != SQL_SUCCESS) && (res != SQL_SUCCESS_WITH_INFO) && (res != SQL_NO_DATA)) { - if (res == SQL_ERROR) { - ast_odbc_print_errors(SQL_HANDLE_STMT, stmt, "SQL Execute"); - } - - ast_log(LOG_WARNING, "SQL Execute error %d!\n", res); - SQLFreeHandle(SQL_HANDLE_STMT, stmt); - stmt = NULL; + res = SQLExecute(stmt); + if ((res != SQL_SUCCESS) && (res != SQL_SUCCESS_WITH_INFO) && (res != SQL_NO_DATA)) { + if (res == SQL_ERROR) { + ast_odbc_print_errors(SQL_HANDLE_STMT, stmt, "SQL Execute"); } + + ast_log(LOG_WARNING, "SQL Execute error %d!\n", res); + SQLFreeHandle(SQL_HANDLE_STMT, stmt); + stmt = NULL; } return stmt; @@ -468,7 +489,7 @@ static int load_odbc_config(void) struct ast_variable *v; char *cat; const char *dsn, *username, *password, *sanitysql; - int enabled, bse, conntimeout, forcecommit, isolation; + int enabled, bse, conntimeout, forcecommit, isolation, maxconnections; struct timeval ncache = { 0, 0 }; int preconnect = 0, res = 0; struct ast_flags config_flags = { 0 }; @@ -495,6 +516,7 @@ static int load_odbc_config(void) conntimeout = 10; forcecommit = 0; isolation = SQL_TXN_READ_COMMITTED; + maxconnections = 1; for (v = ast_variable_browse(config, cat); v; v = v->next) { if (!strcasecmp(v->name, "pooling") || !strncasecmp(v->name, "share", 5) || @@ -538,6 +560,11 @@ static int load_odbc_config(void) ast_log(LOG_ERROR, "Unrecognized value for 'isolation': '%s' in section '%s'\n", v->value, cat); isolation = SQL_TXN_READ_COMMITTED; } + } else if (!strcasecmp(v->name, "max_connections")) { + if (sscanf(v->value, "%30d", &maxconnections) != 1 || maxconnections < 1) { + ast_log(LOG_WARNING, "max_connections must be a positive integer\n"); + maxconnections = 1; + } } } @@ -563,6 +590,7 @@ static int load_odbc_config(void) new->isolation = isolation; new->conntimeout = conntimeout; new->negative_connection_cache = ncache; + new->maxconnections = maxconnections; if (cat) ast_copy_string(new->name, cat, sizeof(new->name)); @@ -581,6 +609,9 @@ static int load_odbc_config(void) break; } + ast_mutex_init(&new->lock); + ast_cond_init(&new->cond, NULL); + odbc_register_class(new, preconnect); ast_log(LOG_NOTICE, "Registered ODBC class '%s' dsn->[%s]\n", cat, dsn); ao2_ref(new, -1); @@ -641,6 +672,7 @@ static char *handle_cli_odbc_show(struct ast_cli_entry *e, int cmd, struct ast_c ast_strftime(timestr, sizeof(timestr), "%Y-%m-%d %T", &tm); ast_cli(a->fd, " Name: %s\n DSN: %s\n", class->name, class->dsn); ast_cli(a->fd, " Last connection attempt: %s\n", timestr); + ast_cli(a->fd, " Number of active connections: %zd (out of %d)\n", class->connection_cnt, class->maxconnections); ast_cli(a->fd, "\n"); } ao2_ref(class, -1); @@ -654,38 +686,47 @@ static struct ast_cli_entry cli_odbc[] = { AST_CLI_DEFINE(handle_cli_odbc_show, "List ODBC DSN(s)") }; -static int odbc_register_class(struct odbc_class *class, int preconnect) +static void odbc_register_class(struct odbc_class *class, int preconnect) { struct odbc_obj *obj; - if (class) { - ao2_link(class_container, class); - /* I still have a reference in the caller, so a deref is NOT missing here. */ - - if (preconnect) { - /* Request and release builds a connection */ - obj = ast_odbc_request_obj(class->name, 0); - if (obj) { - ast_odbc_release_obj(obj); - } - } - return 0; - } else { - ast_log(LOG_WARNING, "Attempted to register a NULL class?\n"); - return -1; + ao2_link(class_container, class); + /* I still have a reference in the caller, so a deref is NOT missing here. */ + + if (!preconnect) { + return; } + + /* Request and release builds a connection */ + obj = ast_odbc_request_obj(class->name, 0); + if (obj) { + ast_odbc_release_obj(obj); + } + + return; } void ast_odbc_release_obj(struct odbc_obj *obj) { - ast_debug(2, "Releasing ODBC handle %p\n", obj); + struct odbc_class *class = obj->parent; -#ifdef DEBUG_THREADS - obj->file[0] = '\0'; - obj->function[0] = '\0'; - obj->lineno = 0; -#endif - ao2_ref(obj, -1); + ast_debug(2, "Releasing ODBC handle %p into pool\n", obj); + + /* The odbc_obj only holds a reference to the class when it is + * actively being used. This guarantees no circular reference + * between odbc_class and odbc_obj. Since it is being released + * we also release our class reference. If a reload occurred before + * the class will go away automatically once all odbc_obj are + * released back. + */ + obj->parent = NULL; + + ast_mutex_lock(&class->lock); + AST_LIST_INSERT_HEAD(&class->connections, obj, list); + ast_cond_signal(&class->cond); + ast_mutex_unlock(&class->lock); + + ao2_ref(class, -1); } int ast_odbc_backslash_is_escape(struct odbc_obj *obj) @@ -703,6 +744,50 @@ static int aoro2_class_cb(void *obj, void *arg, int flags) return 0; } +/* + * \brief Determine if the connection has died. + * + * \param connection The connection to check + * \param class The ODBC class + * \retval 1 Yep, it's dead + * \retval 0 It's alive and well + */ +static int connection_dead(struct odbc_obj *connection, struct odbc_class *class) +{ + char *test_sql = "select 1"; + SQLINTEGER dead; + SQLRETURN res; + SQLHSTMT stmt; + + res = SQLGetConnectAttr(connection->con, SQL_ATTR_CONNECTION_DEAD, &dead, 0, 0); + if (SQL_SUCCEEDED(res)) { + return dead == SQL_CD_TRUE ? 1 : 0; + } + + /* If the Driver doesn't support SQL_ATTR_CONNECTION_DEAD do a + * probing query instead + */ + res = SQLAllocHandle(SQL_HANDLE_STMT, connection->con, &stmt); + if (!SQL_SUCCEEDED(res)) { + return 1; + } + + if (!ast_strlen_zero(class->sanitysql)) { + test_sql = class->sanitysql; + } + + res = SQLPrepare(stmt, (unsigned char *)test_sql, SQL_NTS); + if (!SQL_SUCCEEDED(res)) { + SQLFreeHandle(SQL_HANDLE_STMT, stmt); + return 1; + } + + res = SQLExecute(stmt); + SQLFreeHandle(SQL_HANDLE_STMT, stmt); + + return SQL_SUCCEEDED(res) ? 0 : 1; +} + struct odbc_obj *_ast_odbc_request_obj2(const char *name, struct ast_flags flags, const char *file, const char *function, int lineno) { struct odbc_obj *obj = NULL; @@ -713,17 +798,60 @@ struct odbc_obj *_ast_odbc_request_obj2(const char *name, struct ast_flags flags return NULL; } - /* XXX ODBC connection objects do not have shared ownership, so there is no reason - * to use refcounted objects here. - */ - obj = ao2_alloc(sizeof(*obj), odbc_obj_destructor); - /* Inherit reference from the ao2_callback from before */ - obj->parent = class; - if (odbc_obj_connect(obj) == ODBC_FAIL) { - ao2_ref(obj, -1); - return NULL; + ast_mutex_lock(&class->lock); + + while (!obj) { + obj = AST_LIST_REMOVE_HEAD(&class->connections, list); + + if (!obj) { + if (class->connection_cnt < class->maxconnections) { + /* If no connection is immediately available establish a new + * one if allowed. If we try and fail we give up completely as + * we could go into an infinite loop otherwise. + */ + obj = ao2_alloc(sizeof(*obj), odbc_obj_destructor); + if (!obj) { + break; + } + + obj->parent = ao2_bump(class); + if (odbc_obj_connect(obj) == ODBC_FAIL) { + ao2_ref(obj->parent, -1); + ao2_ref(obj, -1); + obj = NULL; + break; + } + + class->connection_cnt++; + ast_debug(2, "Created ODBC handle %p on class '%s', new count is %zd\n", obj, + name, class->connection_cnt); + } else { + /* Otherwise if we're not allowed to create a new one we + * wait for another thread to give up the connection they + * own. + */ + ast_cond_wait(&class->cond, &class->lock); + } + } else if (connection_dead(obj, class)) { + /* If the connection is dead try to grab another functional one from the + * pool instead of trying to resurrect this one. + */ + ao2_ref(obj, -1); + obj = NULL; + class->connection_cnt--; + ast_debug(2, "ODBC handle %p dead - removing from class '%s', new count is %zd\n", + obj, name, class->connection_cnt); + } else { + /* We successfully grabbed a connection from the pool and all is well! + */ + obj->parent = ao2_bump(class); + ast_debug(2, "Reusing ODBC handle %p from class '%s'\n", obj, name); + } } + ast_mutex_unlock(&class->lock); + ao2_ref(class, -1); + return obj; } @@ -755,14 +883,6 @@ static odbc_status odbc_obj_disconnect(struct odbc_obj *obj) obj->con = NULL; res = SQLDisconnect(con); - if (obj->parent) { - if (res == SQL_SUCCESS || res == SQL_SUCCESS_WITH_INFO) { - ast_debug(3, "Disconnected %d from %s [%s](%p)\n", res, obj->parent->name, obj->parent->dsn, obj); - } else { - ast_debug(3, "res_odbc: %s [%s](%p) already disconnected\n", obj->parent->name, obj->parent->dsn, obj); - } - } - if ((res = SQLFreeHandle(SQL_HANDLE_DBC, con)) == SQL_SUCCESS) { ast_debug(3, "Database handle %p (connection %p) deallocated\n", obj, con); } else { diff --git a/res/res_pjsip/location.c b/res/res_pjsip/location.c index f55bd0fb4..bf08d8e86 100644 --- a/res/res_pjsip/location.c +++ b/res/res_pjsip/location.c @@ -25,6 +25,7 @@ #include "asterisk/astobj2.h" #include "asterisk/paths.h" #include "asterisk/sorcery.h" +#include "asterisk/taskprocessor.h" #include "include/res_pjsip_private.h" #include "asterisk/res_pjsip_cli.h" #include "asterisk/statsd.h" @@ -1119,6 +1120,8 @@ int ast_sip_initialize_sorcery_location(void) ast_pjproject_get_buildopt("PJSIP_MAX_URL_SIZE", "%d", &pjsip_max_url_size); ast_sorcery_apply_default(sorcery, "contact", "astdb", "registrar"); + ast_sorcery_object_set_congestion_levels(sorcery, "contact", -1, + 3 * AST_TASKPROCESSOR_HIGH_WATER_LEVEL); ast_sorcery_apply_default(sorcery, "aor", "config", "pjsip.conf,criteria=type=aor"); if (ast_sorcery_object_register(sorcery, "contact", contact_alloc, NULL, contact_apply_handler) || diff --git a/res/res_pjsip/pjsip_distributor.c b/res/res_pjsip/pjsip_distributor.c index 3867eaea0..e8ed89361 100644 --- a/res/res_pjsip/pjsip_distributor.c +++ b/res/res_pjsip/pjsip_distributor.c @@ -59,6 +59,12 @@ struct unidentified_request{ char src_name[]; }; +/*! Number of serializers in pool if one not otherwise known. (Best if prime number) */ +#define DISTRIBUTOR_POOL_SIZE 31 + +/*! Pool of serializers to use if not supplied. */ +static struct ast_taskprocessor *distributor_pool[DISTRIBUTOR_POOL_SIZE]; + /*! * \internal * \brief Record the task's serializer name on the tdata structure. @@ -278,6 +284,83 @@ static pjsip_dialog *find_dialog(pjsip_rx_data *rdata) return dlg; } +/*! + * \internal + * \brief Compute a hash value on a pjlib string + * \since 13.10.0 + * + * \param[in] str The pjlib string to add to the hash + * \param[in] hash The hash value to add to + * + * \details + * This version of the function is for when you need to compute a + * string hash of more than one string. + * + * This famous hash algorithm was written by Dan Bernstein and is + * commonly used. + * + * \sa http://www.cse.yorku.ca/~oz/hash.html + */ +static int pjstr_hash_add(pj_str_t *str, int hash) +{ + size_t len; + const char *pos; + + len = pj_strlen(str); + pos = pj_strbuf(str); + while (len--) { + hash = hash * 33 ^ *pos++; + } + + return hash; +} + +/*! + * \internal + * \brief Compute a hash value on a pjlib string + * \since 13.10.0 + * + * \param[in] str The pjlib string to hash + * + * This famous hash algorithm was written by Dan Bernstein and is + * commonly used. + * + * http://www.cse.yorku.ca/~oz/hash.html + */ +static int pjstr_hash(pj_str_t *str) +{ + return pjstr_hash_add(str, 5381); +} + +struct ast_taskprocessor *ast_sip_get_distributor_serializer(pjsip_rx_data *rdata) +{ + int hash; + pj_str_t *remote_tag; + struct ast_taskprocessor *serializer; + + if (!rdata->msg_info.msg) { + return NULL; + } + + if (rdata->msg_info.msg->type == PJSIP_REQUEST_MSG) { + remote_tag = &rdata->msg_info.from->tag; + } else { + remote_tag = &rdata->msg_info.to->tag; + } + + /* Compute the hash from the SIP message call-id and remote-tag */ + hash = pjstr_hash(&rdata->msg_info.cid->id); + hash = pjstr_hash_add(remote_tag, hash); + hash = abs(hash); + + serializer = ao2_bump(distributor_pool[hash % ARRAY_LEN(distributor_pool)]); + if (serializer) { + ast_debug(3, "Calculated serializer %s to use for %s\n", + ast_taskprocessor_name(serializer), pjsip_rx_data_get_info(rdata)); + } + return serializer; +} + static pj_bool_t endpoint_lookup(pjsip_rx_data *rdata); static pjsip_module endpoint_mod = { @@ -286,15 +369,22 @@ static pjsip_module endpoint_mod = { .on_rx_request = endpoint_lookup, }; -#define SIP_MAX_QUEUE (AST_TASKPROCESSOR_HIGH_WATER_LEVEL * 3) - static pj_bool_t distributor(pjsip_rx_data *rdata) { - pjsip_dialog *dlg = find_dialog(rdata); + pjsip_dialog *dlg; struct distributor_dialog_data *dist = NULL; struct ast_taskprocessor *serializer = NULL; pjsip_rx_data *clone; + if (!ast_test_flag(&ast_options, AST_OPT_FLAG_FULLY_BOOTED)) { + /* + * Ignore everything until we are fully booted. Let the + * peer retransmit messages until we are ready. + */ + return PJ_TRUE; + } + + dlg = find_dialog(rdata); if (dlg) { ast_debug(3, "Searching for serializer on dialog %s for %s\n", dlg->obj_name, pjsip_rx_data_get_info(rdata)); @@ -315,12 +405,45 @@ static pj_bool_t distributor(pjsip_rx_data *rdata) ast_debug(3, "No dialog serializer for response %s. Using request transaction as basis\n", pjsip_rx_data_get_info(rdata)); serializer = find_request_serializer(rdata); + if (!serializer) { + if (ast_taskprocessor_alert_get()) { + /* We're overloaded, ignore the unmatched response. */ + ast_debug(3, "Taskprocessor overload alert: Ignoring unmatched '%s'.\n", + pjsip_rx_data_get_info(rdata)); + return PJ_TRUE; + } + + /* + * Pick a serializer for the unmatched response. Maybe + * the stack can figure out what it is for, or we really + * should just toss it regardless. + */ + serializer = ast_sip_get_distributor_serializer(rdata); + } } else if (!pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_cancel_method) || !pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_bye_method)) { /* We have a BYE or CANCEL request without a serializer. */ pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, PJSIP_SC_CALL_TSX_DOES_NOT_EXIST, NULL, NULL, NULL); return PJ_TRUE; + } else { + if (ast_taskprocessor_alert_get()) { + /* + * When taskprocessors get backed up, there is a good chance that + * we are being overloaded and need to defer adding new work to + * the system. To defer the work we will ignore the request and + * rely on the peer's transport layer to retransmit the message. + * We usually work off the overload within a few seconds. The + * alternative is to send back a 503 response to these requests + * and be done with it. + */ + ast_debug(3, "Taskprocessor overload alert: Ignoring '%s'.\n", + pjsip_rx_data_get_info(rdata)); + return PJ_TRUE; + } + + /* Pick a serializer for the out-of-dialog request. */ + serializer = ast_sip_get_distributor_serializer(rdata); } pjsip_rx_data_clone(rdata, 0, &clone); @@ -329,18 +452,9 @@ static pj_bool_t distributor(pjsip_rx_data *rdata) clone->endpt_info.mod_data[endpoint_mod.id] = ao2_bump(dist->endpoint); } - if (ast_sip_threadpool_queue_size() > SIP_MAX_QUEUE) { - /* When the threadpool is backed up this much, there is a good chance that we have encountered - * some sort of terrible condition and don't need to be adding more work to the threadpool. - * It's in our best interest to send back a 503 response and be done with it. - */ - if (rdata->msg_info.msg->type == PJSIP_REQUEST_MSG) { - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 503, NULL, NULL, NULL); - } + if (ast_sip_push_task(serializer, distribute, clone)) { ao2_cleanup(clone->endpt_info.mod_data[endpoint_mod.id]); pjsip_rx_data_free_cloned(clone); - } else { - ast_sip_push_task(serializer, distribute, clone); } ast_taskprocessor_unreference(serializer); @@ -787,6 +901,7 @@ static int cli_unid_print_header(void *obj, void *arg, int flags) return 0; } + static int cli_unid_print_body(void *obj, void *arg, int flags) { struct unidentified_request *unid = obj; @@ -877,6 +992,47 @@ static struct ast_sorcery_observer global_observer = { .loaded = global_loaded, }; +/*! + * \internal + * \brief Shutdown the serializers in the distributor pool. + * \since 13.10.0 + * + * \return Nothing + */ +static void distributor_pool_shutdown(void) +{ + int idx; + + for (idx = 0; idx < ARRAY_LEN(distributor_pool); ++idx) { + ast_taskprocessor_unreference(distributor_pool[idx]); + distributor_pool[idx] = NULL; + } +} + +/*! + * \internal + * \brief Setup the serializers in the distributor pool. + * \since 13.10.0 + * + * \retval 0 on success. + * \retval -1 on error. + */ +static int distributor_pool_setup(void) +{ + char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1]; + int idx; + + for (idx = 0; idx < ARRAY_LEN(distributor_pool); ++idx) { + /* Create name with seq number appended. */ + ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/distributor"); + + distributor_pool[idx] = ast_sip_create_serializer_named(tps_name); + if (!distributor_pool[idx]) { + return -1; + } + } + return 0; +} int ast_sip_initialize_distributor(void) { @@ -886,6 +1042,11 @@ int ast_sip_initialize_distributor(void) return -1; } + if (distributor_pool_setup()) { + ast_sip_destroy_distributor(); + return -1; + } + prune_context = ast_sched_context_create(); if (!prune_context) { ast_sip_destroy_distributor(); @@ -918,8 +1079,10 @@ int ast_sip_initialize_distributor(void) return -1; } - unid_formatter = ao2_alloc(sizeof(struct ast_sip_cli_formatter_entry), NULL); + unid_formatter = ao2_alloc_options(sizeof(struct ast_sip_cli_formatter_entry), NULL, + AO2_ALLOC_OPT_LOCK_NOLOCK); if (!unid_formatter) { + ast_sip_destroy_distributor(); ast_log(LOG_ERROR, "Unable to allocate memory for unid_formatter\n"); return -1; } @@ -931,6 +1094,7 @@ int ast_sip_initialize_distributor(void) unid_formatter->get_id = cli_unid_get_id; unid_formatter->retrieve_by_id = cli_unid_retrieve_by_id; ast_sip_register_cli_formatter(unid_formatter); + ast_cli_register_multiple(cli_commands, ARRAY_LEN(cli_commands)); return 0; @@ -941,17 +1105,20 @@ void ast_sip_destroy_distributor(void) ast_cli_unregister_multiple(cli_commands, ARRAY_LEN(cli_commands)); ast_sip_unregister_cli_formatter(unid_formatter); - internal_sip_unregister_service(&distributor_mod); - internal_sip_unregister_service(&endpoint_mod); internal_sip_unregister_service(&auth_mod); + internal_sip_unregister_service(&endpoint_mod); + internal_sip_unregister_service(&distributor_mod); ao2_cleanup(artificial_auth); ao2_cleanup(artificial_endpoint); - ao2_cleanup(unidentified_requests); ast_sorcery_observer_remove(ast_sip_get_sorcery(), "global", &global_observer); if (prune_context) { ast_sched_context_destroy(prune_context); } + + distributor_pool_shutdown(); + + ao2_cleanup(unidentified_requests); } diff --git a/res/res_pjsip/pjsip_options.c b/res/res_pjsip/pjsip_options.c index d73766cb2..70bbaf908 100644 --- a/res/res_pjsip/pjsip_options.c +++ b/res/res_pjsip/pjsip_options.c @@ -31,6 +31,7 @@ #include "asterisk/test.h" #include "asterisk/statsd.h" #include "include/res_pjsip_private.h" +#include "asterisk/taskprocessor.h" #define DEFAULT_LANGUAGE "en" #define DEFAULT_ENCODING "text/plain" @@ -1015,6 +1016,8 @@ int ast_sip_initialize_sorcery_qualify(void) /* initialize sorcery ast_sip_contact_status resource */ ast_sorcery_apply_default(sorcery, CONTACT_STATUS, "memory", NULL); + ast_sorcery_object_set_congestion_levels(sorcery, CONTACT_STATUS, -1, + 3 * AST_TASKPROCESSOR_HIGH_WATER_LEVEL); if (ast_sorcery_internal_object_register(sorcery, CONTACT_STATUS, contact_status_alloc, NULL, NULL)) { diff --git a/res/res_pjsip_pubsub.c b/res/res_pjsip_pubsub.c index 7ed804acf..79019fb25 100644 --- a/res/res_pjsip_pubsub.c +++ b/res/res_pjsip_pubsub.c @@ -355,7 +355,7 @@ struct ast_sip_publication { struct subscription_persistence { /*! Sorcery object details */ SORCERY_OBJECT(details); - /*! The name of the endpoint involved in the subscrption */ + /*! The name of the endpoint involved in the subscription */ char *endpoint; /*! SIP message that creates the subscription */ char packet[PJSIP_MAX_PKT_LEN]; @@ -1229,10 +1229,9 @@ static void subscription_setup_dialog(struct sip_subscription_tree *sub_tree, pj pjsip_dlg_inc_session(dlg, &pubsub_module); } -static struct sip_subscription_tree *allocate_subscription_tree(struct ast_sip_endpoint *endpoint) +static struct sip_subscription_tree *allocate_subscription_tree(struct ast_sip_endpoint *endpoint, pjsip_rx_data *rdata) { struct sip_subscription_tree *sub_tree; - char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1]; sub_tree = ao2_alloc(sizeof *sub_tree, subscription_tree_destructor); if (!sub_tree) { @@ -1241,11 +1240,24 @@ static struct sip_subscription_tree *allocate_subscription_tree(struct ast_sip_e ast_module_ref(ast_module_info->self); - /* Create name with seq number appended. */ - ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/pubsub/%s", - ast_sorcery_object_get_id(endpoint)); + if (rdata) { + /* + * We must continue using the serializer that the original + * SUBSCRIBE came in on for the dialog. There may be + * retransmissions already enqueued in the original + * serializer that can result in reentrancy and message + * sequencing problems. + */ + sub_tree->serializer = ast_sip_get_distributor_serializer(rdata); + } else { + char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1]; - sub_tree->serializer = ast_sip_create_serializer_named(tps_name); + /* Create name with seq number appended. */ + ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/pubsub/%s", + ast_sorcery_object_get_id(endpoint)); + + sub_tree->serializer = ast_sip_create_serializer_named(tps_name); + } if (!sub_tree->serializer) { ao2_ref(sub_tree, -1); return NULL; @@ -1286,7 +1298,7 @@ static struct sip_subscription_tree *create_subscription_tree(const struct ast_s pjsip_dialog *dlg; struct subscription_persistence *persistence; - sub_tree = allocate_subscription_tree(endpoint); + sub_tree = allocate_subscription_tree(endpoint, rdata); if (!sub_tree) { *dlg_status = PJ_ENOMEM; return NULL; @@ -1335,109 +1347,176 @@ static struct sip_subscription_tree *create_subscription_tree(const struct ast_s static int initial_notify_task(void *obj); static int send_notify(struct sip_subscription_tree *sub_tree, unsigned int force_full_state); -/*! \brief Callback function to perform the actual recreation of a subscription */ -static int subscription_persistence_recreate(void *obj, void *arg, int flags) +/*! Persistent subscription recreation continuation under distributor serializer data */ +struct persistence_recreate_data { + struct subscription_persistence *persistence; + pjsip_rx_data *rdata; +}; + +/*! + * \internal + * \brief subscription_persistence_recreate continuation under distributor serializer. + * \since 13.10.0 + * + * \retval 0 on success. + * \retval -1 on error. + */ +static int sub_persistence_recreate(void *obj) { - struct subscription_persistence *persistence = obj; - pj_pool_t *pool = arg; - pjsip_rx_data rdata = { { 0, }, }; - RAII_VAR(struct ast_sip_endpoint *, endpoint, NULL, ao2_cleanup); + struct persistence_recreate_data *recreate_data = obj; + struct subscription_persistence *persistence = recreate_data->persistence; + pjsip_rx_data *rdata = recreate_data->rdata; + struct ast_sip_endpoint *endpoint; struct sip_subscription_tree *sub_tree; struct ast_sip_pubsub_body_generator *generator; - int resp; + struct ast_sip_subscription_handler *handler; char *resource; - size_t resource_size; pjsip_sip_uri *request_uri; + size_t resource_size; + int resp; struct resource_tree tree; pjsip_expires_hdr *expires_header; - struct ast_sip_subscription_handler *handler; - /* If this subscription has already expired remove it */ - if (ast_tvdiff_ms(persistence->expires, ast_tvnow()) <= 0) { - ast_sorcery_delete(ast_sip_get_sorcery(), persistence); - return 0; - } + request_uri = pjsip_uri_get_uri(rdata->msg_info.msg->line.req.uri); + resource_size = pj_strlen(&request_uri->user) + 1; + resource = ast_alloca(resource_size); + ast_copy_pj_str(resource, &request_uri->user, resource_size); - endpoint = ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(), "endpoint", persistence->endpoint); - if (!endpoint) { - ast_log(LOG_WARNING, "A subscription for '%s' could not be recreated as the endpoint was not found\n", + handler = subscription_get_handler_from_rdata(rdata); + if (!handler || !handler->notifier) { + ast_log(LOG_WARNING, "Failed recreating '%s' subscription: Could not get subscription handler.\n", persistence->endpoint); ast_sorcery_delete(ast_sip_get_sorcery(), persistence); return 0; } - pj_pool_reset(pool); - rdata.tp_info.pool = pool; - - if (ast_sip_create_rdata(&rdata, persistence->packet, persistence->src_name, persistence->src_port, - persistence->transport_key, persistence->local_name, persistence->local_port)) { - ast_log(LOG_WARNING, "A subscription for '%s' could not be recreated as the message could not be parsed\n", + generator = subscription_get_generator_from_rdata(rdata, handler); + if (!generator) { + ast_log(LOG_WARNING, "Failed recreating '%s' subscription: Body generator not available.\n", persistence->endpoint); ast_sorcery_delete(ast_sip_get_sorcery(), persistence); return 0; } - if (rdata.msg_info.msg->type != PJSIP_REQUEST_MSG) { - ast_log(LOG_NOTICE, "Endpoint %s persisted a SIP response instead of a subscribe request. Unable to reload subscription.\n", - ast_sorcery_object_get_id(endpoint)); + ast_sip_mod_data_set(rdata->tp_info.pool, rdata->endpt_info.mod_data, + pubsub_module.id, MOD_DATA_PERSISTENCE, persistence); + + /* Getting the endpoint may take some time that can affect the expiration. */ + endpoint = ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(), "endpoint", + persistence->endpoint); + if (!endpoint) { + ast_log(LOG_WARNING, "Failed recreating '%s' subscription: The endpoint was not found\n", + persistence->endpoint); ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + ao2_ref(endpoint, -1); return 0; } - request_uri = pjsip_uri_get_uri(rdata.msg_info.msg->line.req.uri); - resource_size = pj_strlen(&request_uri->user) + 1; - resource = ast_alloca(resource_size); - ast_copy_pj_str(resource, &request_uri->user, resource_size); - /* Update the expiration header with the new expiration */ - expires_header = pjsip_msg_find_hdr(rdata.msg_info.msg, PJSIP_H_EXPIRES, rdata.msg_info.msg->hdr.next); + expires_header = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_EXPIRES, + rdata->msg_info.msg->hdr.next); if (!expires_header) { - expires_header = pjsip_expires_hdr_create(pool, 0); + expires_header = pjsip_expires_hdr_create(rdata->tp_info.pool, 0); if (!expires_header) { + ast_log(LOG_WARNING, "Failed recreating '%s' subscription: Could not update expires header.\n", + persistence->endpoint); ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + ao2_ref(endpoint, -1); return 0; } - pjsip_msg_add_hdr(rdata.msg_info.msg, (pjsip_hdr*)expires_header); + pjsip_msg_add_hdr(rdata->msg_info.msg, (pjsip_hdr *) expires_header); } expires_header->ivalue = (ast_tvdiff_ms(persistence->expires, ast_tvnow()) / 1000); - - handler = subscription_get_handler_from_rdata(&rdata); - if (!handler || !handler->notifier) { - ast_sorcery_delete(ast_sip_get_sorcery(), persistence); - return 0; - } - - generator = subscription_get_generator_from_rdata(&rdata, handler); - if (!generator) { + if (expires_header->ivalue <= 0) { + /* The subscription expired since we started recreating the subscription. */ ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + ao2_ref(endpoint, -1); return 0; } - ast_sip_mod_data_set(rdata.tp_info.pool, rdata.endpt_info.mod_data, - pubsub_module.id, MOD_DATA_PERSISTENCE, persistence); - memset(&tree, 0, sizeof(tree)); resp = build_resource_tree(endpoint, handler, resource, &tree, - ast_sip_pubsub_has_eventlist_support(&rdata)); + ast_sip_pubsub_has_eventlist_support(rdata)); if (PJSIP_IS_STATUS_IN_CLASS(resp, 200)) { pj_status_t dlg_status; - sub_tree = create_subscription_tree(handler, endpoint, &rdata, resource, generator, &tree, &dlg_status); + sub_tree = create_subscription_tree(handler, endpoint, rdata, resource, generator, + &tree, &dlg_status); if (!sub_tree) { - ast_sorcery_delete(ast_sip_get_sorcery(), persistence); - ast_log(LOG_WARNING, "Failed to re-create subscription for %s\n", persistence->endpoint); - return 0; - } - sub_tree->persistence = ao2_bump(persistence); - subscription_persistence_update(sub_tree, &rdata); - if (ast_sip_push_task(sub_tree->serializer, initial_notify_task, ao2_bump(sub_tree))) { - pjsip_evsub_terminate(sub_tree->evsub, PJ_TRUE); - ao2_ref(sub_tree, -1); + if (dlg_status != PJ_EEXISTS) { + ast_log(LOG_WARNING, "Failed recreating '%s' subscription: Could not create subscription tree.\n", + persistence->endpoint); + ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + } + } else { + sub_tree->persistence = ao2_bump(persistence); + subscription_persistence_update(sub_tree, rdata); + if (ast_sip_push_task(sub_tree->serializer, initial_notify_task, + ao2_bump(sub_tree))) { + /* Could not send initial subscribe NOTIFY */ + pjsip_evsub_terminate(sub_tree->evsub, PJ_TRUE); + ao2_ref(sub_tree, -1); + } } } else { ast_sorcery_delete(ast_sip_get_sorcery(), persistence); } resource_tree_destroy(&tree); + ao2_ref(endpoint, -1); + + return 0; +} + +/*! \brief Callback function to perform the actual recreation of a subscription */ +static int subscription_persistence_recreate(void *obj, void *arg, int flags) +{ + struct subscription_persistence *persistence = obj; + pj_pool_t *pool = arg; + struct ast_taskprocessor *serializer; + pjsip_rx_data rdata; + struct persistence_recreate_data recreate_data; + + /* If this subscription has already expired remove it */ + if (ast_tvdiff_ms(persistence->expires, ast_tvnow()) <= 0) { + ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + return 0; + } + + memset(&rdata, 0, sizeof(rdata)); + pj_pool_reset(pool); + rdata.tp_info.pool = pool; + + if (ast_sip_create_rdata(&rdata, persistence->packet, persistence->src_name, persistence->src_port, + persistence->transport_key, persistence->local_name, persistence->local_port)) { + ast_log(LOG_WARNING, "Failed recreating '%s' subscription: The message could not be parsed\n", + persistence->endpoint); + ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + return 0; + } + + if (rdata.msg_info.msg->type != PJSIP_REQUEST_MSG) { + ast_log(LOG_NOTICE, "Failed recreating '%s' subscription: Stored a SIP response instead of a request.\n", + persistence->endpoint); + ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + return 0; + } + + /* Continue the remainder in the distributor serializer */ + serializer = ast_sip_get_distributor_serializer(&rdata); + if (!serializer) { + ast_log(LOG_WARNING, "Failed recreating '%s' subscription: Could not get distributor serializer.\n", + persistence->endpoint); + ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + return 0; + } + recreate_data.persistence = persistence; + recreate_data.rdata = &rdata; + if (ast_sip_push_task_synchronous(serializer, sub_persistence_recreate, &recreate_data)) { + ast_log(LOG_WARNING, "Failed recreating '%s' subscription: Could not continue under distributor serializer.\n", + persistence->endpoint); + ast_sorcery_delete(ast_sip_get_sorcery(), persistence); + } + ast_taskprocessor_unreference(serializer); return 0; } @@ -1593,7 +1672,7 @@ struct ast_sip_subscription *ast_sip_create_subscription(const struct ast_sip_su pjsip_evsub *evsub; struct sip_subscription_tree *sub_tree = NULL; - sub_tree = allocate_subscription_tree(endpoint); + sub_tree = allocate_subscription_tree(endpoint, NULL); if (!sub_tree) { return NULL; } diff --git a/res/res_pjsip_registrar.c b/res/res_pjsip_registrar.c index 838f4616d..aef0e164b 100644 --- a/res/res_pjsip_registrar.c +++ b/res/res_pjsip_registrar.c @@ -231,155 +231,11 @@ static void registrar_add_date_header(pjsip_tx_data *tdata) ast_sip_add_header(tdata, "Date", date); } -#define SERIALIZER_BUCKETS 59 - -static struct ao2_container *serializers; - -/*! \brief Serializer with associated aor key */ -struct serializer { - /* Serializer to distribute tasks to */ - struct ast_taskprocessor *serializer; - /* The name of the aor to associate with the serializer */ - char aor_name[0]; -}; - -static void serializer_destroy(void *obj) -{ - struct serializer *ser = obj; - - ast_taskprocessor_unreference(ser->serializer); -} - -static struct serializer *serializer_create(const char *aor_name) -{ - char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1]; - size_t size = strlen(aor_name) + 1; - struct serializer *ser = ao2_alloc( - sizeof(*ser) + size, serializer_destroy); - - if (!ser) { - return NULL; - } - - /* Create name with seq number appended. */ - ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/aor/%s", - aor_name); - - if (!(ser->serializer = ast_sip_create_serializer_named(tps_name))) { - ao2_ref(ser, -1); - return NULL; - } - - strcpy(ser->aor_name, aor_name); - return ser; -} - -static struct serializer *serializer_find_or_create(const char *aor_name) -{ - struct serializer *ser = ao2_find(serializers, aor_name, OBJ_SEARCH_KEY); - - if (ser) { - return ser; - } - - if (!(ser = serializer_create(aor_name))) { - return NULL; - } - - ao2_link(serializers, ser); - return ser; -} - -static int serializer_hash(const void *obj, const int flags) -{ - const struct serializer *object; - const char *key; - - switch (flags & OBJ_SEARCH_MASK) { - case OBJ_SEARCH_KEY: - key = obj; - return ast_str_hash(key); - case OBJ_SEARCH_OBJECT: - object = obj; - return ast_str_hash(object->aor_name); - default: - /* Hash can only work on something with a full key. */ - ast_assert(0); - return 0; - } -} - -static int serializer_cmp(void *obj_left, void *obj_right, int flags) -{ - const struct serializer *object_left = obj_left; - const struct serializer *object_right = obj_right; - const char *right_key = obj_right; - int cmp; - - switch (flags & OBJ_SEARCH_MASK) { - case OBJ_SEARCH_OBJECT: - right_key = object_right->aor_name; - /* Fall through */ - case OBJ_SEARCH_KEY: - cmp = strcmp(object_left->aor_name, right_key); - break; - case OBJ_SEARCH_PARTIAL_KEY: - /* - * We could also use a partial key struct containing a length - * so strlen() does not get called for every comparison instead. - */ - cmp = strncmp(object_left->aor_name, right_key, strlen(right_key)); - break; - default: - cmp = 0; - break; - } - - return cmp ? 0 : CMP_MATCH; -} - -struct rx_task_data { - pjsip_rx_data *rdata; - struct ast_sip_endpoint *endpoint; - struct ast_sip_aor *aor; -}; - -static void rx_task_data_destroy(void *obj) -{ - struct rx_task_data *task_data = obj; - - pjsip_rx_data_free_cloned(task_data->rdata); - ao2_cleanup(task_data->endpoint); - ao2_cleanup(task_data->aor); -} - -static struct rx_task_data *rx_task_data_create(pjsip_rx_data *rdata, - struct ast_sip_endpoint *endpoint, - struct ast_sip_aor *aor) -{ - struct rx_task_data *task_data = ao2_alloc( - sizeof(*task_data), rx_task_data_destroy); - - if (!task_data) { - return NULL; - } - - pjsip_rx_data_clone(rdata, 0, &task_data->rdata); - - task_data->endpoint = endpoint; - ao2_ref(task_data->endpoint, +1); - - task_data->aor = aor; - ao2_ref(task_data->aor, +1); - - return task_data; -} - static const pj_str_t path_hdr_name = { "Path", 4 }; -static int build_path_data(struct rx_task_data *task_data, struct ast_str **path_str) +static int build_path_data(pjsip_rx_data *rdata, struct ast_str **path_str) { - pjsip_generic_string_hdr *path_hdr = pjsip_msg_find_hdr_by_name(task_data->rdata->msg_info.msg, &path_hdr_name, NULL); + pjsip_generic_string_hdr *path_hdr = pjsip_msg_find_hdr_by_name(rdata->msg_info.msg, &path_hdr_name, NULL); if (!path_hdr) { return 0; @@ -392,24 +248,24 @@ static int build_path_data(struct rx_task_data *task_data, struct ast_str **path ast_str_set(path_str, 0, "%.*s", (int)path_hdr->hvalue.slen, path_hdr->hvalue.ptr); - while ((path_hdr = (pjsip_generic_string_hdr *) pjsip_msg_find_hdr_by_name(task_data->rdata->msg_info.msg, &path_hdr_name, path_hdr->next))) { + while ((path_hdr = (pjsip_generic_string_hdr *) pjsip_msg_find_hdr_by_name(rdata->msg_info.msg, &path_hdr_name, path_hdr->next))) { ast_str_append(path_str, 0, ",%.*s", (int)path_hdr->hvalue.slen, path_hdr->hvalue.ptr); } return 0; } -static int registrar_validate_path(struct rx_task_data *task_data, struct ast_str **path_str) +static int registrar_validate_path(pjsip_rx_data *rdata, struct ast_sip_aor *aor, struct ast_str **path_str) { const pj_str_t path_supported_name = { "path", 4 }; pjsip_supported_hdr *supported_hdr; int i; - if (!task_data->aor->support_path) { + if (!aor->support_path) { return 0; } - if (build_path_data(task_data, path_str)) { + if (build_path_data(rdata, path_str)) { return -1; } @@ -417,7 +273,7 @@ static int registrar_validate_path(struct rx_task_data *task_data, struct ast_st return 0; } - supported_hdr = pjsip_msg_find_hdr(task_data->rdata->msg_info.msg, PJSIP_H_SUPPORTED, NULL); + supported_hdr = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_SUPPORTED, NULL); if (!supported_hdr) { return -1; } @@ -433,8 +289,11 @@ static int registrar_validate_path(struct rx_task_data *task_data, struct ast_st return -1; } -static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *contacts, - const char *aor_name) +static int register_aor_core(pjsip_rx_data *rdata, + struct ast_sip_endpoint *endpoint, + struct ast_sip_aor *aor, + const char *aor_name, + struct ao2_container *contacts) { static const pj_str_t USER_AGENT = { "User-Agent", 10 }; @@ -458,38 +317,38 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co /* So we don't count static contacts against max_contacts we prune them out from the container */ ao2_callback(contacts, OBJ_NODATA | OBJ_UNLINK | OBJ_MULTIPLE, registrar_prune_static, NULL); - if (registrar_validate_contacts(task_data->rdata, contacts, task_data->aor, &added, &updated, &deleted)) { + if (registrar_validate_contacts(rdata, contacts, aor, &added, &updated, &deleted)) { /* The provided Contact headers do not conform to the specification */ - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), task_data->rdata, 400, NULL, NULL, NULL); - ast_sip_report_failed_acl(task_data->endpoint, task_data->rdata, "registrar_invalid_contacts_provided"); + pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 400, NULL, NULL, NULL); + ast_sip_report_failed_acl(endpoint, rdata, "registrar_invalid_contacts_provided"); ast_log(LOG_WARNING, "Failed to validate contacts in REGISTER request from '%s'\n", - ast_sorcery_object_get_id(task_data->endpoint)); + ast_sorcery_object_get_id(endpoint)); return PJ_TRUE; } - if (registrar_validate_path(task_data, &path_str)) { + if (registrar_validate_path(rdata, aor, &path_str)) { /* Ensure that intervening proxies did not make invalid modifications to the request */ - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), task_data->rdata, 420, NULL, NULL, NULL); + pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 420, NULL, NULL, NULL); ast_log(LOG_WARNING, "Invalid modifications made to REGISTER request from '%s' by intervening proxy\n", - ast_sorcery_object_get_id(task_data->endpoint)); + ast_sorcery_object_get_id(endpoint)); return PJ_TRUE; } - if ((MAX(added - deleted, 0) + (!task_data->aor->remove_existing ? ao2_container_count(contacts) : 0)) > task_data->aor->max_contacts) { + if ((MAX(added - deleted, 0) + (!aor->remove_existing ? ao2_container_count(contacts) : 0)) > aor->max_contacts) { /* Enforce the maximum number of contacts */ - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), task_data->rdata, 403, NULL, NULL, NULL); - ast_sip_report_failed_acl(task_data->endpoint, task_data->rdata, "registrar_attempt_exceeds_maximum_configured_contacts"); + pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 403, NULL, NULL, NULL); + ast_sip_report_failed_acl(endpoint, rdata, "registrar_attempt_exceeds_maximum_configured_contacts"); ast_log(LOG_WARNING, "Registration attempt from endpoint '%s' to AOR '%s' will exceed max contacts of %u\n", - ast_sorcery_object_get_id(task_data->endpoint), ast_sorcery_object_get_id(task_data->aor), task_data->aor->max_contacts); + ast_sorcery_object_get_id(endpoint), aor_name, aor->max_contacts); return PJ_TRUE; } if (!(details.pool = pjsip_endpt_create_pool(ast_sip_get_pjsip_endpoint(), "Contact Comparison", 256, 256))) { - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), task_data->rdata, 500, NULL, NULL, NULL); + pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 500, NULL, NULL, NULL); return PJ_TRUE; } - user_agent_hdr = pjsip_msg_find_hdr_by_name(task_data->rdata->msg_info.msg, &USER_AGENT, NULL); + user_agent_hdr = pjsip_msg_find_hdr_by_name(rdata->msg_info.msg, &USER_AGENT, NULL); if (user_agent_hdr) { alloc_size = pj_strlen(&user_agent_hdr->hvalue) + 1; user_agent = ast_alloca(alloc_size); @@ -497,10 +356,10 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co } /* Find the first Via header */ - via_hdr = via_hdr_last = (pjsip_via_hdr*) pjsip_msg_find_hdr(task_data->rdata->msg_info.msg, PJSIP_H_VIA, NULL); + via_hdr = via_hdr_last = (pjsip_via_hdr*) pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_VIA, NULL); if (via_hdr) { /* Find the last Via header */ - while ( (via_hdr = (pjsip_via_hdr*) pjsip_msg_find_hdr(task_data->rdata->msg_info.msg, + while ( (via_hdr = (pjsip_via_hdr*) pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_VIA, via_hdr->next)) != NULL) { via_hdr_last = via_hdr; } @@ -510,7 +369,7 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co via_port=via_hdr_last->sent_by.port; } - call_id_hdr = (pjsip_cid_hdr*) pjsip_msg_find_hdr(task_data->rdata->msg_info.msg, PJSIP_H_CALL_ID, NULL); + call_id_hdr = (pjsip_cid_hdr*) pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_CALL_ID, NULL); if (call_id_hdr) { alloc_size = pj_strlen(&call_id_hdr->id) + 1; call_id = ast_alloca(alloc_size); @@ -518,7 +377,7 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co } /* Iterate each provided Contact header and add, update, or delete */ - while ((contact_hdr = pjsip_msg_find_hdr(task_data->rdata->msg_info.msg, PJSIP_H_CONTACT, contact_hdr ? contact_hdr->next : NULL))) { + while ((contact_hdr = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_CONTACT, contact_hdr ? contact_hdr->next : NULL))) { int expiration; char contact_uri[pjsip_max_url_size]; RAII_VAR(struct ast_sip_contact *, contact, NULL, ao2_cleanup); @@ -534,7 +393,7 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co continue; } - expiration = registrar_get_expiration(task_data->aor, contact_hdr, task_data->rdata); + expiration = registrar_get_expiration(aor, contact_hdr, rdata); details.uri = pjsip_uri_get_uri(contact_hdr->uri); pjsip_uri_print(PJSIP_URI_IN_CONTACT_HDR, details.uri, contact_uri, sizeof(contact_uri)); @@ -546,9 +405,9 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co continue; } - if (ast_sip_location_add_contact_nolock(task_data->aor, contact_uri, ast_tvadd(ast_tvnow(), + if (ast_sip_location_add_contact_nolock(aor, contact_uri, ast_tvadd(ast_tvnow(), ast_samp2tv(expiration, 1)), path_str ? ast_str_buffer(path_str) : NULL, - user_agent, via_addr, via_port, call_id, task_data->endpoint)) { + user_agent, via_addr, via_port, call_id, endpoint)) { ast_log(LOG_ERROR, "Unable to bind contact '%s' to AOR '%s'\n", contact_uri, aor_name); continue; @@ -576,8 +435,8 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co } contact_update->expiration_time = ast_tvadd(ast_tvnow(), ast_samp2tv(expiration, 1)); - contact_update->qualify_frequency = task_data->aor->qualify_frequency; - contact_update->authenticate_qualify = task_data->aor->authenticate_qualify; + contact_update->qualify_frequency = aor->qualify_frequency; + contact_update->authenticate_qualify = aor->authenticate_qualify; if (path_str) { ast_string_field_set(contact_update, path, ast_str_buffer(path_str)); } @@ -625,16 +484,16 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co /* If the AOR is configured to remove any existing contacts that have not been updated/added as a result of this REGISTER * do so */ - if (task_data->aor->remove_existing) { + if (aor->remove_existing) { ao2_callback(contacts, OBJ_NODATA | OBJ_MULTIPLE, registrar_delete_contact, NULL); } /* Re-retrieve contacts. Caller will clean up the original container. */ - contacts = ast_sip_location_retrieve_aor_contacts_nolock(task_data->aor); + contacts = ast_sip_location_retrieve_aor_contacts_nolock(aor); response_contact = ao2_callback(contacts, 0, NULL, NULL); /* Send a response containing all of the contacts (including static) that are present on this AOR */ - if (ast_sip_create_response(task_data->rdata, 200, response_contact, &tdata) != PJ_SUCCESS) { + if (ast_sip_create_response(rdata, 200, response_contact, &tdata) != PJ_SUCCESS) { ao2_cleanup(response_contact); ao2_cleanup(contacts); return PJ_TRUE; @@ -647,44 +506,42 @@ static int rx_task_core(struct rx_task_data *task_data, struct ao2_container *co ao2_callback(contacts, 0, registrar_add_contact, tdata); ao2_cleanup(contacts); - if ((expires_hdr = pjsip_msg_find_hdr(task_data->rdata->msg_info.msg, PJSIP_H_EXPIRES, NULL))) { - expires_hdr = pjsip_expires_hdr_create(tdata->pool, registrar_get_expiration(task_data->aor, NULL, task_data->rdata)); + if ((expires_hdr = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_EXPIRES, NULL))) { + expires_hdr = pjsip_expires_hdr_create(tdata->pool, registrar_get_expiration(aor, NULL, rdata)); pjsip_msg_add_hdr(tdata->msg, (pjsip_hdr*)expires_hdr); } - ast_sip_send_stateful_response(task_data->rdata, tdata, task_data->endpoint); + ast_sip_send_stateful_response(rdata, tdata, endpoint); return PJ_TRUE; } -static int rx_task(void *data) +static int register_aor(pjsip_rx_data *rdata, + struct ast_sip_endpoint *endpoint, + struct ast_sip_aor *aor, + const char *aor_name) { int res; - struct rx_task_data *task_data = data; struct ao2_container *contacts = NULL; struct ast_named_lock *lock; - const char *aor_name = ast_sorcery_object_get_id(task_data->aor); lock = ast_named_lock_get(AST_NAMED_LOCK_TYPE_RWLOCK, "aor", aor_name); if (!lock) { - ao2_cleanup(task_data); return PJ_TRUE; } ao2_wrlock(lock); - contacts = ast_sip_location_retrieve_aor_contacts_nolock(task_data->aor); + contacts = ast_sip_location_retrieve_aor_contacts_nolock(aor); if (!contacts) { ao2_unlock(lock); ast_named_lock_put(lock); - ao2_cleanup(task_data); return PJ_TRUE; } - res = rx_task_core(task_data, contacts, aor_name); + res = register_aor_core(rdata, endpoint, aor, aor_name, contacts); ao2_cleanup(contacts); ao2_unlock(lock); ast_named_lock_put(lock); - ao2_cleanup(task_data); return res; } @@ -748,44 +605,20 @@ static char *find_aor_name(const char *username, const char *domain, const char return NULL; } -static pj_bool_t registrar_on_rx_request(struct pjsip_rx_data *rdata) +static struct ast_sip_aor *find_registrar_aor(struct pjsip_rx_data *rdata, struct ast_sip_endpoint *endpoint) { - RAII_VAR(struct serializer *, ser, NULL, ao2_cleanup); - struct rx_task_data *task_data; - - RAII_VAR(struct ast_sip_endpoint *, endpoint, - ast_pjsip_rdata_get_endpoint(rdata), ao2_cleanup); - RAII_VAR(struct ast_sip_aor *, aor, NULL, ao2_cleanup); - char *domain_name = NULL; + struct ast_sip_aor *aor = NULL; + char *aor_name = NULL; + char *domain_name; char *username = NULL; - RAII_VAR(char *, aor_name, NULL, ast_free); int i; - if (pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_register_method) || !endpoint) { - return PJ_FALSE; - } - - if (ast_strlen_zero(endpoint->aors)) { - /* Short circuit early if the endpoint has no AORs configured on it, which means no registration possible */ - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 403, NULL, NULL, NULL); - ast_sip_report_failed_acl(endpoint, rdata, "registrar_attempt_without_configured_aors"); - ast_log(LOG_WARNING, "Endpoint '%s' has no configured AORs\n", ast_sorcery_object_get_id(endpoint)); - return PJ_TRUE; - } - - if (!PJSIP_URI_SCHEME_IS_SIP(rdata->msg_info.to->uri) && !PJSIP_URI_SCHEME_IS_SIPS(rdata->msg_info.to->uri)) { - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 416, NULL, NULL, NULL); - ast_sip_report_failed_acl(endpoint, rdata, "registrar_invalid_uri_in_to_received"); - ast_log(LOG_WARNING, "Endpoint '%s' attempted to register to an AOR with a non-SIP URI\n", ast_sorcery_object_get_id(endpoint)); - return PJ_TRUE; - } - - for (i = 0; i < AST_VECTOR_SIZE(&endpoint->ident_method_order); i++) { + for (i = 0; i < AST_VECTOR_SIZE(&endpoint->ident_method_order); ++i) { pjsip_sip_uri *uri; pjsip_authorization_hdr *header = NULL; switch (AST_VECTOR_GET(&endpoint->ident_method_order, i)) { - case AST_SIP_ENDPOINT_IDENTIFY_BY_USERNAME : + case AST_SIP_ENDPOINT_IDENTIFY_BY_USERNAME: uri = pjsip_uri_get_uri(rdata->msg_info.to->uri); domain_name = ast_alloca(uri->host.slen + 1); @@ -798,7 +631,7 @@ static pj_bool_t registrar_on_rx_request(struct pjsip_rx_data *rdata) ast_debug(3, "Matched aor '%s' by To username\n", aor_name); } break; - case AST_SIP_ENDPOINT_IDENTIFY_BY_AUTH_USERNAME : + case AST_SIP_ENDPOINT_IDENTIFY_BY_AUTH_USERNAME: while ((header = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_AUTHORIZATION, header ? header->next : NULL))) { if (header && !pj_stricmp2(&header->scheme, "digest")) { @@ -828,42 +661,57 @@ static pj_bool_t registrar_on_rx_request(struct pjsip_rx_data *rdata) /* The provided AOR name was not found (be it within the configuration or sorcery itself) */ pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 404, NULL, NULL, NULL); ast_sip_report_req_no_support(endpoint, rdata, "registrar_requested_aor_not_found"); - ast_log(LOG_WARNING, "AOR '%s' not found for endpoint '%s'\n", username, ast_sorcery_object_get_id(endpoint)); - return PJ_TRUE; + ast_log(LOG_WARNING, "AOR '%s' not found for endpoint '%s'\n", + username ?: "", ast_sorcery_object_get_id(endpoint)); } + ast_free(aor_name); + return aor; +} - if (!aor->max_contacts) { - /* Registration is not permitted for this AOR */ +static pj_bool_t registrar_on_rx_request(struct pjsip_rx_data *rdata) +{ + RAII_VAR(struct ast_sip_endpoint *, endpoint, + ast_pjsip_rdata_get_endpoint(rdata), ao2_cleanup); + struct ast_sip_aor *aor; + const char *aor_name; + + if (pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_register_method) || !endpoint) { + return PJ_FALSE; + } + + if (ast_strlen_zero(endpoint->aors)) { + /* Short circuit early if the endpoint has no AORs configured on it, which means no registration possible */ pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 403, NULL, NULL, NULL); - ast_sip_report_req_no_support(endpoint, rdata, "registrar_attempt_without_registration_permitted"); - ast_log(LOG_WARNING, "AOR '%s' has no configured max_contacts. Endpoint '%s' unable to register\n", - ast_sorcery_object_get_id(aor), ast_sorcery_object_get_id(endpoint)); + ast_sip_report_failed_acl(endpoint, rdata, "registrar_attempt_without_configured_aors"); + ast_log(LOG_WARNING, "Endpoint '%s' has no configured AORs\n", ast_sorcery_object_get_id(endpoint)); return PJ_TRUE; } - if (!(ser = serializer_find_or_create(aor_name))) { - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 403, NULL, NULL, NULL); - ast_sip_report_mem_limit(endpoint, rdata); - ast_log(LOG_WARNING, "Endpoint '%s' unable to register on AOR '%s' - could not get serializer\n", - ast_sorcery_object_get_id(endpoint), ast_sorcery_object_get_id(aor)); + if (!PJSIP_URI_SCHEME_IS_SIP(rdata->msg_info.to->uri) && !PJSIP_URI_SCHEME_IS_SIPS(rdata->msg_info.to->uri)) { + pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 416, NULL, NULL, NULL); + ast_sip_report_failed_acl(endpoint, rdata, "registrar_invalid_uri_in_to_received"); + ast_log(LOG_WARNING, "Endpoint '%s' attempted to register to an AOR with a non-SIP URI\n", ast_sorcery_object_get_id(endpoint)); return PJ_TRUE; } - if (!(task_data = rx_task_data_create(rdata, endpoint, aor))) { - pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 403, NULL, NULL, NULL); - ast_sip_report_mem_limit(endpoint, rdata); - ast_log(LOG_WARNING, "Endpoint '%s' unable to register on AOR '%s' - could not create rx_task_data\n", - ast_sorcery_object_get_id(endpoint), ast_sorcery_object_get_id(aor)); + aor = find_registrar_aor(rdata, endpoint); + if (!aor) { + /* We've already responded about not finding an AOR. */ return PJ_TRUE; } - if (ast_sip_push_task(ser->serializer, rx_task, task_data)) { + aor_name = ast_sorcery_object_get_id(aor); + + if (!aor->max_contacts) { + /* Registration is not permitted for this AOR */ pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 403, NULL, NULL, NULL); - ast_sip_report_mem_limit(endpoint, rdata); - ast_log(LOG_WARNING, "Endpoint '%s' unable to register on AOR '%s' - could not serialize task\n", - ast_sorcery_object_get_id(endpoint), ast_sorcery_object_get_id(aor)); - ao2_ref(task_data, -1); + ast_sip_report_req_no_support(endpoint, rdata, "registrar_attempt_without_registration_permitted"); + ast_log(LOG_WARNING, "AOR '%s' has no configured max_contacts. Endpoint '%s' unable to register\n", + aor_name, ast_sorcery_object_get_id(endpoint)); + } else { + register_aor(rdata, endpoint, aor, aor_name); } + ao2_ref(aor, -1); return PJ_TRUE; } @@ -952,11 +800,6 @@ static int load_module(void) CHECK_PJSIP_MODULE_LOADED(); - if (!(serializers = ao2_container_alloc( - SERIALIZER_BUCKETS, serializer_hash, serializer_cmp))) { - return AST_MODULE_LOAD_DECLINE; - } - if (ast_sip_register_service(®istrar_module)) { return AST_MODULE_LOAD_DECLINE; } @@ -976,8 +819,6 @@ static int unload_module(void) { ast_manager_unregister(AMI_SHOW_REGISTRATIONS); ast_sip_unregister_service(®istrar_module); - - ao2_cleanup(serializers); return 0; } diff --git a/res/res_pjsip_session.c b/res/res_pjsip_session.c index d66a819d7..a4108d566 100644 --- a/res/res_pjsip_session.c +++ b/res/res_pjsip_session.c @@ -1403,12 +1403,11 @@ struct ast_sip_channel_pvt *ast_sip_channel_pvt_alloc(void *pvt, struct ast_sip_ } struct ast_sip_session *ast_sip_session_alloc(struct ast_sip_endpoint *endpoint, - struct ast_sip_contact *contact, pjsip_inv_session *inv_session) + struct ast_sip_contact *contact, pjsip_inv_session *inv_session, pjsip_rx_data *rdata) { RAII_VAR(struct ast_sip_session *, session, NULL, ao2_cleanup); struct ast_sip_session_supplement *iter; int dsp_features = 0; - char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1]; session = ao2_alloc(sizeof(*session), session_destructor); if (!session) { @@ -1429,11 +1428,24 @@ struct ast_sip_session *ast_sip_session_alloc(struct ast_sip_endpoint *endpoint, /* fill session->media with available types */ ao2_callback(sdp_handlers, OBJ_NODATA, add_session_media, session); - /* Create name with seq number appended. */ - ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/session/%s", - ast_sorcery_object_get_id(endpoint)); + if (rdata) { + /* + * We must continue using the serializer that the original + * INVITE came in on for the dialog. There may be + * retransmissions already enqueued in the original + * serializer that can result in reentrancy and message + * sequencing problems. + */ + session->serializer = ast_sip_get_distributor_serializer(rdata); + } else { + char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1]; - session->serializer = ast_sip_create_serializer_named(tps_name); + /* Create name with seq number appended. */ + ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/outsess/%s", + ast_sorcery_object_get_id(endpoint)); + + session->serializer = ast_sip_create_serializer_named(tps_name); + } if (!session->serializer) { return NULL; } @@ -1731,7 +1743,9 @@ struct ast_sip_session *ast_sip_session_create_outgoing(struct ast_sip_endpoint timer.sess_expires = endpoint->extensions.timer.sess_expires; pjsip_timer_init_session(inv_session, &timer); - if (!(session = ast_sip_session_alloc(endpoint, found_contact ? found_contact : contact, inv_session))) { + session = ast_sip_session_alloc(endpoint, found_contact ? found_contact : contact, + inv_session, NULL); + if (!session) { pjsip_inv_terminate(inv_session, 500, PJ_FALSE); return NULL; } @@ -2142,7 +2156,7 @@ static void handle_new_invite_request(pjsip_rx_data *rdata) return; } - session = ast_sip_session_alloc(endpoint, NULL, inv_session); + session = ast_sip_session_alloc(endpoint, NULL, inv_session, rdata); if (!session) { if (pjsip_inv_initial_answer(inv_session, rdata, 500, NULL, NULL, &tdata) == PJ_SUCCESS) { pjsip_inv_terminate(inv_session, 500, PJ_FALSE); diff --git a/rest-api/api-docs/channels.json b/rest-api/api-docs/channels.json index cb41fb681..8eaa5eb9b 100644 --- a/rest-api/api-docs/channels.json +++ b/rest-api/api-docs/channels.json @@ -128,6 +128,14 @@ "required": false, "allowMultiple": false, "dataType": "string" + }, + { + "name": "formats", + "description": "The format name capability list to use if originator is not specified. Ex. \"ulaw,slin16\". Format names can be found with \"core show codecs\".", + "paramType": "query", + "required": false, + "allowMultiple": false, + "dataType": "string" } ], "errorResponses": [ @@ -276,6 +284,14 @@ "required": false, "allowMultiple": false, "dataType": "string" + }, + { + "name": "formats", + "description": "The format name capability list to use if originator is not specified. Ex. \"ulaw,slin16\". Format names can be found with \"core show codecs\".", + "paramType": "query", + "required": false, + "allowMultiple": false, + "dataType": "string" } ], "errorResponses": [ @@ -284,7 +300,6 @@ "reason": "Invalid parameters for originating a channel." } ] - }, { "httpMethod": "DELETE", diff --git a/tests/test_netsock2.c b/tests/test_netsock2.c index e182b0a75..fec1ae2d5 100644 --- a/tests/test_netsock2.c +++ b/tests/test_netsock2.c @@ -75,7 +75,7 @@ AST_TEST_DEFINE(parsing) }; size_t x; - struct ast_sockaddr addr = { { 0, 0, } }; + struct ast_sockaddr addr; int parse_result; switch (cmd) { @@ -91,15 +91,17 @@ AST_TEST_DEFINE(parsing) } for (x = 0; x < ARRAY_LEN(test_vals); x++) { + memset(&addr, 0, sizeof(addr)); if ((parse_result = ast_sockaddr_parse(&addr, test_vals[x].address, 0)) != test_vals[x].expected_result) { ast_test_status_update(test, "On '%s' expected %d but got %d\n", test_vals[x].address, test_vals[x].expected_result, parse_result); res = AST_TEST_FAIL; } if (parse_result) { - struct ast_sockaddr tmp_addr = { { 0, 0, } }; + struct ast_sockaddr tmp_addr; const char *tmp; tmp = ast_sockaddr_stringify(&addr); + memset(&tmp_addr, 0, sizeof(tmp_addr)); ast_sockaddr_parse(&tmp_addr, tmp, 0); if (ast_sockaddr_cmp_addr(&addr, &tmp_addr)) { char buf[64]; |