X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=f7ffb0febf5b7bcdc338b30975ae7e3379724935;hb=a32bd7757cc9f642b3977897819982ed1eb3be80;hp=8bc69006b0e1f30744ea41524e95d7ed1af3de0b;hpb=acdb9057878ddbd9c112206f3c1c4c2104093088;p=lttng-tools.git

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 8bc69006b..f7ffb0feb 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -185,21 +185,21 @@ error:
 static int send_stream_to_thread(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx)
 {
-	int ret, stream_pipe;
+	int ret;
+	struct lttng_pipe *stream_pipe;
 
 	/* Get the right pipe where the stream will be sent. */
 	if (stream->metadata_flag) {
-		stream_pipe = ctx->consumer_metadata_pipe[1];
+		stream_pipe = ctx->consumer_metadata_pipe;
 	} else {
-		stream_pipe = lttng_pipe_get_writefd(ctx->consumer_data_pipe);
+		stream_pipe = ctx->consumer_data_pipe;
 	}
 
-	do {
-		ret = write(stream_pipe, &stream, sizeof(stream));
-	} while (ret < 0 && errno == EINTR);
+	ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream));
 	if (ret < 0) {
-		PERROR("Consumer write %s stream to pipe %d",
-				stream->metadata_flag ? "metadata" : "data", stream_pipe);
+		ERR("Consumer write %s stream to pipe %d",
+				stream->metadata_flag ? "metadata" : "data",
+				lttng_pipe_get_writefd(stream_pipe));
 	}
 
 	return ret;
@@ -491,10 +491,6 @@ static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
 
 	channel->wait_fd = ustctl_channel_get_wait_fd(channel->uchan);
 
-	if (ret < 0) {
-		goto error;
-	}
-
 	/* Open all streams for this channel. */
 	ret = create_ust_streams(channel, ctx);
 	if (ret < 0) {
@@ -558,6 +554,11 @@ int lttng_ustconsumer_push_metadata(struct lttng_consumer_channel *metadata,
 
 	DBG("UST consumer writing metadata to channel %s", metadata->name);
 
+	if (!metadata->metadata_stream) {
+		ret = 0;
+		goto error;
+	}
+
 	assert(target_offset <= metadata->metadata_cache->max_offset);
 	ret = ustctl_write_metadata_to_channel(metadata->uchan,
 			metadata_str + target_offset, len);
@@ -629,11 +630,17 @@ static int close_metadata(uint64_t chan_key)
 	}
 
 	pthread_mutex_lock(&consumer_data.lock);
-	if (!cds_lfht_is_node_deleted(&channel->node.node)) {
-		if (channel->switch_timer_enabled == 1) {
-			DBG("Deleting timer on metadata channel");
-			consumer_timer_switch_stop(channel);
-		}
+
+	if (cds_lfht_is_node_deleted(&channel->node.node)) {
+		goto error_unlock;
+	}
+
+	if (channel->switch_timer_enabled == 1) {
+		DBG("Deleting timer on metadata channel");
+		consumer_timer_switch_stop(channel);
+	}
+
+	if (channel->metadata_stream) {
 		ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream);
 		if (ret < 0) {
 			ERR("UST consumer unable to close fd of metadata (ret: %d)", ret);
@@ -728,13 +735,33 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
 		goto end_free;
 	}
 
+	/*
+	 * XXX: The consumer data lock is acquired before calling metadata cache
+	 * write which calls push metadata that MUST be protected by the consumer
+	 * lock in order to be able to check the validity of the metadata stream of
+	 * the channel.
+	 *
+	 * Note that this will be subject to change to better fine grained locking
+	 * and ultimately try to get rid of this global consumer data lock.
+	 */
+	pthread_mutex_lock(&consumer_data.lock);
+
 	pthread_mutex_lock(&channel->metadata_cache->lock);
 	ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
 	if (ret < 0) {
 		/* Unable to handle metadata. Notify session daemon. */
 		ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
+		/*
+		 * Skip metadata flush on write error since the offset and len might
+		 * not have been updated which could create an infinite loop below when
+		 * waiting for the metadata cache to be flushed.
+		 */
+		pthread_mutex_unlock(&channel->metadata_cache->lock);
+		pthread_mutex_unlock(&consumer_data.lock);
+		goto end_free;
 	}
 	pthread_mutex_unlock(&channel->metadata_cache->lock);
+	pthread_mutex_unlock(&consumer_data.lock);
 
 	while (consumer_metadata_cache_flushed(channel, offset + len)) {
 		DBG("Waiting for metadata to be flushed");