X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=0b0b8af56de69c029f8ba419a1d0e190d09004e6;hp=81f6859124c1353d89dbb36b38f8549761f33d53;hb=749a5b3f83da6d2859d3ac964ec38342870d1e0f;hpb=d7ba13889c8692b14f99238ddf2721ed78df89d2

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 81f685912..0b0b8af56 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -1,6 +1,7 @@
 /*
  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2 only,
@@ -16,7 +17,6 @@
  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
-#define _GNU_SOURCE
 #define _LGPL_SOURCE
 #include <assert.h>
 #include <lttng/ust-ctl.h>
@@ -32,6 +32,7 @@
 #include <unistd.h>
 #include <urcu/list.h>
 #include <signal.h>
+#include <stdbool.h>
 
 #include <bin/lttng-consumerd/health-consumerd.h>
 #include <common/common.h>
@@ -39,17 +40,18 @@
 #include <common/relayd/relayd.h>
 #include <common/compat/fcntl.h>
 #include <common/compat/endian.h>
-#include <common/consumer-metadata-cache.h>
-#include <common/consumer-stream.h>
-#include <common/consumer-timer.h>
+#include <common/consumer/consumer-metadata-cache.h>
+#include <common/consumer/consumer-stream.h>
+#include <common/consumer/consumer-timer.h>
 #include <common/utils.h>
 #include <common/index/index.h>
 
 #include "ust-consumer.h"
 
+#define INT_MAX_STR_LEN 12	/* includes \0 */
+
 extern struct lttng_consumer_global_data consumer_data;
 extern int consumer_poll_timeout;
-extern volatile int consumer_quit;
 
 /*
  * Free channel object and all streams associated with it. This MUST be used
@@ -80,6 +82,7 @@ static void destroy_channel(struct lttng_consumer_channel *channel)
 	 */
 	if (channel->uchan) {
 		lttng_ustconsumer_del_channel(channel);
+		lttng_ustconsumer_free_channel(channel);
 	}
 	free(channel);
 }
@@ -125,7 +128,7 @@ static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
 		uint64_t tracefile_size, uint64_t tracefile_count,
 		uint64_t session_id_per_pid, unsigned int monitor,
 		unsigned int live_timer_interval,
-		const char *shm_path)
+		const char *root_shm_path, const char *shm_path)
 {
 	assert(pathname);
 	assert(name);
@@ -133,7 +136,7 @@ static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
 	return consumer_allocate_channel(key, session_id, pathname, name, uid,
 			gid, relayd_id, output, tracefile_size,
 			tracefile_count, session_id_per_pid, monitor,
-			live_timer_interval, shm_path);
+			live_timer_interval, root_shm_path, shm_path);
 }
 
 /*
@@ -144,7 +147,8 @@ static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
  */
 static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
 		struct lttng_consumer_channel *channel,
-		struct lttng_consumer_local_data *ctx, int *_alloc_ret)
+		struct lttng_consumer_local_data *ctx, int *_alloc_ret,
+		uint64_t trace_archive_id)
 {
 	int alloc_ret;
 	struct lttng_consumer_stream *stream = NULL;
@@ -163,7 +167,8 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
 			cpu,
 			&alloc_ret,
 			channel->type,
-			channel->monitor);
+			channel->monitor,
+			trace_archive_id);
 	if (stream == NULL) {
 		switch (alloc_ret) {
 		case -ENOENT:
@@ -182,6 +187,7 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
 		goto error;
 	}
 
+	consumer_stream_update_channel_attributes(stream, channel);
 	stream->chan = channel;
 
 error:
@@ -204,28 +210,21 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream,
 
 	/* Get the right pipe where the stream will be sent. */
 	if (stream->metadata_flag) {
-		ret = consumer_add_metadata_stream(stream);
-		if (ret) {
-			ERR("Consumer add metadata stream %" PRIu64 " failed.",
-					stream->key);
-			goto error;
-		}
+		consumer_add_metadata_stream(stream);
 		stream_pipe = ctx->consumer_metadata_pipe;
 	} else {
-		ret = consumer_add_data_stream(stream);
-		if (ret) {
-			ERR("Consumer add stream %" PRIu64 " failed.",
-					stream->key);
-			goto error;
-		}
+		consumer_add_data_stream(stream);
 		stream_pipe = ctx->consumer_data_pipe;
 	}
 
 	/*
 	 * From this point on, the stream's ownership has been moved away from
-	 * the channel and becomes globally visible.
+	 * the channel and it becomes globally visible. Hence, remove it from
+	 * the local stream list to prevent the stream from being both local and
+	 * global.
 	 */
 	stream->globally_visible = 1;
+	cds_list_del(&stream->send_node);
 
 	ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream));
 	if (ret < 0) {
@@ -237,18 +236,41 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream,
 		} else {
 			consumer_del_stream_for_data(stream);
 		}
+		goto error;
 	}
+
 error:
 	return ret;
 }
 
+static
+int get_stream_shm_path(char *stream_shm_path, const char *shm_path, int cpu)
+{
+	char cpu_nr[INT_MAX_STR_LEN];  /* int max len */
+	int ret;
+
+	strncpy(stream_shm_path, shm_path, PATH_MAX);
+	stream_shm_path[PATH_MAX - 1] = '\0';
+	ret = snprintf(cpu_nr, INT_MAX_STR_LEN, "%i", cpu);
+	if (ret < 0) {
+		PERROR("snprintf");
+		goto end;
+	}
+	strncat(stream_shm_path, cpu_nr,
+		PATH_MAX - strlen(stream_shm_path) - 1);
+	ret = 0;
+end:
+	return ret;
+}
+
 /*
  * Create streams for the given channel using liblttng-ust-ctl.
  *
  * Return 0 on success else a negative value.
  */
 static int create_ust_streams(struct lttng_consumer_channel *channel,
-		struct lttng_consumer_local_data *ctx)
+		struct lttng_consumer_local_data *ctx,
+		uint64_t trace_archive_id)
 {
 	int ret, cpu = 0;
 	struct ustctl_consumer_stream *ustream;
@@ -279,7 +301,8 @@ static int create_ust_streams(struct lttng_consumer_channel *channel,
 		}
 
 		/* Allocate consumer stream object. */
-		stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret);
+		stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret,
+				trace_archive_id);
 		if (!stream) {
 			goto error_alloc;
 		}
@@ -330,8 +353,12 @@ static int create_ust_streams(struct lttng_consumer_channel *channel,
 		/* Keep stream reference when creating metadata. */
 		if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) {
 			channel->metadata_stream = stream;
-			stream->ust_metadata_poll_pipe[0] = ust_metadata_pipe[0];
-			stream->ust_metadata_poll_pipe[1] = ust_metadata_pipe[1];
+			if (channel->monitor) {
+				/* Set metadata poll pipe if we created one */
+				memcpy(stream->ust_metadata_poll_pipe,
+						ust_metadata_pipe,
+						sizeof(ust_metadata_pipe));
+			}
 		}
 	}
 
