X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=1af9840cd5184b0cfebad0bf680257fbbadac9a5;hp=2f09eac80763e2c20ae60b83e38e4593d5f84346;hb=6f9449c22eef59294cf1e1dc3610a5cbf14baec0;hpb=23d565989350c270c68e9a6c8edfbe2dd6a6895d

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 2f09eac80..1af9840cd 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -7,7 +7,6 @@
  *
  */
 
-#include <stdint.h>
 #define _LGPL_SOURCE
 #include <assert.h>
 #include <lttng/ust-ctl.h>
@@ -24,6 +23,7 @@
 #include <urcu/list.h>
 #include <signal.h>
 #include <stdbool.h>
+#include <stdint.h>
 
 #include <bin/lttng-consumerd/health-consumerd.h>
 #include <common/common.h>
@@ -36,6 +36,7 @@
 #include <common/consumer/consumer-timer.h>
 #include <common/utils.h>
 #include <common/index/index.h>
+#include <common/consumer/consumer.h>
 
 #include "ust-consumer.h"
 
@@ -127,7 +128,7 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
 	assert(channel);
 	assert(ctx);
 
-	stream = consumer_allocate_stream(
+	stream = consumer_stream_create(
 			channel,
 			channel->key,
 			key,
@@ -1040,7 +1041,7 @@ static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel,
 	do {
 		health_code_update();
 
-		ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
+		ret = lttng_consumer_read_subbuffer(metadata_stream, ctx, true);
 		if (ret < 0) {
 			goto error_stream;
 		}
@@ -1230,8 +1231,7 @@ static int snapshot_channel(struct lttng_consumer_channel *channel,
 			subbuf_view = lttng_buffer_view_init(
 					subbuf_addr, 0, padded_len);
 			read_len = lttng_consumer_on_read_subbuffer_mmap(ctx,
-					stream, &subbuf_view, padded_len - len,
-					NULL);
+					stream, &subbuf_view, padded_len - len);
 			if (use_relayd) {
 				if (read_len != len) {
 					ret = -EPERM;
@@ -2428,115 +2428,16 @@ int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream)
 	return ustctl_stream_close_wakeup_fd(stream->ustream);
 }
 
-/*
- * Populate index values of a UST stream. Values are set in big endian order.
- *
- * Return 0 on success or else a negative value.
- */
-static int get_index_values(struct ctf_packet_index *index,
-		struct ustctl_consumer_stream *ustream)
-{
-	int ret;
-	uint64_t packet_size, content_size, timestamp_begin, timestamp_end,
-			events_discarded, stream_id, stream_instance_id,
-			packet_seq_num;
-
-	ret = ustctl_get_timestamp_begin(ustream, &timestamp_begin);
-	if (ret < 0) {
-		PERROR("ustctl_get_timestamp_begin");
-		goto error;
-	}
-
-	ret = ustctl_get_timestamp_end(ustream, &timestamp_end);
-	if (ret < 0) {
-		PERROR("ustctl_get_timestamp_end");
-		goto error;
-	}
-
-	ret = ustctl_get_events_discarded(ustream, &events_discarded);
-	if (ret < 0) {
-		PERROR("ustctl_get_events_discarded");
-		goto error;
-	}
-
-	ret = ustctl_get_content_size(ustream, &content_size);
-	if (ret < 0) {
-		PERROR("ustctl_get_content_size");
-		goto error;
-	}
-
-	ret = ustctl_get_packet_size(ustream, &packet_size);
-	if (ret < 0) {
-		PERROR("ustctl_get_packet_size");
-		goto error;
-	}
-
-	ret = ustctl_get_stream_id(ustream, &stream_id);
-	if (ret < 0) {
-		PERROR("ustctl_get_stream_id");
-		goto error;
-	}
-
-	ret = ustctl_get_instance_id(ustream, &stream_instance_id);
-	if (ret < 0) {
-		PERROR("ustctl_get_instance_id");
-		goto error;
-	}
-
-	ret = ustctl_get_sequence_number(ustream, &packet_seq_num);
-	if (ret < 0) {
-		PERROR("ustctl_get_sequence_number");
-		goto error;
-	}
-
-	*index = (typeof(*index)) {
-		.offset = index->offset,
-		.packet_size = htobe64(packet_size),
-		.content_size = htobe64(content_size),
-		.timestamp_begin = htobe64(timestamp_begin),
-		.timestamp_end = htobe64(timestamp_end),
-		.events_discarded = htobe64(events_discarded),
-		.stream_id = htobe64(stream_id),
-		.stream_instance_id = htobe64(stream_instance_id),
-		.packet_seq_num = htobe64(packet_seq_num),
-	};
-
-error:
-	return ret;
-}
-
 static
-void metadata_stream_reset_cache(struct lttng_consumer_stream *stream,
-		struct consumer_metadata_cache *cache)
+void metadata_stream_reset_cache(struct lttng_consumer_stream *stream)
 {
-	DBG("Metadata stream update to version %" PRIu64,
-			cache->version);
+	DBG("Reset metadata cache of session %" PRIu64,
+			stream->chan->session_id);
 	stream->ust_metadata_pushed = 0;
-	stream->metadata_version = cache->version;
+	stream->metadata_version = stream->chan->metadata_cache->version;
 	stream->reset_metadata_flag = 1;
 }
 
-/*
- * Check if the version of the metadata stream and metadata cache match.
- * If the cache got updated, reset the metadata stream.
- * The stream lock and metadata cache lock MUST be held.
- * Return 0 on success, a negative value on error.
- */
-static
-int metadata_stream_check_version(struct lttng_consumer_stream *stream)
-{
-	int ret = 0;
-	struct consumer_metadata_cache *cache = stream->chan->metadata_cache;
-
-	if (cache->version == stream->metadata_version) {
-		goto end;
-	}
-	metadata_stream_reset_cache(stream, cache);
-
-end:
-	return ret;
-}
-
 /*
  * Write up to one packet from the metadata cache to the channel.
  *
@@ -2550,10 +2451,6 @@ int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
 	int ret;
 
 	pthread_mutex_lock(&stream->chan->metadata_cache->lock);
-	ret = metadata_stream_check_version(stream);
-	if (ret < 0) {
-		goto end;
-	}
 	if (stream->chan->metadata_cache->max_offset
 			== stream->ust_metadata_pushed) {
 		ret = 0;
@@ -2723,261 +2620,264 @@ end:
 	return ret;
 }
 
-static
-int update_stream_stats(struct lttng_consumer_stream *stream)
+static int consumer_stream_ust_on_wake_up(struct lttng_consumer_stream *stream)
 {
-	int ret;
-	uint64_t seq, discarded;
+	int ret = 0;
 
-	ret = ustctl_get_sequence_number(stream->ustream, &seq);
-	if (ret < 0) {
-		PERROR("ustctl_get_sequence_number");
-		goto end;
-	}
 	/*
-	 * Start the sequence when we extract the first packet in case we don't
-	 * start at 0 (for example if a consumer is not connected to the
-	 * session immediately after the beginning).
+	 * We can consume the 1 byte written into the wait_fd by
+	 * UST. Don't trigger error if we cannot read this one byte
+	 * (read returns 0), or if the error is EAGAIN or EWOULDBLOCK.
+	 *
+	 * This is only done when the stream is monitored by a thread,
+	 * before the flush is done after a hangup and if the stream
+	 * is not flagged with data since there might be nothing to
+	 * consume in the wait fd but still have data available
+	 * flagged by the consumer wake up pipe.
 	 */
-	if (stream->last_sequence_number == -1ULL) {
-		stream->last_sequence_number = seq;
-	} else if (seq > stream->last_sequence_number) {
-		stream->chan->lost_packets += seq -
-				stream->last_sequence_number - 1;
-	} else {
-		/* seq <= last_sequence_number */
-		ERR("Sequence number inconsistent : prev = %" PRIu64
-				", current = %" PRIu64,
-				stream->last_sequence_number, seq);
-		ret = -1;
-		goto end;
+	if (stream->monitor && !stream->hangup_flush_done && !stream->has_data) {
+		char dummy;
+		ssize_t readlen;
+
+		readlen = lttng_read(stream->wait_fd, &dummy, 1);
+		if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
+			ret = readlen;
+		}
 	}
-	stream->last_sequence_number = seq;
 
-	ret = ustctl_get_events_discarded(stream->ustream, &discarded);
-	if (ret < 0) {
-		PERROR("kernctl_get_events_discarded");
+	return ret;
+}
+
+static int extract_common_subbuffer_info(struct lttng_consumer_stream *stream,
+		struct stream_subbuffer *subbuf)
+{
+	int ret;
+
+	ret = ustctl_get_subbuf_size(
+			stream->ustream, &subbuf->info.data.subbuf_size);
+	if (ret) {
 		goto end;
 	}
-	if (discarded < stream->last_discarded_events) {
-		/*
-		 * Overflow has occurred. We assume only one wrap-around
-		 * has occurred.
-		 */
-		stream->chan->discarded_events +=
-				(1ULL << (CAA_BITS_PER_LONG - 1)) -
-				stream->last_discarded_events + discarded;
-	} else {
-		stream->chan->discarded_events += discarded -
-				stream->last_discarded_events;
+
+	ret = ustctl_get_padded_subbuf_size(
+			stream->ustream, &subbuf->info.data.padded_subbuf_size);
+	if (ret) {
+		goto end;
 	}
-	stream->last_discarded_events = discarded;
-	ret = 0;
 
 end:
 	return ret;
 }
 
-/*
- * Read subbuffer from the given stream.
- *
- * Stream and channel locks MUST be acquired by the caller.
- *
- * Return 0 on success else a negative value.
- */
-int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
-		struct lttng_consumer_local_data *ctx)
+static int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream,
+		struct stream_subbuffer *subbuf)
 {
-	unsigned long len, subbuf_size, padding;
-	int err, write_index = 1;
-	long ret = 0;
-	struct ustctl_consumer_stream *ustream;
-	struct ctf_packet_index index;
-	const char *subbuf_addr;
-	struct lttng_buffer_view subbuf_view;
+	int ret;
 
-	assert(stream);
-	assert(stream->ustream);
-	assert(ctx);
+	ret = extract_common_subbuffer_info(stream, subbuf);
+	if (ret) {
+		goto end;
+	}
 
-	DBG("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd,
-			stream->name);
+	subbuf->info.metadata.version = stream->chan->metadata_cache->version;
 
-	/* Ease our life for what's next. */
-	ustream = stream->ustream;
+end:
+	return ret;
+}
 
-	/*
-	 * We can consume the 1 byte written into the wait_fd by UST. Don't trigger
-	 * error if we cannot read this one byte (read returns 0), or if the error
-	 * is EAGAIN or EWOULDBLOCK.
-	 *
-	 * This is only done when the stream is monitored by a thread, before the
-	 * flush is done after a hangup and if the stream is not flagged with data
-	 * since there might be nothing to consume in the wait fd but still have
-	 * data available flagged by the consumer wake up pipe.
-	 */
-	if (stream->monitor && !stream->hangup_flush_done && !stream->has_data) {
-		char dummy;
-		ssize_t readlen;
+static int extract_data_subbuffer_info(struct lttng_consumer_stream *stream,
+		struct stream_subbuffer *subbuf)
+{
+	int ret;
 
-		readlen = lttng_read(stream->wait_fd, &dummy, 1);
-		if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
-			ret = readlen;
-			goto error;
-		}
+	ret = extract_common_subbuffer_info(stream, subbuf);
+	if (ret) {
+		goto end;
 	}
 
-retry:
-	/* Get the next subbuffer */
-	err = ustctl_get_next_subbuf(ustream);
-	if (err != 0) {
-		/*
-		 * Populate metadata info if the existing info has
-		 * already been read.
-		 */
-		if (stream->metadata_flag) {
-			ret = commit_one_metadata_packet(stream);
-			if (ret <= 0) {
-				goto error;
-			}
-			goto retry;
-		}
+	ret = ustctl_get_packet_size(
+			stream->ustream, &subbuf->info.data.packet_size);
+	if (ret < 0) {
+		PERROR("Failed to get sub-buffer packet size");
+		goto end;
+	}
 
-		ret = err;	/* ustctl_get_next_subbuf returns negative, caller expect positive. */
-		/*
-		 * This is a debug message even for single-threaded consumer,
-		 * because poll() have more relaxed criterions than get subbuf,
-		 * so get_subbuf may fail for short race windows where poll()
-		 * would issue wakeups.
-		 */
-		DBG("Reserving sub buffer failed (everything is normal, "
-				"it is due to concurrency) [ret: %d]", err);
-		goto error;
+	ret = ustctl_get_content_size(
+			stream->ustream, &subbuf->info.data.content_size);
+	if (ret < 0) {
+		PERROR("Failed to get sub-buffer content size");
+		goto end;
 	}
-	assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
 
-	if (!stream->metadata_flag) {
-		index.offset = htobe64(stream->out_fd_offset);
-		ret = get_index_values(&index, ustream);
-		if (ret < 0) {
-			err = ustctl_put_subbuf(ustream);
-			assert(err == 0);
-			goto error;
-		}
+	ret = ustctl_get_timestamp_begin(
+			stream->ustream, &subbuf->info.data.timestamp_begin);
+	if (ret < 0) {
+		PERROR("Failed to get sub-buffer begin timestamp");
+		goto end;
+	}
 
-		/* Update the stream's sequence and discarded events count. */
-		ret = update_stream_stats(stream);
-		if (ret < 0) {
-			PERROR("kernctl_get_events_discarded");
-			err = ustctl_put_subbuf(ustream);
-			assert(err == 0);
-			goto error;
+	ret = ustctl_get_timestamp_end(
+			stream->ustream, &subbuf->info.data.timestamp_end);
+	if (ret < 0) {
+		PERROR("Failed to get sub-buffer end timestamp");
+		goto end;
+	}
+
+	ret = ustctl_get_events_discarded(
+			stream->ustream, &subbuf->info.data.events_discarded);
+	if (ret) {
+		PERROR("Failed to get sub-buffer events discarded count");
+		goto end;
+	}
+
+	ret = ustctl_get_sequence_number(stream->ustream,
+			&subbuf->info.data.sequence_number.value);
+	if (ret) {
+		/* May not be supported by older LTTng-modules. */
+		if (ret != -ENOTTY) {
+			PERROR("Failed to get sub-buffer sequence number");
+			goto end;
 		}
 	} else {
-		write_index = 0;
+		subbuf->info.data.sequence_number.is_set = true;
 	}
 
-	/* Get the full padded subbuffer size */
-	err = ustctl_get_padded_subbuf_size(ustream, &len);
-	assert(err == 0);
+	ret = ustctl_get_stream_id(
+			stream->ustream, &subbuf->info.data.stream_id);
+	if (ret < 0) {
+		PERROR("Failed to get stream id");
+		goto end;
+	}
 
-	/* Get subbuffer data size (without padding) */
-	err = ustctl_get_subbuf_size(ustream, &subbuf_size);
-	assert(err == 0);
+	ret = ustctl_get_instance_id(stream->ustream,
+			&subbuf->info.data.stream_instance_id.value);
+	if (ret) {
+		/* May not be supported by older LTTng-modules. */
+		if (ret != -ENOTTY) {
+			PERROR("Failed to get stream instance id");
+			goto end;
+		}
+	} else {
+		subbuf->info.data.stream_instance_id.is_set = true;
+	}
+end:
+	return ret;
+}
 
-	/* Make sure we don't get a subbuffer size bigger than the padded */
-	assert(len >= subbuf_size);
+static int get_next_subbuffer_common(struct lttng_consumer_stream *stream,
+		struct stream_subbuffer *subbuffer)
+{
+	int ret;
+	const char *addr;
 
-	padding = len - subbuf_size;
+	ret = stream->read_subbuffer_ops.extract_subbuffer_info(
+			stream, subbuffer);
+	if (ret) {
+		goto end;
+	}
 
-	ret = get_current_subbuf_addr(stream, &subbuf_addr);
+	ret = get_current_subbuf_addr(stream, &addr);
 	if (ret) {
-		write_index = 0;
-		goto error_put_subbuf;
+		goto end;
 	}
 
-	subbuf_view = lttng_buffer_view_init(subbuf_addr, 0, len);
+	subbuffer->buffer.buffer = lttng_buffer_view_init(
+			addr, 0, subbuffer->info.data.padded_subbuf_size);
+	assert(subbuffer->buffer.buffer.data != NULL);
+end:
+	return ret;
+}
 
-	/* write the subbuffer to the tracefile */
-	ret = lttng_consumer_on_read_subbuffer_mmap(
-			ctx, stream, &subbuf_view, padding, &index);
-	/*
-	 * The mmap operation should write subbuf_size amount of data when
-	 * network streaming or the full padding (len) size when we are _not_
-	 * streaming.
-	 */
-	if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) ||
-			(ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) {
-		/*
-		 * Display the error but continue processing to try to release the
-		 * subbuffer. This is a DBG statement since any unexpected kill or
-		 * signal, the application gets unregistered, relayd gets closed or
-		 * anything that affects the buffer lifetime will trigger this error.
-		 * So, for the sake of the user, don't print this error since it can
-		 * happen and it is OK with the code flow.
-		 */
-		DBG("Error writing to tracefile "
-				"(ret: %ld != len: %lu != subbuf_size: %lu)",
-				ret, len, subbuf_size);
-		write_index = 0;
-	}
-error_put_subbuf:
-	err = ustctl_put_next_subbuf(ustream);
-	assert(err == 0);
+static int get_next_subbuffer(struct lttng_consumer_stream *stream,
+		struct stream_subbuffer *subbuffer)
+{
+	int ret;
 
-	/*
-	 * This will consumer the byte on the wait_fd if and only if there is not
-	 * next subbuffer to be acquired.
-	 */
-	if (!stream->metadata_flag) {
-		ret = notify_if_more_data(stream, ctx);
-		if (ret < 0) {
-			goto error;
-		}
+	ret = ustctl_get_next_subbuf(stream->ustream);
+	if (ret) {
+		goto end;
 	}
 
-	/* Write index if needed. */
-	if (!write_index) {
+	ret = get_next_subbuffer_common(stream, subbuffer);
+	if (ret) {
 		goto end;
 	}
+end:
+	return ret;
+}
 
-	if (stream->chan->live_timer_interval && !stream->metadata_flag) {
-		/*
-		 * In live, block until all the metadata is sent.
-		 */
-		pthread_mutex_lock(&stream->metadata_timer_lock);
-		assert(!stream->missed_metadata_flush);
-		stream->waiting_on_metadata = true;
-		pthread_mutex_unlock(&stream->metadata_timer_lock);
-
-		err = consumer_stream_sync_metadata(ctx, stream->session_id);
-
-		pthread_mutex_lock(&stream->metadata_timer_lock);
-		stream->waiting_on_metadata = false;
-		if (stream->missed_metadata_flush) {
-			stream->missed_metadata_flush = false;
-			pthread_mutex_unlock(&stream->metadata_timer_lock);
-			(void) consumer_flush_ust_index(stream);
-		} else {
-			pthread_mutex_unlock(&stream->metadata_timer_lock);
+static int get_next_subbuffer_metadata(struct lttng_consumer_stream *stream,
+		struct stream_subbuffer *subbuffer)
+{
+	int ret;
+
+	ret = ustctl_get_next_subbuf(stream->ustream);
+	if (ret) {
+		ret = commit_one_metadata_packet(stream);
+		if (ret < 0) {
+			goto end;
+		} else if (ret == 0) {
+			/* Not an error, the cache is empty. */
+			ret = -ENODATA;
+			goto end;
 		}
 
-		if (err < 0) {
-			goto error;
+		ret = ustctl_get_next_subbuf(stream->ustream);
+		if (ret) {
+			goto end;
 		}
 	}
 
-	assert(!stream->metadata_flag);
-	err = consumer_stream_write_index(stream, &index);
-	if (err < 0) {
-		goto error;
+	ret = get_next_subbuffer_common(stream, subbuffer);
+	if (ret) {
+		goto end;
 	}
-
 end:
-error:
 	return ret;
 }
 
+static int put_next_subbuffer(struct lttng_consumer_stream *stream,
+		struct stream_subbuffer *subbuffer)
+{
+	const int ret = ustctl_put_next_subbuf(stream->ustream);
+
+	assert(ret == 0);
+	return ret;
+}
+
+static int signal_metadata(struct lttng_consumer_stream *stream,
+		struct lttng_consumer_local_data *ctx)
+{
+	return pthread_cond_broadcast(&stream->metadata_rdv) ? -errno : 0;
+}
+
+static void lttng_ustconsumer_set_stream_ops(
+		struct lttng_consumer_stream *stream)
+{
+	stream->read_subbuffer_ops.on_wake_up = consumer_stream_ust_on_wake_up;
+	if (stream->metadata_flag) {
+		stream->read_subbuffer_ops.get_next_subbuffer =
+				get_next_subbuffer_metadata;
+		stream->read_subbuffer_ops.extract_subbuffer_info =
+				extract_metadata_subbuffer_info;
+		stream->read_subbuffer_ops.reset_metadata =
+				metadata_stream_reset_cache;
+		stream->read_subbuffer_ops.on_sleep = signal_metadata;
+	} else {
+		stream->read_subbuffer_ops.get_next_subbuffer =
+				get_next_subbuffer;
+		stream->read_subbuffer_ops.extract_subbuffer_info =
+				extract_data_subbuffer_info;
+		stream->read_subbuffer_ops.on_sleep = notify_if_more_data;
+		if (stream->chan->is_live) {
+			stream->read_subbuffer_ops.send_live_beacon =
+					consumer_flush_ust_index;
+		}
+	}
+
+	stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer;
+}
+
 /*
  * Called when a stream is created.
  *
@@ -3000,6 +2900,8 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 			goto error;
 		}
 	}
+
+	lttng_ustconsumer_set_stream_ops(stream);
 	ret = 0;
 
 error: