From 87550706c8839e29ba13e57c532cdd94fd5039a1 Mon Sep 17 00:00:00 2001
From: George Joseph <gjoseph@digium.com>
Date: Thu, 30 Aug 2018 12:08:05 -0600
Subject: [PATCH] stasis_cache: Prune stasis_subscription_change messages
The stasis cache provides a way to reconstruct the current state
of topic subscribers. Unfortunately, since every subscribe and
unsubscribe is cached, the cache continues to grow unabated while
asterisk is running. This patch removes subscribe messages from
the cache when the corresponding unsubscribe is received.
This patch also registers the cache containers with ao2 so that if
AO2_DEBUG is turned on, you can list the container and get its
stats from the CLI.
ASTERISK-27121
Change-Id: I3d18905e477f3721815da91f30da8d3fbb2d4f56
---
main/stasis_cache.c | 51 +++++++++++++++++++++++++++++++++++++++++++--
1 file changed, 49 insertions(+), 2 deletions(-)
diff --git a/main/stasis_cache.c b/main/stasis_cache.c
index a34e9bebda..3e2ab1722e 100644
--- a/main/stasis_cache.c
+++ b/main/stasis_cache.c
@@ -50,6 +50,7 @@ struct stasis_cache {
snapshot_get_id id_fn;
cache_aggregate_calc_fn aggregate_calc_fn;
cache_aggregate_publish_fn aggregate_publish_fn;
+ int registered;
};
/*! \internal */
@@ -71,6 +72,8 @@ static void stasis_caching_topic_dtor(void *obj)
* be bad. */
ast_assert(stasis_subscription_is_done(caching_topic->sub));
+ ao2_container_unregister(stasis_topic_name(caching_topic->topic));
+
ao2_cleanup(caching_topic->sub);
caching_topic->sub = NULL;
ao2_cleanup(caching_topic->cache);
@@ -815,7 +818,31 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub,
}
msg_type = stasis_message_type(message);
- if (stasis_cache_clear_type() == msg_type) {
+
+ if (stasis_subscription_change_type() == msg_type) {
+ struct stasis_subscription_change *change = stasis_message_data(message);
+
+ /*
+ * If this change type is an unsubscribe, we need to find the original
+ * subscribe and remove it from the cache otherwise the cache will
+ * continue to grow unabated.
+ */
+ if (strcmp(change->description, "Unsubscribe") == 0) {
+ struct stasis_cache_entry *sub;
+
+ ao2_wrlock(caching_topic->cache->entries);
+ sub = cache_find(caching_topic->cache->entries, stasis_subscription_change_type(), change->uniqueid);
+ if (sub) {
+ cache_remove(caching_topic->cache->entries, sub, stasis_message_eid(message));
+ ao2_cleanup(sub);
+ }
+ ao2_unlock(caching_topic->cache->entries);
+ ao2_cleanup(caching_topic_needs_unref);
+ return;
+ }
+ msg_put = message;
+ msg = message;
+ } else if (stasis_cache_clear_type() == msg_type) {
/* Cache clear event. */
msg_put = NULL;
msg = stasis_message_data(message);
@@ -868,6 +895,17 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub,
ao2_cleanup(caching_topic_needs_unref);
}
+static void print_cache_entry(void *v_obj, void *where, ao2_prnt_fn *prnt)
+{
+ struct stasis_cache_entry *entry = v_obj;
+
+ if (!entry) {
+ return;
+ }
+ prnt(where, "Type: %s ID: %s Hash: %u", stasis_message_type_name(entry->key.type),
+ entry->key.id, entry->key.hash);
+}
+
struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
{
struct stasis_caching_topic *caching_topic;
@@ -888,15 +926,24 @@ struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *or
}
caching_topic->topic = stasis_topic_create(new_name);
- ast_free(new_name);
if (caching_topic->topic == NULL) {
ao2_ref(caching_topic, -1);
+ ast_free(new_name);
return NULL;
}
ao2_ref(cache, +1);
caching_topic->cache = cache;
+ if (!cache->registered) {
+ if (ao2_container_register(new_name, cache->entries, print_cache_entry)) {
+ ast_log(LOG_ERROR, "Stasis cache container '%p' for '%s' did not register\n",
+ cache->entries, new_name);
+ } else {
+ cache->registered = 1;
+ }
+ }
+ ast_free(new_name);
caching_topic->sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0, 0);
if (caching_topic->sub == NULL) {
--
GitLab