2013-03-08 15:15:13 +00:00
|
|
|
/*
|
|
|
|
* Asterisk -- An open source telephony toolkit.
|
|
|
|
*
|
|
|
|
* Copyright (C) 2013, Digium, Inc.
|
|
|
|
*
|
|
|
|
* David M. Lee, II <dlee@digium.com>
|
|
|
|
*
|
|
|
|
* See http://www.asterisk.org for more information about
|
|
|
|
* the Asterisk project. Please do not directly contact
|
|
|
|
* any of the maintainers of this project for assistance;
|
|
|
|
* the project provides a web site, mailing lists and IRC
|
|
|
|
* channels for your use.
|
|
|
|
*
|
|
|
|
* This program is free software, distributed under the terms of
|
|
|
|
* the GNU General Public License Version 2. See the LICENSE file
|
|
|
|
* at the top of the source tree.
|
|
|
|
*/
|
|
|
|
|
|
|
|
/*! \file
|
|
|
|
*
|
|
|
|
* \brief Stasis Message API.
|
|
|
|
*
|
|
|
|
* \author David M. Lee, II <dlee@digium.com>
|
|
|
|
*/
|
|
|
|
|
|
|
|
/*** MODULEINFO
|
|
|
|
<support_level>core</support_level>
|
|
|
|
***/
|
|
|
|
|
|
|
|
#include "asterisk.h"
|
|
|
|
|
|
|
|
ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
|
|
|
|
|
|
|
|
#include "asterisk/astobj2.h"
|
|
|
|
#include "asterisk/hashtab.h"
|
2013-06-17 03:00:38 +00:00
|
|
|
#include "asterisk/stasis_internal.h"
|
2013-03-08 15:15:13 +00:00
|
|
|
#include "asterisk/stasis.h"
|
|
|
|
#include "asterisk/utils.h"
|
|
|
|
|
|
|
|
#ifdef LOW_MEMORY
|
|
|
|
#define NUM_CACHE_BUCKETS 17
|
|
|
|
#else
|
|
|
|
#define NUM_CACHE_BUCKETS 563
|
|
|
|
#endif
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
/*! \internal */
|
|
|
|
struct stasis_cache {
|
|
|
|
struct ao2_container *entries;
|
|
|
|
snapshot_get_id id_fn;
|
|
|
|
};
|
|
|
|
|
2013-03-15 17:35:16 +00:00
|
|
|
/*! \internal */
|
2013-03-08 15:15:13 +00:00
|
|
|
struct stasis_caching_topic {
|
2013-08-01 13:49:34 +00:00
|
|
|
struct stasis_cache *cache;
|
2013-03-08 15:15:13 +00:00
|
|
|
struct stasis_topic *topic;
|
2013-06-13 11:02:16 +00:00
|
|
|
struct stasis_topic *original_topic;
|
2013-03-08 15:15:13 +00:00
|
|
|
struct stasis_subscription *sub;
|
|
|
|
};
|
|
|
|
|
|
|
|
static void stasis_caching_topic_dtor(void *obj) {
|
|
|
|
struct stasis_caching_topic *caching_topic = obj;
|
Stasis: address refcount races; implementation comments
Change r395954 reordered some stasis object destruction, which should
have been fine. Unfortunately, it caused some hard to reproduce issues
related to objects being accessed after they had been destroyed. The
patch in r396329 fixed the destruction order problem; this patch
addresses the underlying issue. A few other stasis-related fixes were
also added.
* Add ref-bumps around areas where objects may get transitively
destroyed. (For example, where we lock a topic, unref a subscription,
which unrefs the topic, which explodes the topic when we try to
unlock it.)
* Wrote an extensive doxygen page about Stasis implementation,
relationships between objects, lifecycles of objects, how the
refcounting works, etc. Many other comments were added, corrected, or
cleaned up.
* Added an assert to the topic dtor to catch extra ref decrements.
* Fixed type used after destruction errors for graceful shutdown in
stasis_channels.c.
* I added two unit tests in an attempt to catch destruction order
issues. Since the underlying cause is a race condition, though, the
tests rarely failed even when the code was wrong.
* Fixed a leak in stasis_cache_pattern.c.
(closes issue ASTERISK-22243)
Review: https://reviewboard.asterisk.org/r/2746/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@396842 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-08-16 16:03:34 +00:00
|
|
|
|
|
|
|
/* Caching topics contain subscriptions, and must be manually
|
|
|
|
* unsubscribed. */
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
|
Stasis: address refcount races; implementation comments
Change r395954 reordered some stasis object destruction, which should
have been fine. Unfortunately, it caused some hard to reproduce issues
related to objects being accessed after they had been destroyed. The
patch in r396329 fixed the destruction order problem; this patch
addresses the underlying issue. A few other stasis-related fixes were
also added.
* Add ref-bumps around areas where objects may get transitively
destroyed. (For example, where we lock a topic, unref a subscription,
which unrefs the topic, which explodes the topic when we try to
unlock it.)
* Wrote an extensive doxygen page about Stasis implementation,
relationships between objects, lifecycles of objects, how the
refcounting works, etc. Many other comments were added, corrected, or
cleaned up.
* Added an assert to the topic dtor to catch extra ref decrements.
* Fixed type used after destruction errors for graceful shutdown in
stasis_channels.c.
* I added two unit tests in an attempt to catch destruction order
issues. Since the underlying cause is a race condition, though, the
tests rarely failed even when the code was wrong.
* Fixed a leak in stasis_cache_pattern.c.
(closes issue ASTERISK-22243)
Review: https://reviewboard.asterisk.org/r/2746/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@396842 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-08-16 16:03:34 +00:00
|
|
|
/* If there are any messages in flight to this subscription; that would
|
|
|
|
* be bad. */
|
2013-05-17 21:10:32 +00:00
|
|
|
ast_assert(stasis_subscription_is_done(caching_topic->sub));
|
Stasis: address refcount races; implementation comments
Change r395954 reordered some stasis object destruction, which should
have been fine. Unfortunately, it caused some hard to reproduce issues
related to objects being accessed after they had been destroyed. The
patch in r396329 fixed the destruction order problem; this patch
addresses the underlying issue. A few other stasis-related fixes were
also added.
* Add ref-bumps around areas where objects may get transitively
destroyed. (For example, where we lock a topic, unref a subscription,
which unrefs the topic, which explodes the topic when we try to
unlock it.)
* Wrote an extensive doxygen page about Stasis implementation,
relationships between objects, lifecycles of objects, how the
refcounting works, etc. Many other comments were added, corrected, or
cleaned up.
* Added an assert to the topic dtor to catch extra ref decrements.
* Fixed type used after destruction errors for graceful shutdown in
stasis_channels.c.
* I added two unit tests in an attempt to catch destruction order
issues. Since the underlying cause is a race condition, though, the
tests rarely failed even when the code was wrong.
* Fixed a leak in stasis_cache_pattern.c.
(closes issue ASTERISK-22243)
Review: https://reviewboard.asterisk.org/r/2746/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@396842 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-08-16 16:03:34 +00:00
|
|
|
|
2013-05-17 21:10:32 +00:00
|
|
|
ao2_cleanup(caching_topic->sub);
|
2013-03-08 15:15:13 +00:00
|
|
|
caching_topic->sub = NULL;
|
|
|
|
ao2_cleanup(caching_topic->cache);
|
|
|
|
caching_topic->cache = NULL;
|
|
|
|
ao2_cleanup(caching_topic->topic);
|
|
|
|
caching_topic->topic = NULL;
|
2013-06-13 11:02:16 +00:00
|
|
|
ao2_cleanup(caching_topic->original_topic);
|
|
|
|
caching_topic->original_topic = NULL;
|
2013-03-08 15:15:13 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
|
|
|
|
{
|
|
|
|
return caching_topic->topic;
|
|
|
|
}
|
|
|
|
|
2013-03-15 12:58:23 +00:00
|
|
|
struct stasis_caching_topic *stasis_caching_unsubscribe(struct stasis_caching_topic *caching_topic)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
if (caching_topic) {
|
2013-07-18 14:08:21 +00:00
|
|
|
RAII_VAR(struct stasis_caching_topic *, hold_ref, NULL,
|
|
|
|
ao2_cleanup);
|
|
|
|
|
|
|
|
/* The subscription may hold the last reference to this caching
|
|
|
|
* topic, but we want to make sure the unsubscribe finishes
|
|
|
|
* before kicking of the caching topic's dtor.
|
|
|
|
*/
|
|
|
|
ao2_ref(caching_topic, +1);
|
|
|
|
hold_ref = caching_topic;
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
if (stasis_subscription_is_subscribed(caching_topic->sub)) {
|
2013-05-17 21:10:32 +00:00
|
|
|
/* Increment the reference to hold on to it past the
|
2013-07-18 14:08:21 +00:00
|
|
|
* unsubscribe. Will be cleaned up in dtor. */
|
2013-05-17 21:10:32 +00:00
|
|
|
ao2_ref(caching_topic->sub, +1);
|
2013-03-08 15:15:13 +00:00
|
|
|
stasis_unsubscribe(caching_topic->sub);
|
|
|
|
} else {
|
|
|
|
ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
|
|
|
|
}
|
|
|
|
}
|
2013-03-15 12:58:23 +00:00
|
|
|
return NULL;
|
2013-03-08 15:15:13 +00:00
|
|
|
}
|
|
|
|
|
2013-05-17 21:10:32 +00:00
|
|
|
struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(struct stasis_caching_topic *caching_topic)
|
|
|
|
{
|
|
|
|
if (!caching_topic) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Hold a ref past the unsubscribe */
|
|
|
|
ao2_ref(caching_topic, +1);
|
|
|
|
stasis_caching_unsubscribe(caching_topic);
|
|
|
|
stasis_subscription_join(caching_topic->sub);
|
|
|
|
ao2_cleanup(caching_topic);
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
struct cache_entry {
|
|
|
|
struct stasis_message_type *type;
|
|
|
|
char *id;
|
|
|
|
struct stasis_message *snapshot;
|
|
|
|
};
|
|
|
|
|
|
|
|
static void cache_entry_dtor(void *obj)
|
|
|
|
{
|
|
|
|
struct cache_entry *entry = obj;
|
|
|
|
ao2_cleanup(entry->type);
|
|
|
|
entry->type = NULL;
|
|
|
|
ast_free(entry->id);
|
|
|
|
entry->id = NULL;
|
|
|
|
ao2_cleanup(entry->snapshot);
|
|
|
|
entry->snapshot = NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct cache_entry *, entry, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_assert(type != NULL);
|
|
|
|
ast_assert(id != NULL);
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor,
|
|
|
|
AO2_ALLOC_OPT_LOCK_NOLOCK);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (!entry) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
entry->id = ast_strdup(id);
|
|
|
|
if (!entry->id) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
ao2_ref(type, +1);
|
|
|
|
entry->type = type;
|
|
|
|
if (snapshot != NULL) {
|
|
|
|
ao2_ref(snapshot, +1);
|
|
|
|
entry->snapshot = snapshot;
|
|
|
|
}
|
|
|
|
|
|
|
|
ao2_ref(entry, +1);
|
|
|
|
return entry;
|
|
|
|
}
|
|
|
|
|
|
|
|
static int cache_entry_hash(const void *obj, int flags)
|
|
|
|
{
|
|
|
|
const struct cache_entry *entry = obj;
|
|
|
|
int hash = 0;
|
|
|
|
|
|
|
|
ast_assert(!(flags & OBJ_KEY));
|
|
|
|
|
|
|
|
hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
|
|
|
|
hash += ast_hashtab_hash_string(entry->id);
|
|
|
|
return hash;
|
|
|
|
}
|
|
|
|
|
|
|
|
static int cache_entry_cmp(void *obj, void *arg, int flags)
|
|
|
|
{
|
|
|
|
const struct cache_entry *left = obj;
|
|
|
|
const struct cache_entry *right = arg;
|
|
|
|
|
|
|
|
ast_assert(!(flags & OBJ_KEY));
|
|
|
|
|
|
|
|
if (left->type == right->type && strcmp(left->id, right->id) == 0) {
|
|
|
|
return CMP_MATCH | CMP_STOP;
|
|
|
|
}
|
|
|
|
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
static void cache_dtor(void *obj)
|
|
|
|
{
|
|
|
|
struct stasis_cache *cache = obj;
|
|
|
|
|
|
|
|
ao2_cleanup(cache->entries);
|
|
|
|
cache->entries = NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
|
|
|
|
AO2_ALLOC_OPT_LOCK_NOLOCK);
|
|
|
|
if (!cache) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
cache->entries = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash,
|
|
|
|
cache_entry_cmp);
|
|
|
|
if (!cache->entries) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
cache->id_fn = id_fn;
|
|
|
|
|
|
|
|
ao2_ref(cache, +1);
|
|
|
|
return cache;
|
|
|
|
}
|
|
|
|
|
|
|
|
static struct stasis_message *cache_put(struct stasis_cache *cache,
|
|
|
|
struct stasis_message_type *type, const char *id,
|
|
|
|
struct stasis_message *new_snapshot)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
RAII_VAR(struct cache_entry *, new_entry, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
|
|
|
|
struct stasis_message *old_snapshot = NULL;
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
ast_assert(cache->entries != NULL);
|
|
|
|
ast_assert(new_snapshot == NULL ||
|
|
|
|
type == stasis_message_type(new_snapshot));
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
new_entry = cache_entry_create(type, id, new_snapshot);
|
|
|
|
|
|
|
|
if (new_snapshot == NULL) {
|
|
|
|
/* Remove entry from cache */
|
2013-08-01 13:49:34 +00:00
|
|
|
cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_UNLINK);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (cached_entry) {
|
|
|
|
old_snapshot = cached_entry->snapshot;
|
|
|
|
cached_entry->snapshot = NULL;
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
/* Insert/update cache */
|
2013-08-01 13:49:34 +00:00
|
|
|
SCOPED_AO2LOCK(lock, cache->entries);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_NOLOCK);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (cached_entry) {
|
|
|
|
/* Update cache. Because objects are moving, no need to update refcounts. */
|
|
|
|
old_snapshot = cached_entry->snapshot;
|
|
|
|
cached_entry->snapshot = new_entry->snapshot;
|
|
|
|
new_entry->snapshot = NULL;
|
|
|
|
} else {
|
|
|
|
/* Insert into the cache */
|
2013-08-01 13:49:34 +00:00
|
|
|
ao2_link_flags(cache->entries, new_entry, OBJ_NOLOCK);
|
2013-03-08 15:15:13 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
return old_snapshot;
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
ast_assert(cache->entries != NULL);
|
2013-06-13 11:02:16 +00:00
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
search_entry = cache_entry_create(type, id, NULL);
|
|
|
|
if (search_entry == NULL) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
cached_entry = ao2_find(cache->entries, search_entry, OBJ_POINTER);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (cached_entry == NULL) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
ast_assert(cached_entry->snapshot != NULL);
|
|
|
|
ao2_ref(cached_entry->snapshot, +1);
|
|
|
|
return cached_entry->snapshot;
|
|
|
|
}
|
|
|
|
|
2013-03-08 16:00:14 +00:00
|
|
|
struct cache_dump_data {
|
|
|
|
struct ao2_container *cached;
|
|
|
|
struct stasis_message_type *type;
|
|
|
|
};
|
|
|
|
|
|
|
|
static int cache_dump_cb(void *obj, void *arg, int flags)
|
|
|
|
{
|
|
|
|
struct cache_dump_data *cache_dump = arg;
|
|
|
|
struct cache_entry *entry = obj;
|
|
|
|
|
|
|
|
if (!cache_dump->type || entry->type == cache_dump->type) {
|
|
|
|
ao2_link(cache_dump->cached, entry->snapshot);
|
|
|
|
}
|
|
|
|
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type)
|
2013-03-08 16:00:14 +00:00
|
|
|
{
|
|
|
|
struct cache_dump_data cache_dump;
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
ast_assert(cache->entries != NULL);
|
2013-03-08 16:00:14 +00:00
|
|
|
|
|
|
|
cache_dump.type = type;
|
2013-08-01 13:49:34 +00:00
|
|
|
cache_dump.cached = ao2_container_alloc_options(
|
|
|
|
AO2_ALLOC_OPT_LOCK_NOLOCK, 1, NULL, NULL);
|
2013-03-08 16:00:14 +00:00
|
|
|
if (!cache_dump.cached) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_cb, &cache_dump);
|
2013-03-08 16:00:14 +00:00
|
|
|
return cache_dump.cached;
|
|
|
|
}
|
|
|
|
|
2013-05-15 02:37:22 +00:00
|
|
|
STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
|
|
|
|
STASIS_MESSAGE_TYPE_DEFN(stasis_cache_update_type);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
2013-06-07 12:56:56 +00:00
|
|
|
struct stasis_message *stasis_cache_clear_create(struct stasis_message *id_message)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
|
|
|
|
|
2013-06-07 12:56:56 +00:00
|
|
|
msg = stasis_message_create(stasis_cache_clear_type(), id_message);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (!msg) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
ao2_ref(msg, +1);
|
|
|
|
return msg;
|
|
|
|
}
|
|
|
|
|
|
|
|
static void stasis_cache_update_dtor(void *obj)
|
|
|
|
{
|
|
|
|
struct stasis_cache_update *update = obj;
|
|
|
|
ao2_cleanup(update->topic);
|
|
|
|
update->topic = NULL;
|
|
|
|
ao2_cleanup(update->old_snapshot);
|
|
|
|
update->old_snapshot = NULL;
|
|
|
|
ao2_cleanup(update->new_snapshot);
|
|
|
|
update->new_snapshot = NULL;
|
|
|
|
ao2_cleanup(update->type);
|
|
|
|
update->type = NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
static struct stasis_message *update_create(struct stasis_topic *topic, struct stasis_message *old_snapshot, struct stasis_message *new_snapshot)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_cache_update *, update, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_assert(topic != NULL);
|
|
|
|
ast_assert(old_snapshot != NULL || new_snapshot != NULL);
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
update = ao2_alloc_options(sizeof(*update), stasis_cache_update_dtor,
|
|
|
|
AO2_ALLOC_OPT_LOCK_NOLOCK);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (!update) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
ao2_ref(topic, +1);
|
|
|
|
update->topic = topic;
|
|
|
|
if (old_snapshot) {
|
|
|
|
ao2_ref(old_snapshot, +1);
|
|
|
|
update->old_snapshot = old_snapshot;
|
|
|
|
if (!new_snapshot) {
|
|
|
|
ao2_ref(stasis_message_type(old_snapshot), +1);
|
|
|
|
update->type = stasis_message_type(old_snapshot);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if (new_snapshot) {
|
|
|
|
ao2_ref(new_snapshot, +1);
|
|
|
|
update->new_snapshot = new_snapshot;
|
|
|
|
ao2_ref(stasis_message_type(new_snapshot), +1);
|
|
|
|
update->type = stasis_message_type(new_snapshot);
|
|
|
|
}
|
|
|
|
|
2013-03-28 15:45:18 +00:00
|
|
|
msg = stasis_message_create(stasis_cache_update_type(), update);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (!msg) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
ao2_ref(msg, +1);
|
|
|
|
return msg;
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
static void caching_topic_exec(void *data, struct stasis_subscription *sub,
|
|
|
|
struct stasis_topic *topic, struct stasis_message *message)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
|
|
|
|
struct stasis_caching_topic *caching_topic = data;
|
|
|
|
const char *id = NULL;
|
|
|
|
|
2013-07-18 14:08:21 +00:00
|
|
|
ast_assert(caching_topic != NULL);
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_assert(caching_topic->topic != NULL);
|
2013-08-01 13:49:34 +00:00
|
|
|
ast_assert(caching_topic->cache != NULL);
|
|
|
|
ast_assert(caching_topic->cache->id_fn != NULL);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
if (stasis_subscription_final_message(sub, message)) {
|
|
|
|
caching_topic_needs_unref = caching_topic;
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Handle cache clear event */
|
2013-05-08 13:39:08 +00:00
|
|
|
if (stasis_cache_clear_type() == stasis_message_type(message)) {
|
2013-03-08 15:15:13 +00:00
|
|
|
RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
|
2013-06-07 12:56:56 +00:00
|
|
|
struct stasis_message *clear_msg = stasis_message_data(message);
|
2013-08-01 13:49:34 +00:00
|
|
|
const char *clear_id = caching_topic->cache->id_fn(clear_msg);
|
2013-06-07 12:56:56 +00:00
|
|
|
struct stasis_message_type *clear_type = stasis_message_type(clear_msg);
|
|
|
|
|
|
|
|
ast_assert(clear_type != NULL);
|
|
|
|
|
|
|
|
if (clear_id) {
|
2013-08-01 13:49:34 +00:00
|
|
|
old_snapshot = cache_put(caching_topic->cache, clear_type, clear_id, NULL);
|
2013-06-07 12:56:56 +00:00
|
|
|
if (old_snapshot) {
|
|
|
|
update = update_create(topic, old_snapshot, NULL);
|
|
|
|
stasis_publish(caching_topic->topic, update);
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
ast_log(LOG_ERROR,
|
|
|
|
"Attempting to remove an item from the %s cache that isn't there: %s %s\n",
|
|
|
|
stasis_topic_name(caching_topic->topic), stasis_message_type_name(clear_type), clear_id);
|
|
|
|
return;
|
2013-03-08 15:15:13 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
id = caching_topic->cache->id_fn(message);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (id == NULL) {
|
2013-08-06 14:28:23 +00:00
|
|
|
/* Object isn't cached; discard */
|
2013-03-08 15:15:13 +00:00
|
|
|
} else {
|
|
|
|
/* Update the cache */
|
|
|
|
RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
old_snapshot = cache_put(caching_topic->cache, stasis_message_type(message), id, message);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
update = update_create(topic, old_snapshot, message);
|
|
|
|
if (update == NULL) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
stasis_publish(caching_topic->topic, update);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
|
|
|
|
struct stasis_subscription *sub;
|
|
|
|
RAII_VAR(char *, new_name, NULL, free);
|
|
|
|
int ret;
|
|
|
|
|
|
|
|
ret = asprintf(&new_name, "%s-cached", stasis_topic_name(original_topic));
|
|
|
|
if (ret < 0) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
caching_topic = ao2_alloc_options(sizeof(*caching_topic),
|
|
|
|
stasis_caching_topic_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (caching_topic == NULL) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
caching_topic->topic = stasis_topic_create(new_name);
|
|
|
|
if (caching_topic->topic == NULL) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
ao2_ref(cache, +1);
|
|
|
|
caching_topic->cache = cache;
|
2013-03-08 15:15:13 +00:00
|
|
|
|
2013-06-17 03:00:38 +00:00
|
|
|
sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0);
|
2013-03-08 15:15:13 +00:00
|
|
|
if (sub == NULL) {
|
|
|
|
return NULL;
|
|
|
|
}
|
2013-06-13 11:02:16 +00:00
|
|
|
|
|
|
|
ao2_ref(original_topic, +1);
|
|
|
|
caching_topic->original_topic = original_topic;
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
/* This is for the reference contained in the subscription above */
|
|
|
|
ao2_ref(caching_topic, +1);
|
|
|
|
caching_topic->sub = sub;
|
|
|
|
|
2013-07-18 14:08:21 +00:00
|
|
|
/* The subscription holds the reference, so no additional ref bump. */
|
2013-03-08 15:15:13 +00:00
|
|
|
return caching_topic;
|
|
|
|
}
|
|
|
|
|
Avoid unnecessary cleanups during immediate shutdown
This patch addresses issues during immediate shutdowns, where modules
are not unloaded, but Asterisk atexit handlers are run.
In the typical case, this usually isn't a big deal. But the
introduction of the Stasis message bus makes it much more likely for
asynchronous activity to be happening off in some thread during
shutdown.
During an immediate shutdown, Asterisk skips unloading modules. But
while it is processing the atexit handlers, there is a window of time
where some of the core message types have been cleaned up, but the
message bus is still running. Specifically, it's still running
module subscriptions that might be using the core message types. If a
message is received by that subscription in that window, it will
attempt to use a message type that has been cleaned up.
To solve this problem, this patch introduces ast_register_cleanup().
This function operates identically to ast_register_atexit(), except
that cleanup calls are not invoked on an immediate shutdown. All of
the core message type and topic cleanup was moved from atexit handlers
to cleanup handlers.
This ensures that core type and topic cleanup only happens if the
modules that used them are first unloaded.
This patch also changes the ast_assert() when accessing a cleaned up
or uninitialized message type to an error log message. Message type
functions are actually NULL safe across the board, so the assert was a
bit heavy handed. Especially for anyone with DO_CRASH enabled.
Review: https://reviewboard.asterisk.org/r/2562/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@390122 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-05-30 17:05:53 +00:00
|
|
|
static void stasis_cache_cleanup(void)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
2013-05-15 02:37:22 +00:00
|
|
|
STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
|
|
|
|
STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
|
2013-03-08 15:15:13 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
int stasis_cache_init(void)
|
|
|
|
{
|
Avoid unnecessary cleanups during immediate shutdown
This patch addresses issues during immediate shutdowns, where modules
are not unloaded, but Asterisk atexit handlers are run.
In the typical case, this usually isn't a big deal. But the
introduction of the Stasis message bus makes it much more likely for
asynchronous activity to be happening off in some thread during
shutdown.
During an immediate shutdown, Asterisk skips unloading modules. But
while it is processing the atexit handlers, there is a window of time
where some of the core message types have been cleaned up, but the
message bus is still running. Specifically, it's still running
module subscriptions that might be using the core message types. If a
message is received by that subscription in that window, it will
attempt to use a message type that has been cleaned up.
To solve this problem, this patch introduces ast_register_cleanup().
This function operates identically to ast_register_atexit(), except
that cleanup calls are not invoked on an immediate shutdown. All of
the core message type and topic cleanup was moved from atexit handlers
to cleanup handlers.
This ensures that core type and topic cleanup only happens if the
modules that used them are first unloaded.
This patch also changes the ast_assert() when accessing a cleaned up
or uninitialized message type to an error log message. Message type
functions are actually NULL safe across the board, so the assert was a
bit heavy handed. Especially for anyone with DO_CRASH enabled.
Review: https://reviewboard.asterisk.org/r/2562/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@390122 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-05-30 17:05:53 +00:00
|
|
|
ast_register_cleanup(stasis_cache_cleanup);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
2013-05-15 02:37:22 +00:00
|
|
|
if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_clear_type) != 0) {
|
2013-03-08 15:15:13 +00:00
|
|
|
return -1;
|
|
|
|
}
|
|
|
|
|
2013-05-15 02:37:22 +00:00
|
|
|
if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_update_type) != 0) {
|
2013-03-08 15:15:13 +00:00
|
|
|
return -1;
|
|
|
|
}
|
|
|
|
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|