X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=5f1f93b0e1024c6c1eee422d80930b71ccaaf1a8;hp=1a7de1cfd59cf0a095f80794e0ff96d462386a93;hb=b1316da1ffbd276fc8271e7a9438e683ad352781;hpb=ff588497b3dfc3138c9ce005e9270ed5568c05df

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 1a7de1cfd..5f1f93b0e 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -1283,6 +1283,17 @@ error_unlock:
 	return ret;
 }
 
+static
+void metadata_stream_reset_cache_consumed_position(
+		struct lttng_consumer_stream *stream)
+{
+	ASSERT_LOCKED(stream->lock);
+
+	DBG("Reset metadata cache of session %" PRIu64,
+			stream->chan->session_id);
+	stream->ust_metadata_pushed = 0;
+}
+
 /*
  * Receive the metadata updates from the sessiond. Supports receiving
  * overlapping metadata, but is needs to always belong to a contiguous
@@ -1297,6 +1308,7 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
 {
 	int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
 	char *metadata_str;
+	enum consumer_metadata_cache_write_status cache_write_status;
 
 	DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len);
 
@@ -1320,9 +1332,40 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
 	health_code_update();
 
 	pthread_mutex_lock(&channel->metadata_cache->lock);
-	ret = consumer_metadata_cache_write(channel, offset, len, version,
-			metadata_str);
-	if (ret < 0) {
+	cache_write_status = consumer_metadata_cache_write(
+			channel, offset, len, version, metadata_str);
+	pthread_mutex_unlock(&channel->metadata_cache->lock);
+	switch (cache_write_status) {
+	case CONSUMER_METADATA_CACHE_WRITE_STATUS_NO_CHANGE:
+		/*
+		 * The write entirely overlapped with existing contents of the
+		 * same metadata version (same content); there is nothing to do.
+		 */
+		break;
+	case CONSUMER_METADATA_CACHE_WRITE_STATUS_INVALIDATED:
+		/*
+		 * The metadata cache was invalidated (previously pushed
+		 * content has been overwritten). Reset the stream's consumed
+		 * metadata position to ensure the metadata poll thread consumes
+		 * the whole cache.
+		 */
+		pthread_mutex_lock(&channel->metadata_stream->lock);
+		metadata_stream_reset_cache_consumed_position(
+				channel->metadata_stream);
+		pthread_mutex_unlock(&channel->metadata_stream->lock);
+		/* Fall-through. */
+	case CONSUMER_METADATA_CACHE_WRITE_STATUS_APPENDED_CONTENT:
+		/*
+		 * In both cases, the metadata poll thread has new data to
+		 * consume.
+		 */
+		ret = consumer_metadata_wakeup_pipe(channel);
+		if (ret) {
+			ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
+			goto end_free;
+		}
+		break;
+	case CONSUMER_METADATA_CACHE_WRITE_STATUS_ERROR:
 		/* Unable to handle metadata. Notify session daemon. */
 		ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
 		/*
@@ -1330,10 +1373,10 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
 		 * 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);
 		goto end_free;
+	default:
+		abort();
 	}
-	pthread_mutex_unlock(&channel->metadata_cache->lock);
 
 	if (!wait) {
 		goto end_free;
@@ -2465,15 +2508,6 @@ int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream)
 	return ustctl_stream_close_wakeup_fd(stream->ustream);
 }
 
-static
-void metadata_stream_reset_cache_consumed_position(
-		struct lttng_consumer_stream *stream)
-{
-	DBG("Reset metadata cache of session %" PRIu64,
-			stream->chan->session_id);
-	stream->ust_metadata_pushed = 0;
-}
-
 /*
  * Write up to one packet from the metadata cache to the channel.
  *
@@ -2966,6 +3000,7 @@ static int put_next_subbuffer(struct lttng_consumer_stream *stream,
 static int signal_metadata(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx)
 {
+	ASSERT_LOCKED(stream->metadata_rdv_lock);
 	return pthread_cond_broadcast(&stream->metadata_rdv) ? -errno : 0;
 }
 
@@ -3051,6 +3086,7 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
 
 	assert(stream);
 	assert(stream->ustream);
+	ASSERT_LOCKED(stream->lock);
 
 	DBG("UST consumer checking data pending");
 
@@ -3063,7 +3099,9 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
 		uint64_t contiguous, pushed;
 
 		/* Ease our life a bit. */
+		pthread_mutex_lock(&stream->chan->metadata_cache->lock);
 		contiguous = stream->chan->metadata_cache->max_offset;
+		pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
 		pushed = stream->ust_metadata_pushed;
 
 		/*