From 10a5031171c7bca5b4498c871b119e5a88b6a3fb Mon Sep 17 00:00:00 2001
From: Julien Desfossez <jdesfossez@efficios.com>
Date: Tue, 18 Jun 2013 15:25:53 -0400
Subject: [PATCH] Add UST snapshot support

Signed-off-by: Julien Desfossez <jdesfossez@efficios.com>
Signed-off-by: David Goulet <dgoulet@efficios.com>
---
 src/bin/lttng-sessiond/cmd.c                 |  19 +-
 src/bin/lttng-sessiond/consumer.c            |  23 +-
 src/bin/lttng-sessiond/snapshot.c            |   5 -
 src/bin/lttng-sessiond/snapshot.h            |   6 +
 src/bin/lttng-sessiond/ust-consumer.c        |  14 +-
 src/common/consumer-stream.c                 | 131 +++++--
 src/common/consumer-stream.h                 |   6 +
 src/common/consumer.c                        |  90 +++--
 src/common/consumer.h                        |  11 +-
 src/common/kernel-consumer/kernel-consumer.c |  83 +----
 src/common/ust-consumer/ust-consumer.c       | 347 ++++++++++++++++++-
 src/common/ust-consumer/ust-consumer.h       |   2 +
 12 files changed, 561 insertions(+), 176 deletions(-)

diff --git a/src/bin/lttng-sessiond/cmd.c b/src/bin/lttng-sessiond/cmd.c
index ed4d6c86f..a6a81c5db 100644
--- a/src/bin/lttng-sessiond/cmd.c
+++ b/src/bin/lttng-sessiond/cmd.c
@@ -25,6 +25,7 @@
 #include <common/common.h>
 #include <common/sessiond-comm/sessiond-comm.h>
 #include <common/relayd/relayd.h>
+#include <common/utils.h>
 
 #include "channel.h"
 #include "consumer.h"