@@ -342,20 +369,85 @@ error_alloc:
 	return ret;
 }
 
+/*
+ * create_posix_shm is never called concurrently within a process.
+ */
+static
+int create_posix_shm(void)
+{
+	char tmp_name[NAME_MAX];
+	int shmfd, ret;
+
+	ret = snprintf(tmp_name, NAME_MAX, "/ust-shm-consumer-%d", getpid());
+	if (ret < 0) {
+		PERROR("snprintf");
+		return -1;
+	}
+	/*
+	 * Allocate shm, and immediately unlink its shm oject, keeping
+	 * only the file descriptor as a reference to the object.
+	 * We specifically do _not_ use the / at the beginning of the
+	 * pathname so that some OS implementations can keep it local to
+	 * the process (POSIX leaves this implementation-defined).
+	 */
+	shmfd = shm_open(tmp_name, O_CREAT | O_EXCL | O_RDWR, 0700);
+	if (shmfd < 0) {
+		PERROR("shm_open");
+		goto error_shm_open;
+	}
+	ret = shm_unlink(tmp_name);
+	if (ret < 0 && errno != ENOENT) {
+		PERROR("shm_unlink");
+		goto error_shm_release;
+	}
+	return shmfd;
+
+error_shm_release:
+	ret = close(shmfd);
+	if (ret) {
+		PERROR("close");
+	}
+error_shm_open:
+	return -1;
+}
+
+static int open_ust_stream_fd(struct lttng_consumer_channel *channel, int cpu)
+{
+	char shm_path[PATH_MAX];
+	int ret;
+
+	if (!channel->shm_path[0]) {
+		return create_posix_shm();
+	}
+	ret = get_stream_shm_path(shm_path, channel->shm_path, cpu);
+	if (ret) {
+		goto error_shm_path;
+	}
+	return run_as_open(shm_path,
+		O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR,
+		channel->uid, channel->gid);
+
+error_shm_path:
+	return -1;
+}
+
 /*
  * Create an UST channel with the given attributes and send it to the session
  * daemon using the ust ctl API.
  *
  * Return 0 on success or else a negative value.
  */
-static int create_ust_channel(struct ustctl_consumer_channel_attr *attr,
-		struct ustctl_consumer_channel **chanp)
+static int create_ust_channel(struct lttng_consumer_channel *channel,
+		struct ustctl_consumer_channel_attr *attr,
+		struct ustctl_consumer_channel **ust_chanp)
 {
-	int ret;
-	struct ustctl_consumer_channel *channel;
+	int ret, nr_stream_fds, i, j;
+	int *stream_fds;
+	struct ustctl_consumer_channel *ust_channel;
 
+	assert(channel);
 	assert(attr);
-	assert(chanp);
+	assert(ust_chanp);
 
 	DBG3("Creating channel to ustctl with attr: [overwrite: %d, "
 			"subbuf_size: %" PRIu64 ", num_subbuf: %" PRIu64 ", "
@@ -364,17 +456,64 @@ static int create_ust_channel(struct ustctl_consumer_channel_attr *attr,
 			attr->num_subbuf, attr->switch_timer_interval,
 			attr->read_timer_interval, attr->output, attr->type);
 
-	channel = ustctl_create_channel(attr);
-	if (!channel) {
+	if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA)
+		nr_stream_fds = 1;
+	else
+		nr_stream_fds = ustctl_get_nr_stream_per_channel();
+	stream_fds = zmalloc(nr_stream_fds * sizeof(*stream_fds));
+	if (!stream_fds) {
+		ret = -1;
+		goto error_alloc;
+	}
+	for (i = 0; i < nr_stream_fds; i++) {
+		stream_fds[i] = open_ust_stream_fd(channel, i);
+		if (stream_fds[i] < 0) {
+			ret = -1;
+			goto error_open;
+		}
+	}
+	ust_channel = ustctl_create_channel(attr, stream_fds, nr_stream_fds);
+	if (!ust_channel) {
 		ret = -1;
 		goto error_create;
 	}
-
-	*chanp = channel;
+	channel->nr_stream_fds = nr_stream_fds;
+	channel->stream_fds = stream_fds;
+	*ust_chanp = ust_channel;
 
 	return 0;
 
 error_create:
+error_open:
+	for (j = i - 1; j >= 0; j--) {
+		int closeret;
+
+		closeret = close(stream_fds[j]);
+		if (closeret) {
+			PERROR("close");
+		}
+		if (channel->shm_path[0]) {
+			char shm_path[PATH_MAX];
+
+			closeret = get_stream_shm_path(shm_path,
+					channel->shm_path, j);
+			if (closeret) {
+				ERR("Cannot get stream shm path");
+			}
+			closeret = run_as_unlink(shm_path,
+					channel->uid, channel->gid);
+			if (closeret) {
+				PERROR("unlink %s", shm_path);
+			}
+		}
+	}
+	/* Try to rmdir all directories under shm_path root. */
+	if (channel->root_shm_path[0]) {
+		(void) run_as_rmdir_recursive(channel->root_shm_path,
+				channel->uid, channel->gid);
+	}
+	free(stream_fds);
+error_alloc:
 	return ret;
 }
 
@@ -403,11 +542,11 @@ error:
 }
 
 /*
- * Send channel to sessiond.
+ * Send channel to sessiond and relayd if applicable.
  *
  * Return 0 on success or else a negative value.
  */
