X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=be78e256f489661eda948fc8b143a25b69c23a0c;hp=f093f0cf5caab19bdf0108cc940021711447b43d;hb=c869f647b0c4476645ab9ee01e362401fb8c1e42;hpb=fb3a43a9284f3300e9b66edc2f2c2d2767895423

diff --git a/src/common/consumer.c b/src/common/consumer.c
index f093f0cf5..be78e256f 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -62,20 +62,23 @@ volatile int consumer_quit = 0;
  * 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 *consumer_find_stream(int key,
+		struct lttng_ht *ht)
 {
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_ulong *node;
 	struct lttng_consumer_stream *stream = NULL;
 
+	assert(ht);
+
 	/* Negative keys are lookup failures */
-	if (key < 0)
+	if (key < 0) {
 		return NULL;
+	}
 
 	rcu_read_lock();
 
-	lttng_ht_lookup(consumer_data.stream_ht, (void *)((unsigned long) key),
-			&iter);
+	lttng_ht_lookup(ht, (void *)((unsigned long) key), &iter);
 	node = lttng_ht_iter_get_node_ulong(&iter);
 	if (node != NULL) {
 		stream = caa_container_of(node, struct lttng_consumer_stream, node);
@@ -86,12 +89,12 @@ static struct lttng_consumer_stream *consumer_find_stream(int key)
 	return stream;
 }
 
-static void consumer_steal_stream_key(int key)
+void consumer_steal_stream_key(int key, struct lttng_ht *ht)
 {
 	struct lttng_consumer_stream *stream;
 
 	rcu_read_lock();
-	stream = consumer_find_stream(key);
+	stream = consumer_find_stream(key, ht);
 	if (stream) {
 		stream->key = -1;
 		/*
@@ -111,8 +114,9 @@ static struct lttng_consumer_channel *consumer_find_channel(int key)
 	struct lttng_consumer_channel *channel = NULL;
 
 	/* Negative keys are lookup failures */
-	if (key < 0)
+	if (key < 0) {
 		return NULL;
+	}
 
 	rcu_read_lock();
 
@@ -157,6 +161,17 @@ void consumer_free_stream(struct rcu_head *head)
 	free(stream);
 }
 
+static
+void consumer_free_metadata_stream(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, waitfd_node);
+
+	free(stream);
+}
+
 /*
  * RCU protected relayd socket pair free.
  */
@@ -175,7 +190,7 @@ static void consumer_rcu_free_relayd(struct rcu_head *head)
  *
  * This function MUST be called with the consumer_data lock acquired.
  */
-void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd)
+static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 {
 	int ret;
 	struct lttng_ht_iter iter;
@@ -218,7 +233,7 @@ void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
 
 	/* Destroy the relayd if refcount is 0 */
 	if (uatomic_read(&relayd->refcount) == 0) {
-		consumer_destroy_relayd(relayd);
+		destroy_relayd(relayd);
 	}
 }
 
@@ -226,7 +241,8 @@ void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
  * Remove a stream from the global list protected by a mutex. This
  * function is also responsible for freeing its data structures.
  */
