From: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Date: Fri, 26 Apr 2013 14:27:51 +0000 (-0400)
Subject: Fix: channel management thread should hold a refcount
X-Git-Tag: v2.2.0-rc2~36
X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=commitdiff_plain;h=f2ad556de9fb7e0920938f1e3d4b1fcb0709beb6

Fix: channel management thread should hold a refcount

In order to handle teardown caused by relayd being unresponsive, we need
to ensure that we correctly handle that the streams can be cleaned up
before the channel wakeup pipe gets closed. We achieve this by making
the channel management thread hold a refcount on the channel object, so
it only gets destroyed when all streams, _and_ the channel management
thread, have released their reference.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Signed-off-by: David Goulet <dgoulet@efficios.com>
---

diff --git a/src/common/consumer.c b/src/common/consumer.c
index cc6c97868..343855797 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -516,8 +516,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 	}
 	rcu_read_unlock();
 
-	uatomic_dec(&stream->chan->refcount);
-	if (!uatomic_read(&stream->chan->refcount)
+	if (!uatomic_sub_return(&stream->chan->refcount, 1)
 			&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
 		free_chan = stream->chan;
 	}
@@ -659,6 +658,8 @@ static int add_stream(struct lttng_consumer_stream *stream,
 	 * stream.
 	 */
 	if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+		/* Increment refcount before decrementing nb_init_stream_left */
+		cmm_smp_wmb();
 		uatomic_dec(&stream->chan->nb_init_stream_left);
 	}
 
@@ -1937,8 +1938,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	rcu_read_unlock();
 
 	/* Atomically decrement channel refcount since other threads can use it. */
-	uatomic_dec(&stream->chan->refcount);
-	if (!uatomic_read(&stream->chan->refcount)
+	if (!uatomic_sub_return(&stream->chan->refcount, 1)
 			&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
 		/* Go for channel deletion! */
 		free_chan = stream->chan;
@@ -2008,6 +2008,8 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 	 * stream.
 	 */
 	if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+		/* Increment refcount before decrementing nb_init_stream_left */
+		cmm_smp_wmb();
 		uatomic_dec(&stream->chan->nb_init_stream_left);
 	}
 
@@ -2557,6 +2559,13 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
 			ht->hash_fct(&channel->key, lttng_ht_seed),
 			ht->match_fct, &channel->key,
 			&iter.iter, stream, node_channel_id.node) {
+		/*
+		 * Protect against teardown with mutex.
+		 */
+		pthread_mutex_lock(&stream->lock);
+		if (cds_lfht_is_node_deleted(&stream->node.node)) {
+			goto next;
+		}
 		switch (consumer_data.type) {
 		case LTTNG_CONSUMER_KERNEL:
 			break;
@@ -2573,6 +2582,8 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
 			ERR("Unknown consumer_data type");
 			assert(0);
 		}
+	next:
+		pthread_mutex_unlock(&stream->lock);
 	}
 	rcu_read_unlock();
 }
@@ -2737,6 +2748,12 @@ restart:
 				ret = lttng_ht_del(channel_ht, &iter);
 				assert(ret == 0);
 				consumer_close_channel_streams(chan);
+
+				/* Release our own refcount */
+				if (!uatomic_sub_return(&chan->refcount, 1)
+						&& !uatomic_read(&chan->nb_init_stream_left)) {
+					consumer_del_channel(chan);
+				}
 			}
 
 			/* Release RCU lock for the channel looked up */
diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 89b946896..2bbc3f5e4 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -859,6 +859,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			goto end_channel_error;
 		}
 
+		/*
+		 * Set refcount to 1 for owner. Below, we will pass
+		 * ownership to the consumer_thread_channel_poll()
+		 * thread.
+		 */
+		channel->refcount = 1;
+
 		/* Build channel attributes from received message. */
 		attr.subbuf_size = msg.u.ask_channel.subbuf_size;
 		attr.num_subbuf = msg.u.ask_channel.num_subbuf;