X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=300fd2a2fc896108ac60c43ce920a5f633be111b;hp=af99333c8d2ae1bacc73f7f4af5fcd231e79e104;hb=d8ef542d25837bdfb960e5df2a91c5d18f5ef401;hpb=d21b0d71990ac6ec4272c1f80f0ca544103628b3

diff --git a/src/common/consumer.c b/src/common/consumer.c
index af99333c8..300fd2a2f 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -47,6 +47,16 @@ struct lttng_consumer_global_data consumer_data = {
 	.type = LTTNG_CONSUMER_UNKNOWN,
 };
 
+enum consumer_channel_action {
+	CONSUMER_CHANNEL_ADD,
+	CONSUMER_CHANNEL_QUIT,
+};
+
+struct consumer_channel_msg {
+	enum consumer_channel_action action;
+	struct lttng_consumer_channel *chan;
+};
+
 /*
  * Flag to inform the polling thread to quit when all fd hung up. Updated by
  * the consumer_thread_receive_fds when it notices that all fds has hung up.
@@ -63,21 +73,6 @@ volatile int consumer_quit;
 static struct lttng_ht *metadata_ht;
 static struct lttng_ht *data_ht;
 
-/*
- * This hash table contains the mapping between the session id of the sessiond
- * and the relayd session id. Element of the ht are indexed by sessiond session
- * id.
- *
- * Node can be added when a relayd communication is opened in the sessiond
- * thread.
- *
- * Note that a session id of the session daemon is unique to a tracing session
- * and not to a domain session. However, a domain session has one consumer
- * which forces the 1-1 mapping between a consumer and a domain session (ex:
- * UST). This means that we can't have duplicate in this ht.
- */
-static struct lttng_ht *relayd_session_id_ht;
-
 /*
  * Notify a thread pipe to poll back again. This usually means that some global
  * state has changed so we just send back the thread in a poll wait call.
@@ -93,28 +88,59 @@ static void notify_thread_pipe(int wpipe)
 	} while (ret < 0 && errno == EINTR);
 }
 
+static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
+		struct lttng_consumer_channel *chan,
+		enum consumer_channel_action action)
+{
+	struct consumer_channel_msg msg;
+	int ret;
+
+	msg.action = action;
+	msg.chan = chan;
+	do {
+		ret = write(ctx->consumer_channel_pipe[1], &msg, sizeof(msg));
+	} while (ret < 0 && errno == EINTR);
+}
+
+static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
+		struct lttng_consumer_channel **chan,
+		enum consumer_channel_action *action)
+{
+	struct consumer_channel_msg msg;
+	int ret;
+
+	do {
+		ret = read(ctx->consumer_channel_pipe[0], &msg, sizeof(msg));
+	} while (ret < 0 && errno == EINTR);
+	if (ret > 0) {
+		*action = msg.action;
+		*chan = msg.chan;
+	}
+	return ret;
+}
+
 /*
  * Find a stream. The consumer_data.lock must be locked during this
  * call.
  */
-static struct lttng_consumer_stream *consumer_find_stream(int key,
+static struct lttng_consumer_stream *find_stream(uint64_t key,
 		struct lttng_ht *ht)
 {
 	struct lttng_ht_iter iter;
-	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_node_u64 *node;
 	struct lttng_consumer_stream *stream = NULL;
 
 	assert(ht);
 
-	/* Negative keys are lookup failures */
-	if (key < 0) {
+	/* -1ULL keys are lookup failures */
+	if (key == (uint64_t) -1ULL) {
 		return NULL;
 	}
 
 	rcu_read_lock();
 
-	lttng_ht_lookup(ht, (void *)((unsigned long) key), &iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
+	lttng_ht_lookup(ht, &key, &iter);
+	node = lttng_ht_iter_get_node_u64(&iter);
 	if (node != NULL) {
 		stream = caa_container_of(node, struct lttng_consumer_stream, node);
 	}
@@ -124,20 +150,20 @@ static struct lttng_consumer_stream *consumer_find_stream(int key,
 	return stream;
 }
 
-void consumer_steal_stream_key(int key, struct lttng_ht *ht)
+static void steal_stream_key(int key, struct lttng_ht *ht)
 {
 	struct lttng_consumer_stream *stream;
 
 	rcu_read_lock();
-	stream = consumer_find_stream(key, ht);
+	stream = find_stream(key, ht);
 	if (stream) {
-		stream->key = -1;
+		stream->key = -1ULL;
 		/*
 		 * We don't want the lookup to match, but we still need
 		 * to iterate on this stream when iterating over the hash table. Just
 		 * change the node key.
 		 */
-		stream->node.key = -1;
+		stream->node.key = -1ULL;
 	}
 	rcu_read_unlock();
 }
@@ -148,20 +174,19 @@ void consumer_steal_stream_key(int key, struct lttng_ht *ht)
  * RCU read side lock MUST be acquired before calling this function and
  * protects the channel ptr.
  */
-static struct lttng_consumer_channel *consumer_find_channel(int key)
+struct lttng_consumer_channel *consumer_find_channel(uint64_t key)
 {
 	struct lttng_ht_iter iter;
-	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_node_u64 *node;
 	struct lttng_consumer_channel *channel = NULL;
 
-	/* Negative keys are lookup failures */
-	if (key < 0) {
+	/* -1ULL keys are lookup failures */
+	if (key == (uint64_t) -1ULL) {
 		return NULL;
 	}
 
-	lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) key),
-			&iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
+	lttng_ht_lookup(consumer_data.channel_ht, &key, &iter);
+	node = lttng_ht_iter_get_node_u64(&iter);
 	if (node != NULL) {
 		channel = caa_container_of(node, struct lttng_consumer_channel, node);
 	}
@@ -169,42 +194,33 @@ static struct lttng_consumer_channel *consumer_find_channel(int key)
 	return channel;
 }
 
-static void consumer_steal_channel_key(int key)
+static void free_stream_rcu(struct rcu_head *head)
 {
-	struct lttng_consumer_channel *channel;
+	struct lttng_ht_node_u64 *node =
+		caa_container_of(head, struct lttng_ht_node_u64, head);
+	struct lttng_consumer_stream *stream =
+		caa_container_of(node, struct lttng_consumer_stream, node);
 
-	rcu_read_lock();
-	channel = consumer_find_channel(key);
-	if (channel) {
-		channel->key = -1;
-		/*
-		 * We don't want the lookup to match, but we still need
-		 * to iterate on this channel when iterating over the hash table. Just
-		 * change the node key.
-		 */
-		channel->node.key = -1;
-	}
-	rcu_read_unlock();
+	free(stream);
 }
 
-static
-void consumer_free_stream(struct rcu_head *head)
+static void free_channel_rcu(struct rcu_head *head)
 {
-	struct lttng_ht_node_ulong *node =
-		caa_container_of(head, struct lttng_ht_node_ulong, head);
-	struct lttng_consumer_stream *stream =
-		caa_container_of(node, struct lttng_consumer_stream, node);
+	struct lttng_ht_node_u64 *node =
+		caa_container_of(head, struct lttng_ht_node_u64, head);
+	struct lttng_consumer_channel *channel =
+		caa_container_of(node, struct lttng_consumer_channel, node);
 
-	free(stream);
+	free(channel);
 }
 
 /*
  * RCU protected relayd socket pair free.
  */
-static void consumer_rcu_free_relayd(struct rcu_head *head)
+static void free_relayd_rcu(struct rcu_head *head)
 {
-	struct lttng_ht_node_ulong *node =
-		caa_container_of(head, struct lttng_ht_node_ulong, head);
+	struct lttng_ht_node_u64 *node =
+		caa_container_of(head, struct lttng_ht_node_u64, head);
 	struct consumer_relayd_sock_pair *relayd =
 		caa_container_of(node, struct consumer_relayd_sock_pair, node);
 
@@ -231,7 +247,6 @@ static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 {
 	int ret;
 	struct lttng_ht_iter iter;
-	struct lttng_ht_node_ulong *node;
 
 	if (relayd == NULL) {
 		return;
@@ -239,22 +254,6 @@ static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 
 	DBG("Consumer destroy and close relayd socket pair");
 
-	/* Loockup for a relayd node in the session id map hash table. */
-	lttng_ht_lookup(relayd_session_id_ht,
-			(void *)((unsigned long) relayd->sessiond_session_id), &iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
-	if (node == NULL) {
-		/* We assume the relayd is being or is destroyed */
-		return;
-	}
-
-	/*
-	 * Try to delete it from the relayd session id ht. The return value is of
-	 * no importance since either way we are going to try to delete the relayd
-	 * from the global relayd_ht.
-	 */
-	lttng_ht_del(relayd_session_id_ht, &iter);
-
 	iter.iter.node = &relayd->node.node;
 	ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
 	if (ret != 0) {
@@ -263,7 +262,44 @@ static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 	}
 
 	/* RCU free() call */
-	call_rcu(&relayd->node.head, consumer_rcu_free_relayd);
+	call_rcu(&relayd->node.head, free_relayd_rcu);
+}
+
+/*
+ * Remove a channel from the global list protected by a mutex. This function is
+ * also responsible for freeing its data structures.
+ */
+void consumer_del_channel(struct lttng_consumer_channel *channel)
+{
+	int ret;
+	struct lttng_ht_iter iter;
+
+	DBG("Consumer delete channel key %" PRIu64, channel->key);
+
+	pthread_mutex_lock(&consumer_data.lock);
+
+	switch (consumer_data.type) {
+	case LTTNG_CONSUMER_KERNEL:
+		break;
+	case LTTNG_CONSUMER32_UST:
+	case LTTNG_CONSUMER64_UST:
+		lttng_ustconsumer_del_channel(channel);
+		break;
+	default:
+		ERR("Unknown consumer_data type");
+		assert(0);
+		goto end;
+	}
+
+	rcu_read_lock();
+	iter.iter.node = &channel->node.node;
+	ret = lttng_ht_del(consumer_data.channel_ht, &iter);
+	assert(!ret);
+	rcu_read_unlock();
+
+	call_rcu(&channel->node.head, free_channel_rcu);
+end:
+	pthread_mutex_unlock(&consumer_data.lock);
 }
 
 /*
@@ -283,8 +319,6 @@ static void cleanup_relayd_ht(void)
 	}
 
 	lttng_ht_destroy(consumer_data.relayd_ht);
-	/* The destroy_relayd call makes sure that this ht is empty here. */
-	lttng_ht_destroy(relayd_session_id_ht);
 
 	rcu_read_unlock();
 }
@@ -402,7 +436,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 
 	if (ht == NULL) {
 		/* Means the stream was allocated but not successfully added */
-		goto free_stream;
+		goto free_stream_rcu;
 	}
 
 	pthread_mutex_lock(&consumer_data.lock);
@@ -432,7 +466,10 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 	ret = lttng_ht_del(ht, &iter);
 	assert(!ret);
 
-	/* Remove node session id from the consumer_data stream ht */
+	iter.iter.node = &stream->node_channel_id.node;
+	ret = lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
+	assert(!ret);
+
 	iter.iter.node = &stream->node_session_id.node;
 	ret = lttng_ht_del(consumer_data.stream_list_ht, &iter);
 	assert(!ret);
@@ -447,18 +484,6 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 			PERROR("close");
 		}
 	}
-	if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
-		ret = close(stream->wait_fd);
-		if (ret) {
-			PERROR("close");
-		}
-	}
-	if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
-		ret = close(stream->shm_fd);
-		if (ret) {
-			PERROR("close");
-		}
-	}
 
 	/* Check and cleanup relayd */
 	rcu_read_lock();
@@ -492,7 +517,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 
 	uatomic_dec(&stream->chan->refcount);
 	if (!uatomic_read(&stream->chan->refcount)
-			&& !uatomic_read(&stream->chan->nb_init_streams)) {
+			&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
 		free_chan = stream->chan;
 	}
 
@@ -505,91 +530,70 @@ end:
 		consumer_del_channel(free_chan);
 	}
 