-void consumer_del_stream(struct lttng_consumer_stream *stream)
+void consumer_del_stream(struct lttng_consumer_stream *stream,
+		struct lttng_ht *ht)
 {
 	int ret;
 	struct lttng_ht_iter iter;
@@ -235,6 +251,11 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 
 	assert(stream);
 
+	if (ht == NULL) {
+		/* Means the stream was allocated but not successfully added */
+		goto free_stream;
+	}
+
 	pthread_mutex_lock(&consumer_data.lock);
 
 	switch (consumer_data.type) {
@@ -242,7 +263,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 		if (stream->mmap_base != NULL) {
 			ret = munmap(stream->mmap_base, stream->mmap_len);
 			if (ret != 0) {
-				perror("munmap");
+				PERROR("munmap");
 			}
 		}
 		break;
@@ -258,7 +279,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 
 	rcu_read_lock();
 	iter.iter.node = &stream->node.node;
-	ret = lttng_ht_del(consumer_data.stream_ht, &iter);
+	ret = lttng_ht_del(ht, &iter);
 	assert(!ret);
 
 	rcu_read_unlock();
@@ -314,23 +335,27 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 		/* Both conditions are met, we destroy the relayd. */
 		if (uatomic_read(&relayd->refcount) == 0 &&
 				uatomic_read(&relayd->destroy_flag)) {
-			consumer_destroy_relayd(relayd);
+			destroy_relayd(relayd);
 		}
 	}
 	rcu_read_unlock();
 
-	if (!--stream->chan->refcount) {
+	uatomic_dec(&stream->chan->refcount);
+	if (!uatomic_read(&stream->chan->refcount)
+			&& !uatomic_read(&stream->chan->nb_init_streams)) {
 		free_chan = stream->chan;
 	}
 
-
-	call_rcu(&stream->node.head, consumer_free_stream);
 end:
 	consumer_data.need_update = 1;
 	pthread_mutex_unlock(&consumer_data.lock);
 
-	if (free_chan)
+	if (free_chan) {
 		consumer_del_channel(free_chan);
+	}
+
+free_stream:
+	call_rcu(&stream->node.head, consumer_free_stream);
 }
 
 struct lttng_consumer_stream *consumer_allocate_stream(
@@ -343,22 +368,29 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 		uid_t uid,
 		gid_t gid,
 		int net_index,
-		int metadata_flag)
+		int metadata_flag,
+		int *alloc_ret)
 {
 	struct lttng_consumer_stream *stream;
-	int ret;
 
 	stream = zmalloc(sizeof(*stream));
 	if (stream == NULL) {
-		perror("malloc struct lttng_consumer_stream");
+		PERROR("malloc struct lttng_consumer_stream");
+		*alloc_ret = -ENOMEM;
 		goto end;
 	}
+
+	/*
+	 * 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) {
-		perror("Unable to find channel key");
-		goto end;
+		*alloc_ret = -ENOENT;
+		ERR("Unable to find channel for stream %d", stream_key);
+		goto error;
 	}
-	stream->chan->refcount++;
+
 	stream->key = stream_key;
 	stream->shm_fd = shm_fd;
 	stream->wait_fd = wait_fd;
@@ -374,35 +406,28 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 	stream->metadata_flag = metadata_flag;
 	strncpy(stream->path_name, path_name, sizeof(stream->path_name));
 	stream->path_name[sizeof(stream->path_name) - 1] = '\0';
-	lttng_ht_node_init_ulong(&stream->node, stream->key);
 	lttng_ht_node_init_ulong(&stream->waitfd_node, stream->wait_fd);
+	lttng_ht_node_init_ulong(&stream->node, stream->key);
 
-	switch (consumer_data.type) {
-	case LTTNG_CONSUMER_KERNEL:
-		break;
-	case LTTNG_CONSUMER32_UST:
-	case LTTNG_CONSUMER64_UST:
-		stream->cpu = stream->chan->cpucount++;
-		ret = lttng_ustconsumer_allocate_stream(stream);
-		if (ret) {
-			free(stream);
-			return NULL;
-		}
-		break;
-	default:
-		ERR("Unknown consumer_data type");
-		assert(0);
-		goto end;
-	}
-	DBG("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, out_fd %d, net_seq_idx %d)",
-			stream->path_name, stream->key,
-			stream->shm_fd,
-			stream->wait_fd,
-			(unsigned long long) stream->mmap_len,
-			stream->out_fd,
+	/*
+	 * 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);
+
+	DBG3("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu,"
+			" out_fd %d, net_seq_idx %d)", stream->path_name, stream->key,
+			stream->shm_fd, stream->wait_fd,
+			(unsigned long long) stream->mmap_len, stream->out_fd,
 			stream->net_seq_idx);
-end:
 	return stream;
+
+error:
+	free(stream);
+end:
+	return NULL;
 }
 
 /*
@@ -411,23 +436,14 @@ end:
 int consumer_add_stream(struct lttng_consumer_stream *stream)
 {
 	int ret = 0;
-	struct lttng_ht_node_ulong *node;
-	struct lttng_ht_iter iter;
 	struct consumer_relayd_sock_pair *relayd;
 
-	pthread_mutex_lock(&consumer_data.lock);
-	/* Steal stream identifier, for UST */
-	consumer_steal_stream_key(stream->key);
+	assert(stream);
+
+	DBG3("Adding consumer stream %d", stream->key);
 
+	pthread_mutex_lock(&consumer_data.lock);
 	rcu_read_lock();
-	lttng_ht_lookup(consumer_data.stream_ht,
-			(void *)((unsigned long) stream->key), &iter);
-	node = lttng_ht_iter_get_node_ulong(&iter);
-	if (node != NULL) {
-		rcu_read_unlock();
-		/* Stream already exist. Ignore the insertion */
-		goto end;
-	}
 
 	lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
 
@@ -436,13 +452,26 @@ int consumer_add_stream(struct lttng_consumer_stream *stream)
 	if (relayd != NULL) {
 		uatomic_inc(&relayd->refcount);
 	}
-	rcu_read_unlock();
 
-	/* Update consumer data */
+	/* Update channel refcount once added without error(s). */
+	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
+	 * 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);
+	}
+
+	/* Update consumer data once the node is inserted. */
 	consumer_data.stream_count++;
 	consumer_data.need_update = 1;
 
-end:
+	rcu_read_unlock();
 	pthread_mutex_unlock(&consumer_data.lock);
 
 	return ret;
@@ -452,8 +481,7 @@ end:
  * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
  * be acquired before calling this.
  */
