X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer-stream.c;h=1f0aad00b5b05a5fd37ce2646d5e56a24a95a421;hb=27e2fa5df07c5bed5a077bdf475c93e9a4d4459d;hp=4243991a47b98fb0dddbdb3d4d7aeff9252df187;hpb=6f1177cf8de79015ed38f48749029c81240b0eb6;p=lttng-tools.git

diff --git a/src/common/consumer/consumer-stream.c b/src/common/consumer/consumer-stream.c
index 4243991a4..1f0aad00b 100644
--- a/src/common/consumer/consumer-stream.c
+++ b/src/common/consumer/consumer-stream.c
@@ -31,6 +31,7 @@
 #include <common/utils.h>
 #include <common/consumer/consumer.h>
 #include <common/consumer/consumer-timer.h>
+#include <common/consumer/metadata-bucket.h>
 
 #include "consumer-stream.h"
 
@@ -78,8 +79,7 @@ static int consumer_stream_update_stats(struct lttng_consumer_stream *stream,
 {
 	int ret = 0;
 	uint64_t sequence_number;
-	const uint64_t discarded_events =
-			LTTNG_OPTIONAL_GET(subbuf->info.data.sequence_number);
+	const uint64_t discarded_events = subbuf->info.data.events_discarded;
 
 	if (!subbuf->info.data.sequence_number.is_set) {
 		/* Command not supported by the tracer. */
@@ -163,7 +163,7 @@ static ssize_t consumer_stream_consume_mmap(
 			subbuffer->info.data.subbuf_size;
 
 	return lttng_consumer_on_read_subbuffer_mmap(
-			ctx, stream, &subbuffer->buffer.buffer, padding_size);
+			stream, &subbuffer->buffer.buffer, padding_size);
 }
 
 static ssize_t consumer_stream_consume_splice(
@@ -206,6 +206,7 @@ static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 		struct lttng_consumer_local_data *ctx)
 {
 	int ret;
+	enum sync_metadata_status status;
 
 	assert(metadata);
 	assert(metadata->metadata_flag);
@@ -253,7 +254,7 @@ static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 			/*
 			 * Empty the metadata cache and flush the current stream.
 			 */
-			ret = lttng_kconsumer_sync_metadata(metadata);
+			status = lttng_kconsumer_sync_metadata(metadata);
 			break;
 		case LTTNG_CONSUMER32_UST:
 		case LTTNG_CONSUMER64_UST:
@@ -261,18 +262,23 @@ static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 			 * Ask the sessiond if we have new metadata waiting and update the
 			 * consumer metadata cache.
 			 */
-			ret = lttng_ustconsumer_sync_metadata(ctx, metadata);
+			status = lttng_ustconsumer_sync_metadata(ctx, metadata);
 			break;
 		default:
-			assert(0);
-			ret = -1;
-			break;
+			abort();
 		}
-		/*
-		 * Error or no new metadata, we exit here.
-		 */
-		if (ret <= 0 || ret == ENODATA) {
+
+		switch (status) {
+		case SYNC_METADATA_STATUS_NEW_DATA:
+			break;
+		case SYNC_METADATA_STATUS_NO_DATA:
+			ret = 0;
 			goto end_unlock_mutex;
+		case SYNC_METADATA_STATUS_ERROR:
+			ret = -1;
+			goto end_unlock_mutex;
+		default:
+			abort();
 		}
 
 		/*
@@ -294,7 +300,7 @@ static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 		 */
 		pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
 		pthread_mutex_unlock(&metadata->metadata_rdv_lock);
-	} while (ret == EAGAIN);
+	} while (status == SYNC_METADATA_STATUS_NEW_DATA);
 
 	/* Success */
 	return 0;
@@ -402,8 +408,8 @@ int metadata_stream_check_version(struct lttng_consumer_stream *stream,
 	}
 
 	DBG("New metadata version detected");
-	stream->metadata_version = subbuffer->info.metadata.version;
-	stream->reset_metadata_flag = 1;
+	consumer_stream_metadata_set_version(stream,
+			subbuffer->info.metadata.version);
 
 	if (stream->read_subbuffer_ops.reset_metadata) {
 		stream->read_subbuffer_ops.reset_metadata(stream);
@@ -736,6 +742,7 @@ void consumer_stream_free(struct lttng_consumer_stream *stream)
 {
 	assert(stream);
 
+	metadata_bucket_destroy(stream->metadata_bucket);
 	call_rcu(&stream->node.head, free_stream_rcu);
 }
 
@@ -1001,3 +1008,71 @@ bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
 	assert(stream);
 	return cds_lfht_is_node_deleted(&stream->node.node);
 }
+
+static ssize_t metadata_bucket_flush(
+		const struct stream_subbuffer *buffer, void *data)
+{
+	ssize_t ret;
+	struct lttng_consumer_stream *stream = data;
+
+	ret = consumer_stream_consume_mmap(NULL, stream, buffer);
+	if (ret < 0) {
+		goto end;
+	}
+end:
+	return ret;
+}
+
+static ssize_t metadata_bucket_consume(
+		struct lttng_consumer_local_data *unused,
+		struct lttng_consumer_stream *stream,
+		const struct stream_subbuffer *subbuffer)
+{
+	ssize_t ret;
+	enum metadata_bucket_status status;
+
+	status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
+	switch (status) {
+	case METADATA_BUCKET_STATUS_OK:
+		/* Return consumed size. */
+		ret = subbuffer->buffer.buffer.size;
+		break;
+	default:
+		ret = -1;
+	}
+
+	return ret;
+}
+
+int consumer_stream_enable_metadata_bucketization(
+		struct lttng_consumer_stream *stream)
+{
+	int ret = 0;
+
+	assert(stream->metadata_flag);
+	assert(!stream->metadata_bucket);
+	assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
+
+	stream->metadata_bucket = metadata_bucket_create(
+			metadata_bucket_flush, stream);
+	if (!stream->metadata_bucket) {
+		ret = -1;
+		goto end;
+	}
+
+	stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
+end:
+	return ret;
+}
+
+void consumer_stream_metadata_set_version(
+		struct lttng_consumer_stream *stream, uint64_t new_version)
+{
+	assert(new_version > stream->metadata_version);
+	stream->metadata_version = new_version;
+	stream->reset_metadata_flag = 1;
+
+	if (stream->metadata_bucket) {
+		metadata_bucket_reset(stream->metadata_bucket);
+	}
+}