[svn-commits] dlee: branch dlee/stasis-core r381805 - /team/dlee/stasis-core/main/stasis.c

[SVN commits to the Digium repositories]

Author: dlee
Date: Tue Feb 19 16:25:08 2013
New Revision: 381805

URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=381805
Log:
Handle unsubscribe/dispatch race conditions

Modified:
    team/dlee/stasis-core/main/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=381805&r1=381804&r2=381805
==============================================================================
--- team/dlee/stasis-core/main/stasis.c (original)
+++ team/dlee/stasis-core/main/stasis.c Tue Feb 19 16:25:08 2013
@@ -58,7 +58,6 @@
 /* Forward declarations for the tightly-coupled subscription object */
 struct stasis_subscription;
 static int topic_add_subscription(struct stasis_topic *topic, struct stasis_subscription *sub);
-static int topic_remove_subscription(struct stasis_topic *topic, struct stasis_subscription *sub);
 
 static void topic_dtor(void *obj)
 {
@@ -66,7 +65,7 @@
 	ast_free(topic->name);
 	topic->name = NULL;
 	while (topic->num_subscribers_current > 0) {
-		topic_remove_subscription(topic, topic->subscribers[0]);
+		stasis_unsubscribe(topic->subscribers[0]);
 	}
 	ast_free(topic->subscribers);
 	topic->subscribers = NULL;
@@ -170,10 +169,26 @@
 
 void stasis_unsubscribe(struct stasis_subscription *sub)
 {
-	if (sub && sub->topic) {
-		if (topic_remove_subscription(sub->topic, sub) != 0) {
-			/* If our topic pointer is wrong, something is seriously messed up */
-			ast_assert(0);
+	size_t i;
+
+	if (sub) {
+		SCOPED_AO2LOCK(lock, sub);
+
+		if (sub->topic) {
+			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];
+					ao2_cleanup(sub);
+					return;
+				}
+			}
+
+			ast_log(LOG_ERROR, "Internal error: subscription has invalid topic\n");
 		}
 	}
 }
@@ -204,31 +219,6 @@
 	ao2_ref(sub, +1);
 	topic->subscribers[topic->num_subscribers_current++] = sub;
 	return 0;
-}
-
-/*!
- * \brief Remove a subscriber from a topic.
- * \param topic Topic
- * \param sub Subscriber
- * \return 0 on success
- * \return Non-zero is subscriber is not subscribed to the topic.
- */
-static int topic_remove_subscription(struct stasis_topic *topic, struct stasis_subscription *sub)
-{
-	size_t i;
-	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];
-			ao2_cleanup(sub);
-			return 0;
-		}
-	}
-
-	return -1;
 }
 
 /*!
@@ -286,10 +276,25 @@
 static int dispatch_exec(void *data)
 {
 	RAII_VAR(struct dispatch *, dispatch, data, ao2_cleanup);
-
-	ast_assert(dispatch->sub->topic != NULL);
-
-	dispatch->sub->callback(dispatch->sub->data, dispatch->sub->topic, dispatch->message);
+	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);
+	}
+
+	dispatch->sub->callback(dispatch->sub->data,
+				sub_topic,
+				dispatch->message);
+
 	return 0;
 }