-static int send_sessiond_channel(int sock,
+static int send_channel_to_sessiond_and_relayd(int sock,
 		struct lttng_consumer_channel *channel,
 		struct lttng_consumer_local_data *ctx, int *relayd_error)
 {
@@ -427,6 +566,8 @@ static int send_sessiond_channel(int sock,
 			health_code_update();
 
 			/* Try to send the stream to the relayd if one is available. */
+			DBG("Sending stream %" PRIu64 " of channel \"%s\" to relayd",
+					stream->key, channel->name);
 			ret = consumer_send_relayd_stream(stream, stream->chan->pathname);
 			if (ret < 0) {
 				/*
@@ -502,9 +643,10 @@ error:
  * Return 0 on success or else, a negative value is returned and the channel
  * MUST be destroyed by consumer_del_channel().
  */
-static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
+static int ask_channel(struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_channel *channel,
-		struct ustctl_consumer_channel_attr *attr)
+		struct ustctl_consumer_channel_attr *attr,
+		uint64_t trace_archive_id)
 {
 	int ret;
 
@@ -528,7 +670,7 @@ static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
 	channel->nb_init_stream_left = 0;
 
 	/* The reply msg status is handled in the following call. */
-	ret = create_ust_channel(attr, &channel->uchan);
+	ret = create_ust_channel(channel, attr, &channel->uchan);
 	if (ret < 0) {
 		goto end;
 	}
@@ -545,7 +687,7 @@ static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
 	}
 
 	/* Open all streams for this channel. */
-	ret = create_ust_streams(channel, ctx);
+	ret = create_ust_streams(channel, ctx, trace_archive_id);
 	if (ret < 0) {
 		goto end;
 	}
@@ -581,14 +723,8 @@ static int send_streams_to_thread(struct lttng_consumer_channel *channel,
 			 * If we are unable to send the stream to the thread, there is
 			 * a big problem so just stop everything.
 			 */
-			/* Remove node from the channel stream list. */
-			cds_list_del(&stream->send_node);
 			goto error;
 		}
-
-		/* Remove node from the channel stream list. */
-		cds_list_del(&stream->send_node);
-
 	}
 
 error:
@@ -627,7 +763,54 @@ static int flush_channel(uint64_t chan_key)
 
 		health_code_update();
 
-		ustctl_flush_buffer(stream->ustream, 1);
+		pthread_mutex_lock(&stream->lock);
+		if (!stream->quiescent) {
+			lttng_ustconsumer_flush_buffer(stream, 0);
+			stream->quiescent = true;
+		}
+		pthread_mutex_unlock(&stream->lock);
+	}
+error:
+	rcu_read_unlock();
+	return ret;
+}
+
+/*
+ * Clear quiescent state from channel's streams using the given key to
+ * retrieve the channel.
+ *
+ * Return 0 on success else an LTTng error code.
+ */
+static int clear_quiescent_channel(uint64_t chan_key)
+{
+	int ret = 0;
+	struct lttng_consumer_channel *channel;
+	struct lttng_consumer_stream *stream;
+	struct lttng_ht *ht;
+	struct lttng_ht_iter iter;
+
+	DBG("UST consumer clear quiescent channel key %" PRIu64, chan_key);
+
+	rcu_read_lock();
+	channel = consumer_find_channel(chan_key);
+	if (!channel) {
+		ERR("UST consumer clear quiescent channel %" PRIu64 " not found", chan_key);
+		ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+		goto error;
+	}
+
+	ht = consumer_data.stream_per_chan_id_ht;
+
+	/* For each stream of the channel id, clear quiescent state. */
+	cds_lfht_for_each_entry_duplicate(ht->ht,
+			ht->hash_fct(&channel->key, lttng_ht_seed), ht->match_fct,
+			&channel->key, &iter.iter, stream, node_channel_id.node) {
+
+		health_code_update();
+
+		pthread_mutex_lock(&stream->lock);
+		stream->quiescent = false;
+		pthread_mutex_unlock(&stream->lock);
 	}
 error:
 	rcu_read_unlock();
@@ -644,6 +827,7 @@ static int close_metadata(uint64_t chan_key)
 {
 	int ret = 0;
 	struct lttng_consumer_channel *channel;
+	unsigned int channel_monitor;
 
 	DBG("UST consumer close metadata key %" PRIu64, chan_key);
 
@@ -662,13 +846,48 @@ static int close_metadata(uint64_t chan_key)
 
 	pthread_mutex_lock(&consumer_data.lock);
 	pthread_mutex_lock(&channel->lock);
-
+	channel_monitor = channel->monitor;
 	if (cds_lfht_is_node_deleted(&channel->node.node)) {
 		goto error_unlock;
 	}
 
 	lttng_ustconsumer_close_metadata(channel);
+	pthread_mutex_unlock(&channel->lock);
+	pthread_mutex_unlock(&consumer_data.lock);
 
+	/*
+	 * The ownership of a metadata channel depends on the type of
+	 * session to which it belongs. In effect, the monitor flag is checked
+	 * to determine if this metadata channel is in "snapshot" mode or not.
+	 *
+	 * In the non-snapshot case, the metadata channel is created along with
+	 * a single stream which will remain present until the metadata channel
+	 * is destroyed (on the destruction of its session). In this case, the
+	 * metadata stream in "monitored" by the metadata poll thread and holds
+	 * the ownership of its channel.
+	 *
+	 * Closing the metadata will cause the metadata stream's "metadata poll
+	 * pipe" to be closed. Closing this pipe will wake-up the metadata poll
+	 * thread which will teardown the metadata stream which, in return,
+	 * deletes the metadata channel.
+	 *
+	 * In the snapshot case, the metadata stream is created and destroyed
+	 * on every snapshot record. Since the channel doesn't have an owner
+	 * other than the session daemon, it is safe to destroy it immediately
+	 * on reception of the CLOSE_METADATA command.
+	 */
+	if (!channel_monitor) {
+		/*
+		 * The channel and consumer_data locks must be
+		 * released before this call since consumer_del_channel
+		 * re-acquires the channel and consumer_data locks to teardown
+		 * the channel and queue its reclamation by the "call_rcu"
+		 * worker thread.
+		 */
+		consumer_del_channel(channel);
+	}
+
+	return ret;
 error_unlock:
 	pthread_mutex_unlock(&channel->lock);
 	pthread_mutex_unlock(&consumer_data.lock);
@@ -731,6 +950,10 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
 		}
 	}
 
+	/*
+	 * Ownership of metadata stream is passed along. Freeing is handled by
+	 * the callee.
+	 */
 	ret = send_streams_to_thread(metadata, ctx);
 	if (ret < 0) {
 		/*
@@ -738,7 +961,7 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
 		 * a big problem so just stop everything.
 		 */
 		ret = LTTCOMM_CONSUMERD_FATAL;
-		goto error;
+		goto send_streams_error;
 	}
 	/* List MUST be empty after or else it could be reused. */
 	assert(cds_list_empty(&metadata->streams.head));
@@ -756,6 +979,7 @@ error:
 	consumer_stream_destroy(metadata->metadata_stream, NULL);
 	cds_list_del(&metadata->metadata_stream->send_node);
 	metadata->metadata_stream = NULL;
+send_streams_error:
 error_no_stream:
 end:
 	return ret;
@@ -763,14 +987,17 @@ end:
 
 /*
  * Snapshot the whole metadata.
+ * RCU read-side lock must be held across this function to ensure existence of
+ * metadata_channel.
  *
  * Returns 0 on success, < 0 on error
  */