-free_stream:
-	call_rcu(&stream->node.head, consumer_free_stream);
+free_stream_rcu:
+	call_rcu(&stream->node.head, free_stream_rcu);
 }
 
-struct lttng_consumer_stream *consumer_allocate_stream(
-		int channel_key, int stream_key,
-		int shm_fd, int wait_fd,
+struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
+		uint64_t stream_key,
 		enum lttng_consumer_stream_state state,
-		uint64_t mmap_len,
-		enum lttng_event_output output,
-		const char *path_name,
+		const char *channel_name,
 		uid_t uid,
 		gid_t gid,
-		int net_index,
-		int metadata_flag,
+		int relayd_id,
 		uint64_t session_id,
-		int *alloc_ret)
+		int cpu,
+		int *alloc_ret,
+		enum consumer_channel_type type)
 {
+	int ret;
 	struct lttng_consumer_stream *stream;
 
 	stream = zmalloc(sizeof(*stream));
 	if (stream == NULL) {
 		PERROR("malloc struct lttng_consumer_stream");
-		*alloc_ret = -ENOMEM;
+		ret = -ENOMEM;
 		goto end;
 	}
 
 	rcu_read_lock();
 
-	/*
-	 * Get stream's channel reference. Needed when adding the stream to the
-	 * global hash table.
-	 */
-	stream->chan = consumer_find_channel(channel_key);
-	if (!stream->chan) {
-		*alloc_ret = -ENOENT;
-		ERR("Unable to find channel for stream %d", stream_key);
-		goto error;
-	}
-
 	stream->key = stream_key;
-	stream->shm_fd = shm_fd;
-	stream->wait_fd = wait_fd;
 	stream->out_fd = -1;
 	stream->out_fd_offset = 0;
 	stream->state = state;
-	stream->mmap_len = mmap_len;
-	stream->mmap_base = NULL;
-	stream->output = output;
 	stream->uid = uid;
 	stream->gid = gid;
-	stream->net_seq_idx = net_index;
-	stream->metadata_flag = metadata_flag;
+	stream->net_seq_idx = relayd_id;
 	stream->session_id = session_id;
-	strncpy(stream->path_name, path_name, sizeof(stream->path_name));
-	stream->path_name[sizeof(stream->path_name) - 1] = '\0';
 	pthread_mutex_init(&stream->lock, NULL);
 
-	/*
-	 * Index differently the metadata node because the thread is using an
-	 * internal hash table to match streams in the metadata_ht to the epoll set
-	 * file descriptor.
-	 */
-	if (metadata_flag) {
-		lttng_ht_node_init_ulong(&stream->node, stream->wait_fd);
+	/* If channel is the metadata, flag this stream as metadata. */
+	if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
+		stream->metadata_flag = 1;
+		/* Metadata is flat out. */
+		strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
 	} else {
-		lttng_ht_node_init_ulong(&stream->node, stream->key);
+		/* Format stream name to <channel_name>_<cpu_number> */
+		ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
+				channel_name, cpu);
+		if (ret < 0) {
+			PERROR("snprintf stream name");
+			goto error;
+		}
 	}
 
-	/* Init session id node with the stream session id */
-	lttng_ht_node_init_ulong(&stream->node_session_id, stream->session_id);
+	/* Key is always the wait_fd for streams. */
+	lttng_ht_node_init_u64(&stream->node, stream->key);
 
-	/*
-	 * The cpu number is needed before using any ustctl_* actions. Ignored for
-	 * the kernel so the value does not matter.
-	 */
-	pthread_mutex_lock(&consumer_data.lock);
-	stream->cpu = stream->chan->cpucount++;
-	pthread_mutex_unlock(&consumer_data.lock);
+	/* Init node per channel id key */
+	lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
 
-	DBG3("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu,"
-			" out_fd %d, net_seq_idx %d, session_id %" PRIu64,
-			stream->path_name, stream->key, stream->shm_fd, stream->wait_fd,
-			(unsigned long long) stream->mmap_len, stream->out_fd,
-			stream->net_seq_idx, stream->session_id);
+	/* Init session id node with the stream session id */
+	lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
+
+	DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64 " relayd_id %" PRIu64 ", session_id %" PRIu64,
+			stream->name, stream->key, channel_key, stream->net_seq_idx, stream->session_id);
 
 	rcu_read_unlock();
 	return stream;
