X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=0e9c52e835dc594d2fa773452045208a26ea3cac;hb=f96e4545bd7a4dd9c58eb7e8a42eafe0b18089ad;hp=cc6c978686ffaf18697c7a54a84ff581f7bf579b;hpb=d4298c99f2fc4864cdabb3a90fed4d1117308a51;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index cc6c97868..0e9c52e83 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -319,9 +319,9 @@ static void cleanup_relayd_ht(void)
 		destroy_relayd(relayd);
 	}
 
-	lttng_ht_destroy(consumer_data.relayd_ht);
-
 	rcu_read_unlock();
+
+	lttng_ht_destroy(consumer_data.relayd_ht);
 }
 
 /*
@@ -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);
 	}
 
@@ -719,6 +720,8 @@ struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
 	obj->net_seq_idx = net_seq_idx;
 	obj->refcount = 0;
 	obj->destroy_flag = 0;
+	obj->control_sock.sock.fd = -1;
+	obj->data_sock.sock.fd = -1;
 	lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
 	pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
 
@@ -1937,8 +1940,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 +2010,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 +2561,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 +2584,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 +2750,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 */