-static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
-		struct lttng_consumer_local_data *ctx)
+static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel,
+		uint64_t key, char *path, uint64_t relayd_id,
+		struct lttng_consumer_local_data *ctx,
+		uint64_t trace_archive_id)
 {
 	int ret = 0;
-	struct lttng_consumer_channel *metadata_channel;
 	struct lttng_consumer_stream *metadata_stream;
 
 	assert(path);
@@ -781,13 +1008,6 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
 
 	rcu_read_lock();
 
-	metadata_channel = consumer_find_channel(key);
-	if (!metadata_channel) {
-		ERR("UST snapshot metadata channel not found for key %" PRIu64,
-			key);
-		ret = -1;
-		goto error;
-	}
 	assert(!metadata_channel->monitor);
 
 	health_code_update();
@@ -807,7 +1027,7 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
 	 * The metadata stream is NOT created in no monitor mode when the channel
 	 * is created on a sessiond ask channel command.
 	 */
-	ret = create_ust_streams(metadata_channel, ctx);
+	ret = create_ust_streams(metadata_channel, ctx, trace_archive_id);
 	if (ret < 0) {
 		goto error;
 	}
@@ -858,16 +1078,19 @@ error:
 
 /*
  * Take a snapshot of all the stream of a channel.
+ * RCU read-side lock must be held across this function to ensure existence of
+ * channel.
  *
  * Returns 0 on success, < 0 on error
  */
-static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
-		uint64_t nb_packets_per_stream, struct lttng_consumer_local_data *ctx)
+static int snapshot_channel(struct lttng_consumer_channel *channel,
+		uint64_t key, char *path, uint64_t relayd_id,
+		uint64_t nb_packets_per_stream,
+		struct lttng_consumer_local_data *ctx)
 {
 	int ret;
 	unsigned use_relayd = 0;
 	unsigned long consumed_pos, produced_pos;
-	struct lttng_consumer_channel *channel;
 	struct lttng_consumer_stream *stream;
 
 	assert(path);
@@ -879,17 +1102,10 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 		use_relayd = 1;
 	}
 
-	channel = consumer_find_channel(key);
-	if (!channel) {
-		ERR("UST snapshot channel not found for key %" PRIu64, key);
-		ret = -1;
-		goto error;
-	}
 	assert(!channel->monitor);
 	DBG("UST consumer snapshot channel %" PRIu64, key);
 
 	cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
-
 		health_code_update();
 
 		/* Lock stream because we are about to change its state. */
@@ -915,14 +1131,14 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 			DBG("UST consumer snapshot stream %s/%s (%" PRIu64 ")", path,
 					stream->name, stream->key);
 		}
-		if (relayd_id != -1ULL) {
-			ret = consumer_send_relayd_streams_sent(relayd_id);
-			if (ret < 0) {
-				goto error_unlock;
-			}
-		}
 
-		ustctl_flush_buffer(stream->ustream, 1);
+		/*
+		 * If tracing is active, we want to perform a "full" buffer flush.
+		 * Else, if quiescent, it has already been done by the prior stop.
+		 */
+		if (!stream->quiescent) {
+			lttng_ustconsumer_flush_buffer(stream, 0);
+		}
 
 		ret = lttng_ustconsumer_take_snapshot(stream);
 		if (ret < 0) {
@@ -968,6 +1184,7 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 				}
 				DBG("UST consumer get subbuf failed. Skipping it.");
 				consumed_pos += stream->max_sb_size;
+				stream->chan->lost_packets++;
 				continue;
 			}
 
@@ -1021,17 +1238,21 @@ error_close_stream:
 	consumer_stream_close(stream);
 error_unlock:
 	pthread_mutex_unlock(&stream->lock);
-error:
 	rcu_read_unlock();
 	return ret;
 }
 
 /*
- * Receive the metadata updates from the sessiond.
+ * Receive the metadata updates from the sessiond. Supports receiving
+ * overlapping metadata, but is needs to always belong to a contiguous
+ * range starting from 0.
+ * Be careful about the locks held when calling this function: it needs
+ * the metadata cache flush to concurrently progress in order to
+ * complete.
  */
 int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
