stasis: Add setting subscription congestion levels.

Stasis subscriptions and message routers create taskprocessors to process
the event messages.  API calls are needed to be able to set the congestion
levels of these taskprocessors for selected subscriptions and message
routers.

* Updated CDR, CEL, and manager's stasis subscription congestion levels
based upon stress testing.  Increased the congestion levels to reduce the
potential for bursty call setup/teardown activity from triggering the
taskprocessor overload alert.  CDRs in particular need an extra high
congestion level because they can take awhile to process the stasis
messages.

ASTERISK-26088
Reported by:  Richard Mudgett

Change-Id: Id0a716394b4eee746dd158acc63d703902450244

5 files changed, 33 insertions, 0 deletions

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 d9fcc5f6b..a26a93918 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">
@@ -1575,6 +1576,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/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/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)