X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=6a508143e0d7ab22796e3e9ad6fcf233258ae2bc;hb=84a182ce022e128395c8d1e5762883f5825582d7;hp=6a692b9ba5a8bffd7826bd164dfc3f1de9f8c9df;hpb=be2b50c7b31b4ee99c62cc60ed1be74cef3fc61e;p=lttng-tools.git

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 6a692b9ba..6a508143e 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -40,6 +40,7 @@
 #include <common/consumer-stream.h>
 #include <common/consumer-timer.h>
 #include <common/utils.h>
+#include <common/index/index.h>
 
 #include "ust-consumer.h"
 
@@ -116,14 +117,15 @@ static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
 		const char *pathname, const char *name, uid_t uid, gid_t gid,
 		uint64_t relayd_id, uint64_t key, enum lttng_event_output output,
 		uint64_t tracefile_size, uint64_t tracefile_count,
-		uint64_t session_id_per_pid, unsigned int monitor)
+		uint64_t session_id_per_pid, unsigned int monitor,
+		unsigned int live_timer_interval)
 {
 	assert(pathname);
 	assert(name);
 
 	return consumer_allocate_channel(key, session_id, pathname, name, uid,
 			gid, relayd_id, output, tracefile_size,
-			tracefile_count, session_id_per_pid, monitor);
+			tracefile_count, session_id_per_pid, monitor, live_timer_interval);
 }
 
 /*
@@ -798,7 +800,7 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
 	 * Ask the sessiond if we have new metadata waiting and update the
 	 * consumer metadata cache.
 	 */
-	ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0);
+	ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 1);
 	if (ret < 0) {
 		goto error;
 	}
@@ -825,7 +827,7 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
 		ret = utils_create_stream_file(path, metadata_stream->name,
 				metadata_stream->chan->tracefile_size,
 				metadata_stream->tracefile_count_current,
-				metadata_stream->uid, metadata_stream->gid);
+				metadata_stream->uid, metadata_stream->gid, NULL);
 		if (ret < 0) {
 			goto error_stream;
 		}
@@ -833,18 +835,13 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
 		metadata_stream->tracefile_size_current = 0;
 	}
 
-	pthread_mutex_lock(&metadata_channel->metadata_cache->lock);
-
 	do {
 		ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
 		if (ret < 0) {
-			goto error_unlock;
+			goto error_stream;
 		}
 	} while (ret > 0);
 
-error_unlock:
-	pthread_mutex_unlock(&metadata_channel->metadata_cache->lock);
-
 error_stream:
 	/*
 	 * Clean up the stream completly because the next snapshot will use a new
@@ -905,7 +902,7 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 			ret = utils_create_stream_file(path, stream->name,
 					stream->chan->tracefile_size,
 					stream->tracefile_count_current,
-					stream->uid, stream->gid);
+					stream->uid, stream->gid, NULL);
 			if (ret < 0) {
 				goto error_unlock;
 			}
@@ -975,7 +972,7 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 			}
 
 			read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
-					padded_len - len);
+					padded_len - len, NULL);
 			if (use_relayd) {
 				if (read_len != len) {
 					ret = -EPERM;
@@ -1022,7 +1019,7 @@ error:
  */
 int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
 		uint64_t len, struct lttng_consumer_channel *channel,
-		int timer)
+		int timer, int wait)
 {
 	int ret, ret_code = LTTNG_OK;
 	char *metadata_str;
@@ -1059,6 +1056,9 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
 	}
 	pthread_mutex_unlock(&channel->metadata_cache->lock);
 
+	if (!wait) {
+		goto end_free;
+	}
 	while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
 		DBG("Waiting for metadata to be flushed");
 		usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
@@ -1118,7 +1118,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		/* Session daemon status message are handled in the following call. */
 		ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
 				msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
-				&msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id);
+				&msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
+				msg.u.relayd_sock.relayd_session_id);
 		goto end_nosignal;
 	}
 	case LTTNG_CONSUMER_DESTROY_RELAYD:
@@ -1193,7 +1194,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				msg.u.ask_channel.tracefile_size,
 				msg.u.ask_channel.tracefile_count,
 				msg.u.ask_channel.session_id_per_pid,
-				msg.u.ask_channel.monitor);
+				msg.u.ask_channel.monitor,
+				msg.u.ask_channel.live_timer_interval);
 		if (!channel) {
 			goto end_channel_error;
 		}
@@ -1249,6 +1251,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			}
 			consumer_timer_switch_start(channel, attr.switch_timer_interval);
 			attr.switch_timer_interval = 0;