@@ -2403,6 +2404,14 @@ static int record_kernel_snapshot(struct ltt_kernel_session *ksess,
 	assert(output);
 	assert(session);
 
+	/* Get the datetime for the snapshot output directory. */
+	ret = utils_get_current_time_str("%Y%m%d-%H%M%S", output->datetime,
+			sizeof(output->datetime));
+	if (!ret) {
+		ret = -EINVAL;
+		goto error;
+	}
+
 	if (!output->kernel_sockets_copied) {
 		ret = consumer_copy_sockets(output->consumer, ksess->consumer);
 		if (ret < 0) {
@@ -2439,6 +2448,14 @@ static int record_ust_snapshot(struct ltt_ust_session *usess,
 	assert(output);
 	assert(session);
 
+	/* Get the datetime for the snapshot output directory. */
+	ret = utils_get_current_time_str("%Y%m%d-%H%M%S", output->datetime,
+			sizeof(output->datetime));
+	if (!ret) {
+		ret = -EINVAL;
+		goto error;
+	}
+
 	if (!output->ust_sockets_copied) {
 		ret = consumer_copy_sockets(output->consumer, usess->consumer);
 		if (ret < 0) {
@@ -2555,7 +2572,7 @@ int cmd_snapshot_record(struct ltt_session *session,
 			rcu_read_lock();
 			cds_lfht_for_each_entry(session->snapshot.output_ht->ht,
 					&iter.iter, sout, node.node) {
-				ret = record_ust_snapshot(usess, tmp_sout, session, wait);
+				ret = record_ust_snapshot(usess, sout, session, wait);
 				if (ret < 0) {
 					rcu_read_unlock();
 					goto error;
diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c
index 368837669..d3c561c9e 100644
--- a/src/bin/lttng-sessiond/consumer.c
+++ b/src/bin/lttng-sessiond/consumer.c
@@ -28,7 +28,6 @@
 #include <common/common.h>
 #include <common/defaults.h>
 #include <common/uri.h>
-#include <common/utils.h>
 
 #include "consumer.h"
 #include "health.h"
@@ -752,9 +751,11 @@ void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 
 	memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 
-	strncpy(msg->u.ask_channel.pathname, pathname,
-			sizeof(msg->u.ask_channel.pathname));
-	msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
+	if (pathname) {
+		strncpy(msg->u.ask_channel.pathname, pathname,
+				sizeof(msg->u.ask_channel.pathname));
+		msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
+	}
 
 	strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 	msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
@@ -1198,7 +1199,6 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
 		const char *session_path, int wait)
 {
 	int ret;
-	char datetime[16];
 	struct lttcomm_consumer_msg msg;
 
 	assert(socket);
@@ -1208,13 +1208,6 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
 
 	DBG("Consumer snapshot channel key %" PRIu64, key);
 
-	ret = utils_get_current_time_str("%Y%m%d-%H%M%S", datetime,
-			sizeof(datetime));
-	if (!ret) {
-		ret = -EINVAL;
-		goto error;
-	}
-
 	memset(&msg, 0, sizeof(msg));
 	msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
 	msg.u.snapshot_channel.key = key;
@@ -1226,7 +1219,7 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
 		msg.u.snapshot_channel.use_relayd = 1;
 		ret = snprintf(msg.u.snapshot_channel.pathname,
 				sizeof(msg.u.snapshot_channel.pathname), "%s/%s-%s%s",
-				output->consumer->subdir, output->name, datetime,
+				output->consumer->subdir, output->name, output->datetime,
 				session_path);
 		if (ret < 0) {
 			ret = -LTTNG_ERR_NOMEM;
@@ -1235,8 +1228,8 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
 	} else {
 		ret = snprintf(msg.u.snapshot_channel.pathname,
 				sizeof(msg.u.snapshot_channel.pathname), "%s/%s-%s%s",
-				output->consumer->dst.trace_path, output->name, datetime,
-				session_path);
+				output->consumer->dst.trace_path, output->name,
+				output->datetime, session_path);
 		if (ret < 0) {
 			ret = -LTTNG_ERR_NOMEM;
 			goto error;
diff --git a/src/bin/lttng-sessiond/snapshot.c b/src/bin/lttng-sessiond/snapshot.c
index 5265cf557..b14a398c3 100644
--- a/src/bin/lttng-sessiond/snapshot.c
+++ b/src/bin/lttng-sessiond/snapshot.c
@@ -214,11 +214,6 @@ error:
 	return output;
 }
 
-struct snapshot *snapshot_alloc(void)
-{
-	return zmalloc(sizeof(struct snapshot));
-}
-
 /*
  * Initialized a snapshot object that was already allocated.
  *
diff --git a/src/bin/lttng-sessiond/snapshot.h b/src/bin/lttng-sessiond/snapshot.h
index 91215272a..bf594b5fe 100644
--- a/src/bin/lttng-sessiond/snapshot.h
+++ b/src/bin/lttng-sessiond/snapshot.h
@@ -36,6 +36,12 @@ struct snapshot_output {
 	struct consumer_output *consumer;
 	int kernel_sockets_copied;
 	int ust_sockets_copied;
+	/*
+	 * Contains the string with "<date>-<time>" for when the snapshot command
+	 * is triggered. This is to make sure every streams will use the same time
+	 * for the directory output.
+	 */
+	char datetime[16];
 
 	/* Indexed by ID. */
 	struct lttng_ht_node_ulong node;
diff --git a/src/bin/lttng-sessiond/ust-consumer.c b/src/bin/lttng-sessiond/ust-consumer.c
index d85e32aa0..9f3557cd0 100644
--- a/src/bin/lttng-sessiond/ust-consumer.c
+++ b/src/bin/lttng-sessiond/ust-consumer.c
@@ -118,10 +118,12 @@ static int ask_channel_creation(struct ust_app_session *ua_sess,
 	DBG2("Asking UST consumer for channel");
 
 	/* Get and create full trace path of session. */
-	pathname = setup_trace_path(consumer, ua_sess);
-	if (!pathname) {
-		ret = -1;
-		goto error;
+	if (ua_sess->output_traces) {
+		pathname = setup_trace_path(consumer, ua_sess);
+		if (!pathname) {
+			ret = -1;
+			goto error;
+		}
 	}
 
 	/* Depending on the buffer type, a different channel key is used. */
@@ -176,7 +178,9 @@ static int ask_channel_creation(struct ust_app_session *ua_sess,
 	/* Communication protocol error. */
 	assert(key == ua_chan->key);
 	/* We need at least one where 1 stream for 1 cpu. */
-	assert(ua_chan->expected_stream_count > 0);
+	if (ua_sess->output_traces) {
+		assert(ua_chan->expected_stream_count > 0);
+	}
 
 	DBG2("UST ask channel %" PRIu64 " successfully done with %u stream(s)", key,
 			ua_chan->expected_stream_count);
diff --git a/src/common/consumer-stream.c b/src/common/consumer-stream.c
index b5ab6c6be..027d75150 100644
--- a/src/common/consumer-stream.c
+++ b/src/common/consumer-stream.c
@@ -19,6 +19,7 @@
 
 #define _GNU_SOURCE
 #include <assert.h>
+#include <inttypes.h>
 #include <sys/mman.h>
 #include <unistd.h>
 
@@ -80,6 +81,7 @@ void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
 			uatomic_read(&relayd->destroy_flag)) {
 		consumer_destroy_relayd(relayd);
 	}
+	stream->net_seq_idx = (uint64_t) -1ULL;
 }
 
 /*
@@ -110,11 +112,11 @@ void consumer_stream_close(struct lttng_consumer_stream *stream)
 			if (ret) {
 				PERROR("close");
 			}
+			stream->wait_fd = -1;
 		}
 		break;
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
-		lttng_ustconsumer_del_stream(stream);
 		break;
 	default:
 		ERR("Unknown consumer_data type");
@@ -127,6 +129,7 @@ void consumer_stream_close(struct lttng_consumer_stream *stream)
 		if (ret) {
 			PERROR("close");
 		}
+		stream->out_fd = -1;
 	}
 
 	/* Check and cleanup relayd if needed. */
@@ -151,6 +154,8 @@ void consumer_stream_delete(struct lttng_consumer_stream *stream,
 	struct lttng_ht_iter iter;
 
 	assert(stream);
+	/* Should NEVER be called not in monitor mode. */
+	assert(stream->chan->monitor);
 
 	rcu_read_lock();
 
@@ -177,15 +182,9 @@ void consumer_stream_delete(struct lttng_consumer_stream *stream,
 
 	rcu_read_unlock();
 
-	/*
-	 * For a *non* monitored stream, we MUST NOT decrement or else the data
-	 * thread will use the wrong value or stream for its local stream set.
-	 */
-	if (stream->chan->monitor) {
-		/* Decrement the stream count of the global consumer data. */
-		assert(consumer_data.stream_count > 0);
-		consumer_data.stream_count--;
-	}
+	/* Decrement the stream count of the global consumer data. */
+	assert(consumer_data.stream_count > 0);
+	consumer_data.stream_count--;
 }
 
 /*
@@ -199,44 +198,110 @@ void consumer_stream_free(struct lttng_consumer_stream *stream)
 }
 
 /*
- * Destroy a stream completely. This will delete, close and free the stream.
- * Once return, the stream is NO longer usable. Its channel may get destroyed
- * if conditions are met.
- *
- * This MUST be called WITHOUT the consumer data and stream lock acquired.
+ * Destroy the stream's buffers of the tracer.
  */
-void consumer_stream_destroy(struct lttng_consumer_stream *stream,
-		struct lttng_ht *ht)
+void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 {
-	struct lttng_consumer_channel *free_chan = NULL;
+	assert(stream);
+
+	switch (consumer_data.type) {
+	case LTTNG_CONSUMER_KERNEL:
+		break;
+	case LTTNG_CONSUMER32_UST:
+	case LTTNG_CONSUMER64_UST:
+		lttng_ustconsumer_del_stream(stream);
+		break;
+	default:
+		ERR("Unknown consumer_data type");
+		assert(0);
+	}
+}
 
+/*
+ * Destroy a stream in no monitor mode.
+ *
+ * We need a separate function because this can be called inside a destroy
+ * channel path which have the consumer data lock acquired. Also, in no monitor
+ * mode, the channel refcount is NOT incremented per stream since the ownership
+ * of those streams are INSIDE the channel making the lazy destroy channel not
+ * possible for a non monitor stream.
+ *
+ * Furthermore, there is no need to delete the stream from the global hash
+ * table so we avoid useless calls.
+ */
+static void destroy_no_monitor(struct lttng_consumer_stream *stream)
+{
 	assert(stream);
 
-	DBG("Consumer stream destroy - wait_fd: %d", stream->wait_fd);
+	DBG("Consumer stream destroy unmonitored key: %" PRIu64, stream->key);
+
+	/* Destroy tracer buffers of the stream. */
+	consumer_stream_destroy_buffers(stream);
+	/* Close down everything including the relayd if one. */
+	consumer_stream_close(stream);
+}
 
-	pthread_mutex_lock(&consumer_data.lock);
-	pthread_mutex_lock(&stream->lock);
+/*
+ * Destroy a stream in monitor mode.
+ */
+static void destroy_monitor(struct lttng_consumer_stream *stream,
+		struct lttng_ht *ht)
+{
+	assert(stream);
+
+	DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
 
 	/* Remove every reference of the stream in the consumer. */
 	consumer_stream_delete(stream, ht);
-
+	/* Destroy tracer buffers of the stream. */
+	consumer_stream_destroy_buffers(stream);
 	/* Close down everything including the relayd if one. */
 	consumer_stream_close(stream);
+}
 
-	/* Update refcount of channel and see if we need to destroy it. */
-	if (!uatomic_sub_return(&stream->chan->refcount, 1)
-			&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
-		free_chan = stream->chan;
-	}
+/*
+ * Destroy a stream completely. This will delete, close and free the stream.
+ * Once return, the stream is NO longer usable. Its channel may get destroyed
+ * if conditions are met for a monitored stream.
+ *
+ * This MUST be called WITHOUT the consumer data and stream lock acquired if
+ * the stream is in _monitor_ mode else it does not matter.
+ */
+void consumer_stream_destroy(struct lttng_consumer_stream *stream,
+		struct lttng_ht *ht)
+{
+	assert(stream);
+
+	/* Stream is in monitor mode. */
+	if (stream->chan->monitor) {
+		struct lttng_consumer_channel *free_chan = NULL;
 
-	/* Indicates that the consumer data state MUST be updated after this. */
-	consumer_data.need_update = 1;
+		pthread_mutex_lock(&consumer_data.lock);
+		pthread_mutex_lock(&stream->lock);
 
-	pthread_mutex_unlock(&stream->lock);
-	pthread_mutex_unlock(&consumer_data.lock);
+		destroy_monitor(stream, ht);
 
-	if (free_chan) {
-		consumer_del_channel(free_chan);
+		/* Update refcount of channel and see if we need to destroy it. */
+		if (!uatomic_sub_return(&stream->chan->refcount, 1)
+				&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
+			free_chan = stream->chan;
+		}
+
+		/* Indicates that the consumer data state MUST be updated after this. */
+		consumer_data.need_update = 1;
+
+		pthread_mutex_unlock(&stream->lock);
+		pthread_mutex_unlock(&consumer_data.lock);
+
+		if (free_chan) {
+			consumer_del_channel(free_chan);
+		}
+	} else {
+		/*
+		 * No monitor mode the stream's ownership is in its channel thus we
+		 * don't have to handle the channel refcount nor the lazy deletion.
+		 */
+		destroy_no_monitor(stream);
 	}
 
 	/* Free stream within a RCU call. */
diff --git a/src/common/consumer-stream.h b/src/common/consumer-stream.h
index 3036b2c77..3096e1e78 100644
--- a/src/common/consumer-stream.h
+++ b/src/common/consumer-stream.h
@@ -62,4 +62,10 @@ void consumer_stream_free(struct lttng_consumer_stream *stream);
 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 		struct lttng_ht *ht);
 
+/*
+ * Destroy the stream's buffers on the tracer side. This is also called in a
+ * stream destroy.
+ */
+void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream);
+
 #endif /* LTTNG_CONSUMER_STREAM_H */
diff --git a/src/common/consumer.c b/src/common/consumer.c
index 1f96fa9d4..0005b1039 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -301,8 +301,11 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 		cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
 				send_node) {
 			cds_list_del(&stream->send_node);
-			lttng_ustconsumer_del_stream(stream);
-			free(stream);
+			/*
+			 * Once a stream is added to this list, the buffers were created so
+			 * we have a guarantee that this call will succeed.
+			 */
+			consumer_stream_destroy(stream, NULL);
 		}
 		lttng_ustconsumer_del_channel(channel);
 		break;
@@ -312,25 +315,6 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 		goto end;
 	}
 
-	/* Empty no monitor streams list. */
-	if (!channel->monitor) {
-		struct lttng_consumer_stream *stream, *stmp;
-
-		/*
-		 * So, these streams are not visible to any data thread. This is why we
-		 * close and free them because they were never added to any data
-		 * structure apart from this one.
-		 */
-		cds_list_for_each_entry_safe(stream, stmp,
-				&channel->stream_no_monitor_list.head, no_monitor_node) {
-			cds_list_del(&stream->no_monitor_node);
-			/* Close everything in that stream. */
-			consumer_stream_close(stream);
-			/* Free the ressource. */
-			consumer_stream_free(stream);
-		}
-	}
-
 	rcu_read_lock();
 	iter.iter.node = &channel->node.node;
 	ret = lttng_ht_del(consumer_data.channel_ht, &iter);
@@ -693,6 +677,66 @@ error:
 	return relayd;
 }
 
+/*
+ * Find a relayd and send the stream
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
+		char *path)
+{
+	int ret = 0;
+	struct consumer_relayd_sock_pair *relayd;
+
+	assert(stream);
+	assert(stream->net_seq_idx != -1ULL);
+	assert(path);
+
+	/* The stream is not metadata. Get relayd reference if exists. */
+	rcu_read_lock();
+	relayd = consumer_find_relayd(stream->net_seq_idx);
+	if (relayd != NULL) {
+		/* Add stream on the relayd */
+		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+		ret = relayd_add_stream(&relayd->control_sock, stream->name,
+				path, &stream->relayd_stream_id,
+				stream->chan->tracefile_size, stream->chan->tracefile_count);
+		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+		if (ret < 0) {
+			goto end;
+		}
+		uatomic_inc(&relayd->refcount);
+	} else {
+		ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
+				stream->key, stream->net_seq_idx);
+		ret = -1;
+		goto end;
+	}
+
+	DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
+			stream->name, stream->key, stream->net_seq_idx);
+
+end:
+	rcu_read_unlock();
+	return ret;
+}
+
+/*
+ * Find a relayd and close the stream
+ */
+void close_relayd_stream(struct lttng_consumer_stream *stream)
+{
+	struct consumer_relayd_sock_pair *relayd;
+
+	/* The stream is not metadata. Get relayd reference if exists. */
+	rcu_read_lock();
+	relayd = consumer_find_relayd(stream->net_seq_idx);
+	if (relayd) {
+		consumer_stream_relayd_close(stream, relayd);
+	}
+	rcu_read_unlock();
+}
+
 /*
  * Handle stream for relayd transmission if the stream applies for network
  * streaming where the net sequence index is set.
@@ -816,7 +860,6 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 	channel->wait_fd = -1;
 
 	CDS_INIT_LIST_HEAD(&channel->streams.head);
-	CDS_INIT_LIST_HEAD(&channel->stream_no_monitor_list.head);
 
 	DBG("Allocated channel (key %" PRIu64 ")", channel->key)
 
@@ -856,7 +899,7 @@ end:
 	pthread_mutex_unlock(&consumer_data.lock);
 
 	if (!ret && channel->wait_fd != -1 &&
-			channel->metadata_stream == NULL) {
+			channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
 		notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
 	}
 	return ret;
@@ -2744,7 +2787,6 @@ restart:
 				lttng_poll_del(&events, chan->wait_fd);
 				ret = lttng_ht_del(channel_ht, &iter);
 				assert(ret == 0);
-				assert(cds_list_empty(&chan->streams.head));
 				consumer_close_channel_streams(chan);
 
 				/* Release our own refcount */
diff --git a/src/common/consumer.h b/src/common/consumer.h
index fa503ea11..4cf6e910d 100644
--- a/src/common/consumer.h
+++ b/src/common/consumer.h
@@ -140,12 +140,6 @@ struct lttng_consumer_channel {
 	 */
 	struct stream_list streams;
 
-	/*
-	 * List of streams in no monitor mode for this channel. Used ONLY for
-	 * snapshots recording.
-	 */
-	struct stream_list stream_no_monitor_list;
-
 	/*
 	 * Set if the channel is metadata. We keep a reference to the stream
 	 * because we have to flush data once pushed by the session daemon. For a
@@ -265,9 +259,6 @@ struct lttng_consumer_stream {
 	/* On-disk circular buffer */
 	uint64_t tracefile_size_current;
 	uint64_t tracefile_count_current;
-
-	/* Node for the no monitor stream list in a channel. */
-	struct cds_list_head no_monitor_node;
 };
 
 /*
@@ -512,6 +503,8 @@ void consumer_del_channel(struct lttng_consumer_channel *channel);
 struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
 		uint64_t net_seq_idx);
 struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key);
+int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, char *path);
+void close_relayd_stream(struct lttng_consumer_stream *stream);
 struct lttng_consumer_channel *consumer_find_channel(uint64_t key);
 int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
 		size_t data_size);
diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 02198bd1b..a6306291a 100644
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -104,72 +104,6 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
 	return ret;
 }
 
-/*
- * Find a relayd and send the stream
- *
- * Returns 0 on success, < 0 on error
- */
-static int send_relayd_stream(struct lttng_consumer_stream *stream,
-		char *path)
-{
-	int ret = 0;
-	const char *stream_path;
-	struct consumer_relayd_sock_pair *relayd;
-
-	assert(stream);
-	assert(stream->net_seq_idx != -1ULL);
-
-	if (path != NULL) {
-		stream_path = path;
-	} else {
-		stream_path = stream->chan->pathname;
-	}
-
-	/* The stream is not metadata. Get relayd reference if exists. */
-	rcu_read_lock();
-	relayd = consumer_find_relayd(stream->net_seq_idx);
-	if (relayd != NULL) {
-		/* Add stream on the relayd */
-		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
-		ret = relayd_add_stream(&relayd->control_sock, stream->name,
-				stream_path, &stream->relayd_stream_id,
-				stream->chan->tracefile_size, stream->chan->tracefile_count);
-		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-		if (ret < 0) {
-			goto end;
-		}
-		uatomic_inc(&relayd->refcount);
-	} else {
-		ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
-				stream->key, stream->net_seq_idx);
-		ret = -1;
-		goto end;
-	}
-
-	DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
-			stream->name, stream->key, stream->net_seq_idx);
-
-end:
-	rcu_read_unlock();
-	return ret;
-}
-
-/*
- * Find a relayd and close the stream
- */
-static void close_relayd_stream(struct lttng_consumer_stream *stream)
-{
-	struct consumer_relayd_sock_pair *relayd;
-
-	/* The stream is not metadata. Get relayd reference if exists. */
-	rcu_read_lock();
-	relayd = consumer_find_relayd(stream->net_seq_idx);
-	if (relayd) {
-		consumer_stream_relayd_close(stream, relayd);
-	}
-	rcu_read_unlock();
-}
-
 /*
  * Take a snapshot of all the stream of a channel
  *
@@ -201,8 +135,7 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
 		goto end;
 	}
 
-	cds_list_for_each_entry(stream, &channel->stream_no_monitor_list.head,
-			no_monitor_node) {
+	cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
 		/*
 		 * Lock stream because we are about to change its state.
 		 */
@@ -215,14 +148,15 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
 		stream->net_seq_idx = relayd_id;
 		channel->relayd_id = relayd_id;
 		if (relayd_id != (uint64_t) -1ULL) {
-			ret = send_relayd_stream(stream, path);
+			ret = consumer_send_relayd_stream(stream, path);
 			if (ret < 0) {
 				ERR("sending stream to relayd");
 				goto end_unlock;
 			}
 		} else {
 			ret = utils_create_stream_file(path, stream->name,
-					stream->chan->tracefile_size, stream->tracefile_count_current,
+					stream->chan->tracefile_size,
+					stream->tracefile_count_current,
 					stream->uid, stream->gid);
 			if (ret < 0) {
 				ERR("utils_create_stream_file");
@@ -238,7 +172,7 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
 
 		ret = kernctl_buffer_flush(stream->wait_fd);
 		if (ret < 0) {
-			ERR("Failed to flush kernel metadata stream");
+			ERR("Failed to flush kernel stream");
 			goto end_unlock;
 		}
 
@@ -391,7 +325,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
 	}
 
 	if (use_relayd) {
-		ret = send_relayd_stream(metadata_stream, path);
+		ret = consumer_send_relayd_stream(metadata_stream, path);
 		if (ret < 0) {
 			goto error;
 		}
@@ -685,12 +619,11 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			DBG("Kernel consumer add stream %s in no monitor mode with "
 					"relayd id %" PRIu64, new_stream->name,
 					new_stream->net_seq_idx);
-			cds_list_add(&new_stream->no_monitor_node,
-					&channel->stream_no_monitor_list.head);
+			cds_list_add(&new_stream->send_node, &channel->streams.head);
 			break;
 		}
 
-		ret = send_relayd_stream(new_stream, NULL);
+		ret = consumer_send_relayd_stream(new_stream, NULL);
 		if (ret < 0) {
 			consumer_stream_free(new_stream);
 			goto end_nosignal;
diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 18eb507de..f318af153 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -37,6 +37,7 @@
 #include <common/relayd/relayd.h>
 #include <common/compat/fcntl.h>
 #include <common/consumer-metadata-cache.h>
+#include <common/consumer-stream.h>
 #include <common/consumer-timer.h>
 #include <common/utils.h>
 
@@ -284,7 +285,9 @@ static int create_ust_streams(struct lttng_consumer_channel *channel,
 		 * Increment channel refcount since the channel reference has now been
 		 * assigned in the allocation process above.
 		 */
-		uatomic_inc(&stream->chan->refcount);
+		if (stream->chan->monitor) {
+			uatomic_inc(&stream->chan->refcount);
+		}
 
 		/*
 		 * Order is important this is why a list is used. On error, the caller
@@ -503,18 +506,27 @@ static int ask_channel(struct lttng_consumer_local_data *ctx, int sock,
 	/* The reply msg status is handled in the following call. */
 	ret = create_ust_channel(attr, &channel->uchan);
 	if (ret < 0) {
-		goto error;
+		goto end;
 	}
 
 	channel->wait_fd = ustctl_channel_get_wait_fd(channel->uchan);
 
+	/*
+	 * For the snapshots (no monitor), we create the metadata streams
+	 * on demand, not during the channel creation.
+	 */
+	if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && !channel->monitor) {
+		ret = 0;
+		goto end;
+	}
+
 	/* Open all streams for this channel. */
 	ret = create_ust_streams(channel, ctx);
 	if (ret < 0) {
-		goto error;
+		goto end;
 	}
 
-error:
+end:
 	return ret;
 }
 
@@ -694,7 +706,17 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
 	if (!metadata) {
 		ERR("UST consumer push metadata %" PRIu64 " not found", key);
 		ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
-		goto error_find;
+		goto end;
+	}
+
+	/*
+	 * In no monitor mode, the metadata channel has no stream(s) so skip the
+	 * ownership transfer to the metadata thread.
+	 */
+	if (!metadata->monitor) {
+		DBG("Metadata channel in no monitor");
+		ret = 0;
+		goto end;
 	}
 
 	/*
@@ -726,7 +748,8 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
 	/* List MUST be empty after or else it could be reused. */
 	assert(cds_list_empty(&metadata->streams.head));
 
-	return 0;
+	ret = 0;
+	goto end;
 
 error:
 	/*
@@ -736,7 +759,268 @@ error:
 	 * will make sure to clean that list.
 	 */
 	consumer_del_channel(metadata);
-error_find:
+end:
+	return ret;
+}
+
+/*
+ * Snapshot the whole metadata.
+ *
+ * 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)
+{
+	int ret = 0;
+	ssize_t write_len;
+	uint64_t total_len = 0;
+	struct lttng_consumer_channel *metadata_channel;
+	struct lttng_consumer_stream *metadata_stream;
+
+	assert(path);
+	assert(ctx);
+
+	DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s",
+			key, path);
+
+	rcu_read_lock();
+
+	metadata_channel = consumer_find_channel(key);
+	if (!metadata_channel) {
+		ERR("UST snapshot metadata channel not found for key %lu", key);
+		ret = -1;
+		goto error;
+	}
+	assert(!metadata_channel->monitor);
+
+	/*
+	 * Ask the sessiond if we have new metadata waiting and update the
+	 * consumer metadata cache.
+	 */
+	ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel);
+	if (ret < 0) {
+		goto error;
+	}
+
+	/*
+	 * 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);
+	if (ret < 0) {
+		goto error;
+	}
+
+	metadata_stream = metadata_channel->metadata_stream;
+	assert(metadata_stream);
+
+	if (relayd_id != (uint64_t) -1ULL) {
+		metadata_stream->net_seq_idx = relayd_id;
+		ret = consumer_send_relayd_stream(metadata_stream, path);
+		if (ret < 0) {
+			goto error_stream;
+		}
+	} else {
+		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);
+		if (ret < 0) {
+			goto error_stream;
+		}
+		metadata_stream->out_fd = ret;
+		metadata_stream->tracefile_size_current = 0;
+	}
+
+	pthread_mutex_lock(&metadata_channel->metadata_cache->lock);
+	while (total_len < metadata_channel->metadata_cache->total_bytes_written) {
+		/*
+		 * Write at most one packet of metadata into the channel
+		 * to avoid blocking here.
+		 */
+		write_len = ustctl_write_one_packet_to_channel(metadata_channel->uchan,
+				metadata_channel->metadata_cache->data,
+				metadata_channel->metadata_cache->total_bytes_written);
+		if (write_len < 0) {
+			ERR("UST consumer snapshot writing metadata packet");
+			ret = -1;
+			goto error_unlock;
+		}
+		total_len += write_len;
+
+		DBG("Written %" PRIu64 " bytes to metadata (left: %" PRIu64 ")",
+				write_len,
+				metadata_channel->metadata_cache->total_bytes_written - write_len);
+		ustctl_flush_buffer(metadata_stream->ustream, 1);
+		ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
+		if (ret < 0) {
+			goto error_unlock;
+		}
+	}
+
+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
+	 * metadata stream.
+	 */
+	cds_list_del(&metadata_stream->send_node);
+	consumer_stream_destroy(metadata_stream, NULL);
+	metadata_channel->metadata_stream = NULL;
+
+error:
+	rcu_read_unlock();
+	return ret;
+}
+
+/*
+ * Take a snapshot of all the stream of a channel.
+ *
+ * Returns 0 on success, < 0 on error
+ */
+static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
+		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);
+	assert(ctx);
+
+	rcu_read_lock();
+
+	if (relayd_id != (uint64_t) -1ULL) {
+		use_relayd = 1;
+	}
+
+	channel = consumer_find_channel(key);
+	if (!channel) {
+		ERR("UST snapshot channel not found for key %lu", key);
+		ret = -1;
+		goto error;
+	}
+	assert(!channel->monitor);
+	DBG("UST consumer snapshot channel %lu", key);
+
+	cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+		/* Lock stream because we are about to change its state. */
+		pthread_mutex_lock(&stream->lock);
+		stream->net_seq_idx = relayd_id;
+
+		if (use_relayd) {
+			ret = consumer_send_relayd_stream(stream, path);
+			if (ret < 0) {
+				goto error_unlock;
+			}
+		} else {
+			ret = utils_create_stream_file(path, stream->name,
+					stream->chan->tracefile_size,
+					stream->tracefile_count_current,
+					stream->uid, stream->gid);
+			if (ret < 0) {
+				goto error_unlock;
+			}
+			stream->out_fd = ret;
+			stream->tracefile_size_current = 0;
+
+			DBG("UST consumer snapshot stream %s/%s (%" PRIu64 ")", path,
+					stream->name, stream->key);
+		}
+
+		ustctl_flush_buffer(stream->ustream, 1);
+
+		ret = lttng_ustconsumer_take_snapshot(stream);
+		if (ret < 0) {
+			ERR("Taking UST snapshot");
+			goto error_unlock;
+		}
+
+		ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
+		if (ret < 0) {
+			ERR("Produced UST snapshot position");
+			goto error_unlock;
+		}
+
+		ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
+		if (ret < 0) {
+			ERR("Consumerd UST snapshot position");
+			goto error_unlock;
+		}
+
+		while (consumed_pos < produced_pos) {
+			ssize_t read_len;
+			unsigned long len, padded_len;
+
+			DBG("UST consumer taking snapshot at pos %lu", consumed_pos);
+
+			ret = ustctl_get_subbuf(stream->ustream, &consumed_pos);
+			if (ret < 0) {
+				if (ret != -EAGAIN) {
+					PERROR("ustctl_get_subbuf snapshot");
+					goto error_close_stream;
+				}
+				DBG("UST consumer get subbuf failed. Skipping it.");
+				consumed_pos += stream->max_sb_size;
+				continue;
+			}
+
+			ret = ustctl_get_subbuf_size(stream->ustream, &len);
+			if (ret < 0) {
+				ERR("Snapshot ustctl_get_subbuf_size");
+				goto error_put_subbuf;
+			}
+
+			ret = ustctl_get_padded_subbuf_size(stream->ustream, &padded_len);
+			if (ret < 0) {
+				ERR("Snapshot ustctl_get_padded_subbuf_size");
+				goto error_put_subbuf;
+			}
+
+			read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
+					padded_len - len);
+			if (use_relayd) {
+				if (read_len != len) {
+					ret = -1;
+					goto error_put_subbuf;
+				}
+			} else {
+				if (read_len != padded_len) {
+					ret = -1;
+					goto error_put_subbuf;
+				}
+			}
+
+			ret = ustctl_put_subbuf(stream->ustream);
+			if (ret < 0) {
+				ERR("Snapshot ustctl_put_subbuf");
+				goto error_close_stream;
+			}
+			consumed_pos += stream->max_sb_size;
+		}
+
+		/* Simply close the stream so we can use it on the next snapshot. */
+		consumer_stream_close(stream);
+		pthread_mutex_unlock(&stream->lock);
+	}
+
+	rcu_read_unlock();
+	return 0;
+
+error_put_subbuf:
+	if (ustctl_put_subbuf(stream->ustream) < 0) {
+		ERR("Snapshot ustctl_put_subbuf");
+	}
+error_close_stream:
+	consumer_stream_close(stream);
+error_unlock:
+	pthread_mutex_unlock(&stream->lock);
+error:
+	rcu_read_unlock();
 	return ret;
 }
 