-
-int consumer_add_relayd(struct consumer_relayd_sock_pair *relayd)
+static int add_relayd(struct consumer_relayd_sock_pair *relayd)
 {
 	int ret = 0;
 	struct lttng_ht_node_ulong *node;
@@ -542,7 +570,8 @@ error:
  * Return destination file descriptor or negative value on error.
  */
 static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
-		size_t data_size, struct consumer_relayd_sock_pair *relayd)
+		size_t data_size, unsigned long padding,
+		struct consumer_relayd_sock_pair *relayd)
 {
 	int outfd = -1, ret;
 	struct lttcomm_relayd_data_hdr data_hdr;
@@ -567,6 +596,7 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
 		/* Set header with stream information */
 		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++);
 		/* Other fields are zeroed previously */
 
@@ -593,7 +623,7 @@ void consumer_change_stream_state(int stream_key,
 	struct lttng_consumer_stream *stream;
 
 	pthread_mutex_lock(&consumer_data.lock);
-	stream = consumer_find_stream(stream_key);
+	stream = consumer_find_stream(stream_key, consumer_data.stream_ht);
 	if (stream) {
 		stream->state = state;
 	}
@@ -645,7 +675,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 	if (channel->mmap_base != NULL) {
 		ret = munmap(channel->mmap_base, channel->mmap_len);
 		if (ret != 0) {
-			perror("munmap");
+			PERROR("munmap");
 		}
 	}
 	if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
@@ -670,14 +700,15 @@ 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)
+		uint64_t max_sb_size,
+		unsigned int nb_init_streams)
 {
 	struct lttng_consumer_channel *channel;
 	int ret;
 
 	channel = zmalloc(sizeof(*channel));
 	if (channel == NULL) {
-		perror("malloc struct lttng_consumer_channel");
+		PERROR("malloc struct lttng_consumer_channel");
 		goto end;
 	}
 	channel->key = channel_key;
@@ -686,6 +717,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(
 	channel->mmap_len = mmap_len;
 	channel->max_sb_size = max_sb_size;
 	channel->refcount = 0;
+	channel->nb_init_streams = nb_init_streams;
 	lttng_ht_node_init_ulong(&channel->node, channel->key);
 
 	switch (consumer_data.type) {
@@ -800,7 +832,7 @@ restart:
 		if (errno == EINTR) {
 			goto restart;
 		}
-		perror("Poll error");
+		PERROR("Poll error");
 		goto exit;
 	}
 	if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
@@ -865,7 +897,7 @@ void lttng_consumer_cleanup(void)
 			node) {
 		struct lttng_consumer_stream *stream =
 			caa_container_of(node, struct lttng_consumer_stream, node);
-		consumer_del_stream(stream);
+		consumer_del_stream(stream, consumer_data.stream_ht);
 	}
 
 	cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
@@ -892,7 +924,7 @@ void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
 		ret = write(ctx->consumer_should_quit[1], "4", 1);
 	} while (ret < 0 && errno == EINTR);
 	if (ret < 0) {
-		perror("write consumer quit");
+		PERROR("write consumer quit");
 	}
 }
 
@@ -964,7 +996,7 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 
 	ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
 	if (ctx == NULL) {
-		perror("allocating context");
+		PERROR("allocating context");
 		goto error;
 	}
 
@@ -977,33 +1009,33 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 
 	ret = pipe(ctx->consumer_poll_pipe);
 	if (ret < 0) {
-		perror("Error creating poll pipe");
+		PERROR("Error creating poll pipe");
 		goto error_poll_pipe;
 	}
 
 	/* set read end of the pipe to non-blocking */
 	ret = fcntl(ctx->consumer_poll_pipe[0], F_SETFL, O_NONBLOCK);
 	if (ret < 0) {
-		perror("fcntl O_NONBLOCK");
+		PERROR("fcntl O_NONBLOCK");
 		goto error_poll_fcntl;
 	}
 
 	/* set write end of the pipe to non-blocking */
 	ret = fcntl(ctx->consumer_poll_pipe[1], F_SETFL, O_NONBLOCK);
 	if (ret < 0) {
-		perror("fcntl O_NONBLOCK");
+		PERROR("fcntl O_NONBLOCK");
 		goto error_poll_fcntl;
 	}
 
 	ret = pipe(ctx->consumer_should_quit);
 	if (ret < 0) {
-		perror("Error creating recv pipe");
+		PERROR("Error creating recv pipe");
 		goto error_quit_pipe;
 	}
 
 	ret = pipe(ctx->consumer_thread_pipe);
 	if (ret < 0) {
-		perror("Error creating thread pipe");
+		PERROR("Error creating thread pipe");
 		goto error_thread_pipe;
 	}
 
@@ -1094,22 +1126,23 @@ 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)
+		struct consumer_relayd_sock_pair *relayd,
+		unsigned long padding)
 {
 	int ret;
-	uint64_t metadata_id;
+	struct lttcomm_relayd_metadata_payload hdr;
 
-	metadata_id = htobe64(stream->relayd_stream_id);
+	hdr.stream_id = htobe64(stream->relayd_stream_id);
+	hdr.padding_size = htobe32(padding);
 	do {
-		ret = write(fd, (void *) &metadata_id,
-				sizeof(stream->relayd_stream_id));
+		ret = write(fd, (void *) &hdr, sizeof(hdr));
 	} while (ret < 0 && errno == EINTR);
 	if (ret < 0) {
 		PERROR("write metadata stream id");
 		goto end;
 	}
-	DBG("Metadata stream id %" PRIu64 " written before data",
-			stream->relayd_stream_id);
+	DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
+			stream->relayd_stream_id, padding);
 
 end:
 	return ret;
