[svn-commits] rmudgett: trunk r410185 - in /trunk: ./ include/asterisk/ main/ tests/
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Fri Mar 7 14:41:20 CST 2014
Author: rmudgett
Date: Fri Mar 7 14:41:13 2014
New Revision: 410185
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=410185
Log:
stasis cache: Enhance to keep track of an item from different entities.
A stasis cache entry now contains more than a single message/snapshot. It
contains messages/snapshots for the local entity as well as any remote
entities that post to the cached item. In addition callbacks can be
supplied when the cache is created to compute and post the aggregate
message/snapshot representing all entities stored in the cache entry.
* All stasis messages now have an eid to indicate what entity posted it.
* The stasis cache enhancements allow device state to cache and aggregate
the device states from local and remote entities in a single operation.
The cached aggregate device state is available immediately after it is
posted to the stasis bus. This improves performance by eliminating a
cache dump and associated ao2 container traversals to calculate the
aggregate state.
(closes issue ASTERISK-23204)
Reported by: Mark Michelson
Review: https://reviewboard.asterisk.org/r/3281/
........
Merged revisions 410184 from http://svn.asterisk.org/svn/asterisk/branches/12
Modified:
trunk/ (props changed)
trunk/include/asterisk/devicestate.h
trunk/include/asterisk/stasis.h
trunk/main/app.c
trunk/main/devicestate.c
trunk/main/stasis_cache.c
trunk/main/stasis_message.c
trunk/tests/test_devicestate.c
trunk/tests/test_stasis.c
Propchange: trunk/
------------------------------------------------------------------------------
Binary property 'branch-12-merged' - no diff available.
Modified: trunk/include/asterisk/devicestate.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/devicestate.h?view=diff&rev=410185&r1=410184&r2=410185
==============================================================================
--- trunk/include/asterisk/devicestate.h (original)
+++ trunk/include/asterisk/devicestate.h Fri Mar 7 14:41:13 2014
@@ -274,13 +274,20 @@
* \since 12
*/
struct ast_device_state_message {
- AST_DECLARE_STRING_FIELDS(
- AST_STRING_FIELD(cache_id); /*!< A unique ID used for hashing */
- AST_STRING_FIELD(device); /*!< The name of the device */
- );
- enum ast_device_state state; /*!< The state of the device */
- struct ast_eid *eid; /*!< The EID of the server where this message originated, NULL EID means aggregate state */
- enum ast_devstate_cache cachable; /*!< Flag designating the cachability of this device state */
+ /*! The name of the device */
+ const char *device;
+ /*!
+ * \brief The EID of the server where this message originated.
+ *
+ * \note A NULL EID means aggregate state.
+ */
+ const struct ast_eid *eid;
+ /*! The state of the device */
+ enum ast_device_state state;
+ /*! Flag designating the cachability of this device state */
+ enum ast_devstate_cache cachable;
+ /*! The device and eid data is stuffed here when the struct is allocated. */
+ struct ast_eid stuff[0];
};
/*!
Modified: trunk/include/asterisk/stasis.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/stasis.h?view=diff&rev=410185&r1=410184&r2=410185
==============================================================================
--- trunk/include/asterisk/stasis.h (original)
+++ trunk/include/asterisk/stasis.h Fri Mar 7 14:41:13 2014
@@ -290,11 +290,49 @@
*
* \param type Type of the message
* \param data Immutable data that is the actual contents of the message
+ *
* \return New message
* \return \c NULL on error
+ *
* \since 12
*/
struct stasis_message *stasis_message_create(struct stasis_message_type *type, void *data);
+
+/*!
+ * \brief Create a new message for an entity.
+ *
+ * This message is an \c ao2 object, and must be ao2_cleanup()'ed when you are done
+ * with it. Messages are also immutable, and must not be modified after they
+ * are initialized. Especially the \a data in the message.
+ *
+ * \param type Type of the message
+ * \param data Immutable data that is the actual contents of the message
+ * \param eid What entity originated this message. (NULL for aggregate)
+ *
+ * \note An aggregate message is a combined representation of the local
+ * and remote entities publishing the message data. e.g., An aggregate
+ * device state represents the combined device state from the local and
+ * any remote entities publishing state for a device. e.g., An aggregate
+ * MWI message is the old/new MWI counts accumulated from the local and
+ * any remote entities publishing to a mailbox.
+ *
+ * \retval New message
+ * \retval \c NULL on error
+ *
+ * \since 12.2.0
+ */
+struct stasis_message *stasis_message_create_full(struct stasis_message_type *type, void *data, const struct ast_eid *eid);
+
+/*!
+ * \brief Get the entity id for a \ref stasis_message.
+ * \since 12.2.0
+ *
+ * \param msg Message to get eid.
+ *
+ * \retval Entity id of \a msg
+ * \retval \c NULL if \a msg is an aggregate or \a msg is \c NULL.
+ */
+const struct ast_eid *stasis_message_eid(const struct stasis_message *msg);
/*!
* \brief Get the message type for a \ref stasis_message.
@@ -503,8 +541,8 @@
* \brief Create a subscription which forwards all messages from one topic to
* another.
*
- * Note that the \a topic parameter of the invoked callback will the be \a topic
- * the message was sent to, not the topic the subscriber subscribed to.
+ * Note that the \a topic parameter of the invoked callback will the be the
+ * \a topic the message was sent to, not the topic the subscriber subscribed to.
*
* \param from_topic Topic to forward.
* \param to_topic Destination topic of forwarded messages.
@@ -640,6 +678,9 @@
*/
struct stasis_cache;
+/*! Cache entry used for calculating the aggregate snapshot. */
+struct stasis_cache_entry;
+
/*!
* \brief A topic wrapper, which caches certain messages.
* \since 12
@@ -661,6 +702,101 @@
typedef const char *(*snapshot_get_id)(struct stasis_message *message);
/*!
+ * \brief Callback to calculate the aggregate 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.
+ *
+ * \note An aggregate message is a combined representation of the local
+ * and remote entities publishing the message data. e.g., An aggregate
+ * device state represents the combined device state from the local and
+ * any remote entities publishing state for a device. e.g., An aggregate
+ * MWI message is the old/new MWI counts accumulated from the local and
+ * any remote entities publishing to a mailbox.
+ *
+ * \return New aggregate-snapshot calculated on success.
+ * Caller has a reference on return.
+ */
+typedef struct stasis_message *(*cache_aggregate_calc_fn)(struct stasis_cache_entry *entry, struct stasis_message *new_snapshot);
+
+/*!
+ * \brief Callback to publish the aggregate cache entry message.
+ * \since 12.2.0
+ *
+ * \details
+ * Once an aggregate message is calculated. This callback publishes the
+ * message so subscribers will know the new value of an aggregated state.
+ *
+ * \param topic The aggregate message may be published to this topic.
+ * It is the topic to which the cache itself is subscribed.
+ * \param aggregate The aggregate shapshot message to publish.
+ *
+ * \note It is up to the function to determine if there is a better topic
+ * the aggregate message should be published over.
+ *
+ * \note An aggregate message is a combined representation of the local
+ * and remote entities publishing the message data. e.g., An aggregate
+ * device state represents the combined device state from the local and
+ * any remote entities publishing state for a device. e.g., An aggregate
+ * MWI message is the old/new MWI counts accumulated from the local and
+ * any remote entities publishing to a mailbox.
+ *
+ * \return Nothing
+ */
+typedef void (*cache_aggregate_publish_fn)(struct stasis_topic *topic, struct stasis_message *aggregate);
+
+/*!
+ * \brief Get the aggregate cache entry snapshot.
+ * \since 12.2.0
+ *
+ * \param entry Cache entry to get the aggregate snapshot.
+ *
+ * \note A reference is not given to the returned pointer so don't unref it.
+ *
+ * \note An aggregate message is a combined representation of the local
+ * and remote entities publishing the message data. e.g., An aggregate
+ * device state represents the combined device state from the local and
+ * any remote entities publishing state for a device. e.g., An aggregate
+ * MWI message is the old/new MWI counts accumulated from the local and
+ * any remote entities publishing to a mailbox.
+ *
+ * \retval Aggregate-snapshot in cache.
+ * \retval NULL if not present.
+ */
+struct stasis_message *stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry);
+
+/*!
+ * \brief Get the local entity's cache entry snapshot.
+ * \since 12.2.0
+ *
+ * \param entry Cache entry to get the local entity's snapshot.
+ *
+ * \note A reference is not given to the returned pointer so don't unref it.
+ *
+ * \retval Internal-snapshot in cache.
+ * \retval NULL if not present.
+ */
+struct stasis_message *stasis_cache_entry_get_local(struct stasis_cache_entry *entry);
+
+/*!
+ * \brief Get a remote entity's cache entry snapshot by index.
+ * \since 12.2.0
+ *
+ * \param entry Cache entry to get a remote entity's snapshot.
+ * \param idx Which remote entity's snapshot to get.
+ *
+ * \note A reference is not given to the returned pointer so don't unref it.
+ *
+ * \retval Remote-entity-snapshot in cache.
+ * \retval NULL if not present.
+ */
+struct stasis_message *stasis_cache_entry_get_remote(struct stasis_cache_entry *entry, int idx);
+
+/*!
* \brief Create a cache.
*
* This is the backend store for a \ref stasis_caching_topic. The cache is
@@ -669,11 +805,39 @@
* The returned object is AO2 managed, so ao2_cleanup() when you're done.
*
* \param id_fn Callback to extract the id from a snapshot message.
- * \return New cache indexed by \a id_fn.
- * \return \c NULL on error
+ *
+ * \retval New cache indexed by \a id_fn.
+ * \retval \c NULL on error
+ *
* \since 12
*/
struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn);
+
+/*!
+ * \brief Create a cache.
+ *
+ * This is the backend store for a \ref stasis_caching_topic. The cache is
+ * thread safe, allowing concurrent reads and writes.
+ *
+ * The returned object is AO2 managed, so ao2_cleanup() when you're done.
+ *
+ * \param id_fn Callback to extract the id from a snapshot message.
+ * \param aggregate_calc_fn Callback to calculate the aggregate cache entry.
+ * \param aggregate_publish_fn Callback to publish the aggregate cache entry.
+ *
+ * \note An aggregate message is a combined representation of the local
+ * and remote entities publishing the message data. e.g., An aggregate
+ * device state represents the combined device state from the local and
+ * any remote entities publishing state for a device. e.g., An aggregate
+ * MWI message is the old/new MWI counts accumulated from the local and
+ * any remote entities publishing to a mailbox.
+ *
+ * \retval New cache indexed by \a id_fn.
+ * \retval \c NULL on error
+ *
+ * \since 12.2.0
+ */
+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);
/*!
* \brief Create a topic which monitors and caches messages from another topic.
@@ -749,31 +913,95 @@
struct stasis_message *stasis_cache_clear_create(struct stasis_message *message);
/*!
- * \brief Retrieve an item from the cache.
+ * \brief Retrieve an item from the cache for the ast_eid_default entity.
*
* The returned item is AO2 managed, so ao2_cleanup() when you're done with it.
*
* \param cache The cache to query.
* \param type Type of message to retrieve.
* \param id Identity of the snapshot to retrieve.
- * \return Message from the cache.
- * \return \c NULL if message is not found.
- * \since 12
- */
-struct stasis_message *stasis_cache_get(
- struct stasis_cache *cache, struct stasis_message_type *type,
- const char *id);
-
-/*!
- * \brief Dump cached items to a subscription
+ *
+ * \retval Message from the cache.
+ * \retval \c NULL if message is not found.
+ *
+ * \since 12
+ */
+struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id);
+
+/*!
+ * \brief Retrieve an item from the cache for a specific entity.
+ *
+ * The returned item is AO2 managed, so ao2_cleanup() when you're done with it.
+ *
+ * \param cache The cache to query.
+ * \param type Type of message to retrieve.
+ * \param id Identity of the snapshot to retrieve.
+ * \param eid Specific entity id to retrieve. NULL for aggregate.
+ *
+ * \note An aggregate message is a combined representation of the local
+ * and remote entities publishing the message data. e.g., An aggregate
+ * device state represents the combined device state from the local and
+ * any remote entities publishing state for a device. e.g., An aggregate
+ * MWI message is the old/new MWI counts accumulated from the local and
+ * any remote entities publishing to a mailbox.
+ *
+ * \retval Message from the cache.
+ * \retval \c NULL if message is not found.
+ *
+ * \since 12.2.0
+ */
+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);
+
+/*!
+ * \brief Retrieve all matching entity items from the cache.
+ * \since 12.2.0
+ *
+ * \param cache The cache to query.
+ * \param type Type of message to retrieve.
+ * \param id Identity of the snapshot to retrieve.
+ *
+ * \retval Container of matching items found.
+ * \retval \c NULL if error.
+ */
+struct ao2_container *stasis_cache_get_all(struct stasis_cache *cache, struct stasis_message_type *type, const char *id);
+
+/*!
+ * \brief Dump cached items to a subscription for the ast_eid_default entity.
+ *
* \param cache The cache to query.
* \param type Type of message to dump (any type if \c NULL).
- * \return ao2_container containing all matches (must be unreffed by caller)
- * \return \c NULL on allocation error
- * \since 12
- */
-struct ao2_container *stasis_cache_dump(struct stasis_cache *cache,
- struct stasis_message_type *type);
+ *
+ * \retval ao2_container containing all matches (must be unreffed by caller)
+ * \retval \c NULL on allocation error
+ *
+ * \since 12
+ */
+struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type);
+
+/*!
+ * \brief Dump cached items to a subscription for a specific entity.
+ * \since 12.2.0
+ *
+ * \param cache The cache to query.
+ * \param type Type of message to dump (any type if \c NULL).
+ * \param eid Specific entity id to retrieve. NULL for aggregate.
+ *
+ * \retval ao2_container containing all matches (must be unreffed by caller)
+ * \retval \c NULL on allocation error
+ */
+struct ao2_container *stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid);
+
+/*!
+ * \brief Dump all entity items from the cache to a subscription.
+ * \since 12.2.0
+ *
+ * \param cache The cache to query.
+ * \param type Type of message to dump (any type if \c NULL).
+ *
+ * \retval ao2_container containing all matches (must be unreffed by caller)
+ * \retval \c NULL on allocation error
+ */
+struct ao2_container *stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type);
/*! @} */
Modified: trunk/main/app.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/app.c?view=diff&rev=410185&r1=410184&r2=410185
==============================================================================
--- trunk/main/app.c (original)
+++ trunk/main/app.c Fri Mar 7 14:41:13 2014
@@ -2821,7 +2821,6 @@
return mwi_state;
}
-
int ast_publish_mwi_state_full(
const char *mailbox,
const char *context,
@@ -2857,10 +2856,19 @@
if (eid) {
mwi_state->eid = *eid;
} else {
- ast_set_default_eid(&mwi_state->eid);
- }
-
- message = stasis_message_create(ast_mwi_state_type(), mwi_state);
+ mwi_state->eid = ast_eid_default;
+ }
+
+ /*
+ * 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 @@
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;
Modified: trunk/main/devicestate.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/devicestate.c?view=diff&rev=410185&r1=410184&r2=410185
==============================================================================
--- trunk/main/devicestate.c (original)
+++ trunk/main/devicestate.c Fri Mar 7 14:41:13 2014
@@ -282,16 +282,20 @@
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;
-
- cached_msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), device);
+ enum ast_device_state state;
+
+ 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);
-
- return device_state->state;
+ state = device_state->state;
+ ao2_cleanup(cached_msg);
+
+ return state;
}
/*! \brief Check device state through channel specific function or generic function */
@@ -522,148 +526,62 @@
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);
-
- if (!new_device_state || ast_string_field_init(new_device_state, 256)) {
+ struct ast_device_state_message *new_device_state;
+ char *pos;
+ size_t stuff_len;
+
+ ast_assert(!ast_strlen_zero(device));
+
+ 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);
+ if (eid) {
+ /* 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 {
+ pos = (char *) &new_device_state->stuff[0];
+ }
+
+ strcpy(pos, device);/* Safe */
+ new_device_state->device = pos;
+
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));
- } else {
- /* no EID makes this an aggregate state */
- ast_string_field_set(new_device_state, cache_id, device);
- }
-
- 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;
-
- cached_aggregate_msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), device);
- if (!cached_aggregate_msg) {
- return 1;
- }
-
- cached_aggregate_device_state = stasis_message_data(cached_aggregate_msg);
- if (cached_aggregate_device_state->state == new_aggregate_state) {
- return 0;
- }
- return 1;
-}
-
-static void devstate_change_collector_cb(void *data, struct stasis_subscription *sub, struct stasis_message *msg)
-{
- enum ast_device_state aggregate_state;
- char *device;
+static void devstate_change_cb(void *data, struct stasis_subscription *sub, struct stasis_message *msg)
+{
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 @@
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);
-
- if (!(cached_msg = stasis_cache_get(ast_device_state_cache(),
- ast_device_state_message_type(), device))) {
+ struct stasis_message *cached_msg;
+ struct stasis_message *msg;
+
+ 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 @@
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 @@
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 @@
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);
+ device_state_topic_cached = stasis_caching_unsubscribe_and_join(device_state_topic_cached);
+
+ ao2_cleanup(device_state_cache);
+ device_state_cache = NULL;
+
+ ao2_cleanup(device_state_topic_pool);
+ device_state_topic_pool = NULL;
+
ao2_cleanup(device_state_topic_all);
device_state_topic_all = NULL;
- 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;
}
int devstate_init(void)
@@ -817,25 +849,32 @@
}
device_state_topic_all = stasis_topic_create("ast_device_state_topic");
if (!device_state_topic_all) {
- return -1;
- }
- device_state_cache = stasis_cache_create(device_state_get_id);
- if (!device_state_cache) {
- return -1;
- }
- device_state_topic_cached = stasis_caching_topic_create(device_state_topic_all, device_state_cache);
- if (!device_state_topic_cached) {
+ devstate_cleanup();
return -1;
}
device_state_topic_pool = stasis_topic_pool_create(ast_device_state_topic_all());
if (!device_state_topic_pool) {
- return -1;
- }
-
- devstate_message_sub = stasis_subscribe(ast_device_state_topic_cached(), devstate_change_collector_cb, NULL);
-
+ devstate_cleanup();
+ return -1;
+ }
+ 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_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_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;
}
Modified: trunk/main/stasis_cache.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/stasis_cache.c?view=diff&rev=410185&r1=410184&r2=410185
==============================================================================
--- trunk/main/stasis_cache.c (original)
+++ trunk/main/stasis_cache.c Fri Mar 7 14:41:13 2014
@@ -36,6 +36,7 @@
#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 @@
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 @@
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;
-}
-
-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);
+ 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 stasis_cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
+{
+ 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 @@
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);
+ 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;
+ }
+
+ if (is_remote) {
+ if (AST_VECTOR_APPEND(&entry->remote, snapshot)) {
+ ao2_cleanup(entry);
+ return NULL;
+ }
+ } else {
+ entry->local = snapshot;
+ }
+ ao2_bump(snapshot);
+
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));
-
- hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
- hash += ast_hashtab_hash_string(entry->id);
+ 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(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;
-
- ast_assert(!(flags & OBJ_KEY));
-
- if (left->type == right->type && strcmp(left->id, right->id) == 0) {
- return CMP_MATCH | CMP_STOP;
- }
-
- return 0;
+ 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;
-
- ao2_cleanup(cache->entries);
- cache->entries = NULL;
+ struct stasis_cache *cache = obj;
+
+ ao2_cleanup(cache->entries);
+ cache->entries = NULL;
+}
+
+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;
+ }
+
+ 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;
}
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,
+ return stasis_cache_create_full(id_fn, NULL, NULL);
+}
+
+struct stasis_message *stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry)
+{
+ return entry->aggregate;
+}
+
+struct stasis_message *stasis_cache_entry_get_local(struct stasis_cache_entry *entry)
+{
+ return entry->local;
+}
+
+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;
+}
+
+/*!
+ * \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.
+ *
[... 1361 lines stripped ...]
More information about the svn-commits
mailing list