+		} else {
+			consumer_timer_live_start(channel,
+					msg.u.ask_channel.live_timer_interval);
 		}
 
 		/*
@@ -1266,6 +1271,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				}
 				consumer_metadata_cache_destroy(channel);
 			}
+			if (channel->live_timer_enabled == 1) {
+				consumer_timer_live_stop(channel);
+			}
 			goto end_channel_error;
 		}
 
@@ -1401,7 +1409,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		}
 
 		ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
-				len, channel, 0);
+				len, channel, 0, 1);
 		if (ret < 0) {
 			/* error receiving from sessiond */
 			goto error_fatal;
@@ -1565,6 +1573,25 @@ int lttng_ustconsumer_get_consumed_snapshot(
 	return ustctl_snapshot_get_consumed(stream->ustream, pos);
 }
 
+void lttng_ustconsumer_flush_buffer(struct lttng_consumer_stream *stream,
+		int producer)
+{
+	assert(stream);
+	assert(stream->ustream);
+
+	ustctl_flush_buffer(stream->ustream, producer);
+}
+
+int lttng_ustconsumer_get_current_timestamp(
+		struct lttng_consumer_stream *stream, uint64_t *ts)
+{
+	assert(stream);
+	assert(stream->ustream);
+	assert(ts);
+
+	return ustctl_get_current_timestamp(stream->ustream, ts);
+}
+
 /*
  * Called when the stream signal the consumer that it has hang up.
  */
@@ -1600,14 +1627,177 @@ void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
 	ustctl_destroy_stream(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 lttng_packet_index *index,
+		struct ustctl_consumer_stream *ustream)
+{
+	int ret;
+
+	ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin);
+	if (ret < 0) {
+		PERROR("ustctl_get_timestamp_begin");
+		goto error;
+	}
+	index->timestamp_begin = htobe64(index->timestamp_begin);
+
+	ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end);
+	if (ret < 0) {
+		PERROR("ustctl_get_timestamp_end");
+		goto error;
+	}
+	index->timestamp_end = htobe64(index->timestamp_end);
+
+	ret = ustctl_get_events_discarded(ustream, &index->events_discarded);
+	if (ret < 0) {
+		PERROR("ustctl_get_events_discarded");
+		goto error;
+	}
+	index->events_discarded = htobe64(index->events_discarded);
+
+	ret = ustctl_get_content_size(ustream, &index->content_size);
+	if (ret < 0) {
+		PERROR("ustctl_get_content_size");
+		goto error;
+	}
+	index->content_size = htobe64(index->content_size);
+
+	ret = ustctl_get_packet_size(ustream, &index->packet_size);
+	if (ret < 0) {
+		PERROR("ustctl_get_packet_size");
+		goto error;
+	}
+	index->packet_size = htobe64(index->packet_size);
+
+	ret = ustctl_get_stream_id(ustream, &index->stream_id);
+	if (ret < 0) {
+		PERROR("ustctl_get_stream_id");
+		goto error;
+	}
+	index->stream_id = htobe64(index->stream_id);
+
+error:
+	return ret;
+}
+
+/*
+ * Write up to one packet from the metadata cache to the channel.
+ *
+ * Returns the number of bytes pushed in the cache, or a negative value
+ * on error.
+ */
+static
+int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
+{
+	ssize_t write_len;
+	int ret;
+
+	pthread_mutex_lock(&stream->chan->metadata_cache->lock);
+	if (stream->chan->metadata_cache->contiguous
+			== stream->ust_metadata_pushed) {
+		ret = 0;
+		goto end;
+	}
+
+	write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
+			&stream->chan->metadata_cache->data[stream->ust_metadata_pushed],
+			stream->chan->metadata_cache->contiguous
+			- stream->ust_metadata_pushed);
+	assert(write_len != 0);
+	if (write_len < 0) {
+		ERR("Writing one metadata packet");
+		ret = -1;
+		goto end;
+	}
+	stream->ust_metadata_pushed += write_len;
+
+	assert(stream->chan->metadata_cache->contiguous >=
+			stream->ust_metadata_pushed);
+	ret = write_len;
+
+end:
+	pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
+	return ret;
+}
+
+
+/*
+ * Sync metadata meaning request them to the session daemon and snapshot to the
+ * metadata thread can consumer them.
+ *
+ * Metadata stream lock MUST be acquired.
+ *
+ * Return 0 if new metadatda is available, EAGAIN if the metadata stream
+ * is empty or a negative value on error.
+ */
+int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
+		struct lttng_consumer_stream *metadata)
+{
+	int ret;
+	int retry = 0;
+
+	assert(ctx);
+	assert(metadata);
+
+	/*
+	 * Request metadata from the sessiond, but don't wait for the flush
+	 * because we locked the metadata thread.
+	 */
+	ret = lttng_ustconsumer_request_metadata(ctx, metadata->chan, 0, 0);
+	if (ret < 0) {
+		goto end;
+	}
+
+	ret = commit_one_metadata_packet(metadata);
+	if (ret <= 0) {
+		goto end;
+	} else if (ret > 0) {
+		retry = 1;
+	}
+
+	ustctl_flush_buffer(metadata->ustream, 1);
+	ret = ustctl_snapshot(metadata->ustream);
+	if (ret < 0) {
+		if (errno != EAGAIN) {
+			ERR("Sync metadata, taking UST snapshot");
+			goto end;
+		}
+		DBG("No new metadata when syncing them.");
+		/* No new metadata, exit. */
+		ret = ENODATA;
+		goto end;
+	}
+
+	/*
+	 * After this flush, we still need to extract metadata.
+	 */
+	if (retry) {
+		ret = EAGAIN;
+	}
+
+end:
+	return ret;
+}
+
+/*
+ * Read subbuffer from the given stream.
+ *
+ * Stream lock MUST be acquired.
+ *
+ * Return 0 on success else a negative value.
+ */
 int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx)
 {
 	unsigned long len, subbuf_size, padding;
-	int err;
+	int err, write_index = 1;
 	long ret = 0;
 	char dummy;
 	struct ustctl_consumer_stream *ustream;
+	struct lttng_packet_index index;
 
 	assert(stream);
 	assert(stream->ustream);
@@ -1641,25 +1831,10 @@ retry:
 		 * already been read.
 		 */
 		if (stream->metadata_flag) {
-			ssize_t write_len;
-
-			if (stream->chan->metadata_cache->contiguous
-					== stream->ust_metadata_pushed) {
-				ret = 0;
-				goto end;
-			}
-
-			write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
-					&stream->chan->metadata_cache->data[stream->ust_metadata_pushed],
-					stream->chan->metadata_cache->contiguous
-						- stream->ust_metadata_pushed);
-			assert(write_len != 0);
-			if (write_len < 0) {
-				ERR("Writing one metadata packet");
-				ret = -1;
+			ret = commit_one_metadata_packet(stream);
+			if (ret <= 0) {
 				goto end;
 			}
-			stream->ust_metadata_pushed += write_len;
 			ustctl_flush_buffer(stream->ustream, 1);
 			goto retry;
 		}
@@ -1676,6 +1851,17 @@ retry:
 		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) {
+			goto end;
+		}
+	} else {
+		write_index = 0;
+	}
+
 	/* Get the full padded subbuffer size */
 	err = ustctl_get_padded_subbuf_size(ustream, &len);
 	assert(err == 0);
