[asterisk-commits] dlee: branch dlee/stasis-core r382458 - in /team/dlee/stasis-core: main/ tests/
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Tue Mar 5 09:20:41 CST 2013
Author: dlee
Date: Tue Mar 5 09:20:38 2013
New Revision: 382458
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=382458
Log:
Reversed the strong/weak pointer relationship between topic and subscriber.
Now subscriptions require an explicit unsubscribe, which causes an
implicit unref. You can still ref/unref the subscription if you need to
pass it around.
Modified:
team/dlee/stasis-core/main/stasis.c
team/dlee/stasis-core/tests/test_stasis.c
Modified: team/dlee/stasis-core/main/stasis.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-core/main/stasis.c?view=diff&rev=382458&r1=382457&r2=382458
==============================================================================
--- team/dlee/stasis-core/main/stasis.c (original)
+++ team/dlee/stasis-core/main/stasis.c Tue Mar 5 09:20:38 2013
@@ -47,7 +47,7 @@
/*! \private */
struct stasis_topic {
char *name;
- /*! Variable length array of the subscribers */
+ /*! Variable length array of the subscribers (raw pointer to avoid cyclic references) */
struct stasis_subscription **subscribers;
/*! Allocated length of the subscribers array */
size_t num_subscribers_max;
@@ -64,9 +64,6 @@
struct stasis_topic *topic = obj;
ast_free(topic->name);
topic->name = NULL;
- while (topic->num_subscribers_current > 0) {
- stasis_unsubscribe(topic->subscribers[0]);
- }
ast_free(topic->subscribers);
topic->subscribers = NULL;
}
@@ -103,7 +100,7 @@
/*! \private */
struct stasis_subscription {
- /*! Native pointer to the topic. AO2 ref could cause a cycle. */
+ /*! Topic subscribed to. */
struct stasis_topic *topic;
/*! Mailbox for processing incoming messages. */
struct ast_taskprocessor *mailbox;
@@ -116,8 +113,8 @@
static void subscription_dtor(void *obj)
{
struct stasis_subscription *sub = obj;
- /* This should never be called until after we've been unsubscribed */
- ast_assert(sub->topic == NULL);
+ ao2_cleanup(sub->topic);
+ sub->topic = NULL;
ao2_cleanup(sub->mailbox);
sub->mailbox = NULL;
}
@@ -146,7 +143,8 @@
}
}
- sub->topic = topic; /* Don't increase the refcount, or it will cause a cyclic ref! */
+ ao2_ref(topic, +1);
+ sub->topic = topic;
sub->callback = callback;
sub->data = data;
@@ -166,27 +164,21 @@
void stasis_unsubscribe(struct stasis_subscription *sub)
{
if (sub) {
- RAII_VAR(struct stasis_subscription *, cleanup_after_unlock, NULL, ao2_cleanup);
- SCOPED_AO2LOCK(lock, sub);
-
- if (sub->topic) {
- size_t i;
- struct stasis_topic *topic = sub->topic;
- SCOPED_AO2LOCK(lock, topic);
-
- for (i = 0; i < topic->num_subscribers_current; ++i) {
- if (topic->subscribers[i] == sub) {
- sub->topic = NULL;
- /* swap [i] with last entry; remove last entry */
- topic->subscribers[i] = topic->subscribers[--topic->num_subscribers_current];
- /* We can't clean up now, since the lock is held. defer to RAII */
- cleanup_after_unlock = sub;
- return;
- }
+ size_t i;
+ struct stasis_topic *topic = sub->topic;
+ SCOPED_AO2LOCK(lock_topic, topic);
+
+ for (i = 0; i < topic->num_subscribers_current; ++i) {
+ if (topic->subscribers[i] == sub) {
+ /* swap [i] with last entry; remove last entry */
+ topic->subscribers[i] = topic->subscribers[--topic->num_subscribers_current];
+ /* Unsubscribing unrefs the subscription */
+ ao2_cleanup(sub);
+ return;
}
-
- ast_log(LOG_ERROR, "Internal error: subscription has invalid topic\n");
- }
+ }
+
+ ast_log(LOG_ERROR, "Internal error: subscription has invalid topic\n");
}
}
@@ -212,7 +204,7 @@
topic->num_subscribers_max *= 2;
}
- ao2_ref(sub, +1);
+ /* Don't ref sub here or we'll cause a reference cycle. */
topic->subscribers[topic->num_subscribers_current++] = sub;
return 0;
}
@@ -274,18 +266,10 @@
RAII_VAR(struct dispatch *, dispatch, data, ao2_cleanup);
RAII_VAR(struct stasis_topic *, sub_topic, NULL, ao2_cleanup);
- /* If the subscription's topic is NULL, then we've been unsubscribed,
- * and don't dispatch. Check this while locking the subscription, but
- * don't hold the lock while dispatching.
- */
- {
- SCOPED_AO2LOCK(lock, dispatch->sub);
- if (!dispatch->sub->topic) {
- return 0;
- }
- sub_topic = dispatch->sub->topic;
- ao2_ref(sub_topic, +1);
- }
+ /* Since sub->topic doesn't change, no need to lock sub */
+ ast_assert(dispatch->sub->topic != NULL);
+ ao2_ref(dispatch->sub->topic, +1);
+ sub_topic = dispatch->sub->topic;
dispatch->sub->callback(dispatch->sub->data,
sub_topic,
@@ -296,7 +280,7 @@
void stasis_forward_message(struct stasis_topic *topic, struct stasis_topic *publisher_topic, struct stasis_message *message)
{
- RAII_VAR(struct stasis_subscription **, subscribers, NULL, ast_free);
+ struct stasis_subscription **subscribers = NULL;
size_t num_subscribers, i;
/* Copy the subscribers, so we don't have to hold the mutex for long */
@@ -305,9 +289,10 @@
num_subscribers = topic->num_subscribers_current;
subscribers = ast_malloc(num_subscribers * sizeof(*subscribers));
if (subscribers) {
- memcpy(subscribers,
- topic->subscribers,
- num_subscribers * sizeof(*subscribers));
+ for (i = 0; i < num_subscribers; ++i) {
+ ao2_ref(topic->subscribers[i], +1);
+ subscribers[i] = topic->subscribers[i];
+ }
}
}
@@ -338,6 +323,12 @@
sub->callback(sub->data, sub->topic, message);
}
}
+
+ for (i = 0; i < num_subscribers; ++i) {
+ ao2_cleanup(subscribers[i]);
+ }
+ ast_free(subscribers);
+
}
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Modified: team/dlee/stasis-core/tests/test_stasis.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-core/tests/test_stasis.c?view=diff&rev=382458&r1=382457&r2=382458
==============================================================================
--- team/dlee/stasis-core/tests/test_stasis.c (original)
+++ team/dlee/stasis-core/tests/test_stasis.c Tue Mar 5 09:20:38 2013
@@ -206,7 +206,7 @@
AST_TEST_DEFINE(subscribe)
{
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_subscription *, uut, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
@@ -253,7 +253,7 @@
{
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_subscription *, uut, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
@@ -280,6 +280,7 @@
ast_test_validate(test, NULL != uut);
stasis_unsubscribe(uut);
+ uut = NULL;
test_data = ao2_alloc(1, NULL);
ast_test_validate(test, NULL != test_data);
@@ -303,9 +304,9 @@
RAII_VAR(struct consumer *, parent_consumer, NULL, ao2_cleanup);
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_subscription *, forward_sub, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_subscription *, parent_sub, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_subscription *, sub, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_subscription *, forward_sub, NULL, stasis_unsubscribe);
+ RAII_VAR(struct stasis_subscription *, parent_sub, NULL, stasis_unsubscribe);
+ RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
@@ -407,7 +408,7 @@
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_subscription *, sub, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
int actual_len;
struct stasis_message_type *actual_type;
@@ -456,7 +457,7 @@
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_subscription *, sub, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
More information about the asterisk-commits
mailing list