X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=6272e19b710b1ff2106c0d709c50bb973222523a;hp=be984a49270fefbba0b55bccd8ebf897ceea4ce7;hb=d0b96690836f4b876096f3dc14801f8e25281a77;hpb=c5c45efab4fc692edd8fd4db88c0df1d5f6cefff

diff --git a/src/common/consumer.c b/src/common/consumer.c
index be984a492..6272e19b7 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -63,21 +63,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.
@@ -97,7 +82,7 @@ static void notify_thread_pipe(int wpipe)
  * 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(int key,
 		struct lttng_ht *ht)
 {
 	struct lttng_ht_iter iter;
@@ -124,12 +109,12 @@ 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;
 		/*
@@ -148,7 +133,7 @@ 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(unsigned long key)
 {
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_ulong *node;
@@ -159,8 +144,7 @@ static struct lttng_consumer_channel *consumer_find_channel(int key)
 		return NULL;
 	}
 
-	lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) key),
-			&iter);
+	lttng_ht_lookup(consumer_data.channel_ht, (void *) key, &iter);
 	node = lttng_ht_iter_get_node_ulong(&iter);
 	if (node != NULL) {
 		channel = caa_container_of(node, struct lttng_consumer_channel, node);
@@ -169,39 +153,30 @@ 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_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);
 
-	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_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);
@@ -231,7 +206,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 +213,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 +221,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 %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();
+
+	call_rcu(&channel->node.head, free_channel_rcu);
+end:
+	pthread_mutex_unlock(&consumer_data.lock);
 }
 
 /*
@@ -283,8 +278,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 +395,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);
@@ -447,18 +440,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 +473,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 +486,67 @@ 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(int channel_key,
+		int 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;
+		}
 	}
 
+	/* Key is always the wait_fd for streams. */
+	lttng_ht_node_init_ulong(&stream->node, stream->key);
+
 	/* Init session id node with the stream session id */
 	lttng_ht_node_init_ulong(&stream->node_session_id, stream->session_id);
 
-	/*
-	 * 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, 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);
+	DBG3("Allocated stream %s (key %d, relayd_id %d, session_id %" PRIu64,
+			stream->name, stream->key, stream->net_seq_idx, stream->session_id);
 
 	rcu_read_unlock();
 	return stream;
@@ -598,13 +555,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;
@@ -620,7 +580,7 @@ static int consumer_add_stream(struct lttng_consumer_stream *stream,
 	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_ulong(ht, &stream->node);
 
@@ -641,14 +601,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. */
@@ -672,16 +632,12 @@ static int add_relayd(struct consumer_relayd_sock_pair *relayd)
 	struct lttng_ht_node_ulong *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);
 	if (node != NULL) {
-		/* Relayd already exist. Ignore the insertion */
 		goto end;
 	}
 	lttng_ht_add_unique_ulong(consumer_data.relayd_ht, &relayd->node);
@@ -782,7 +738,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 +754,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 +764,48 @@ 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;
+	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_ulong(&channel->node, channel->key);
+	CDS_INIT_LIST_HEAD(&channel->streams.head);
+
+	DBG("Allocated channel (key %d)", 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;
 }
@@ -921,12 +815,11 @@ end:
  */
 int consumer_add_channel(struct lttng_consumer_channel *channel)
 {
+	int ret = 0;
 	struct lttng_ht_node_ulong *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,
@@ -934,6 +827,8 @@ int consumer_add_channel(struct lttng_consumer_channel *channel)
 	node = lttng_ht_iter_get_node_ulong(&iter);
 	if (node != NULL) {
 		/* Channel already exist. Ignore the insertion */
+		ERR("Consumer add channel key %d already exists!", channel->key);
+		ret = -1;
 		goto end;
 	}
 
@@ -943,7 +838,7 @@ end:
 	rcu_read_unlock();
 	pthread_mutex_unlock(&consumer_data.lock);
 
-	return 0;
+	return ret;
 }
 
 /*
@@ -953,14 +848,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 +927,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 +946,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 +963,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);
 	}
 
@@ -1099,7 +996,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 +1014,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 +1036,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);
 }
 
 /*
@@ -1306,8 +1203,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 +1213,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 +1255,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 +1277,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,7 +1348,7 @@ 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) {
@@ -1704,21 +1621,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;
 	}
-
 }
 
 /*
@@ -1726,17 +1641,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);
@@ -1836,7 +1749,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);
@@ -1879,20 +1792,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);
@@ -1925,7 +1824,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;
 	}
@@ -1938,15 +1837,15 @@ 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;
@@ -1957,7 +1856,7 @@ static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
 	assert(stream);
 	assert(ht);
 
-	DBG3("Adding metadata stream %d to hash table", stream->wait_fd);
+	DBG3("Adding metadata stream %d to hash table", stream->key);
 
 	pthread_mutex_lock(&consumer_data.lock);
 	pthread_mutex_lock(&stream->lock);
@@ -1973,7 +1872,7 @@ 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);
+	lttng_ht_lookup(ht, (void *)((unsigned long) stream->key), &iter);
 	node = lttng_ht_iter_get_node_ulong(&iter);
 	assert(!node);
 
@@ -1987,14 +1886,14 @@ 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);
@@ -2168,13 +2067,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. */
@@ -2294,14 +2193,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));
@@ -2319,7 +2215,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");
@@ -2362,7 +2258,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 */
@@ -2386,7 +2282,7 @@ 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",
 						new_stream->key);
@@ -2498,14 +2394,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
@@ -2582,7 +2472,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;
 	}
@@ -2625,10 +2515,10 @@ 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");
 
 	/*
 	 * when all fds have hung up, the polling thread
@@ -2709,7 +2599,6 @@ 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);
 }
 
 /*
@@ -2726,7 +2615,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);
 
@@ -2744,6 +2632,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;
@@ -2762,7 +2651,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. */
@@ -2793,32 +2682,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 */
@@ -2840,6 +2725,7 @@ 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;
 	}
 
@@ -2864,10 +2750,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);
 	}
 
@@ -2911,30 +2795,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;
 }
 
@@ -3021,7 +2899,8 @@ int consumer_data_pending(uint64_t id)
 						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) {
@@ -3080,3 +2959,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));
+}