@@ -1689,7 +1875,7 @@ retry:
 
 	padding = len - subbuf_size;
 	/* write the subbuffer to the tracefile */
-	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding);
+	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, 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.
@@ -1707,10 +1893,32 @@ retry:
 		DBG("Error writing to tracefile "
 				"(ret: %ld != len: %lu != subbuf_size: %lu)",
 				ret, len, subbuf_size);
+		write_index = 0;
 	}
 	err = ustctl_put_next_subbuf(ustream);
 	assert(err == 0);
 
+	/* Write index if needed. */
+	if (!write_index) {
+		goto end;
+	}
+
+	if (stream->chan->live_timer_interval && !stream->metadata_flag) {
+		/*
+		 * In live, block until all the metadata is sent.
+		 */
+		err = consumer_stream_sync_metadata(ctx, stream->session_id);
+		if (err < 0) {
+			goto end;
+		}
+	}
+
+	assert(!stream->metadata_flag);
+	err = consumer_stream_write_index(stream, &index);
+	if (err < 0) {
+		goto end;
+	}
+
 end:
 	return ret;
 }
@@ -1730,12 +1938,23 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 	if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) {
 		ret = utils_create_stream_file(stream->chan->pathname, stream->name,
 				stream->chan->tracefile_size, stream->tracefile_count_current,
-				stream->uid, stream->gid);
+				stream->uid, stream->gid, NULL);
 		if (ret < 0) {
 			goto error;
 		}
 		stream->out_fd = ret;
 		stream->tracefile_size_current = 0;
+
+		if (!stream->metadata_flag) {
+			ret = index_create_file(stream->chan->pathname,
+					stream->name, stream->uid, stream->gid,
+					stream->chan->tracefile_size,
+					stream->tracefile_count_current);
+			if (ret < 0) {
+				goto error;
+			}
+			stream->index_fd = ret;
+		}
 	}
 	ret = 0;
 
@@ -1864,7 +2083,7 @@ void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
  * introduces deadlocks.
  */
 int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_channel *channel, int timer)
+		struct lttng_consumer_channel *channel, int timer, int wait)
 {
 	struct lttcomm_metadata_request_msg request;
 	struct lttcomm_consumer_msg msg;
@@ -1958,7 +2177,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
 	}
 
 	ret_code = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
-			key, offset, len, channel, timer);
+			key, offset, len, channel, timer, wait);
 	if (ret_code >= 0) {
 		/*
 		 * Only send the status msg if the sessiond is alive meaning a positive