X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=814d93e76887c64d1ba8e54ad813d8f909f2850b;hb=3b4a46a4284f7d8a3ab12871cea143bc90d7121f;hp=929be6553a7ea76dc54694cb149d988b448efa3b;hpb=e0547b835d76e0a3a2c3082d7c10e6902a1f3e04;p=lttng-tools.git

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index 929be6553..814d93e76 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -47,6 +47,7 @@
 #include <common/consumer/consumer-stream.h>
 #include <common/consumer/consumer-testpoint.h>
 #include <common/align.h>
+#include <common/consumer/consumer-metadata-cache.h>
 
 struct lttng_consumer_global_data consumer_data = {
 	.stream_count = 0,
@@ -1070,7 +1071,7 @@ int consumer_add_channel(struct lttng_consumer_channel *channel,
  */
 static int update_poll_array(struct lttng_consumer_local_data *ctx,
 		struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
-		struct lttng_ht *ht)
+		struct lttng_ht *ht, int *nb_inactive_fd)
 {
 	int i = 0;
 	struct lttng_ht_iter iter;
@@ -1082,6 +1083,7 @@ static int update_poll_array(struct lttng_consumer_local_data *ctx,
 	assert(local_stream);
 
 	DBG("Updating poll fd array");
+	*nb_inactive_fd = 0;
 	rcu_read_lock();
 	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
 		/*
@@ -1092,9 +1094,14 @@ static int update_poll_array(struct lttng_consumer_local_data *ctx,
 		 * just after the check. However, this is OK since the stream(s) will
 		 * be deleted once the thread is notified that the end point state has
 		 * changed where this function will be called back again.
+		 *
+		 * We track the number of inactive FDs because they still need to be
+		 * closed by the polling thread after a wakeup on the data_pipe or
+		 * metadata_pipe.
 		 */
 		if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM ||
 				stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
+			(*nb_inactive_fd)++;
 			continue;
 		}
 		/*
@@ -2052,6 +2059,10 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	pthread_mutex_lock(&consumer_data.lock);
 	pthread_mutex_lock(&stream->chan->lock);
 	pthread_mutex_lock(&stream->lock);
+	if (stream->chan->metadata_cache) {
+		/* Only applicable to userspace consumers. */
+		pthread_mutex_lock(&stream->chan->metadata_cache->lock);
+	}
 
 	/* Remove any reference to that stream. */
 	consumer_stream_delete(stream, ht);
@@ -2075,6 +2086,9 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	 */
 	stream->chan->metadata_stream = NULL;
 
+	if (stream->chan->metadata_cache) {
+		pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
+	}
 	pthread_mutex_unlock(&stream->lock);
 	pthread_mutex_unlock(&stream->chan->lock);
 	pthread_mutex_unlock(&consumer_data.lock);
@@ -2441,6 +2455,8 @@ void *consumer_thread_data_poll(void *data)
 	struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
 	/* local view of consumer_data.fds_count */
 	int nb_fd = 0;
+	/* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
+	int nb_inactive_fd = 0;
 	struct lttng_consumer_local_data *ctx = data;
 	ssize_t len;
 
@@ -2497,7 +2513,7 @@ void *consumer_thread_data_poll(void *data)
 				goto end;
 			}
 			ret = update_poll_array(ctx, &pollfd, local_stream,
-					data_ht);
+					data_ht, &nb_inactive_fd);
 			if (ret < 0) {
 				ERR("Error in allocating pollfd or local_outfds");
 				lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
@@ -2510,7 +2526,7 @@ void *consumer_thread_data_poll(void *data)
 		pthread_mutex_unlock(&consumer_data.lock);
 
 		/* No FDs and consumer_quit, consumer_cleanup the thread */
-		if (nb_fd == 0 && consumer_quit == 1) {
+		if (nb_fd == 0 && consumer_quit == 1 && nb_inactive_fd == 0) {
 			err = 0;	/* All is OK */
 			goto end;
 		}