-		uint64_t len, struct lttng_consumer_channel *channel,
-		int timer, int wait)
+		uint64_t len, uint64_t version,
+		struct lttng_consumer_channel *channel, int timer, int wait)
 {
 	int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
 	char *metadata_str;
@@ -1058,7 +1279,8 @@ 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, metadata_str);
+	ret = consumer_metadata_cache_write(channel, offset, len, version,
+			metadata_str);
 	if (ret < 0) {
 		/* Unable to handle metadata. Notify session daemon. */
 		ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
@@ -1133,7 +1355,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
 	{
 		/* Session daemon status message are handled in the following call. */
-		ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
+		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.relayd_session_id);
@@ -1213,6 +1435,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				msg.u.ask_channel.session_id_per_pid,
 				msg.u.ask_channel.monitor,
 				msg.u.ask_channel.live_timer_interval,
+				msg.u.ask_channel.root_shm_path,
 				msg.u.ask_channel.shm_path);
 		if (!channel) {
 			goto end_channel_error;
@@ -1233,9 +1456,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
 		attr.chan_id = msg.u.ask_channel.chan_id;
 		memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
-		strncpy(attr.shm_path, channel->shm_path,
-			sizeof(attr.shm_path));
-		attr.shm_path[sizeof(attr.shm_path) - 1] = '\0';
+		attr.blocking_timeout= msg.u.ask_channel.blocking_timeout;
 
 		/* Match channel buffer type to the UST abi. */
 		switch (msg.u.ask_channel.output) {
@@ -1268,7 +1489,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
 		health_code_update();
 
-		ret = ask_channel(ctx, sock, channel, &attr);
+		ret = ask_channel(ctx, channel, &attr,
+				msg.u.ask_channel.trace_archive_id);
 		if (ret < 0) {
 			goto end_channel_error;
 		}
@@ -1282,8 +1504,17 @@ 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 {
+			int monitor_start_ret;
+
 			consumer_timer_live_start(channel,
 					msg.u.ask_channel.live_timer_interval);
+			monitor_start_ret = consumer_timer_monitor_start(
+					channel,
+					msg.u.ask_channel.monitor_timer_interval);
+			if (monitor_start_ret < 0) {
+				ERR("Starting channel monitoring timer failed");
+				goto end_channel_error;
+			}
 		}
 
 		health_code_update();
@@ -1306,6 +1537,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			if (channel->live_timer_enabled == 1) {
 				consumer_timer_live_stop(channel);
 			}
+			if (channel->monitor_timer_enabled == 1) {
+				consumer_timer_monitor_stop(channel);
+			}
 			goto end_channel_error;
 		}
 
@@ -1341,8 +1575,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
 		health_code_update();
 
-		/* Send everything to sessiond. */
-		ret = send_sessiond_channel(sock, channel, ctx, &relayd_err);
+		/* Send the channel to sessiond (and relayd, if applicable). */
+		ret = send_channel_to_sessiond_and_relayd(sock, channel, ctx,
+				&relayd_err);
 		if (ret < 0) {
 			if (relayd_err) {
 				/*
@@ -1416,12 +1651,25 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
 		goto end_msg_sessiond;
 	}
+	case LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL:
+	{
+		int ret;
+
+		ret = clear_quiescent_channel(
+				msg.u.clear_quiescent_channel.key);
+		if (ret != 0) {
+			ret_code = ret;
+		}
+
+		goto end_msg_sessiond;
+	}
 	case LTTNG_CONSUMER_PUSH_METADATA:
 	{
 		int ret;
 		uint64_t len = msg.u.push_metadata.len;
 		uint64_t key = msg.u.push_metadata.key;
 		uint64_t offset = msg.u.push_metadata.target_offset;
+		uint64_t version = msg.u.push_metadata.version;
 		struct lttng_consumer_channel *channel;
 
 		DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key,
@@ -1443,6 +1691,15 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
 		health_code_update();
 
+		if (!len) {
+			/*
+			 * There is nothing to receive. We have simply
+			 * checked whether the channel can be found.
+			 */
+			ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+			goto end_msg_sessiond;
+		}
+
 		/* Tell session daemon we are ready to receive the metadata. */
 		ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
 		if (ret < 0) {
@@ -1463,7 +1720,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		health_code_update();
 
 		ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
-				len, channel, 0, 1);
+				len, version, channel, 0, 1);
 		if (ret < 0) {
 			/* error receiving from sessiond */
 			goto error_fatal;
@@ -1484,34 +1741,397 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	}
 	case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
 	{
-		if (msg.u.snapshot_channel.metadata) {
-			ret = snapshot_metadata(msg.u.snapshot_channel.key,
-					msg.u.snapshot_channel.pathname,
-					msg.u.snapshot_channel.relayd_id,
-					ctx);
-			if (ret < 0) {
-				ERR("Snapshot metadata failed");
-				ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
+		struct lttng_consumer_channel *channel;
+		uint64_t key = msg.u.snapshot_channel.key;
+
+		channel = consumer_find_channel(key);
+		if (!channel) {
+			DBG("UST snapshot channel not found for key %" PRIu64, key);
+			ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+		} else {
+			if (msg.u.snapshot_channel.metadata) {
+				ret = snapshot_metadata(channel, key,
+						msg.u.snapshot_channel.pathname,
+						msg.u.snapshot_channel.relayd_id,
+						ctx,
+						msg.u.snapshot_channel.trace_archive_id);
+				if (ret < 0) {
+					ERR("Snapshot metadata failed");
+					ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
+				}
+			} else {
+				ret = snapshot_channel(channel, key,
+						msg.u.snapshot_channel.pathname,
+						msg.u.snapshot_channel.relayd_id,
+						msg.u.snapshot_channel.nb_packets_per_stream,
+						ctx);
+				if (ret < 0) {
+					ERR("Snapshot channel failed");
+					ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
+				}
 			}
+		}
+		health_code_update();
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+		health_code_update();
+		break;
+	}
+	case LTTNG_CONSUMER_DISCARDED_EVENTS:
+	{
+		int ret = 0;
+		uint64_t discarded_events;
+		struct lttng_ht_iter iter;
+		struct lttng_ht *ht;
+		struct lttng_consumer_stream *stream;
+		uint64_t id = msg.u.discarded_events.session_id;
+		uint64_t key = msg.u.discarded_events.channel_key;
+
+		DBG("UST consumer discarded events command for session id %"
+				PRIu64, id);
+		rcu_read_lock();
+		pthread_mutex_lock(&consumer_data.lock);
+
+		ht = consumer_data.stream_list_ht;
+
+		/*
+		 * We only need a reference to the channel, but they are not
+		 * directly indexed, so we just use the first matching stream
+		 * to extract the information we need, we default to 0 if not
+		 * found (no events are dropped if the channel is not yet in
+		 * use).
+		 */
+		discarded_events = 0;
+		cds_lfht_for_each_entry_duplicate(ht->ht,
+				ht->hash_fct(&id, lttng_ht_seed),
+				ht->match_fct, &id,
+				&iter.iter, stream, node_session_id.node) {
+			if (stream->chan->key == key) {
+				discarded_events = stream->chan->discarded_events;
+				break;
+			}
+		}
+		pthread_mutex_unlock(&consumer_data.lock);
+		rcu_read_unlock();
+
+		DBG("UST consumer discarded events command for session id %"
+				PRIu64 ", channel key %" PRIu64, id, key);
+
+		health_code_update();
+
+		/* Send back returned value to session daemon */
+		ret = lttcomm_send_unix_sock(sock, &discarded_events, sizeof(discarded_events));
+		if (ret < 0) {
+			PERROR("send discarded events");
+			goto error_fatal;
+		}
+
+		break;
+	}
+	case LTTNG_CONSUMER_LOST_PACKETS:
+	{
+	        int ret;
+		uint64_t lost_packets;
+		struct lttng_ht_iter iter;
+		struct lttng_ht *ht;
+		struct lttng_consumer_stream *stream;
+		uint64_t id = msg.u.lost_packets.session_id;
+		uint64_t key = msg.u.lost_packets.channel_key;
+
+		DBG("UST consumer lost packets command for session id %"
+				PRIu64, id);
+		rcu_read_lock();
+		pthread_mutex_lock(&consumer_data.lock);
+
+		ht = consumer_data.stream_list_ht;
+
+		/*
+		 * We only need a reference to the channel, but they are not
+		 * directly indexed, so we just use the first matching stream
+		 * to extract the information we need, we default to 0 if not
+		 * found (no packets lost if the channel is not yet in use).
+		 */
+	        lost_packets = 0;
+		cds_lfht_for_each_entry_duplicate(ht->ht,
+				ht->hash_fct(&id, lttng_ht_seed),
+				ht->match_fct, &id,
+				&iter.iter, stream, node_session_id.node) {
+			if (stream->chan->key == key) {
+			        lost_packets = stream->chan->lost_packets;
+				break;
+			}
+		}
+		pthread_mutex_unlock(&consumer_data.lock);
+		rcu_read_unlock();
+
+		DBG("UST consumer lost packets command for session id %"
+				PRIu64 ", channel key %" PRIu64, id, key);
+
+		health_code_update();
+
+		/* Send back returned value to session daemon */
+		ret = lttcomm_send_unix_sock(sock, &lost_packets,
+			        sizeof(lost_packets));
+		if (ret < 0) {
+			PERROR("send lost packets");
+			goto error_fatal;
+		}
+
+		break;
+	}
+	case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
+	{
+		int channel_monitor_pipe;
+
+		ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+		/* Successfully received the command's type. */
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			goto error_fatal;
+		}
+
+		ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe,
+				1);
+		if (ret != sizeof(channel_monitor_pipe)) {
+			ERR("Failed to receive channel monitor pipe");
+			goto error_fatal;
+		}
+
+		DBG("Received channel monitor pipe (%d)", channel_monitor_pipe);
+		ret = consumer_timer_thread_set_channel_monitor_pipe(
+				channel_monitor_pipe);
+		if (!ret) {
+			int flags;
+
+			ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+			/* Set the pipe as non-blocking. */
+			ret = fcntl(channel_monitor_pipe, F_GETFL, 0);
+			if (ret == -1) {
+				PERROR("fcntl get flags of the channel monitoring pipe");
+				goto error_fatal;
+			}
+			flags = ret;
+
+			ret = fcntl(channel_monitor_pipe, F_SETFL,
+					flags | O_NONBLOCK);
+			if (ret == -1) {
+				PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe");
+				goto error_fatal;
+			}
+			DBG("Channel monitor pipe set as non-blocking");
 		} else {
-			ret = snapshot_channel(msg.u.snapshot_channel.key,
-					msg.u.snapshot_channel.pathname,
-					msg.u.snapshot_channel.relayd_id,
-					msg.u.snapshot_channel.nb_packets_per_stream,
+			ret_code = LTTCOMM_CONSUMERD_ALREADY_SET;
+		}
+		goto end_msg_sessiond;
+	}
+	case LTTNG_CONSUMER_ROTATE_CHANNEL:
+	{
+		struct lttng_consumer_channel *channel;
+		uint64_t key = msg.u.rotate_channel.key;
+
+		channel = consumer_find_channel(key);
+		if (!channel) {
+			DBG("Channel %" PRIu64 " not found", key);
+			ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+		} else {
+			/*
+			 * Sample the rotate position of all the streams in
+			 * this channel.
+			 */
+			ret = lttng_consumer_rotate_sample_channel(channel, key,
+					msg.u.rotate_channel.pathname,
+					msg.u.rotate_channel.relayd_id,
+					msg.u.rotate_channel.metadata,
+					msg.u.rotate_channel.new_chunk_id,
 					ctx);
 			if (ret < 0) {
-				ERR("Snapshot channel failed");
-				ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
+				ERR("Rotate channel failed");
+				ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL;
 			}
+
+			health_code_update();
+		}
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+
+		/*
+		 * Rotate the streams that are ready right now.
+		 * FIXME: this is a second consecutive iteration over the
+		 * streams in a channel, there is probably a better way to
+		 * handle this, but it needs to be after the
+		 * consumer_send_status_msg() call.
+		 */
+		if (channel) {
+			ret = lttng_consumer_rotate_ready_streams(
+					channel, key, ctx);
+			if (ret < 0) {
+				ERR("Rotate channel failed");
+			}
+		}
+		break;
+	}
+	case LTTNG_CONSUMER_ROTATE_RENAME:
+	{
+		DBG("Consumer rename session %" PRIu64 " after rotation",
+				msg.u.rotate_rename.session_id);
+		ret = lttng_consumer_rotate_rename(msg.u.rotate_rename.old_path,
+				msg.u.rotate_rename.new_path,
+				msg.u.rotate_rename.uid,
+				msg.u.rotate_rename.gid,
+				msg.u.rotate_rename.relayd_id);
+		if (ret < 0) {
+			ERR("Rotate rename failed");
+			ret_code = LTTCOMM_CONSUMERD_ROTATE_RENAME_FAILED;
 		}
 
 		health_code_update();
+
 		ret = consumer_send_status_msg(sock, ret_code);
 		if (ret < 0) {
 			/* Somehow, the session daemon is not responding anymore. */
 			goto end_nosignal;
 		}
+		break;
+	}
+	case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_LOCAL:
+	{
+		int pending;
+		uint32_t pending_reply;
+
+		DBG("Perform local check of pending rotation for session id %" PRIu64,
+				msg.u.check_rotation_pending_local.session_id);
+		pending = lttng_consumer_check_rotation_pending_local(
+				msg.u.check_rotation_pending_local.session_id,
+				msg.u.check_rotation_pending_local.chunk_id);
+		if (pending < 0) {
+			ERR("Local rotation pending check failed with code %i", pending);
+			ret_code = LTTCOMM_CONSUMERD_ROTATION_PENDING_LOCAL_FAILED;
+		} else {
+			pending_reply = !!pending;
+		}
+
 		health_code_update();
+
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+
+		if (pending < 0) {
+			/*
+			 * An error occurred while running the command;
+			 * don't send the 'pending' flag as the sessiond
+			 * will not read it.
+			 */
+			break;
+		}
+
+		/* Send back returned value to session daemon */
+		ret = lttcomm_send_unix_sock(sock, &pending_reply,
+				sizeof(pending_reply));
+		if (ret < 0) {
+			PERROR("Failed to send rotation pending return code");
+			goto error_fatal;
+		}
+		break;
+	}
+	case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_RELAY:
+	{
+		int pending;
+		uint32_t pending_reply;
+
+		DBG("Perform relayd check of pending rotation for session id %" PRIu64,
+				msg.u.check_rotation_pending_relay.session_id);
+		pending = lttng_consumer_check_rotation_pending_relay(
+				msg.u.check_rotation_pending_relay.session_id,
+				msg.u.check_rotation_pending_relay.relayd_id,
+				msg.u.check_rotation_pending_relay.chunk_id);
+		if (pending < 0) {
+			ERR("Relayd rotation pending check failed with code %i", pending);
+			ret_code = LTTCOMM_CONSUMERD_ROTATION_PENDING_RELAY_FAILED;
+		} else {
+			pending_reply = !!pending;
+		}
+
+		health_code_update();
+
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+
+		if (pending < 0) {
+			/*
+			 * An error occurred while running the command;
+			 * don't send the 'pending' flag as the sessiond
+			 * will not read it.
+			 */
+			break;
+		}
+
+		/* Send back returned value to session daemon */
+		ret = lttcomm_send_unix_sock(sock, &pending_reply,
+				sizeof(pending_reply));
+		if (ret < 0) {
+			PERROR("Failed to send rotation pending return code");
+			goto error_fatal;
+		}
+		break;
+	}
+	case LTTNG_CONSUMER_MKDIR:
+	{
+		DBG("Consumer mkdir %s in session %" PRIu64,
+				msg.u.mkdir.path,
+				msg.u.mkdir.session_id);
+		ret = lttng_consumer_mkdir(msg.u.mkdir.path,
+				msg.u.mkdir.uid,
+				msg.u.mkdir.gid,
+				msg.u.mkdir.relayd_id);
+		if (ret < 0) {
+			ERR("consumer mkdir failed");
+			ret_code = LTTCOMM_CONSUMERD_MKDIR_FAILED;
+		}
+
+		health_code_update();
+
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
+		break;
+	}
+	case LTTNG_CONSUMER_CLEAR_CHANNEL:
+	{
+		struct lttng_consumer_channel *channel;
+		uint64_t key = msg.u.clear_channel.key;
+
+		channel = consumer_find_channel(key);
+		if (!channel) {
+			DBG("Channel %" PRIu64 " not found", key);
+			ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+		} else {
+			ret = lttng_consumer_clear_channel(channel);
+			if (ret) {
+				ERR("Clear channel failed key %" PRIu64, key);
+				ret_code = ret;
+			}
+
+			health_code_update();
+		}
+		ret = consumer_send_status_msg(sock, ret_code);
+		if (ret < 0) {
+			/* Somehow, the session daemon is not responding anymore. */
+			goto end_nosignal;
+		}
 		break;
 	}
 	default:
@@ -1595,7 +2215,7 @@ void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
 }
 
 /*
- * Take a snapshot for a specific fd
+ * Take a snapshot for a specific stream.
  *
  * Returns 0 on success, < 0 on error
  */
@@ -1607,6 +2227,20 @@ int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
 	return ustctl_snapshot(stream->ustream);
 }
 
