diff options
Diffstat (limited to 'main')
| -rw-r--r-- | main/app.c | 15 | ||||
| -rw-r--r-- | main/devicestate.c | 327 | ||||
| -rw-r--r-- | main/stasis_cache.c | 650 | ||||
| -rw-r--r-- | main/stasis_message.c | 30 |
4 files changed, 744 insertions, 278 deletions
diff --git a/main/app.c b/main/app.c index 6e1a176c6..ce0d16cdd 100644 --- a/main/app.c +++ b/main/app.c @@ -2821,7 +2821,6 @@ struct ast_mwi_state *ast_mwi_create(const char *mailbox, const char *context) return mwi_state; } - int ast_publish_mwi_state_full( const char *mailbox, const char *context, @@ -2857,10 +2856,19 @@ int ast_publish_mwi_state_full( if (eid) { mwi_state->eid = *eid; } else { - ast_set_default_eid(&mwi_state->eid); + mwi_state->eid = ast_eid_default; } - message = stasis_message_create(ast_mwi_state_type(), mwi_state); + /* + * As far as stasis is concerned, all MWI events are internal. + * + * We may in the future want to make MWI aggregate internal/external + * message counts similar to how device state aggregates state. + */ + message = stasis_message_create_full(ast_mwi_state_type(), mwi_state, &ast_eid_default); + if (!message) { + return -1; + } mailbox_specific_topic = ast_mwi_topic(mwi_state->uniqueid); if (!mailbox_specific_topic) { @@ -2911,6 +2919,7 @@ struct stasis_message *ast_mwi_blob_create(struct ast_mwi_state *mwi_state, ao2_ref(obj->mwi_state, +1); obj->blob = ast_json_ref(blob); + /* This is not a normal MWI event. Only used by the MinivmNotify app. */ msg = stasis_message_create(message_type, obj); if (!msg) { return NULL; diff --git a/main/devicestate.c b/main/devicestate.c index 1f0c968aa..5c2340863 100644 --- a/main/devicestate.c +++ b/main/devicestate.c @@ -282,16 +282,20 @@ enum ast_device_state ast_parse_device_state(const char *device) static enum ast_device_state devstate_cached(const char *device) { - RAII_VAR(struct stasis_message *, cached_msg, NULL, ao2_cleanup); + struct stasis_message *cached_msg; struct ast_device_state_message *device_state; + enum ast_device_state state; - cached_msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), device); + cached_msg = stasis_cache_get_by_eid(ast_device_state_cache(), + ast_device_state_message_type(), device, NULL); if (!cached_msg) { return AST_DEVICE_UNKNOWN; } device_state = stasis_message_data(cached_msg); + state = device_state->state; + ao2_cleanup(cached_msg); - return device_state->state; + return state; } /*! \brief Check device state through channel specific function or generic function */ @@ -522,148 +526,62 @@ static void *do_devstate_changes(void *data) return NULL; } -#define MAX_SERVERS 64 -static int devstate_change_aggregator_cb(void *obj, void *arg, void *data, int flags) -{ - struct stasis_message *msg = obj; - struct ast_devstate_aggregate *aggregate = arg; - char *device = data; - struct ast_device_state_message *device_state = stasis_message_data(msg); - - if (!device_state->eid || strcmp(device, device_state->device)) { - /* ignore aggregate states and devices that don't match */ - return 0; - } - ast_debug(1, "Adding per-server state of '%s' for '%s'\n", - ast_devstate2str(device_state->state), device); - ast_devstate_aggregate_add(aggregate, device_state->state); - return 0; -} - -static void device_state_dtor(void *obj) -{ - struct ast_device_state_message *device_state = obj; - ast_string_field_free_memory(device_state); - ast_free(device_state->eid); -} - static struct ast_device_state_message *device_state_alloc(const char *device, enum ast_device_state state, enum ast_devstate_cache cachable, const struct ast_eid *eid) { - RAII_VAR(struct ast_device_state_message *, new_device_state, ao2_alloc(sizeof(*new_device_state), device_state_dtor), ao2_cleanup); + struct ast_device_state_message *new_device_state; + char *pos; + size_t stuff_len; + + ast_assert(!ast_strlen_zero(device)); - if (!new_device_state || ast_string_field_init(new_device_state, 256)) { + stuff_len = strlen(device) + 1; + if (eid) { + stuff_len += sizeof(*eid); + } + new_device_state = ao2_alloc_options(sizeof(*new_device_state) + stuff_len, NULL, + AO2_ALLOC_OPT_LOCK_NOLOCK); + if (!new_device_state) { return NULL; } - ast_string_field_set(new_device_state, device, device); - new_device_state->state = state; - new_device_state->cachable = cachable; - if (eid) { - char eid_str[20]; - struct ast_str *cache_id = ast_str_alloca(256); - - new_device_state->eid = ast_malloc(sizeof(*eid)); - if (!new_device_state->eid) { - return NULL; - } - - *new_device_state->eid = *eid; - ast_eid_to_str(eid_str, sizeof(eid_str), new_device_state->eid); - ast_str_set(&cache_id, 0, "%s%s", eid_str, device); - ast_string_field_set(new_device_state, cache_id, ast_str_buffer(cache_id)); + /* non-aggregate device state. */ + new_device_state->stuff[0] = *eid; + new_device_state->eid = &new_device_state->stuff[0]; + pos = (char *) &new_device_state->stuff[1]; } else { - /* no EID makes this an aggregate state */ - ast_string_field_set(new_device_state, cache_id, device); + pos = (char *) &new_device_state->stuff[0]; } - ao2_ref(new_device_state, +1); - return new_device_state; -} - -static enum ast_device_state get_aggregate_state(char *device) -{ - RAII_VAR(struct ao2_container *, cached, NULL, ao2_cleanup); - struct ast_devstate_aggregate aggregate; - - ast_devstate_aggregate_init(&aggregate); - - cached = stasis_cache_dump(ast_device_state_cache(), NULL); - - ao2_callback_data(cached, OBJ_NODATA, devstate_change_aggregator_cb, &aggregate, device); - - return ast_devstate_aggregate_result(&aggregate); -} - -static int aggregate_state_changed(char *device, enum ast_device_state new_aggregate_state) -{ - RAII_VAR(struct stasis_message *, cached_aggregate_msg, NULL, ao2_cleanup); - struct ast_device_state_message *cached_aggregate_device_state; + strcpy(pos, device);/* Safe */ + new_device_state->device = pos; - cached_aggregate_msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), device); - if (!cached_aggregate_msg) { - return 1; - } + new_device_state->state = state; + new_device_state->cachable = cachable; - cached_aggregate_device_state = stasis_message_data(cached_aggregate_msg); - if (cached_aggregate_device_state->state == new_aggregate_state) { - return 0; - } - return 1; + return new_device_state; } -static void devstate_change_collector_cb(void *data, struct stasis_subscription *sub, struct stasis_message *msg) +static void devstate_change_cb(void *data, struct stasis_subscription *sub, struct stasis_message *msg) { - enum ast_device_state aggregate_state; - char *device; struct ast_device_state_message *device_state; - RAII_VAR(struct stasis_message *, new_aggregate_msg, NULL, ao2_cleanup); - RAII_VAR(struct ast_device_state_message *, new_aggregate_state, NULL, ao2_cleanup); - - if (stasis_cache_update_type() == stasis_message_type(msg)) { - struct stasis_cache_update *update = stasis_message_data(msg); - if (!update->new_snapshot) { - return; - } - msg = update->new_snapshot; - } if (ast_device_state_message_type() != stasis_message_type(msg)) { return; } device_state = stasis_message_data(msg); - - if (!device_state->eid) { - /* ignore aggregate messages */ + if (device_state->cachable == AST_DEVSTATE_CACHABLE || !device_state->eid) { + /* Ignore cacheable and aggregate messages. */ return; } - device = ast_strdupa(device_state->device); - ast_debug(1, "Processing device state change for '%s'\n", device); - - if (device_state->cachable == AST_DEVSTATE_NOT_CACHABLE) { - /* if it's not cachable, there will be no aggregate state to get - * and this should be passed through */ - aggregate_state = device_state->state; - } else { - - aggregate_state = get_aggregate_state(device); - ast_debug(1, "Aggregate devstate result is '%s' for '%s'\n", - ast_devstate2str(aggregate_state), device); - - if (!aggregate_state_changed(device, aggregate_state)) { - /* No change since last reported device state */ - ast_debug(1, "Aggregate state for device '%s' has not changed from '%s'\n", - device, ast_devstate2str(aggregate_state)); - return; - } - } - - ast_debug(1, "Aggregate state for device '%s' has changed to '%s'\n", - device, ast_devstate2str(aggregate_state)); - - ast_publish_device_state_full(device, aggregate_state, device_state->cachable, NULL); + /* + * Non-cacheable device state aggregates are just the + * device state republished as the aggregate. + */ + ast_publish_device_state_full(device_state->device, device_state->state, + device_state->cachable, NULL); } /*! \brief Initialize the device state engine in separate thread */ @@ -738,25 +656,30 @@ struct stasis_topic *ast_device_state_topic(const char *device) int ast_device_state_clear_cache(const char *device) { - RAII_VAR(struct stasis_message *, cached_msg, NULL, ao2_cleanup); - RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup); + struct stasis_message *cached_msg; + struct stasis_message *msg; - if (!(cached_msg = stasis_cache_get(ast_device_state_cache(), - ast_device_state_message_type(), device))) { + cached_msg = stasis_cache_get_by_eid(ast_device_state_cache(), + ast_device_state_message_type(), device, &ast_eid_default); + if (!cached_msg) { /* nothing to clear */ return -1; } msg = stasis_cache_clear_create(cached_msg); - stasis_publish(ast_device_state_topic(device), msg); + if (msg) { + stasis_publish(ast_device_state_topic(device), msg); + } + ao2_cleanup(msg); + ao2_cleanup(cached_msg); return 0; } int ast_publish_device_state_full( - const char *device, - enum ast_device_state state, - enum ast_devstate_cache cachable, - struct ast_eid *eid) + const char *device, + enum ast_device_state state, + enum ast_devstate_cache cachable, + struct ast_eid *eid) { RAII_VAR(struct ast_device_state_message *, device_state, NULL, ao2_cleanup); RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup); @@ -769,7 +692,11 @@ int ast_publish_device_state_full( return -1; } - message = stasis_message_create(ast_device_state_message_type(), device_state); + message = stasis_message_create_full(ast_device_state_message_type(), device_state, + eid); + if (!message) { + return -1; + } device_specific_topic = ast_device_state_topic(device); if (!device_specific_topic) { @@ -783,6 +710,7 @@ int ast_publish_device_state_full( static const char *device_state_get_id(struct stasis_message *message) { struct ast_device_state_message *device_state; + if (ast_device_state_message_type() != stasis_message_type(message)) { return NULL; } @@ -792,20 +720,124 @@ static const char *device_state_get_id(struct stasis_message *message) return NULL; } - return device_state->cache_id; + return device_state->device; +} + +/*! + * \internal + * \brief Callback to publish the aggregate device state cache entry message. + * \since 12.2.0 + * + * \param cache_topic Caching topic the aggregate message may be published over. + * \param aggregate The aggregate shapshot message to publish. + * + * \return Nothing + */ +static void device_state_aggregate_publish(struct stasis_topic *cache_topic, struct stasis_message *aggregate) +{ + const char *device; + struct stasis_topic *device_specific_topic; + + device = device_state_get_id(aggregate); + if (!device) { + return; + } + device_specific_topic = ast_device_state_topic(device); + if (!device_specific_topic) { + return; + } + + stasis_publish(device_specific_topic, aggregate); +} + +/*! + * \internal + * \brief Callback to calculate the aggregate device state cache entry. + * \since 12.2.0 + * + * \param entry Cache entry to calculate a new aggregate snapshot. + * \param new_snapshot The shapshot that is being updated. + * + * \note Return a ref bumped pointer from stasis_cache_entry_get_aggregate() + * if a new aggregate could not be calculated because of error. + * + * \return New aggregate-snapshot calculated on success. + * Caller has a reference on return. + */ +static struct stasis_message *device_state_aggregate_calc(struct stasis_cache_entry *entry, struct stasis_message *new_snapshot) +{ + struct stasis_message *aggregate_snapshot; + struct stasis_message *snapshot; + struct ast_device_state_message *device_state; + const char *device = NULL; + struct ast_devstate_aggregate aggregate; + int idx; + + /* Determine the new aggregate device state. */ + ast_devstate_aggregate_init(&aggregate); + snapshot = stasis_cache_entry_get_local(entry); + if (snapshot) { + device_state = stasis_message_data(snapshot); + device = device_state->device; + ast_devstate_aggregate_add(&aggregate, device_state->state); + } + for (idx = 0; ; ++idx) { + snapshot = stasis_cache_entry_get_remote(entry, idx); + if (!snapshot) { + break; + } + + device_state = stasis_message_data(snapshot); + device = device_state->device; + ast_devstate_aggregate_add(&aggregate, device_state->state); + } + + if (!device) { + /* There are no device states cached. Delete the aggregate. */ + return NULL; + } + + snapshot = stasis_cache_entry_get_aggregate(entry); + if (snapshot) { + device_state = stasis_message_data(snapshot); + if (device_state->state == ast_devstate_aggregate_result(&aggregate)) { + /* Aggregate device state did not change. */ + return ao2_bump(snapshot); + } + } + + device_state = device_state_alloc(device, ast_devstate_aggregate_result(&aggregate), + AST_DEVSTATE_CACHABLE, NULL); + if (!device_state) { + /* Bummer. We have to keep the old aggregate snapshot. */ + return ao2_bump(snapshot); + } + aggregate_snapshot = stasis_message_create_full(ast_device_state_message_type(), + device_state, NULL); + ao2_cleanup(device_state); + if (!aggregate_snapshot) { + /* Bummer. We have to keep the old aggregate snapshot. */ + return ao2_bump(snapshot); + } + + return aggregate_snapshot; } static void devstate_cleanup(void) { devstate_message_sub = stasis_unsubscribe_and_join(devstate_message_sub); - ao2_cleanup(device_state_topic_all); - device_state_topic_all = NULL; + device_state_topic_cached = stasis_caching_unsubscribe_and_join(device_state_topic_cached); + ao2_cleanup(device_state_cache); device_state_cache = NULL; - device_state_topic_cached = stasis_caching_unsubscribe_and_join(device_state_topic_cached); - STASIS_MESSAGE_TYPE_CLEANUP(ast_device_state_message_type); + ao2_cleanup(device_state_topic_pool); device_state_topic_pool = NULL; + + ao2_cleanup(device_state_topic_all); + device_state_topic_all = NULL; + + STASIS_MESSAGE_TYPE_CLEANUP(ast_device_state_message_type); } int devstate_init(void) @@ -817,25 +849,32 @@ int devstate_init(void) } device_state_topic_all = stasis_topic_create("ast_device_state_topic"); if (!device_state_topic_all) { + devstate_cleanup(); return -1; } - device_state_cache = stasis_cache_create(device_state_get_id); - if (!device_state_cache) { + device_state_topic_pool = stasis_topic_pool_create(ast_device_state_topic_all()); + if (!device_state_topic_pool) { + devstate_cleanup(); return -1; } - device_state_topic_cached = stasis_caching_topic_create(device_state_topic_all, device_state_cache); - if (!device_state_topic_cached) { + device_state_cache = stasis_cache_create_full(device_state_get_id, + device_state_aggregate_calc, device_state_aggregate_publish); + if (!device_state_cache) { + devstate_cleanup(); return -1; } - device_state_topic_pool = stasis_topic_pool_create(ast_device_state_topic_all()); - if (!device_state_topic_pool) { + device_state_topic_cached = stasis_caching_topic_create(ast_device_state_topic_all(), + device_state_cache); + if (!device_state_topic_cached) { + devstate_cleanup(); return -1; } - devstate_message_sub = stasis_subscribe(ast_device_state_topic_cached(), devstate_change_collector_cb, NULL); - + devstate_message_sub = stasis_subscribe(ast_device_state_topic_all(), + devstate_change_cb, NULL); if (!devstate_message_sub) { - ast_log(LOG_ERROR, "Failed to create subscription for the device state change collector\n"); + ast_log(LOG_ERROR, "Failed to create subscription creating uncached device state aggregate events.\n"); + devstate_cleanup(); return -1; } diff --git a/main/stasis_cache.c b/main/stasis_cache.c index 1c5053915..22ace60af 100644 --- a/main/stasis_cache.c +++ b/main/stasis_cache.c @@ -36,6 +36,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "asterisk/stasis_internal.h" #include "asterisk/stasis.h" #include "asterisk/utils.h" +#include "asterisk/vector.h" #ifdef LOW_MEMORY #define NUM_CACHE_BUCKETS 17 @@ -47,6 +48,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") struct stasis_cache { struct ao2_container *entries; snapshot_get_id id_fn; + cache_aggregate_calc_fn aggregate_calc_fn; + cache_aggregate_publish_fn aggregate_publish_fn; }; /*! \internal */ @@ -124,29 +127,53 @@ struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(struct stasis_c return NULL; } -struct cache_entry { +struct cache_entry_key { struct stasis_message_type *type; - char *id; - struct stasis_message *snapshot; + const char *id; +}; + +struct stasis_cache_entry { + struct cache_entry_key key; + /*! Aggregate snapshot of the stasis cache. */ + struct stasis_message *aggregate; + /*! Local entity snapshot of the stasis event. */ + struct stasis_message *local; + /*! Remote entity snapshots of the stasis event. */ + AST_VECTOR(, struct stasis_message *) remote; }; 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; + struct stasis_cache_entry *entry = obj; + size_t idx; + + ao2_cleanup(entry->key.type); + entry->key.type = NULL; + ast_free((char *) entry->key.id); + entry->key.id = NULL; + + ao2_cleanup(entry->aggregate); + entry->aggregate = NULL; + ao2_cleanup(entry->local); + entry->local = NULL; + + for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) { + struct stasis_message *remote; + + remote = AST_VECTOR_GET(&entry->remote, idx); + ao2_cleanup(remote); + } + AST_VECTOR_FREE(&entry->remote); } -static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot) +static struct stasis_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); + struct stasis_cache_entry *entry; + int is_remote; ast_assert(type != NULL); ast_assert(id != NULL); + ast_assert(snapshot != NULL); entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK); @@ -154,173 +181,528 @@ static struct cache_entry *cache_entry_create(struct stasis_message_type *type, return NULL; } - entry->id = ast_strdup(id); - if (!entry->id) { + entry->key.id = ast_strdup(id); + if (!entry->key.id) { + ao2_cleanup(entry); + return NULL; + } + entry->key.type = ao2_bump(type); + + is_remote = ast_eid_cmp(&ast_eid_default, stasis_message_eid(snapshot)) ? 1 : 0; + if (AST_VECTOR_INIT(&entry->remote, is_remote)) { + ao2_cleanup(entry); return NULL; } - ao2_ref(type, +1); - entry->type = type; - if (snapshot != NULL) { - ao2_ref(snapshot, +1); - entry->snapshot = snapshot; + if (is_remote) { + if (AST_VECTOR_APPEND(&entry->remote, snapshot)) { + ao2_cleanup(entry); + return NULL; + } + } else { + entry->local = snapshot; } + ao2_bump(snapshot); - ao2_ref(entry, +1); return entry; } static int cache_entry_hash(const void *obj, int flags) { - const struct cache_entry *entry = obj; + const struct stasis_cache_entry *object; + const struct cache_entry_key *key; int hash = 0; - ast_assert(!(flags & OBJ_KEY)); + switch (flags & OBJ_SEARCH_MASK) { + case OBJ_SEARCH_KEY: + key = obj; + break; + case OBJ_SEARCH_OBJECT: + object = obj; + key = &object->key; + break; + default: + /* Hash can only work on something with a full key. */ + ast_assert(0); + return 0; + } - hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type)); - hash += ast_hashtab_hash_string(entry->id); + hash += ast_hashtab_hash_string(stasis_message_type_name(key->type)); + hash += ast_hashtab_hash_string(key->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; + const struct stasis_cache_entry *object_left = obj; + const struct stasis_cache_entry *object_right = arg; + const struct cache_entry_key *right_key = obj; + int cmp; + + switch (flags & OBJ_SEARCH_MASK) { + case OBJ_SEARCH_OBJECT: + right_key = &object_right->key; + /* Fall through */ + case OBJ_SEARCH_KEY: + cmp = object_left->key.type != right_key->type + || strcmp(object_left->key.id, right_key->id); + break; + case OBJ_SEARCH_PARTIAL_KEY: + /* Not supported by container */ + ast_assert(0); + cmp = -1; + break; + default: + /* + * What arg points to is specific to this traversal callback + * and has no special meaning to astobj2. + */ + cmp = 0; + break; + } + if (cmp) { + return 0; + } + /* + * At this point the traversal callback is identical to a sorted + * container. + */ + return CMP_MATCH; +} + +static void cache_dtor(void *obj) +{ + struct stasis_cache *cache = obj; - ast_assert(!(flags & OBJ_KEY)); + ao2_cleanup(cache->entries); + cache->entries = NULL; +} - if (left->type == right->type && strcmp(left->id, right->id) == 0) { - return CMP_MATCH | CMP_STOP; +struct stasis_cache *stasis_cache_create_full(snapshot_get_id id_fn, + cache_aggregate_calc_fn aggregate_calc_fn, + cache_aggregate_publish_fn aggregate_publish_fn) +{ + struct stasis_cache *cache; + + cache = ao2_alloc_options(sizeof(*cache), cache_dtor, + AO2_ALLOC_OPT_LOCK_NOLOCK); + if (!cache) { + return NULL; } - return 0; + cache->entries = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0, + NUM_CACHE_BUCKETS, cache_entry_hash, NULL, cache_entry_cmp); + if (!cache->entries) { + ao2_cleanup(cache); + return NULL; + } + + cache->id_fn = id_fn; + cache->aggregate_calc_fn = aggregate_calc_fn; + cache->aggregate_publish_fn = aggregate_publish_fn; + + return cache; } -static void cache_dtor(void *obj) +struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn) { - struct stasis_cache *cache = obj; + return stasis_cache_create_full(id_fn, NULL, NULL); +} - ao2_cleanup(cache->entries); - cache->entries = NULL; +struct stasis_message *stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry) +{ + return entry->aggregate; } -struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn) +struct stasis_message *stasis_cache_entry_get_local(struct stasis_cache_entry *entry) { - RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup); + return entry->local; +} - cache = ao2_alloc_options(sizeof(*cache), cache_dtor, - AO2_ALLOC_OPT_LOCK_NOLOCK); - if (!cache) { - return NULL; - } +struct stasis_message *stasis_cache_entry_get_remote(struct stasis_cache_entry *entry, int idx) +{ + if (idx < AST_VECTOR_SIZE(&entry->remote)) { + return AST_VECTOR_GET(&entry->remote, idx); + } + return NULL; +} - cache->entries = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash, - cache_entry_cmp); - if (!cache->entries) { - return NULL; - } +/*! + * \internal + * \brief Find the cache entry in the cache entries container. + * + * \param entries Container of cached entries. + * \param type Type of message to retrieve the cache entry. + * \param id Identity of the snapshot to retrieve the cache entry. + * + * \note The entries container is already locked. + * + * \retval Cache-entry on success. + * \retval NULL Not in cache. + */ +static struct stasis_cache_entry *cache_find(struct ao2_container *entries, struct stasis_message_type *type, const char *id) +{ + struct cache_entry_key search_key; + + search_key.type = type; + search_key.id = id; + return ao2_find(entries, &search_key, OBJ_SEARCH_KEY | OBJ_NOLOCK); +} + +/*! + * \internal + * \brief Remove the stasis snapshot in the cache entry determined by eid. + * + * \param entries Container of cached entries. + * \param cached_entry The entry to remove the snapshot from. + * \param eid Which snapshot in the cached entry. + * + * \note The entries container is already locked. + * + * \return Previous stasis entry snapshot. + */ +static struct stasis_message *cache_remove(struct ao2_container *entries, struct stasis_cache_entry *cached_entry, const struct ast_eid *eid) +{ + struct stasis_message *old_snapshot; + int is_remote; + + is_remote = ast_eid_cmp(eid, &ast_eid_default); + if (!is_remote) { + old_snapshot = cached_entry->local; + cached_entry->local = NULL; + } else { + int idx; + + old_snapshot = NULL; + for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) { + struct stasis_message *cur; + + cur = AST_VECTOR_GET(&cached_entry->remote, idx); + if (!ast_eid_cmp(eid, stasis_message_eid(cur))) { + old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx); + break; + } + } + } + + if (!cached_entry->local && !AST_VECTOR_SIZE(&cached_entry->remote)) { + ao2_unlink_flags(entries, cached_entry, OBJ_NOLOCK); + } + + return old_snapshot; +} - cache->id_fn = id_fn; +/*! + * \internal + * \brief Update the stasis snapshot in the cache entry determined by eid. + * + * \param cached_entry The entry to remove the snapshot from. + * \param eid Which snapshot in the cached entry. + * \param new_snapshot Snapshot to replace the old snapshot. + * + * \return Previous stasis entry snapshot. + */ +static struct stasis_message *cache_udpate(struct stasis_cache_entry *cached_entry, const struct ast_eid *eid, struct stasis_message *new_snapshot) +{ + struct stasis_message *old_snapshot; + int is_remote; + int idx; + + is_remote = ast_eid_cmp(eid, &ast_eid_default); + if (!is_remote) { + old_snapshot = cached_entry->local; + cached_entry->local = ao2_bump(new_snapshot); + return old_snapshot; + } + + old_snapshot = NULL; + for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) { + struct stasis_message *cur; - ao2_ref(cache, +1); - return cache; + cur = AST_VECTOR_GET(&cached_entry->remote, idx); + if (!ast_eid_cmp(eid, stasis_message_eid(cur))) { + old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx); + break; + } + } + if (!AST_VECTOR_APPEND(&cached_entry->remote, new_snapshot)) { + ao2_bump(new_snapshot); + } + + return old_snapshot; } -static struct stasis_message *cache_put(struct stasis_cache *cache, - struct stasis_message_type *type, const char *id, +struct cache_put_snapshots { + /*! Old cache eid snapshot. */ + struct stasis_message *old; + /*! Old cache aggregate snapshot. */ + struct stasis_message *aggregate_old; + /*! New cache aggregate snapshot. */ + struct stasis_message *aggregate_new; +}; + +static struct cache_put_snapshots cache_put(struct stasis_cache *cache, + struct stasis_message_type *type, const char *id, const struct ast_eid *eid, struct stasis_message *new_snapshot) { - 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; + struct stasis_cache_entry *cached_entry; + struct cache_put_snapshots snapshots; ast_assert(cache->entries != NULL); + ast_assert(eid != NULL);/* Aggregate snapshots not allowed to be put directly. */ ast_assert(new_snapshot == NULL || type == stasis_message_type(new_snapshot)); - new_entry = cache_entry_create(type, id, new_snapshot); + memset(&snapshots, 0, sizeof(snapshots)); + + ao2_wrlock(cache->entries); + + cached_entry = cache_find(cache->entries, type, id); - if (new_snapshot == NULL) { - /* Remove entry from cache */ - cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_UNLINK); + /* Update the eid snapshot. */ + if (!new_snapshot) { + /* Remove snapshot from cache */ if (cached_entry) { - old_snapshot = cached_entry->snapshot; - cached_entry->snapshot = NULL; + snapshots.old = cache_remove(cache->entries, cached_entry, eid); } + } else if (cached_entry) { + /* Update snapshot in cache */ + snapshots.old = cache_udpate(cached_entry, eid, new_snapshot); } else { - /* Insert/update cache */ - SCOPED_AO2LOCK(lock, cache->entries); - - cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_NOLOCK); + /* Insert into the cache */ + cached_entry = cache_entry_create(type, id, new_snapshot); 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 */ - ao2_link_flags(cache->entries, new_entry, OBJ_NOLOCK); + ao2_link_flags(cache->entries, cached_entry, OBJ_NOLOCK); } + } + /* Update the aggregate snapshot. */ + if (cache->aggregate_calc_fn && cached_entry) { + snapshots.aggregate_new = cache->aggregate_calc_fn(cached_entry, new_snapshot); + snapshots.aggregate_old = cached_entry->aggregate; + cached_entry->aggregate = ao2_bump(snapshots.aggregate_new); } - return old_snapshot; + ao2_unlock(cache->entries); + + ao2_cleanup(cached_entry); + return snapshots; } -struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id) +/*! + * \internal + * \brief Dump all entity snapshots in the cache entry into the given container. + * + * \param snapshots Container to put all snapshots in the cache entry. + * \param entry Cache entry to use. + * + * \retval 0 on success. + * \retval non-zero on error. + */ +static int cache_entry_dump(struct ao2_container *snapshots, const struct stasis_cache_entry *entry) +{ + int idx; + int err = 0; + + ast_assert(snapshots != NULL); + ast_assert(entry != NULL); + + /* The aggregate snapshot is not a snapshot from an entity. */ + + if (entry->local) { + err |= !ao2_link(snapshots, entry->local); + } + + for (idx = 0; !err && idx < AST_VECTOR_SIZE(&entry->remote); ++idx) { + struct stasis_message *snapshot; + + snapshot = AST_VECTOR_GET(&entry->remote, idx); + err |= !ao2_link(snapshots, snapshot); + } + + return err; +} + +struct ao2_container *stasis_cache_get_all(struct stasis_cache *cache, struct stasis_message_type *type, const char *id) { - RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup); - RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup); + struct stasis_cache_entry *cached_entry; + struct ao2_container *found; + ast_assert(cache != NULL); ast_assert(cache->entries != NULL); + ast_assert(type != NULL); + ast_assert(id != NULL); - search_entry = cache_entry_create(type, id, NULL); - if (search_entry == NULL) { + found = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL); + if (!found) { return NULL; } - cached_entry = ao2_find(cache->entries, search_entry, OBJ_POINTER); - if (cached_entry == NULL) { - return NULL; + ao2_rdlock(cache->entries); + + cached_entry = cache_find(cache->entries, type, id); + if (cached_entry && cache_entry_dump(found, cached_entry)) { + ao2_cleanup(found); + found = NULL; + } + + ao2_unlock(cache->entries); + + ao2_cleanup(cached_entry); + return found; +} + +/*! + * \internal + * \brief Retrieve an item from the cache entry for a specific eid. + * + * \param entry Cache entry to use. + * \param eid Specific entity id to retrieve. NULL for aggregate. + * + * \note The returned snapshot has not had its reference bumped. + * + * \retval Snapshot from the cache. + * \retval \c NULL if snapshot is not found. + */ +static struct stasis_message *cache_entry_by_eid(const struct stasis_cache_entry *entry, const struct ast_eid *eid) +{ + int is_remote; + int idx; + + if (!eid) { + /* Get aggregate. */ + return entry->aggregate; + } + + /* Get snapshot with specific eid. */ + is_remote = ast_eid_cmp(eid, &ast_eid_default); + if (!is_remote) { + return entry->local; } - ast_assert(cached_entry->snapshot != NULL); - ao2_ref(cached_entry->snapshot, +1); - return cached_entry->snapshot; + for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) { + struct stasis_message *cur; + + cur = AST_VECTOR_GET(&entry->remote, idx); + if (!ast_eid_cmp(eid, stasis_message_eid(cur))) { + return cur; + } + } + + return NULL; +} + +struct stasis_message *stasis_cache_get_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const char *id, const struct ast_eid *eid) +{ + struct stasis_cache_entry *cached_entry; + struct stasis_message *snapshot = NULL; + + ast_assert(cache != NULL); + ast_assert(cache->entries != NULL); + ast_assert(type != NULL); + ast_assert(id != NULL); + + ao2_rdlock(cache->entries); + + cached_entry = cache_find(cache->entries, type, id); + if (cached_entry) { + snapshot = cache_entry_by_eid(cached_entry, eid); + ao2_bump(snapshot); + } + + ao2_unlock(cache->entries); + + ao2_cleanup(cached_entry); + return snapshot; +} + +struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id) +{ + return stasis_cache_get_by_eid(cache, type, id, &ast_eid_default); } struct cache_dump_data { - struct ao2_container *cached; + struct ao2_container *container; struct stasis_message_type *type; + const struct ast_eid *eid; }; -static int cache_dump_cb(void *obj, void *arg, int flags) +static int cache_dump_by_eid_cb(void *obj, void *arg, int flags) { struct cache_dump_data *cache_dump = arg; - struct cache_entry *entry = obj; + struct stasis_cache_entry *entry = obj; + + if (!cache_dump->type || entry->key.type == cache_dump->type) { + struct stasis_message *snapshot; - if (!cache_dump->type || entry->type == cache_dump->type) { - ao2_link(cache_dump->cached, entry->snapshot); + snapshot = cache_entry_by_eid(entry, cache_dump->eid); + if (snapshot) { + if (!ao2_link(cache_dump->container, snapshot)) { + ao2_cleanup(cache_dump->container); + cache_dump->container = NULL; + return CMP_STOP; + } + } } return 0; } +struct ao2_container *stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid) +{ + struct cache_dump_data cache_dump; + + ast_assert(cache != NULL); + ast_assert(cache->entries != NULL); + + cache_dump.eid = eid; + cache_dump.type = type; + cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL); + if (!cache_dump.container) { + return NULL; + } + + ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_by_eid_cb, &cache_dump); + return cache_dump.container; +} + struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type) { + return stasis_cache_dump_by_eid(cache, type, &ast_eid_default); +} + +static int cache_dump_all_cb(void *obj, void *arg, int flags) +{ + struct cache_dump_data *cache_dump = arg; + struct stasis_cache_entry *entry = obj; + + if (!cache_dump->type || entry->key.type == cache_dump->type) { + if (cache_entry_dump(cache_dump->container, entry)) { + ao2_cleanup(cache_dump->container); + cache_dump->container = NULL; + return CMP_STOP; + } + } + + return 0; +} + +struct ao2_container *stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type) +{ struct cache_dump_data cache_dump; + ast_assert(cache != NULL); ast_assert(cache->entries != NULL); + cache_dump.eid = NULL; cache_dump.type = type; - cache_dump.cached = ao2_container_alloc_options( - AO2_ALLOC_OPT_LOCK_NOLOCK, 1, NULL, NULL); - if (!cache_dump.cached) { + cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL); + if (!cache_dump.container) { return NULL; } - ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_cb, &cache_dump); - return cache_dump.cached; + ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_all_cb, &cache_dump); + return cache_dump.container; } STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type); @@ -380,9 +762,13 @@ static struct stasis_message *update_create(struct stasis_message *old_snapshot, static void caching_topic_exec(void *data, struct stasis_subscription *sub, struct stasis_message *message) { - RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup); + struct stasis_caching_topic *caching_topic_needs_unref; struct stasis_caching_topic *caching_topic = data; - const char *id = NULL; + struct stasis_message *msg; + struct stasis_message *msg_put; + struct stasis_message_type *msg_type; + const struct ast_eid *msg_eid; + const char *msg_id; ast_assert(caching_topic != NULL); ast_assert(caching_topic->topic != NULL); @@ -391,50 +777,62 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub, if (stasis_subscription_final_message(sub, message)) { caching_topic_needs_unref = caching_topic; + } else { + caching_topic_needs_unref = NULL; } - /* Handle cache clear event */ - if (stasis_cache_clear_type() == stasis_message_type(message)) { - RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup); - RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup); - struct stasis_message *clear_msg = stasis_message_data(message); - const char *clear_id = caching_topic->cache->id_fn(clear_msg); - struct stasis_message_type *clear_type = stasis_message_type(clear_msg); + msg_type = stasis_message_type(message); + if (stasis_cache_clear_type() == msg_type) { + /* Cache clear event. */ + msg_put = NULL; + msg = stasis_message_data(message); + msg_type = stasis_message_type(msg); + } else { + /* Normal cache update event. */ + msg_put = message; + msg = message; + } + ast_assert(msg_type != NULL); - ast_assert(clear_type != NULL); + msg_eid = stasis_message_eid(msg);/* msg_eid is NULL for aggregate message. */ + msg_id = caching_topic->cache->id_fn(msg); + if (msg_id && msg_eid) { + struct stasis_message *update; + struct cache_put_snapshots snapshots; - if (clear_id) { - old_snapshot = cache_put(caching_topic->cache, clear_type, clear_id, NULL); - if (old_snapshot) { - update = update_create(old_snapshot, NULL); + /* Update the cache */ + snapshots = cache_put(caching_topic->cache, msg_type, msg_id, msg_eid, msg_put); + if (snapshots.old || msg_put) { + update = update_create(snapshots.old, msg_put); + if (update) { stasis_publish(caching_topic->topic, update); - return; } - + ao2_cleanup(update); + } else { 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; + stasis_topic_name(caching_topic->topic), + stasis_message_type_name(msg_type), msg_id); } - } - - id = caching_topic->cache->id_fn(message); - if (id == NULL) { - /* Object isn't cached; discard */ - } else { - /* Update the cache */ - RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup); - RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup); - old_snapshot = cache_put(caching_topic->cache, stasis_message_type(message), id, message); - - update = update_create(old_snapshot, message); - if (update == NULL) { - return; + if (snapshots.aggregate_old != snapshots.aggregate_new) { + if (snapshots.aggregate_new && caching_topic->cache->aggregate_publish_fn) { + caching_topic->cache->aggregate_publish_fn(caching_topic->original_topic, + snapshots.aggregate_new); + } + update = update_create(snapshots.aggregate_old, snapshots.aggregate_new); + if (update) { + stasis_publish(caching_topic->topic, update); + } + ao2_cleanup(update); } - stasis_publish(caching_topic->topic, update); + ao2_cleanup(snapshots.old); + ao2_cleanup(snapshots.aggregate_old); + ao2_cleanup(snapshots.aggregate_new); } + + ao2_cleanup(caching_topic_needs_unref); } struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache) diff --git a/main/stasis_message.c b/main/stasis_message.c index 240845aff..1db2ae97a 100644 --- a/main/stasis_message.c +++ b/main/stasis_message.c @@ -53,7 +53,7 @@ static void message_type_dtor(void *obj) struct stasis_message_type *stasis_message_type_create(const char *name, struct stasis_message_vtable *vtable) { - RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup); + struct stasis_message_type *type; type = ao2_alloc(sizeof(*type), message_type_dtor); if (!type) { @@ -66,11 +66,11 @@ struct stasis_message_type *stasis_message_type_create(const char *name, type->name = ast_strdup(name); if (!type->name) { + ao2_cleanup(type); return NULL; } type->vtable = vtable; - ao2_ref(type, +1); return type; } @@ -85,8 +85,12 @@ struct stasis_message { struct timeval timestamp; /*! Type of the message */ struct stasis_message_type *type; + /*! Where this message originated. NULL if aggregate message. */ + const struct ast_eid *eid_ptr; /*! Message content */ void *data; + /*! Where this message originated. */ + struct ast_eid eid; }; static void stasis_message_dtor(void *obj) @@ -96,9 +100,9 @@ static void stasis_message_dtor(void *obj) ao2_cleanup(message->data); } -struct stasis_message *stasis_message_create(struct stasis_message_type *type, void *data) +struct stasis_message *stasis_message_create_full(struct stasis_message_type *type, void *data, const struct ast_eid *eid) { - RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup); + struct stasis_message *message; if (type == NULL || data == NULL) { return NULL; @@ -114,11 +118,27 @@ struct stasis_message *stasis_message_create(struct stasis_message_type *type, v message->type = type; ao2_ref(data, +1); message->data = data; + if (eid) { + message->eid_ptr = &message->eid; + message->eid = *eid; + } - ao2_ref(message, +1); return message; } +struct stasis_message *stasis_message_create(struct stasis_message_type *type, void *data) +{ + return stasis_message_create_full(type, data, &ast_eid_default); +} + +const struct ast_eid *stasis_message_eid(const struct stasis_message *msg) +{ + if (msg == NULL) { + return NULL; + } + return msg->eid_ptr; +} + struct stasis_message_type *stasis_message_type(const struct stasis_message *msg) { if (msg == NULL) { |