@@ -1126,7 +1159,8 @@ end:
  */
 ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *stream, unsigned long len)
+		struct lttng_consumer_stream *stream, unsigned long len,
+		unsigned long padding)
 {
 	unsigned long mmap_offset;
 	ssize_t ret = 0, written = 0;
@@ -1178,17 +1212,17 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		if (stream->metadata_flag) {
 			/* Metadata requires the control socket. */
 			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
-			netlen += sizeof(stream->relayd_stream_id);
+			netlen += sizeof(struct lttcomm_relayd_metadata_payload);
 		}
 
-		ret = write_relayd_stream_header(stream, netlen, relayd);
+		ret = write_relayd_stream_header(stream, netlen, padding, relayd);
 		if (ret >= 0) {
 			/* Use the returned socket. */
 			outfd = ret;
 
 			/* Write metadata stream id before payload */
 			if (stream->metadata_flag) {
-				ret = write_relayd_metadata_id(outfd, stream, relayd);
+				ret = write_relayd_metadata_id(outfd, stream, relayd, padding);
 				if (ret < 0) {
 					written = ret;
 					goto end;
@@ -1196,12 +1230,16 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 			}
 		}
 		/* Else, use the default set before which is the filesystem. */
+	} else {
+		/* No streaming, we have to set the len with the full padding */
+		len += padding;
 	}
 
 	while (len > 0) {
 		do {
 			ret = write(outfd, stream->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");
 			if (written == 0) {
@@ -1216,7 +1254,6 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 			len -= ret;
 			mmap_offset += ret;
 		}
-		DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
 
 		/* This call is useless on a socket so better save a syscall. */
 		if (!relayd) {
@@ -1246,7 +1283,8 @@ end:
  */
 ssize_t lttng_consumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *stream, unsigned long len)
+		struct lttng_consumer_stream *stream, unsigned long len,
+		unsigned long padding)
 {
 	ssize_t ret = 0, written = 0, ret_splice = 0;
 	loff_t offset = 0;
@@ -1292,23 +1330,42 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 	}
 
 	/* Write metadata stream id before payload */
-	if (stream->metadata_flag && relayd) {
-		/*
-		 * Lock the control socket for the complete duration of the function
-		 * since from this point on we will use the socket.
-		 */
-		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+	if (relayd) {
+		int total_len = len;
 
-		ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd);
-		if (ret < 0) {
-			written = ret;
+		if (stream->metadata_flag) {
+			/*
+			 * Lock the control socket for the complete duration of the function
+			 * since from this point on we will use the socket.
+			 */
+			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+
+			ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
+					padding);
+			if (ret < 0) {
+				written = ret;
+				goto end;
+			}
+
+			total_len += sizeof(struct lttcomm_relayd_metadata_payload);
+		}
+
+		ret = write_relayd_stream_header(stream, total_len, padding, relayd);
+		if (ret >= 0) {
+			/* Use the returned socket. */
+			outfd = ret;
+		} else {
+			ERR("Remote relayd disconnected. Stopping");
 			goto end;
 		}
+	} else {
+		/* No streaming, we have to set the len with the full padding */
+		len += padding;
 	}
 
 	while (len > 0) {
-		DBG("splice chan to pipe offset %lu of len %lu (fd : %d)",
-				(unsigned long)offset, len, fd);
+		DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
+				(unsigned long)offset, len, fd, splice_pipe[1]);
 		ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
 				SPLICE_F_MOVE | SPLICE_F_MORE);
 		DBG("splice chan to pipe, ret %zd", ret_splice);
@@ -1324,30 +1381,24 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 		/* Handle stream on the relayd if the output is on the network */
 		if (relayd) {
 			if (stream->metadata_flag) {
+				size_t metadata_payload_size =
+					sizeof(struct lttcomm_relayd_metadata_payload);
+
 				/* Update counter to fit the spliced data */
-				ret_splice += sizeof(stream->relayd_stream_id);
-				len += sizeof(stream->relayd_stream_id);
+				ret_splice += metadata_payload_size;
+				len += metadata_payload_size;
 				/*
 				 * We do this so the return value can match the len passed as
 				 * argument to this function.
 				 */
-				written -= sizeof(stream->relayd_stream_id);
-			}
-
-			ret = write_relayd_stream_header(stream, ret_splice, relayd);
-			if (ret >= 0) {
-				/* Use the returned socket. */
-				outfd = ret;
-			} else {
-				ERR("Remote relayd disconnected. Stopping");
-				goto end;
+				written -= metadata_payload_size;
 			}
 		}
 
 		/* Splice data out */
 		ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
 				ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
-		DBG("Kernel consumer splice pipe to file, ret %zd", ret_splice);
+		DBG("Consumer splice pipe to file, ret %zd", ret_splice);
 		if (ret_splice < 0) {
 			PERROR("Error in file splice");
 			if (written == 0) {
@@ -1468,9 +1519,9 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 }
 
 /*
- * Iterate over all stream element of the hashtable and free them. This is race
- * free since the hashtable received MUST be in a race free synchronization
- * state. It's the caller responsability to make sure of that.
+ * Iterate over all streams of the hashtable and free them properly.
+ *
+ * XXX: Should not be only for metadata stream or else use an other name.
  */
 static void destroy_stream_ht(struct lttng_ht *ht)
 {
@@ -1482,12 +1533,14 @@ static void destroy_stream_ht(struct lttng_ht *ht)
 		return;
 	}
 
-	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
+	rcu_read_lock();
+	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, waitfd_node.node) {
 		ret = lttng_ht_del(ht, &iter);
 		assert(!ret);
 
-		free(stream);
+		call_rcu(&stream->waitfd_node.head, consumer_free_metadata_stream);
 	}
+	rcu_read_unlock();
 
 	lttng_ht_destroy(ht);
 }
@@ -1495,9 +1548,11 @@ static void destroy_stream_ht(struct lttng_ht *ht)
 /*
  * Clean up a metadata stream and free its memory.
  */
-static void consumer_del_metadata_stream(struct lttng_consumer_stream *stream)
+void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
+		struct lttng_ht *ht)
 {
 	int ret;
+	struct lttng_ht_iter iter;
 	struct lttng_consumer_channel *free_chan = NULL;
 	struct consumer_relayd_sock_pair *relayd;
 
@@ -1508,6 +1563,13 @@ static void consumer_del_metadata_stream(struct lttng_consumer_stream *stream)
 	 */
 	assert(stream->metadata_flag);
 
+	DBG3("Consumer delete metadata stream %d", stream->wait_fd);
+
+	if (ht == NULL) {
+		/* Means the stream was allocated but not successfully added */
+		goto free_stream;
+	}
+
 	pthread_mutex_lock(&consumer_data.lock);
 	switch (consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
@@ -1525,8 +1587,14 @@ static void consumer_del_metadata_stream(struct lttng_consumer_stream *stream)
 	default:
 		ERR("Unknown consumer_data type");
 		assert(0);
+		goto end;
 	}
-	pthread_mutex_unlock(&consumer_data.lock);
+
+	rcu_read_lock();
+	iter.iter.node = &stream->waitfd_node.node;
+	ret = lttng_ht_del(ht, &iter);
+	assert(!ret);
+	rcu_read_unlock();
 
 	if (stream->out_fd >= 0) {
 		ret = close(stream->out_fd);
@@ -1573,50 +1641,89 @@ static void consumer_del_metadata_stream(struct lttng_consumer_stream *stream)
 		/* Both conditions are met, we destroy the relayd. */
 		if (uatomic_read(&relayd->refcount) == 0 &&
 				uatomic_read(&relayd->destroy_flag)) {
-			consumer_destroy_relayd(relayd);
+			destroy_relayd(relayd);
 		}
 	}
 	rcu_read_unlock();
 
 	/* Atomically decrement channel refcount since other threads can use it. */
 	uatomic_dec(&stream->chan->refcount);
-	if (!uatomic_read(&stream->chan->refcount)) {
+	if (!uatomic_read(&stream->chan->refcount)
+			&& !uatomic_read(&stream->chan->nb_init_streams)) {
+		/* Go for channel deletion! */
 		free_chan = stream->chan;
 	}
 
+end:
+	pthread_mutex_unlock(&consumer_data.lock);
+
 	if (free_chan) {
 		consumer_del_channel(free_chan);
 	}
 
-	free(stream);
+free_stream:
+	call_rcu(&stream->waitfd_node.head, consumer_free_metadata_stream);
 }
 
 /*
  * Action done with the metadata stream when adding it to the consumer internal
  * data structures to handle it.
  */
-static void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
+static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
+		struct lttng_ht *ht)
 {
+	int ret = 0;
 	struct consumer_relayd_sock_pair *relayd;
 
-	/* Find relayd and, if one is found, increment refcount. */
+	assert(stream);
+	assert(ht);
+
+	DBG3("Adding metadata stream %d to hash table", stream->wait_fd);
+
+	pthread_mutex_lock(&consumer_data.lock);
+
+	/*
+	 * From here, refcounts are updated so be _careful_ when returning an error
+	 * after this point.
+	 */
+
 	rcu_read_lock();
+	/* Find relayd and, if one is found, increment refcount. */
 	relayd = consumer_find_relayd(stream->net_seq_idx);
 	if (relayd != NULL) {
 		uatomic_inc(&relayd->refcount);
 	}
+
+	/* Update channel refcount once added without error(s). */
+	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
+	 * 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);
+	}
+
+	lttng_ht_add_unique_ulong(ht, &stream->waitfd_node);
 	rcu_read_unlock();