@@ -598,13 +602,16 @@ error:
 	rcu_read_unlock();
 	free(stream);
 end:
+	if (alloc_ret) {
+		*alloc_ret = ret;
+	}
 	return NULL;
 }
 
 /*
  * Add a stream to the global list protected by a mutex.
  */
-static int consumer_add_stream(struct lttng_consumer_stream *stream,
+static int add_stream(struct lttng_consumer_stream *stream,
 		struct lttng_ht *ht)
 {
 	int ret = 0;
@@ -613,23 +620,26 @@ static int consumer_add_stream(struct lttng_consumer_stream *stream,
 	assert(stream);
 	assert(ht);
 
-	DBG3("Adding consumer stream %d", stream->key);
+	DBG3("Adding consumer stream %" PRIu64, stream->key);
 
 	pthread_mutex_lock(&consumer_data.lock);
 	pthread_mutex_lock(&stream->lock);
 	rcu_read_lock();
 
 	/* Steal stream identifier to avoid having streams with the same key */
-	consumer_steal_stream_key(stream->key, ht);
+	steal_stream_key(stream->key, ht);
+
+	lttng_ht_add_unique_u64(ht, &stream->node);
 
-	lttng_ht_add_unique_ulong(ht, &stream->node);
+	lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
+			&stream->node_channel_id);
 
 	/*
 	 * Add stream to the stream_list_ht of the consumer data. No need to steal
 	 * the key since the HT does not use it and we allow to add redundant keys
 	 * into this table.
 	 */
-	lttng_ht_add_ulong(consumer_data.stream_list_ht, &stream->node_session_id);
+	lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
 
 	/* Check and cleanup relayd */
 	relayd = consumer_find_relayd(stream->net_seq_idx);
@@ -641,14 +651,14 @@ static int consumer_add_stream(struct lttng_consumer_stream *stream,
 	uatomic_inc(&stream->chan->refcount);
 
 	/*
-	 * When nb_init_streams reaches 0, we don't need to trigger any action in
-	 * terms of destroying the associated channel, because the action that
+	 * When nb_init_stream_left reaches 0, we don't need to trigger any action
+	 * in terms of destroying the associated channel, because the action that
 	 * causes the count to become 0 also causes a stream to be added. The
 	 * channel deletion will thus be triggered by the following removal of this
 	 * stream.
 	 */
-	if (uatomic_read(&stream->chan->nb_init_streams) > 0) {
-		uatomic_dec(&stream->chan->nb_init_streams);
+	if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+		uatomic_dec(&stream->chan->nb_init_stream_left);
 	}
 
 	/* Update consumer data once the node is inserted. */
@@ -669,22 +679,18 @@ static int consumer_add_stream(struct lttng_consumer_stream *stream,
 static int add_relayd(struct consumer_relayd_sock_pair *relayd)
 {
 	int ret = 0;
-	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_node_u64 *node;
 	struct lttng_ht_iter iter;
 
-	if (relayd == NULL) {
-		ret = -1;
-		goto end;
-	}
+	assert(relayd);
 
 	lttng_ht_lookup(consumer_data.relayd_ht,
-			(void *)((unsigned long) relayd->net_seq_idx), &iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
+			&relayd->net_seq_idx, &iter);
+	node = lttng_ht_iter_get_node_u64(&iter);
 	if (node != NULL) {
-		/* Relayd already exist. Ignore the insertion */
 		goto end;
 	}
-	lttng_ht_add_unique_ulong(consumer_data.relayd_ht, &relayd->node);
+	lttng_ht_add_unique_u64(consumer_data.relayd_ht, &relayd->node);
 
 end:
 	return ret;
@@ -712,7 +718,7 @@ struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
 	obj->net_seq_idx = net_seq_idx;
 	obj->refcount = 0;
 	obj->destroy_flag = 0;
-	lttng_ht_node_init_ulong(&obj->node, obj->net_seq_idx);
+	lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
 	pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
 
 error:
@@ -726,20 +732,20 @@ error:
  * RCU read-side lock must be held across this call and while using the
  * returned object.
  */
-struct consumer_relayd_sock_pair *consumer_find_relayd(int key)
+struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
 {
 	struct lttng_ht_iter iter;
-	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_node_u64 *node;
 	struct consumer_relayd_sock_pair *relayd = NULL;
 
 	/* Negative keys are lookup failures */
-	if (key < 0) {
+	if (key == (uint64_t) -1ULL) {
 		goto error;
 	}
 
-	lttng_ht_lookup(consumer_data.relayd_ht, (void *)((unsigned long) key),
+	lttng_ht_lookup(consumer_data.relayd_ht, &key,
 			&iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
+	node = lttng_ht_iter_get_node_u64(&iter);
 	if (node != NULL) {
 		relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
 	}
@@ -782,7 +788,14 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
 		data_hdr.stream_id = htobe64(stream->relayd_stream_id);
 		data_hdr.data_size = htobe32(data_size);
 		data_hdr.padding_size = htobe32(padding);
-		data_hdr.net_seq_num = htobe64(stream->next_net_seq_num++);
+		/*
+		 * Note that net_seq_num below is assigned with the *current* value of
+		 * next_net_seq_num and only after that the next_net_seq_num will be
+		 * increment. This is why when issuing a command on the relayd using
+		 * this next value, 1 should always be substracted in order to compare
+		 * the last seen sequence number on the relayd side to the last sent.
+		 */
+		data_hdr.net_seq_num = htobe64(stream->next_net_seq_num);
 		/* Other fields are zeroed previously */
 
 		ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
@@ -791,6 +804,8 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
 			goto error;
 		}
 
+		++stream->next_net_seq_num;
+
 		/* Set to go on data socket */
 		outfd = relayd->data_sock.fd;
 	}
@@ -799,119 +814,51 @@ error:
 	return outfd;
 }
 
-static
-void consumer_free_channel(struct rcu_head *head)
-{
-	struct lttng_ht_node_ulong *node =
-		caa_container_of(head, struct lttng_ht_node_ulong, head);
-	struct lttng_consumer_channel *channel =
-		caa_container_of(node, struct lttng_consumer_channel, node);
-
-	free(channel);
-}
-
 /*
- * Remove a channel from the global list protected by a mutex. This
- * function is also responsible for freeing its data structures.
+ * Allocate and return a new lttng_consumer_channel object using the given key
+ * to initialize the hash table node.
+ *
+ * On error, return NULL.
  */
-void consumer_del_channel(struct lttng_consumer_channel *channel)
-{
-	int ret;
-	struct lttng_ht_iter iter;
-
-	DBG("Consumer delete channel key %d", channel->key);
-
-	pthread_mutex_lock(&consumer_data.lock);
-
-	switch (consumer_data.type) {
-	case LTTNG_CONSUMER_KERNEL:
-		break;
-	case LTTNG_CONSUMER32_UST:
-	case LTTNG_CONSUMER64_UST:
-		lttng_ustconsumer_del_channel(channel);
-		break;
-	default:
-		ERR("Unknown consumer_data type");
-		assert(0);
-		goto end;
-	}
-
-	rcu_read_lock();
-	iter.iter.node = &channel->node.node;
-	ret = lttng_ht_del(consumer_data.channel_ht, &iter);
-	assert(!ret);
-	rcu_read_unlock();
-
-	if (channel->mmap_base != NULL) {
-		ret = munmap(channel->mmap_base, channel->mmap_len);
-		if (ret != 0) {
-			PERROR("munmap");
-		}
-	}
-	if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
-		ret = close(channel->wait_fd);
-		if (ret) {
-			PERROR("close");
-		}
-	}
-	if (channel->shm_fd >= 0 && channel->wait_fd != channel->shm_fd) {
-		ret = close(channel->shm_fd);
-		if (ret) {
-			PERROR("close");
-		}
-	}
-
-	call_rcu(&channel->node.head, consumer_free_channel);
-end:
-	pthread_mutex_unlock(&consumer_data.lock);
-}
-
-struct lttng_consumer_channel *consumer_allocate_channel(
-		int channel_key,
-		int shm_fd, int wait_fd,
-		uint64_t mmap_len,
-		uint64_t max_sb_size,
-		unsigned int nb_init_streams)
+struct lttng_consumer_channel *consumer_allocate_channel(unsigned long key,
+		uint64_t session_id,
+		const char *pathname,
+		const char *name,
+		uid_t uid,
+		gid_t gid,
+		int relayd_id,
+		enum lttng_event_output output)
 {
 	struct lttng_consumer_channel *channel;
-	int ret;
 
 	channel = zmalloc(sizeof(*channel));
 	if (channel == NULL) {
 		PERROR("malloc struct lttng_consumer_channel");
 		goto end;
 	}
-	channel->key = channel_key;
-	channel->shm_fd = shm_fd;
-	channel->wait_fd = wait_fd;
-	channel->mmap_len = mmap_len;
-	channel->max_sb_size = max_sb_size;
+
+	channel->key = key;
 	channel->refcount = 0;
-	channel->nb_init_streams = nb_init_streams;
-	lttng_ht_node_init_ulong(&channel->node, channel->key);
+	channel->session_id = session_id;
+	channel->uid = uid;
+	channel->gid = gid;
+	channel->relayd_id = relayd_id;
+	channel->output = output;
+
+	strncpy(channel->pathname, pathname, sizeof(channel->pathname));
+	channel->pathname[sizeof(channel->pathname) - 1] = '\0';
+
+	strncpy(channel->name, name, sizeof(channel->name));
+	channel->name[sizeof(channel->name) - 1] = '\0';
+
+	lttng_ht_node_init_u64(&channel->node, channel->key);
+
+	channel->wait_fd = -1;
+
+	CDS_INIT_LIST_HEAD(&channel->streams.head);
+
+	DBG("Allocated channel (key %" PRIu64 ")", channel->key)
 
-	switch (consumer_data.type) {
-	case LTTNG_CONSUMER_KERNEL:
-		channel->mmap_base = NULL;
-		channel->mmap_len = 0;
-		break;
-	case LTTNG_CONSUMER32_UST:
-	case LTTNG_CONSUMER64_UST:
-		ret = lttng_ustconsumer_allocate_channel(channel);
-		if (ret) {
-			free(channel);
-			return NULL;
-		}
-		break;
-	default:
-		ERR("Unknown consumer_data type");
-		assert(0);
-		goto end;
-	}
-	DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
-			channel->key, channel->shm_fd, channel->wait_fd,
-			(unsigned long long) channel->mmap_len,
-			(unsigned long long) channel->max_sb_size);
 end:
 	return channel;
 }
@@ -919,31 +866,38 @@ end:
 /*
  * Add a channel to the global list protected by a mutex.
  */
-int consumer_add_channel(struct lttng_consumer_channel *channel)
+int consumer_add_channel(struct lttng_consumer_channel *channel,
+		struct lttng_consumer_local_data *ctx)
 {
-	struct lttng_ht_node_ulong *node;
+	int ret = 0;
+	struct lttng_ht_node_u64 *node;
 	struct lttng_ht_iter iter;
 
 	pthread_mutex_lock(&consumer_data.lock);
-	/* Steal channel identifier, for UST */
-	consumer_steal_channel_key(channel->key);
 	rcu_read_lock();
 
 	lttng_ht_lookup(consumer_data.channel_ht,
-			(void *)((unsigned long) channel->key), &iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
+			&channel->key, &iter);
+	node = lttng_ht_iter_get_node_u64(&iter);
 	if (node != NULL) {
 		/* Channel already exist. Ignore the insertion */
+		ERR("Consumer add channel key %" PRIu64 " already exists!",
+			channel->key);
+		ret = -1;
 		goto end;
 	}
 
-	lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
+	lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
 
 end:
 	rcu_read_unlock();
 	pthread_mutex_unlock(&consumer_data.lock);
 
-	return 0;
+	if (!ret && channel->wait_fd != -1 &&
+			channel->metadata_stream == NULL) {
+		notify_channel_pipe(ctx, channel, CONSUMER_CHANNEL_ADD);
+	}
+	return ret;
 }
 
 /*
@@ -953,14 +907,19 @@ end:
  *
  * Returns the number of fds in the structures.
  */
-static int consumer_update_poll_array(
-		struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
-		struct lttng_consumer_stream **local_stream, struct lttng_ht *ht)
+static int update_poll_array(struct lttng_consumer_local_data *ctx,
+		struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
+		struct lttng_ht *ht)
 {
 	int i = 0;
 	struct lttng_ht_iter iter;
 	struct lttng_consumer_stream *stream;
 
+	assert(ctx);
+	assert(ht);
+	assert(pollfd);
+	assert(local_stream);
+
 	DBG("Updating poll fd array");
 	rcu_read_lock();
 	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
@@ -1027,8 +986,8 @@ exit:
 /*
  * Set the error socket.
  */
-void lttng_consumer_set_error_sock(
-		struct lttng_consumer_local_data *ctx, int sock)
+void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
+		int sock)
 {
 	ctx->consumer_error_socket = sock;
 }
@@ -1046,8 +1005,7 @@ void lttng_consumer_set_command_sock_path(
  * Send return code to the session daemon.
  * If the socket is not defined, we return 0, it is not a fatal error
  */
-int lttng_consumer_send_error(
-		struct lttng_consumer_local_data *ctx, int cmd)
+int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
 {
 	if (ctx->consumer_error_socket > 0) {
 		return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
@@ -1064,14 +1022,12 @@ int lttng_consumer_send_error(
 void lttng_consumer_cleanup(void)
 {
 	struct lttng_ht_iter iter;
-	struct lttng_ht_node_ulong *node;
+	struct lttng_consumer_channel *channel;
 
 	rcu_read_lock();
 
-	cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
-			node) {
-		struct lttng_consumer_channel *channel =
-			caa_container_of(node, struct lttng_consumer_channel, node);
+	cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
+			node.node) {
 		consumer_del_channel(channel);
 	}
 
@@ -1081,6 +1037,8 @@ void lttng_consumer_cleanup(void)
 
 	cleanup_relayd_ht();
 
+	lttng_ht_destroy(consumer_data.stream_per_chan_id_ht);
+
 	/*
 	 * This HT contains streams that are freed by either the metadata thread or
 	 * the data thread so we do *nothing* on the hash table and simply destroy
@@ -1099,7 +1057,7 @@ void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
 	do {
 		ret = write(ctx->consumer_should_quit[1], "4", 1);
 	} while (ret < 0 && errno == EINTR);
-	if (ret < 0) {
+	if (ret < 0 || ret != 1) {
 		PERROR("write consumer quit");
 	}
 
@@ -1117,11 +1075,11 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
 	 * Don't care about error values, as these are just hints and ways to
 	 * limit the amount of page cache used.
 	 */
-	if (orig_offset < stream->chan->max_sb_size) {
+	if (orig_offset < stream->max_sb_size) {
 		return;
 	}
-	lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
-			stream->chan->max_sb_size,
+	lttng_sync_file_range(outfd, orig_offset - stream->max_sb_size,
+			stream->max_sb_size,
 			SYNC_FILE_RANGE_WAIT_BEFORE
 			| SYNC_FILE_RANGE_WRITE
 			| SYNC_FILE_RANGE_WAIT_AFTER);
@@ -1139,8 +1097,8 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
 	 * defined. So it can be expected to lead to lower throughput in
 	 * streaming.
 	 */
-	posix_fadvise(outfd, orig_offset - stream->chan->max_sb_size,
-			stream->chan->max_sb_size, POSIX_FADV_DONTNEED);
+	posix_fadvise(outfd, orig_offset - stream->max_sb_size,
+			stream->max_sb_size, POSIX_FADV_DONTNEED);
 }
 
 /*
@@ -1165,7 +1123,7 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 		int (*recv_stream)(struct lttng_consumer_stream *stream),
 		int (*update_stream)(int stream_key, uint32_t state))
 {
-	int ret, i;
+	int ret;
 	struct lttng_consumer_local_data *ctx;
 
 	assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
@@ -1217,6 +1175,12 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 		goto error_thread_pipe;
 	}
 
+	ret = pipe(ctx->consumer_channel_pipe);
+	if (ret < 0) {
+		PERROR("Error creating channel pipe");
+		goto error_channel_pipe;
+	}
+
 	ret = utils_create_pipe(ctx->consumer_metadata_pipe);
 	if (ret < 0) {
 		goto error_metadata_pipe;
@@ -1232,26 +1196,14 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 error_splice_pipe:
 	utils_close_pipe(ctx->consumer_metadata_pipe);
 error_metadata_pipe:
+	utils_close_pipe(ctx->consumer_channel_pipe);
+error_channel_pipe:
 	utils_close_pipe(ctx->consumer_thread_pipe);
 error_thread_pipe:
-	for (i = 0; i < 2; i++) {
-		int err;
-
-		err = close(ctx->consumer_should_quit[i]);
-		if (err) {
-			PERROR("close");
-		}
-	}
+	utils_close_pipe(ctx->consumer_should_quit);
 error_poll_fcntl:
 error_quit_pipe:
-	for (i = 0; i < 2; i++) {
-		int err;
-
-		err = close(ctx->consumer_data_pipe[i]);
-		if (err) {
-			PERROR("close");
-		}
-	}
+	utils_close_pipe(ctx->consumer_data_pipe);
 error_poll_pipe:
 	free(ctx);
 error:
@@ -1271,30 +1223,10 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
 	if (ret) {
 		PERROR("close");
 	}
-	ret = close(ctx->consumer_thread_pipe[0]);
-	if (ret) {
-		PERROR("close");
-	}
-	ret = close(ctx->consumer_thread_pipe[1]);
-	if (ret) {
-		PERROR("close");
-	}
-	ret = close(ctx->consumer_data_pipe[0]);
-	if (ret) {
-		PERROR("close");
-	}
-	ret = close(ctx->consumer_data_pipe[1]);
-	if (ret) {
-		PERROR("close");
-	}
-	ret = close(ctx->consumer_should_quit[0]);
-	if (ret) {
-		PERROR("close");
-	}
-	ret = close(ctx->consumer_should_quit[1]);
-	if (ret) {
-		PERROR("close");
-	}
+	utils_close_pipe(ctx->consumer_thread_pipe);
+	utils_close_pipe(ctx->consumer_channel_pipe);
+	utils_close_pipe(ctx->consumer_data_pipe);
+	utils_close_pipe(ctx->consumer_should_quit);
 	utils_close_pipe(ctx->consumer_splice_metadata_pipe);
 
 	unlink(ctx->consumer_command_sock_path);
@@ -1306,8 +1238,7 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
  */
 static int write_relayd_metadata_id(int fd,
 		struct lttng_consumer_stream *stream,
-		struct consumer_relayd_sock_pair *relayd,
-		unsigned long padding)
+		struct consumer_relayd_sock_pair *relayd, unsigned long padding)
 {
 	int ret;
 	struct lttcomm_relayd_metadata_payload hdr;
@@ -1317,8 +1248,22 @@ static int write_relayd_metadata_id(int fd,
 	do {
 		ret = write(fd, (void *) &hdr, sizeof(hdr));
 	} while (ret < 0 && errno == EINTR);
-	if (ret < 0) {
-		PERROR("write metadata stream id");
+	if (ret < 0 || ret != sizeof(hdr)) {
+		/*
+		 * This error means that the fd's end is closed so ignore the perror
+		 * not to clubber the error output since this can happen in a normal
+		 * code path.
+		 */
+		if (errno != EPIPE) {
+			PERROR("write metadata stream id");
+		}
+		DBG3("Consumer failed to write relayd metadata id (errno: %d)", errno);
+		/*
+		 * Set ret to a negative value because if ret != sizeof(hdr), we don't
+		 * handle writting the missing part so report that as an error and
+		 * don't lie to the caller.
+		 */
+		ret = -1;
 		goto end;
 	}
 	DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
@@ -1345,6 +1290,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		unsigned long padding)
 {
 	unsigned long mmap_offset;
+	void *mmap_base;
 	ssize_t ret = 0, written = 0;
 	off_t orig_offset = stream->out_fd_offset;
 	/* Default is on the disk */
@@ -1366,12 +1312,18 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 	/* get the offset inside the fd to mmap */
 	switch (consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
+		mmap_base = stream->mmap_base;
 		ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
 		break;
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
-		ret = lttng_ustctl_get_mmap_read_offset(stream->chan->handle,
-				stream->buf, &mmap_offset);
+		mmap_base = lttng_ustctl_get_mmap_base(stream);
+		if (!mmap_base) {
+			ERR("read mmap get mmap base for stream %s", stream->name);
+			written = -1;
+			goto end;
+		}
+		ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
 		break;
 	default:
 		ERR("Unknown consumer_data type");
@@ -1431,11 +1383,17 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 
 	while (len > 0) {
 		do {
-			ret = write(outfd, stream->mmap_base + mmap_offset, len);
+			ret = write(outfd, mmap_base + mmap_offset, len);
 		} while (ret < 0 && errno == EINTR);
 		DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
 		if (ret < 0) {
-			PERROR("Error in file write");
+			/*
+			 * This is possible if the fd is closed on the other side (outfd)
+			 * or any write problem. It can be verbose a bit for a normal
+			 * execution if for instance the relayd is stopped abruptly. This
+			 * can happen so set this to a DBG statement.
+			 */
+			DBG("Error in file write mmap");
 			if (written == 0) {
 				written = ret;
 			}
@@ -1698,21 +1656,19 @@ end:
  *
  * Returns 0 on success, < 0 on error
  */
-int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *stream)
+int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
 {
 	switch (consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
-		return lttng_kconsumer_take_snapshot(ctx, stream);
+		return lttng_kconsumer_take_snapshot(stream);
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
-		return lttng_ustconsumer_take_snapshot(ctx, stream);
+		return lttng_ustconsumer_take_snapshot(stream);
 	default:
 		ERR("Unknown consumer_data type");
 		assert(0);
 		return -ENOSYS;
 	}
-
 }
 
 /*
@@ -1720,17 +1676,15 @@ int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
  *
  * Returns 0 on success, < 0 on error
  */
-int lttng_consumer_get_produced_snapshot(
-		struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *stream,
+int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
 		unsigned long *pos)
 {
 	switch (consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
-		return lttng_kconsumer_get_produced_snapshot(ctx, stream, pos);
+		return lttng_kconsumer_get_produced_snapshot(stream, pos);
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
-		return lttng_ustconsumer_get_produced_snapshot(ctx, stream, pos);
+		return lttng_ustconsumer_get_produced_snapshot(stream, pos);
 	default:
 		ERR("Unknown consumer_data type");
 		assert(0);
@@ -1808,6 +1762,32 @@ static void destroy_stream_ht(struct lttng_ht *ht)
 	lttng_ht_destroy(ht);
 }
 
+void lttng_consumer_close_metadata(void)
+{
+	switch (consumer_data.type) {
+	case LTTNG_CONSUMER_KERNEL:
+		/*
+		 * The Kernel consumer has a different metadata scheme so we don't
+		 * close anything because the stream will be closed by the session
+		 * daemon.
+		 */
+		break;
+	case LTTNG_CONSUMER32_UST:
+	case LTTNG_CONSUMER64_UST:
+		/*
+		 * Close all metadata streams. The metadata hash table is passed and
+		 * this call iterates over it by closing all wakeup fd. This is safe
+		 * because at this point we are sure that the metadata producer is
+		 * either dead or blocked.
+		 */
+		lttng_ustconsumer_close_metadata(metadata_ht);
+		break;
+	default:
+		ERR("Unknown consumer_data type");
+		assert(0);
+	}
+}
+
 /*
  * Clean up a metadata stream and free its memory.
  */
@@ -1830,7 +1810,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 
 	if (ht == NULL) {
 		/* Means the stream was allocated but not successfully added */
-		goto free_stream;
+		goto free_stream_rcu;
 	}
 
 	pthread_mutex_lock(&consumer_data.lock);
@@ -1860,7 +1840,10 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	ret = lttng_ht_del(ht, &iter);
 	assert(!ret);
 
-	/* Remove node session id from the consumer_data stream ht */
+	iter.iter.node = &stream->node_channel_id.node;
+	ret = lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
+	assert(!ret);
+
 	iter.iter.node = &stream->node_session_id.node;
 	ret = lttng_ht_del(consumer_data.stream_list_ht, &iter);
 	assert(!ret);
@@ -1873,20 +1856,6 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 		}
 	}
 
-	if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
-		ret = close(stream->wait_fd);
-		if (ret) {
-			PERROR("close");
-		}
-	}
-
-	if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
-		ret = close(stream->shm_fd);
-		if (ret) {
-			PERROR("close");
-		}
-	}
-
 	/* Check and cleanup relayd */
 	rcu_read_lock();
 	relayd = consumer_find_relayd(stream->net_seq_idx);
@@ -1919,7 +1888,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	/* Atomically decrement channel refcount since other threads can use it. */
 	uatomic_dec(&stream->chan->refcount);
 	if (!uatomic_read(&stream->chan->refcount)
-			&& !uatomic_read(&stream->chan->nb_init_streams)) {
+			&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
 		/* Go for channel deletion! */
 		free_chan = stream->chan;
 	}
@@ -1932,26 +1901,26 @@ end:
 		consumer_del_channel(free_chan);
 	}
 
-free_stream:
-	call_rcu(&stream->node.head, consumer_free_stream);
+free_stream_rcu:
+	call_rcu(&stream->node.head, free_stream_rcu);
 }
 
 /*
  * Action done with the metadata stream when adding it to the consumer internal
  * data structures to handle it.
  */
-static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
+static int add_metadata_stream(struct lttng_consumer_stream *stream,
 		struct lttng_ht *ht)
 {
 	int ret = 0;
 	struct consumer_relayd_sock_pair *relayd;
 	struct lttng_ht_iter iter;
-	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_node_u64 *node;
 
 	assert(stream);
 	assert(ht);
 
-	DBG3("Adding metadata stream %d to hash table", stream->wait_fd);
+	DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
 
 	pthread_mutex_lock(&consumer_data.lock);
 	pthread_mutex_lock(&stream->lock);
@@ -1967,8 +1936,8 @@ static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
 	 * Lookup the stream just to make sure it does not exist in our internal
 	 * state. This should NEVER happen.
 	 */
-	lttng_ht_lookup(ht, (void *)((unsigned long) stream->wait_fd), &iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
+	lttng_ht_lookup(ht, &stream->key, &iter);
+	node = lttng_ht_iter_get_node_u64(&iter);
 	assert(!node);
 
 	/* Find relayd and, if one is found, increment refcount. */
@@ -1981,24 +1950,27 @@ static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
 	uatomic_inc(&stream->chan->refcount);
 
 	/*
-	 * When nb_init_streams reaches 0, we don't need to trigger any action in
-	 * terms of destroying the associated channel, because the action that
+	 * When nb_init_stream_left reaches 0, we don't need to trigger any action
+	 * in terms of destroying the associated channel, because the action that
 	 * causes the count to become 0 also causes a stream to be added. The
 	 * channel deletion will thus be triggered by the following removal of this
 	 * stream.
 	 */
-	if (uatomic_read(&stream->chan->nb_init_streams) > 0) {
-		uatomic_dec(&stream->chan->nb_init_streams);
+	if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+		uatomic_dec(&stream->chan->nb_init_stream_left);
 	}
 
-	lttng_ht_add_unique_ulong(ht, &stream->node);
+	lttng_ht_add_unique_u64(ht, &stream->node);
+
+	lttng_ht_add_unique_u64(consumer_data.stream_per_chan_id_ht,
+		&stream->node_channel_id);
 
 	/*
 	 * Add stream to the stream_list_ht of the consumer data. No need to steal
 	 * the key since the HT does not use it and we allow to add redundant keys
 	 * into this table.
 	 */
-	lttng_ht_add_ulong(consumer_data.stream_list_ht, &stream->node_session_id);
+	lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
 
 	rcu_read_unlock();
 
@@ -2070,17 +2042,17 @@ void *consumer_thread_metadata_poll(void *data)
 	uint32_t revents, nb_fd;
 	struct lttng_consumer_stream *stream = NULL;
 	struct lttng_ht_iter iter;
-	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_node_u64 *node;
 	struct lttng_poll_event events;
 	struct lttng_consumer_local_data *ctx = data;
 	ssize_t len;
 
 	rcu_register_thread();
 
-	metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
+	metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
 	if (!metadata_ht) {
 		/* ENOMEM at this point. Better to bail out. */
-		goto error;
+		goto end_ht;
 	}
 
 	DBG("Thread metadata poll started");
@@ -2089,7 +2061,7 @@ void *consumer_thread_metadata_poll(void *data)
 	ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
 	if (ret < 0) {
 		ERR("Poll set creation failed");
-		goto end;
+		goto end_poll;
 	}
 
 	ret = lttng_poll_add(&events, ctx->consumer_metadata_pipe[0], LPOLLIN);
@@ -2162,13 +2134,13 @@ restart:
 					if (stream == NULL) {
 						/* Check for deleted streams. */
 						validate_endpoint_status_metadata_stream(&events);
-						continue;
+						goto restart;
 					}
 
 					DBG("Adding metadata stream %d to poll set",
 							stream->wait_fd);
 
-					ret = consumer_add_metadata_stream(stream, metadata_ht);
+					ret = add_metadata_stream(stream, metadata_ht);
 					if (ret) {
 						ERR("Unable to add metadata stream");
 						/* Stream was not setup properly. Continuing. */
@@ -2186,9 +2158,12 @@ restart:
 			}
 
 			rcu_read_lock();
-			lttng_ht_lookup(metadata_ht, (void *)((unsigned long) pollfd),
-					&iter);
-			node = lttng_ht_iter_get_node_ulong(&iter);
+			{
+				uint64_t tmp_id = (uint64_t) pollfd;
+
+				lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
+			}
+			node = lttng_ht_iter_get_node_u64(&iter);
 			assert(node);
 
 			stream = caa_container_of(node, struct lttng_consumer_stream,
@@ -2245,10 +2220,11 @@ restart:
 error:
 end:
 	DBG("Metadata poll thread exiting");
-	lttng_poll_clean(&events);
 
+	lttng_poll_clean(&events);
+end_poll:
 	destroy_stream_ht(metadata_ht);
-
+end_ht:
 	rcu_unregister_thread();
 	return NULL;
 }
@@ -2270,7 +2246,7 @@ void *consumer_thread_data_poll(void *data)
 
 	rcu_register_thread();
 
-	data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
+	data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
 	if (data_ht == NULL) {
 		/* ENOMEM at this point. Better to bail out. */
 		goto end;
@@ -2288,14 +2264,11 @@ void *consumer_thread_data_poll(void *data)
 		 */
 		pthread_mutex_lock(&consumer_data.lock);
 		if (consumer_data.need_update) {
-			if (pollfd != NULL) {
-				free(pollfd);
-				pollfd = NULL;
-			}
-			if (local_stream != NULL) {
-				free(local_stream);
-				local_stream = NULL;
-			}
+			free(pollfd);
+			pollfd = NULL;
+
+			free(local_stream);
+			local_stream = NULL;
 
 			/* allocate for all fds + 1 for the consumer_data_pipe */
 			pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
@@ -2313,7 +2286,7 @@ void *consumer_thread_data_poll(void *data)
 				pthread_mutex_unlock(&consumer_data.lock);
 				goto end;
 			}
-			ret = consumer_update_poll_array(ctx, &pollfd, local_stream,
+			ret = update_poll_array(ctx, &pollfd, local_stream,
 					data_ht);
 			if (ret < 0) {
 				ERR("Error in allocating pollfd or local_outfds");
@@ -2356,7 +2329,7 @@ void *consumer_thread_data_poll(void *data)
 		 * array update over low-priority reads.
 		 */
 		if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
-			size_t pipe_readlen;
+			ssize_t pipe_readlen;
 
 			DBG("consumer_data_pipe wake up");
 			/* Consume 1 byte of pipe data */
@@ -2364,6 +2337,11 @@ void *consumer_thread_data_poll(void *data)
 				pipe_readlen = read(ctx->consumer_data_pipe[0], &new_stream,
 						sizeof(new_stream));
 			} while (pipe_readlen == -1 && errno == EINTR);
+			if (pipe_readlen < 0) {
+				PERROR("read consumer data pipe");
+				/* Continue so we can at least handle the current stream(s). */
+				continue;
+			}
 
 			/*
 			 * If the stream is NULL, just ignore it. It's also possible that
@@ -2375,9 +2353,9 @@ void *consumer_thread_data_poll(void *data)
 				continue;
 			}
 
-			ret = consumer_add_stream(new_stream, data_ht);
+			ret = add_stream(new_stream, data_ht);
 			if (ret) {
-				ERR("Consumer add stream %d failed. Continuing",
+				ERR("Consumer add stream %" PRIu64 " failed. Continuing",
 						new_stream->key);
 				/*
 				 * At this point, if the add_stream fails, it is not in the
@@ -2487,14 +2465,8 @@ void *consumer_thread_data_poll(void *data)
 	}
 end:
 	DBG("polling thread exiting");
-	if (pollfd != NULL) {
-		free(pollfd);
-		pollfd = NULL;
-	}
-	if (local_stream != NULL) {
-		free(local_stream);
-		local_stream = NULL;
-	}
+	free(pollfd);
+	free(local_stream);
 
 	/*
 	 * Close the write side of the pipe so epoll_wait() in
@@ -2515,6 +2487,222 @@ end:
 	return NULL;
 }
 
+/*
+ * Close wake-up end of each stream belonging to the channel. This will
+ * allow the poll() on the stream read-side to detect when the
+ * write-side (application) finally closes them.
+ */
+static
+void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
+{
+	struct lttng_ht *ht;
+	struct lttng_consumer_stream *stream;
+	struct lttng_ht_iter iter;
+
+	ht = consumer_data.stream_per_chan_id_ht;
+
+	rcu_read_lock();
+	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) {
+		switch (consumer_data.type) {
+		case LTTNG_CONSUMER_KERNEL:
+			break;
+		case LTTNG_CONSUMER32_UST:
+		case LTTNG_CONSUMER64_UST:
+			/*
+			 * Note: a mutex is taken internally within
+			 * liblttng-ust-ctl to protect timer wakeup_fd
+			 * use from concurrent close.
+			 */
+			lttng_ustconsumer_close_stream_wakeup(stream);
+			break;
+		default:
+			ERR("Unknown consumer_data type");
+			assert(0);
+		}
+	}
+	rcu_read_unlock();
+}
+
+static void destroy_channel_ht(struct lttng_ht *ht)
+{
+	struct lttng_ht_iter iter;
+	struct lttng_consumer_channel *channel;
+	int ret;
+
+	if (ht == NULL) {
+		return;
+	}
+
+	rcu_read_lock();
+	cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
+		ret = lttng_ht_del(ht, &iter);
+		assert(ret != 0);
+	}
+	rcu_read_unlock();
+
+	lttng_ht_destroy(ht);
+}
+
+/*
+ * This thread polls the channel fds to detect when they are being
+ * closed. It closes all related streams if the channel is detected as
+ * closed. It is currently only used as a shim layer for UST because the
+ * consumerd needs to keep the per-stream wakeup end of pipes open for
+ * periodical flush.
+ */
+void *consumer_thread_channel_poll(void *data)
+{
+	int ret, i, pollfd;
+	uint32_t revents, nb_fd;
+	struct lttng_consumer_channel *chan = NULL;
+	struct lttng_ht_iter iter;
+	struct lttng_ht_node_u64 *node;
+	struct lttng_poll_event events;
+	struct lttng_consumer_local_data *ctx = data;
+	struct lttng_ht *channel_ht;
+
+	rcu_register_thread();
+
+	channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!channel_ht) {
+		/* ENOMEM at this point. Better to bail out. */
+		goto end_ht;
+	}
+
+	DBG("Thread channel poll started");
+
+	/* Size is set to 1 for the consumer_channel pipe */
+	ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
+	if (ret < 0) {
+		ERR("Poll set creation failed");
+		goto end_poll;
+	}
+
+	ret = lttng_poll_add(&events, ctx->consumer_channel_pipe[0], LPOLLIN);
+	if (ret < 0) {
+		goto end;
+	}
+
+	/* Main loop */
+	DBG("Channel main loop started");
+
+	while (1) {
+		/* Only the channel pipe is set */
+		if (LTTNG_POLL_GETNB(&events) == 0 && consumer_quit == 1) {
+			goto end;
+		}
+
+restart:
+		DBG("Channel poll wait with %d fd(s)", LTTNG_POLL_GETNB(&events));
+		ret = lttng_poll_wait(&events, -1);
+		DBG("Channel event catched in thread");
+		if (ret < 0) {
+			if (errno == EINTR) {
+				ERR("Poll EINTR catched");
+				goto restart;
+			}
+			goto end;
+		}
+
+		nb_fd = ret;
+
+		/* From here, the event is a channel wait fd */
+		for (i = 0; i < nb_fd; i++) {
+			revents = LTTNG_POLL_GETEV(&events, i);
+			pollfd = LTTNG_POLL_GETFD(&events, i);
+
+			/* Just don't waste time if no returned events for the fd */
+			if (!revents) {
+				continue;
+			}
+			if (pollfd == ctx->consumer_channel_pipe[0]) {
+				if (revents & (LPOLLERR | LPOLLHUP)) {
+					DBG("Channel thread pipe hung up");
+					/*
+					 * Remove the pipe from the poll set and continue the loop
+					 * since their might be data to consume.
+					 */
+					lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
+					continue;
+				} else if (revents & LPOLLIN) {
+					enum consumer_channel_action action;
+
+					ret = read_channel_pipe(ctx, &chan, &action);
+					if (ret <= 0) {
+						ERR("Error reading channel pipe");
+						continue;
+					}
+
+					switch (action) {
+					case CONSUMER_CHANNEL_ADD:
+						DBG("Adding channel %d to poll set",
+							chan->wait_fd);
+
+						lttng_ht_node_init_u64(&chan->wait_fd_node,
+							chan->wait_fd);
+						lttng_ht_add_unique_u64(channel_ht,
+								&chan->wait_fd_node);
+						/* Add channel to the global poll events list */
+						lttng_poll_add(&events, chan->wait_fd,
+								LPOLLIN | LPOLLPRI);
+						break;
+					case CONSUMER_CHANNEL_QUIT:
+						/*
+						 * Remove the pipe from the poll set and continue the loop
+						 * since their might be data to consume.
+						 */
+						lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
+						continue;
+					default:
+						ERR("Unknown action");
+						break;
+					}
+				}
+
+				/* Handle other stream */
+				continue;
+			}
+
+			rcu_read_lock();
+			{
+				uint64_t tmp_id = (uint64_t) pollfd;
+
+				lttng_ht_lookup(channel_ht, &tmp_id, &iter);
+			}
+			node = lttng_ht_iter_get_node_u64(&iter);
+			assert(node);
+
+			chan = caa_container_of(node, struct lttng_consumer_channel,
+					wait_fd_node);
+
+			/* Check for error event */
+			if (revents & (LPOLLERR | LPOLLHUP)) {
+				DBG("Channel fd %d is hup|err.", pollfd);
+
+				lttng_poll_del(&events, chan->wait_fd);
+				ret = lttng_ht_del(channel_ht, &iter);
+				assert(ret == 0);
+				consumer_close_channel_streams(chan);
+			}
+
+			/* Release RCU lock for the channel looked up */
+			rcu_read_unlock();
+		}
+	}
+
+end:
+	lttng_poll_clean(&events);
+end_poll:
+	destroy_channel_ht(channel_ht);
+end_ht:
+	DBG("Channel poll thread exiting");
+	rcu_unregister_thread();
+	return NULL;
+}
+
 /*
  * This thread listens on the consumerd socket and receives the file
  * descriptors from the session daemon.
@@ -2571,7 +2759,7 @@ void *consumer_thread_sessiond_poll(void *data)
 
 	/* Blocking call, waiting for transmission */
 	sock = lttcomm_accept_unix_sock(client_socket);
-	if (sock <= 0) {
+	if (sock < 0) {
 		WARN("On accept");
 		goto end;
 	}
@@ -2614,10 +2802,18 @@ void *consumer_thread_sessiond_poll(void *data)
 			DBG("consumer_thread_receive_fds received quit from signal");
 			goto end;
 		}
-		DBG("received fds on sock");
+		DBG("received command on sock");
 	}
 end:
-	DBG("consumer_thread_receive_fds exiting");
+	DBG("Consumer thread sessiond poll exiting");
+
+	/*
+	 * Close metadata streams since the producer is the session daemon which
+	 * just died.
+	 *
+	 * NOTE: for now, this only applies to the UST tracer.
+	 */
+	lttng_consumer_close_metadata();
 
 	/*
 	 * when all fds have hung up, the polling thread
@@ -2631,6 +2827,8 @@ end:
 	 */
 	notify_thread_pipe(ctx->consumer_data_pipe[1]);
 
+	notify_channel_pipe(ctx, NULL, CONSUMER_CHANNEL_QUIT);
+
 	/* Cleaning up possibly open sockets. */
 	if (sock >= 0) {
 		ret = close(sock);
@@ -2695,10 +2893,10 @@ int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
  */
 void lttng_consumer_init(void)
 {
-	consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
-	consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
-	consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
-	relayd_session_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
+	consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	consumer_data.stream_per_chan_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
 }
 
 /*
@@ -2715,7 +2913,6 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 	int fd = -1, ret = -1, relayd_created = 0;
 	enum lttng_error_code ret_code = LTTNG_OK;
 	struct consumer_relayd_sock_pair *relayd;
-	struct consumer_relayd_session_id *relayd_id_node;
 
 	DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx);
 
@@ -2733,6 +2930,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
 		if (relayd == NULL) {
 			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
+			ret = -1;
 			goto error;
 		}
 		relayd->sessiond_session_id = (uint64_t) sessiond_id;
@@ -2751,7 +2949,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 		ret = -1;
 		fd = -1;	/* Just in case it gets set with an invalid value. */
-		goto error;
+		goto error_close;
 	}
 
 	/* We have the fds without error. Send status back. */
@@ -2782,32 +2980,28 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		relayd->control_sock.fd = fd;
 
 		/*
-		 * Create a session on the relayd and store the returned id. No need to
-		 * grab the socket lock since the relayd object is not yet visible.
+		 * Create a session on the relayd and store the returned id. Lock the
+		 * control socket mutex if the relayd was NOT created before.
 		 */
+		if (!relayd_created) {
+			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+		}
 		ret = relayd_create_session(&relayd->control_sock,
 				&relayd->relayd_session_id);
-		if (ret < 0) {
-			goto error;
+		if (!relayd_created) {
+			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 		}
-
-		/* Set up a relayd session id node. */
-		relayd_id_node = zmalloc(sizeof(struct consumer_relayd_session_id));
-		if (!relayd_id_node) {
-			PERROR("zmalloc relayd id node");
+		if (ret < 0) {
+			/*
+			 * Close all sockets of a relayd object. It will be freed if it was
+			 * created at the error code path or else it will be garbage
+			 * collect.
+			 */
+			(void) relayd_close(&relayd->control_sock);
+			(void) relayd_close(&relayd->data_sock);
 			goto error;
 		}
 
-		relayd_id_node->relayd_id = relayd->relayd_session_id;
-		relayd_id_node->sessiond_id = (uint64_t) sessiond_id;
-
-		/* Indexed by session id of the sessiond. */
-		lttng_ht_node_init_ulong(&relayd_id_node->node,
-				relayd_id_node->sessiond_id);
-		rcu_read_lock();
-		lttng_ht_add_unique_ulong(relayd_session_id_ht, &relayd_id_node->node);
-		rcu_read_unlock();
-
 		break;
 	case LTTNG_STREAM_DATA:
 		/* Copy received lttcomm socket */
@@ -2829,10 +3023,11 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		break;
 	default:
 		ERR("Unknown relayd socket type (%d)", sock_type);
+		ret = -1;
 		goto error;
 	}
 
-	DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
+	DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
 			sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
 			relayd->net_seq_idx, fd);
 
@@ -2853,10 +3048,8 @@ error:
 		}
 	}
 
+error_close:
 	if (relayd_created) {
-		/* We just want to cleanup. Ignore ret value. */
-		(void) relayd_close(&relayd->control_sock);
-		(void) relayd_close(&relayd->data_sock);
 		free(relayd);
 	}
 
@@ -2900,30 +3093,24 @@ end:
 static struct consumer_relayd_sock_pair *find_relayd_by_session_id(uint64_t id)
 {
 	struct lttng_ht_iter iter;
-	struct lttng_ht_node_ulong *node;
 	struct consumer_relayd_sock_pair *relayd = NULL;
-	struct consumer_relayd_session_id *session_id_map;
-
-	/* Get the session id map. */
-	lttng_ht_lookup(relayd_session_id_ht, (void *)((unsigned long) id), &iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
-	if (node == NULL) {
-		goto end;
-	}
-
-	session_id_map = caa_container_of(node, struct consumer_relayd_session_id,
-			node);
 
 	/* Iterate over all relayd since they are indexed by net_seq_idx. */
 	cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
 			node.node) {
-		if (relayd->relayd_session_id == session_id_map->relayd_id) {
+		/*
+		 * Check by sessiond id which is unique here where the relayd session
+		 * id might not be when having multiple relayd.
+		 */
+		if (relayd->sessiond_session_id == id) {
 			/* Found the relayd. There can be only one per id. */
-			break;
+			goto found;
 		}
 	}
 
-end:
+	return NULL;
+
+found:
 	return relayd;
 }
 
@@ -2977,8 +3164,8 @@ int consumer_data_pending(uint64_t id)
 	}
 
 	cds_lfht_for_each_entry_duplicate(ht->ht,
-			ht->hash_fct((void *)((unsigned long) id), lttng_ht_seed),
-			ht->match_fct, (void *)((unsigned long) id),
+			ht->hash_fct(&id, lttng_ht_seed),
+			ht->match_fct, &id,
 			&iter.iter, stream, node_session_id.node) {
 		/* If this call fails, the stream is being used hence data pending. */
 		ret = stream_try_lock(stream);
@@ -3006,10 +3193,12 @@ int consumer_data_pending(uint64_t id)
 		if (relayd) {
 			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 			if (stream->metadata_flag) {
-				ret = relayd_quiescent_control(&relayd->control_sock);
+				ret = relayd_quiescent_control(&relayd->control_sock,
+						stream->relayd_stream_id);
 			} else {
 				ret = relayd_data_pending(&relayd->control_sock,
-						stream->relayd_stream_id, stream->next_net_seq_num);
+						stream->relayd_stream_id,
+						stream->next_net_seq_num - 1);
 			}
 			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 			if (ret == 1) {
@@ -3068,3 +3257,26 @@ int consumer_send_status_msg(int sock, int ret_code)
 
 	return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
 }
+
+/*
+ * Send a channel status message to the sessiond daemon.
+ *
+ * Return the sendmsg() return value.
+ */
+int consumer_send_status_channel(int sock,
+		struct lttng_consumer_channel *channel)
+{
+	struct lttcomm_consumer_status_channel msg;
+
+	assert(sock >= 0);
+
+	if (!channel) {
+		msg.ret_code = -LTTNG_ERR_UST_CHAN_FAIL;
+	} else {
+		msg.ret_code = LTTNG_OK;
+		msg.key = channel->key;
+		msg.stream_count = channel->streams.count;
+	}
+
+	return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
+}