+/*
+ * Sample consumed and produced positions for a specific stream.
+ *
+ * Returns 0 on success, < 0 on error.
+ */
+int lttng_ustconsumer_sample_snapshot_positions(
+		struct lttng_consumer_stream *stream)
+{
+	assert(stream);
+	assert(stream->ustream);
+
+	return ustctl_snapshot_sample_positions(stream->ustream);
+}
+
 /*
  * Get the produced position
  *
@@ -1646,6 +2280,14 @@ void lttng_ustconsumer_flush_buffer(struct lttng_consumer_stream *stream,
 	ustctl_flush_buffer(stream->ustream, producer);
 }
 
+void lttng_ustconsumer_clear_buffer(struct lttng_consumer_stream *stream)
+{
+	assert(stream);
+	assert(stream->ustream);
+
+	ustctl_clear_buffer(stream->ustream);
+}
+
 int lttng_ustconsumer_get_current_timestamp(
 		struct lttng_consumer_stream *stream, uint64_t *ts)
 {
@@ -1656,28 +2298,78 @@ int lttng_ustconsumer_get_current_timestamp(
 	return ustctl_get_current_timestamp(stream->ustream, ts);
 }
 
+int lttng_ustconsumer_get_sequence_number(
+		struct lttng_consumer_stream *stream, uint64_t *seq)
+{
+	assert(stream);
+	assert(stream->ustream);
+	assert(seq);
+
+	return ustctl_get_sequence_number(stream->ustream, seq);
+}
+
 /*
- * Called when the stream signal the consumer that it has hang up.
+ * Called when the stream signals the consumer that it has hung up.
  */
 void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
 {
 	assert(stream);
 	assert(stream->ustream);
 
-	ustctl_flush_buffer(stream->ustream, 0);
+	pthread_mutex_lock(&stream->lock);
+	if (!stream->quiescent) {
+		lttng_ustconsumer_flush_buffer(stream, 0);
+		stream->quiescent = true;
+	}
+	pthread_mutex_unlock(&stream->lock);
 	stream->hangup_flush_done = 1;
 }
 
 void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
 {
+	int i;
+
 	assert(chan);
 	assert(chan->uchan);
 
 	if (chan->switch_timer_enabled == 1) {
 		consumer_timer_switch_stop(chan);
 	}
+	for (i = 0; i < chan->nr_stream_fds; i++) {
+		int ret;
+
+		ret = close(chan->stream_fds[i]);
+		if (ret) {
+			PERROR("close");
+		}
+		if (chan->shm_path[0]) {
+			char shm_path[PATH_MAX];
+
+			ret = get_stream_shm_path(shm_path, chan->shm_path, i);
+			if (ret) {
+				ERR("Cannot get stream shm path");
+			}
+			ret = run_as_unlink(shm_path, chan->uid, chan->gid);
+			if (ret) {
+				PERROR("unlink %s", shm_path);
+			}
+		}
+	}
+}
+
+void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan)
+{
+	assert(chan);
+	assert(chan->uchan);
+
 	consumer_metadata_cache_destroy(chan);
 	ustctl_destroy_channel(chan->uchan);
+	/* Try to rmdir all directories under shm_path root. */
+	if (chan->root_shm_path[0]) {
+		(void) run_as_rmdir_recursive(chan->root_shm_path,
+				chan->uid, chan->gid);
+	}
+	free(chan->stream_fds);
 }
 
 void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