@@ -940,6 +1224,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		attr.switch_timer_interval = msg.u.ask_channel.switch_timer_interval;
 		attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
 		attr.chan_id = msg.u.ask_channel.chan_id;
+		attr.output = msg.u.ask_channel.output;
 		memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
 
 		/* Translate and save channel type. */
@@ -1043,6 +1328,14 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			goto error_fatal;
 		}
 
+		/*
+		 * In no monitor mode, the streams ownership is kept inside the channel
+		 * so don't send them to the data thread.
+		 */
+		if (!channel->monitor) {
+			goto end_msg_sessiond;
+		}
+
 		ret = send_streams_to_thread(channel, ctx);
 		if (ret < 0) {
 			/*
@@ -1053,7 +1346,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		}
 		/* List MUST be empty after or else it could be reused. */
 		assert(cds_list_empty(&channel->streams.head));
-
 		goto end_msg_sessiond;
 	}
 	case LTTNG_CONSUMER_DESTROY_CHANNEL:
@@ -1142,6 +1434,26 @@ 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 = LTTNG_ERR_UST_META_FAIL;
+			}
+		} else {
+			ret = snapshot_channel(msg.u.snapshot_channel.key,
+					msg.u.snapshot_channel.pathname,
+					msg.u.snapshot_channel.relayd_id,
+					ctx);
+			if (ret < 0) {
+				ERR("Snapshot channel failed");
+				ret_code = LTTNG_ERR_UST_CHAN_FAIL;
+			}
+		}
+
 		ret = consumer_send_status_msg(sock, ret_code);
 		if (ret < 0) {
 			/* Somehow, the session daemon is not responding anymore. */
@@ -1249,6 +1561,21 @@ int lttng_ustconsumer_get_produced_snapshot(
 	return ustctl_snapshot_get_produced(stream->ustream, pos);
 }
 
+/*
+ * Get the consumed position
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int lttng_ustconsumer_get_consumed_snapshot(
+		struct lttng_consumer_stream *stream, unsigned long *pos)
+{
+	assert(stream);
+	assert(stream->ustream);
+	assert(pos);
+
+	return ustctl_snapshot_get_consumed(stream->ustream, pos);
+}
+
 /*
  * Called when the stream signal the consumer that it has hang up.
  */
@@ -1379,8 +1706,10 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 {
 	int ret;
 
+	assert(stream);
+
 	/* Don't create anything if this is set for streaming. */
-	if (stream->net_seq_idx == (uint64_t) -1ULL) {
+	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);
diff --git a/src/common/ust-consumer/ust-consumer.h b/src/common/ust-consumer/ust-consumer.h
index 9f0b22dde..191cdabf8 100644
--- a/src/common/ust-consumer/ust-consumer.h
+++ b/src/common/ust-consumer/ust-consumer.h
@@ -30,6 +30,8 @@ int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream);
 
 int lttng_ustconsumer_get_produced_snapshot(
 		struct lttng_consumer_stream *stream, unsigned long *pos);
+int lttng_ustconsumer_get_consumed_snapshot(
+		struct lttng_consumer_stream *stream, unsigned long *pos);
 
 int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		int sock, struct pollfd *consumer_sockpoll);
-- 
2.34.1