+
+	pthread_mutex_unlock(&consumer_data.lock);
+	return ret;
 }
 
 /*
  * Thread polls on metadata file descriptor and write them on disk or on the
  * network.
  */
-void *lttng_consumer_thread_poll_metadata(void *data)
+void *consumer_thread_metadata_poll(void *data)
 {
 	int ret, i, pollfd;
 	uint32_t revents, nb_fd;
-	struct lttng_consumer_stream *stream;
+	struct lttng_consumer_stream *stream = NULL;
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_ulong *node;
 	struct lttng_ht *metadata_ht = NULL;
@@ -1664,18 +1771,24 @@ restart:
 		DBG("Metadata event catched in thread");
 		if (ret < 0) {
 			if (errno == EINTR) {
+				ERR("Poll EINTR catched");
 				goto restart;
 			}
 			goto error;
 		}
 
+		/* From here, the event is a metadata wait fd */
 		for (i = 0; i < nb_fd; i++) {
 			revents = LTTNG_POLL_GETEV(&events, i);
 			pollfd = LTTNG_POLL_GETFD(&events, i);
 
-			/* Check the metadata pipe for incoming metadata. */
+			/* Just don't waste time if no returned events for the fd */
+			if (!revents) {
+				continue;
+			}
+
 			if (pollfd == ctx->consumer_metadata_pipe[0]) {
-				if (revents & (LPOLLERR | LPOLLHUP | LPOLLNVAL)) {
+				if (revents & (LPOLLERR | LPOLLHUP )) {
 					DBG("Metadata thread pipe hung up");
 					/*
 					 * Remove the pipe from the poll set and continue the loop
@@ -1685,20 +1798,13 @@ restart:
 					close(ctx->consumer_metadata_pipe[0]);
 					continue;
 				} else if (revents & LPOLLIN) {
-					stream = zmalloc(sizeof(struct lttng_consumer_stream));
-					if (stream == NULL) {
-						PERROR("zmalloc metadata consumer stream");
-						goto error;
-					}
-
 					do {
-						/* Get the stream and add it to the local hash table */
-						ret = read(pollfd, stream,
-								sizeof(struct lttng_consumer_stream));
+						/* Get the stream pointer received */
+						ret = read(pollfd, &stream, sizeof(stream));
 					} while (ret < 0 && errno == EINTR);
-					if (ret < 0 || ret < sizeof(struct lttng_consumer_stream)) {
+					if (ret < 0 ||
+							ret < sizeof(struct lttng_consumer_stream *)) {
 						PERROR("read metadata stream");
-						free(stream);
 						/*
 						 * Let's continue here and hope we can still work
 						 * without stopping the consumer. XXX: Should we?
@@ -1709,54 +1815,35 @@ restart:
 					DBG("Adding metadata stream %d to poll set",
 							stream->wait_fd);
 
-					/* The node should be init at this point */
-					lttng_ht_add_unique_ulong(metadata_ht,
-							&stream->waitfd_node);
+					ret = consumer_add_metadata_stream(stream, metadata_ht);
+					if (ret) {
+						ERR("Unable to add metadata stream");
+						/* Stream was not setup properly. Continuing. */
+						consumer_del_metadata_stream(stream, NULL);
+						continue;
+					}
 
 					/* Add metadata stream to the global poll events list */
 					lttng_poll_add(&events, stream->wait_fd,
 							LPOLLIN | LPOLLPRI);
-
-					consumer_add_metadata_stream(stream);
 				}
 
-				/* Metadata pipe handled. Continue handling the others */
+				/* Handle other stream */
 				continue;
 			}
 
-			/* From here, the event is a metadata wait fd */
-
+			rcu_read_lock();
 			lttng_ht_lookup(metadata_ht, (void *)((unsigned long) pollfd),
 					&iter);
 			node = lttng_ht_iter_get_node_ulong(&iter);
-			if (node == NULL) {
-				/* FD not found, continue loop */
-				continue;
-			}
+			assert(node);
 
 			stream = caa_container_of(node, struct lttng_consumer_stream,
 					waitfd_node);
 
-			/* Get the data out of the metadata file descriptor */
-			if (revents & (LPOLLIN | LPOLLPRI)) {
-				DBG("Metadata available on fd %d", pollfd);
-				assert(stream->wait_fd == pollfd);
-
-				len = ctx->on_buffer_ready(stream, ctx);
-				/* It's ok to have an unavailable sub-buffer */
-				if (len < 0 && len != -EAGAIN) {
-					goto end;
-				} else if (len > 0) {
-					stream->data_read = 1;
-				}
-			}
-
-			/*
-			 * Remove the stream from the hash table since there is no data
-			 * left on the fd because we previously did a read on the buffer.
-			 */
-			if (revents & (LPOLLERR | LPOLLHUP | LPOLLNVAL)) {
-				DBG("Metadata fd %d is hup|err|nval.", pollfd);
+			/* Check for error event */
+			if (revents & (LPOLLERR | LPOLLHUP)) {
+				DBG("Metadata fd %d is hup|err.", pollfd);
 				if (!stream->hangup_flush_done
 						&& (consumer_data.type == LTTNG_CONSUMER32_UST
 							|| consumer_data.type == LTTNG_CONSUMER64_UST)) {
@@ -1764,18 +1851,40 @@ restart:
 					lttng_ustconsumer_on_stream_hangup(stream);
 
 					/* We just flushed the stream now read it. */
-					len = ctx->on_buffer_ready(stream, ctx);
-					/* It's ok to have an unavailable sub-buffer */
-					if (len < 0 && len != -EAGAIN) {
-						goto end;
-					}
+					do {
+						len = ctx->on_buffer_ready(stream, ctx);
+						/*
+						 * We don't check the return value here since if we get
+						 * a negative len, it means an error occured thus we
+						 * simply remove it from the poll set and free the
+						 * stream.
+						 */
+					} while (len > 0);
 				}
 
-				/* Removing it from hash table, poll set and free memory */
-				lttng_ht_del(metadata_ht, &iter);
 				lttng_poll_del(&events, stream->wait_fd);
-				consumer_del_metadata_stream(stream);
+				/*
+				 * This call update the channel states, closes file descriptors
+				 * and securely free the stream.
+				 */
+				consumer_del_metadata_stream(stream, metadata_ht);
+			} else if (revents & (LPOLLIN | LPOLLPRI)) {
+				/* Get the data out of the metadata file descriptor */
+				DBG("Metadata available on fd %d", pollfd);
+				assert(stream->wait_fd == pollfd);
+
+				len = ctx->on_buffer_ready(stream, ctx);
+				/* It's ok to have an unavailable sub-buffer */
+				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
+					rcu_read_unlock();
+					goto end;
+				} else if (len > 0) {
+					stream->data_read = 1;
+				}
 			}
+
+			/* Release RCU lock for the stream looked up */
+			rcu_read_unlock();
 		}
 	}
 
@@ -1796,29 +1905,19 @@ end:
  * This thread polls the fds in the set to consume the data and write
  * it to tracefile if necessary.
  */
-void *lttng_consumer_thread_poll_fds(void *data)
+void *consumer_thread_data_poll(void *data)
 {
 	int num_rdy, num_hup, high_prio, ret, i;
 	struct pollfd *pollfd = NULL;
 	/* local view of the streams */
-	struct lttng_consumer_stream **local_stream = NULL;
+	struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
 	/* local view of consumer_data.fds_count */
 	int nb_fd = 0;
 	struct lttng_consumer_local_data *ctx = data;
 	ssize_t len;
-	pthread_t metadata_thread;
-	void *status;
 
 	rcu_register_thread();
 
-	/* Start metadata polling thread */
-	ret = pthread_create(&metadata_thread, NULL,
-			lttng_consumer_thread_poll_metadata, (void *) ctx);
-	if (ret < 0) {
-		PERROR("pthread_create metadata thread");
-		goto end;
-	}
-
 	local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
 
 	while (1) {
@@ -1843,7 +1942,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			/* allocate for all fds + 1 for the consumer_poll_pipe */
 			pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
 			if (pollfd == NULL) {
-				perror("pollfd malloc");
+				PERROR("pollfd malloc");
 				pthread_mutex_unlock(&consumer_data.lock);
 				goto end;
 			}
@@ -1852,7 +1951,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			local_stream = zmalloc((consumer_data.stream_count + 1) *
 					sizeof(struct lttng_consumer_stream));
 			if (local_stream == NULL) {
-				perror("local_stream malloc");
+				PERROR("local_stream malloc");
 				pthread_mutex_unlock(&consumer_data.lock);
 				goto end;
 			}
@@ -1884,7 +1983,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			if (errno == EINTR) {
 				goto restart;
 			}
-			perror("Poll error");
+			PERROR("Poll error");
 			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
 			goto end;
 		} else if (num_rdy == 0) {
@@ -1899,13 +1998,35 @@ void *lttng_consumer_thread_poll_fds(void *data)
 		 */
 		if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
 			size_t pipe_readlen;
-			char tmp;
 
 			DBG("consumer_poll_pipe wake up");
 			/* Consume 1 byte of pipe data */
 			do {
-				pipe_readlen = read(ctx->consumer_poll_pipe[0], &tmp, 1);
+				pipe_readlen = read(ctx->consumer_poll_pipe[0], &new_stream,
+						sizeof(new_stream));
 			} while (pipe_readlen == -1 && errno == EINTR);
+
+			/*
+			 * If the stream is NULL, just ignore it. It's also possible that
+			 * the sessiond poll thread changed the consumer_quit state and is
+			 * waking us up to test it.
+			 */
+			if (new_stream == NULL) {
+				continue;
+			}
+
+			ret = consumer_add_stream(new_stream);
+			if (ret) {
+				ERR("Consumer add stream %d failed. Continuing",
+						new_stream->key);
+				/*
+				 * At this point, if the add_stream fails, it is not in the
+				 * hash table thus passing the NULL value here.
+				 */
+				consumer_del_stream(new_stream, NULL);
+			}
+
+			/* Continue to update the local streams and handle prio ones */
 			continue;
 		}
 
@@ -1916,7 +2037,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 				high_prio = 1;
 				len = ctx->on_buffer_ready(local_stream[i], ctx);
 				/* it's ok to have an unavailable sub-buffer */
-				if (len < 0 && len != -EAGAIN) {
+				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					goto end;
 				} else if (len > 0) {
 					local_stream[i]->data_read = 1;
@@ -1939,7 +2060,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 				DBG("Normal read on fd %d", pollfd[i].fd);
 				len = ctx->on_buffer_ready(local_stream[i], ctx);
 				/* it's ok to have an unavailable sub-buffer */
-				if (len < 0 && len != -EAGAIN) {
+				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					goto end;
 				} else if (len > 0) {
 					local_stream[i]->data_read = 1;
@@ -1967,19 +2088,22 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			if ((pollfd[i].revents & POLLHUP)) {
 				DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
-					consumer_del_stream(local_stream[i]);
+					consumer_del_stream(local_stream[i],
+							consumer_data.stream_ht);
 					num_hup++;
 				}
 			} else if (pollfd[i].revents & POLLERR) {
 				ERR("Error returned in polling fd %d.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
-					consumer_del_stream(local_stream[i]);
+					consumer_del_stream(local_stream[i],
+							consumer_data.stream_ht);
 					num_hup++;
 				}
 			} else if (pollfd[i].revents & POLLNVAL) {
 				ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
-					consumer_del_stream(local_stream[i]);
+					consumer_del_stream(local_stream[i],
+							consumer_data.stream_ht);
 					num_hup++;
 				}
 			}
@@ -1999,19 +2123,13 @@ end:
 
 	/*
 	 * Close the write side of the pipe so epoll_wait() in
-	 * lttng_consumer_thread_poll_metadata can catch it. The thread is
-	 * monitoring the read side of the pipe. If we close them both, epoll_wait
-	 * strangely does not return and could create a endless wait period if the
-	 * pipe is the only tracked fd in the poll set. The thread will take care
-	 * of closing the read side.
+	 * consumer_thread_metadata_poll can catch it. The thread is monitoring the
+	 * read side of the pipe. If we close them both, epoll_wait strangely does
+	 * not return and could create a endless wait period if the pipe is the
+	 * only tracked fd in the poll set. The thread will take care of closing
+	 * the read side.
 	 */
 	close(ctx->consumer_metadata_pipe[1]);
-	if (ret) {
-		ret = pthread_join(metadata_thread, &status);
-		if (ret < 0) {
-			PERROR("pthread_join metadata thread");
-		}
-	}
 
 	rcu_unregister_thread();
 	return NULL;
@@ -2021,7 +2139,7 @@ end:
  * This thread listens on the consumerd socket and receives the file
  * descriptors from the session daemon.
  */
-void *lttng_consumer_thread_receive_fds(void *data)
+void *consumer_thread_sessiond_poll(void *data)
 {
 	int sock, client_socket, ret;
 	/*
@@ -2056,7 +2174,7 @@ void *lttng_consumer_thread_receive_fds(void *data)
 
 	ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
 	if (ret < 0) {
-		perror("fcntl O_NONBLOCK");
+		PERROR("fcntl O_NONBLOCK");
 		goto end;
 	}
 
@@ -2079,7 +2197,7 @@ void *lttng_consumer_thread_receive_fds(void *data)
 	}
 	ret = fcntl(sock, F_SETFL, O_NONBLOCK);
 	if (ret < 0) {
-		perror("fcntl O_NONBLOCK");
+		PERROR("fcntl O_NONBLOCK");
 		goto end;
 	}
 
@@ -2128,19 +2246,16 @@ end:
 	consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
 
 	/*
-	 * Wake-up the other end by writing a null byte in the pipe
-	 * (non-blocking). Important note: Because writing into the
-	 * pipe is non-blocking (and therefore we allow dropping wakeup
-	 * data, as long as there is wakeup data present in the pipe
-	 * buffer to wake up the other end), the other end should
-	 * perform the following sequence for waiting:
-	 * 1) empty the pipe (reads).
-	 * 2) perform update operation.
-	 * 3) wait on the pipe (poll).
+	 * Notify the data poll thread to poll back again and test the
+	 * consumer_quit state to quit gracefully.
 	 */
 	do {
-		ret = write(ctx->consumer_poll_pipe[1], "", 1);
+		struct lttng_consumer_stream *null_stream = NULL;
+
+		ret = write(ctx->consumer_poll_pipe[1], &null_stream,
+				sizeof(null_stream));
 	} while (ret < 0 && errno == EINTR);
+
 	rcu_unregister_thread();
 	return NULL;
 }
@@ -2269,7 +2384,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 	 * Add relayd socket pair to consumer data hashtable. If object already
 	 * exists or on error, the function gracefully returns.
 	 */
-	consumer_add_relayd(relayd);
+	add_relayd(relayd);
 
 	/* All good! */
 	ret = 0;