@@ -1759,10 +2451,56 @@ static int get_index_values(struct ctf_packet_index *index,
 	}
 	index->stream_id = htobe64(index->stream_id);
 
+	ret = ustctl_get_instance_id(ustream, &index->stream_instance_id);
+	if (ret < 0) {
+		PERROR("ustctl_get_instance_id");
+		goto error;
+	}
+	index->stream_instance_id = htobe64(index->stream_instance_id);
+
+	ret = ustctl_get_sequence_number(ustream, &index->packet_seq_num);
+	if (ret < 0) {
+		PERROR("ustctl_get_sequence_number");
+		goto error;
+	}
+	index->packet_seq_num = htobe64(index->packet_seq_num);
+
 error:
 	return ret;
 }
 
+static
+void metadata_stream_reset_cache(struct lttng_consumer_stream *stream,
+		struct consumer_metadata_cache *cache)
+{
+	DBG("Metadata stream update to version %" PRIu64,
+			cache->version);
+	stream->ust_metadata_pushed = 0;
+	stream->metadata_version = 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.
  *
@@ -1776,7 +2514,11 @@ int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
 	int ret;
 
 	pthread_mutex_lock(&stream->chan->metadata_cache->lock);
-	if (stream->chan->metadata_cache->contiguous
+	ret = metadata_stream_check_version(stream);
+	if (ret < 0) {
+		goto end;
+	}
+	if (stream->chan->metadata_cache->max_offset
 			== stream->ust_metadata_pushed) {
 		ret = 0;
 		goto end;
@@ -1784,7 +2526,7 @@ int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
 
 	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->chan->metadata_cache->max_offset
 			- stream->ust_metadata_pushed);
 	assert(write_len != 0);
 	if (write_len < 0) {
@@ -1794,7 +2536,7 @@ int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
 	}
 	stream->ust_metadata_pushed += write_len;
 
-	assert(stream->chan->metadata_cache->contiguous >=
+	assert(stream->chan->metadata_cache->max_offset >=
 			stream->ust_metadata_pushed);
 	ret = write_len;
 
@@ -1808,7 +2550,9 @@ end:
  * Sync metadata meaning request them to the session daemon and snapshot to the
  * metadata thread can consumer them.
  *
- * Metadata stream lock MUST be acquired.
+ * Metadata stream lock is held here, but we need to release it when
+ * interacting with sessiond, else we cause a deadlock with live
+ * awaiting on metadata to be pushed out.
  *
  * Return 0 if new metadatda is available, EAGAIN if the metadata stream
  * is empty or a negative value on error.
@@ -1822,11 +2566,13 @@ int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
 	assert(ctx);
 	assert(metadata);
 
+	pthread_mutex_unlock(&metadata->lock);
 	/*
 	 * 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);
+	pthread_mutex_lock(&metadata->lock);
 	if (ret < 0) {
 		goto end;
 	}
@@ -1838,7 +2584,7 @@ int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
 		retry = 1;
 	}
 
-	ustctl_flush_buffer(metadata->ustream, 1);
+	lttng_ustconsumer_flush_buffer(metadata, 1);
 	ret = ustctl_snapshot(metadata->ustream);
 	if (ret < 0) {
 		if (errno != EAGAIN) {
@@ -1913,6 +2659,61 @@ end:
 	return ret;
 }
 
+static
+int update_stream_stats(struct lttng_consumer_stream *stream)
+{
+	int ret;
+	uint64_t seq, discarded;
+
+	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).
+	 */
+	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;
+	}
+	stream->last_sequence_number = seq;
+
+	ret = ustctl_get_events_discarded(stream->ustream, &discarded);
+	if (ret < 0) {
+		PERROR("kernctl_get_events_discarded");
+		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;
+	}
+	stream->last_discarded_events = discarded;
+	ret = 0;
+
+end:
+	return ret;
+}
+
 /*
  * Read subbuffer from the given stream.
  *
@@ -1921,10 +2722,10 @@ end:
  * Return 0 on success else a negative value.
  */
 int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
-		struct lttng_consumer_local_data *ctx)
+		struct lttng_consumer_local_data *ctx, bool *rotated)
 {
 	unsigned long len, subbuf_size, padding;
-	int err, write_index = 1;
+	int err, write_index = 1, rotation_ret;
 	long ret = 0;
 	struct ustctl_consumer_stream *ustream;
 	struct ctf_packet_index index;
@@ -1956,7 +2757,21 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		readlen = lttng_read(stream->wait_fd, &dummy, 1);
 		if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
 			ret = readlen;
-			goto end;
+			goto error;
+		}
+	}
+
+	/*
+	 * If the stream was flagged to be ready for rotation before we extract the
+	 * next packet, rotate it now.
+	 */
+	if (stream->rotate_ready) {
+		DBG("Rotate stream before extracting data");
+		rotation_ret = lttng_consumer_rotate_stream(ctx, stream, rotated);
+		if (rotation_ret < 0) {
+			ERR("Stream rotation error");
+			ret = -1;
+			goto error;
 		}
 	}
 
