X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=dd5cf1761a4992c9e7384d48675038c5438ece0e;hp=e06f3b3e079aae44f494ba6d4409f02120018af4;hb=f5ba75b4f0c0b44092c76bc931b25b24a2e62718;hpb=6f9449c22eef59294cf1e1dc3610a5cbf14baec0

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index e06f3b3e0..dd5cf1761 100644
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -19,6 +19,7 @@
 #include <inttypes.h>
 #include <unistd.h>
 #include <sys/stat.h>
+#include <stdint.h>
 
 #include <bin/lttng-consumerd/health-consumerd.h>
 #include <common/common.h>
@@ -36,7 +37,7 @@
 #include <common/optional.h>
 #include <common/buffer-view.h>
 #include <common/consumer/consumer.h>
-#include <stdint.h>
+#include <common/consumer/metadata-bucket.h>
 
 #include "kernel-consumer.h"
 
@@ -287,7 +288,7 @@ static int lttng_kconsumer_snapshot_channel(
 
 			subbuf_view = lttng_buffer_view_init(
 					subbuf_addr, 0, padded_len);
-			read_len = lttng_consumer_on_read_subbuffer_mmap(ctx,
+			read_len = lttng_consumer_on_read_subbuffer_mmap(
 					stream, &subbuf_view,
 					padded_len - len);
 			/*
@@ -1558,6 +1559,42 @@ end:
 	return ret;
 }
 
+static
+int get_next_subbuffer_metadata_check(struct lttng_consumer_stream *stream,
+		struct stream_subbuffer *subbuffer)
+{
+	int ret;
+	const char *addr;
+	bool coherent;
+
+	ret = kernctl_get_next_subbuf_metadata_check(stream->wait_fd,
+			&coherent);
+	if (ret) {
+		goto end;
+	}
+
+	ret = stream->read_subbuffer_ops.extract_subbuffer_info(
+			stream, subbuffer);
+	if (ret) {
+		goto end;
+	}
+
+	LTTNG_OPTIONAL_SET(&subbuffer->info.metadata.coherent, coherent);
+
+	ret = get_current_subbuf_addr(stream, &addr);
+	if (ret) {
+		goto end;
+	}
+
+	subbuffer->buffer.buffer = lttng_buffer_view_init(
+			addr, 0, subbuffer->info.data.padded_subbuf_size);
+	DBG("Got metadata packet with padded_subbuf_size = %lu, coherent = %s",
+			subbuffer->info.metadata.padded_subbuf_size,
+			coherent ? "true" : "false");
+end:
+	return ret;
+}
+
 static
 int put_next_subbuffer(struct lttng_consumer_stream *stream,
 		struct stream_subbuffer *subbuffer)
@@ -1576,15 +1613,53 @@ int put_next_subbuffer(struct lttng_consumer_stream *stream,
 	return ret;
 }
 
-static void lttng_kconsumer_set_stream_ops(
+static
+bool is_get_next_check_metadata_available(int tracer_fd)
+{
+	return kernctl_get_next_subbuf_metadata_check(tracer_fd, NULL) !=
+			-ENOTTY;
+}
+
+static
+int lttng_kconsumer_set_stream_ops(
 		struct lttng_consumer_stream *stream)
 {
-	if (stream->chan->output == CONSUMER_CHANNEL_MMAP) {
-		stream->read_subbuffer_ops.get_next_subbuffer =
-				get_next_subbuffer_mmap;
-	} else {
-		stream->read_subbuffer_ops.get_next_subbuffer =
-				get_next_subbuffer_splice;
+	int ret = 0;
+
+	if (stream->metadata_flag && stream->chan->is_live) {
+		DBG("Attempting to enable metadata bucketization for live consumers");
+		if (is_get_next_check_metadata_available(stream->wait_fd)) {
+			DBG("Kernel tracer supports get_next_subbuffer_metadata_check, metadata will be accumulated until a coherent state is reached");
+			stream->read_subbuffer_ops.get_next_subbuffer =
+					get_next_subbuffer_metadata_check;
+			ret = consumer_stream_enable_metadata_bucketization(
+					stream);
+			if (ret) {
+				goto end;
+			}
+		} else {
+			/*
+			 * The kernel tracer version is too old to indicate
+			 * when the metadata stream has reached a "coherent"
+			 * (parseable) point.
+			 *
+			 * This means that a live viewer may see an incoherent
+			 * sequence of metadata and fail to parse it.
+			 */
+			WARN("Kernel tracer does not support get_next_subbuffer_metadata_check which may cause live clients to fail to parse the metadata stream");
+			metadata_bucket_destroy(stream->metadata_bucket);
+			stream->metadata_bucket = NULL;
+		}
+	}
+
+	if (!stream->read_subbuffer_ops.get_next_subbuffer) {
+		if (stream->chan->output == CONSUMER_CHANNEL_MMAP) {
+			stream->read_subbuffer_ops.get_next_subbuffer =
+					get_next_subbuffer_mmap;
+		} else {
+			stream->read_subbuffer_ops.get_next_subbuffer =
+					get_next_subbuffer_splice;
+		}
 	}
 
 	if (stream->metadata_flag) {
@@ -1600,6 +1675,8 @@ static void lttng_kconsumer_set_stream_ops(
 	}
 
 	stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer;
+end:
+	return ret;
 }
 
 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
@@ -1640,7 +1717,10 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 		}
 	}
 
-	lttng_kconsumer_set_stream_ops(stream);
+	ret = lttng_kconsumer_set_stream_ops(stream);
+	if (ret) {
+		goto error_close_fd;
+	}
 
 	/* we return 0 to let the library handle the FD internally */
 	return 0;