@@ -1971,9 +2786,9 @@ retry:
 		if (stream->metadata_flag) {
 			ret = commit_one_metadata_packet(stream);
 			if (ret <= 0) {
-				goto end;
+				goto error;
 			}
-			ustctl_flush_buffer(stream->ustream, 1);
+			lttng_ustconsumer_flush_buffer(stream, 1);
 			goto retry;
 		}
 
@@ -1986,7 +2801,7 @@ retry:
 		 */
 		DBG("Reserving sub buffer failed (everything is normal, "
 				"it is due to concurrency) [ret: %d]", err);
-		goto end;
+		goto error;
 	}
 	assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
 
@@ -1994,7 +2809,18 @@ retry:
 		index.offset = htobe64(stream->out_fd_offset);
 		ret = get_index_values(&index, ustream);
 		if (ret < 0) {
-			goto end;
+			err = ustctl_put_subbuf(ustream);
+			assert(err == 0);
+			goto error;
+		}
+
+		/* 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;
 		}
 	} else {
 		write_index = 0;
@@ -2012,6 +2838,7 @@ retry:
 	assert(len >= subbuf_size);
 
 	padding = len - subbuf_size;
+
 	/* write the subbuffer to the tracefile */
 	ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
 	/*
@@ -2043,32 +2870,66 @@ retry:
 	if (!stream->metadata_flag) {
 		ret = notify_if_more_data(stream, ctx);
 		if (ret < 0) {
-			goto end;
+			goto error;
 		}
 	}
 
 	/* Write index if needed. */
 	if (!write_index) {
-		goto end;
+		goto rotate;
 	}
 
 	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);
+		}
+
 		if (err < 0) {
-			goto end;
+			goto error;
 		}
 	}
 
 	assert(!stream->metadata_flag);
 	err = consumer_stream_write_index(stream, &index);
 	if (err < 0) {
-		goto end;
+		goto error;
 	}
 
-end:
+rotate:
+	/*
+	 * After extracting the packet, we check if the stream is now ready to be
+	 * rotated and perform the action immediately.
+	 */
+	rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
+	if (rotation_ret == 1) {
+		rotation_ret = lttng_consumer_rotate_stream(ctx, stream, rotated);
+		if (rotation_ret < 0) {
+			ERR("Stream rotation error");
+			ret = -1;
+			goto error;
+		}
+	} else if (rotation_ret < 0) {
+		ERR("Checking if stream is ready to rotate");
+		ret = -1;
+		goto error;
+	}
+error:
 	return ret;
 }
 
@@ -2095,14 +2956,18 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 		stream->tracefile_size_current = 0;
 
 		if (!stream->metadata_flag) {
-			ret = index_create_file(stream->chan->pathname,
+			struct lttng_index_file *index_file;
+
+			index_file = lttng_index_file_create(stream->chan->pathname,
 					stream->name, stream->uid, stream->gid,
 					stream->chan->tracefile_size,
-					stream->tracefile_count_current);
-			if (ret < 0) {
+					stream->tracefile_count_current,
+					CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
+			if (!index_file) {
 				goto error;
 			}
-			stream->index_fd = ret;
+			assert(!stream->index_file);
+			stream->index_file = index_file;
 		}
 	}
 	ret = 0;
@@ -2137,7 +3002,7 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
 		uint64_t contiguous, pushed;
 
 		/* Ease our life a bit. */
-		contiguous = stream->chan->metadata_cache->contiguous;
+		contiguous = stream->chan->metadata_cache->max_offset;
 		pushed = stream->ust_metadata_pushed;
 
 		/*
@@ -2266,6 +3131,10 @@ void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
  * function or any of its callees. Timers have a very strict locking
  * semantic with respect to teardown. Failure to respect this semantic
  * introduces deadlocks.
+ *
+ * DON'T hold the metadata lock when calling this function, else this
+ * can cause deadlock involving consumer awaiting for metadata to be
+ * pushed out due to concurrent interaction with the session daemon.
  */
 int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_channel *channel, int timer, int wait)
@@ -2273,7 +3142,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
 	struct lttcomm_metadata_request_msg request;
 	struct lttcomm_consumer_msg msg;
 	enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
-	uint64_t len, key, offset;
+	uint64_t len, key, offset, version;
 	int ret;
 
 	assert(channel);
@@ -2305,7 +3174,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
 	request.key = channel->key;
 
 	DBG("Sending metadata request to sessiond, session id %" PRIu64
-			", per-pid %" PRIu64 ", app UID %u and channek key %" PRIu64,
+			", per-pid %" PRIu64 ", app UID %u and channel key %" PRIu64,
 			request.session_id, request.session_id_per_pid, request.uid,
 			request.key);
 
@@ -2353,6 +3222,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
 	len = msg.u.push_metadata.len;
 	key = msg.u.push_metadata.key;
 	offset = msg.u.push_metadata.target_offset;
+	version = msg.u.push_metadata.version;
 
 	assert(key == channel->key);
 	if (len == 0) {
@@ -2375,7 +3245,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
 	health_code_update();
 
 	ret = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
-			key, offset, len, channel, timer, wait);
+			key, offset, len, version, channel, timer, wait);
 	if (ret >= 0) {
 		/*
 		 * Only send the status msg if the sessiond is alive meaning a positive