+ struct lttng_consumer_stream *stream;
+
+ if (ht == NULL) {
+ return;
+ }
+
+ rcu_read_lock();
+ cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
+ ret = lttng_ht_del(ht, &iter);
+ assert(!ret);
+
+ call_rcu(&stream->node.head, consumer_free_stream);
+ }
+ rcu_read_unlock();
+
+ lttng_ht_destroy(ht);
+}
+
+/*
+ * 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)
+{
+ int ret;
+ struct lttng_ht_iter iter;
+ struct lttng_consumer_stream *stream;
+
+ if (ht == NULL) {
+ return;
+ }
+
+ rcu_read_lock();
+ cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
+ ret = lttng_ht_del(ht, &iter);
+ assert(!ret);
+
+ call_rcu(&stream->node.head, consumer_free_stream);
+ }
+ rcu_read_unlock();
+
+ lttng_ht_destroy(ht);
+}
+
+/*
+ * Clean up a metadata stream and free its memory.
+ */
+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;
+
+ assert(stream);
+ /*
+ * This call should NEVER receive regular stream. It must always be
+ * metadata stream and this is crucial for data structure synchronization.
+ */
+ 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(&stream->lock);
+
+ pthread_mutex_lock(&consumer_data.lock);
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ if (stream->mmap_base != NULL) {
+ ret = munmap(stream->mmap_base, stream->mmap_len);
+ if (ret != 0) {
+ PERROR("munmap metadata stream");
+ }
+ }
+ break;
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ lttng_ustconsumer_del_stream(stream);
+ break;
+ default:
+ ERR("Unknown consumer_data type");
+ assert(0);
+ goto end;
+ }
+
+ rcu_read_lock();
+ iter.iter.node = &stream->node.node;
+ ret = lttng_ht_del(ht, &iter);
+ assert(!ret);
+
+ /* Remove node session id from the consumer_data stream ht */
+ iter.iter.node = &stream->node_session_id.node;
+ ret = lttng_ht_del(consumer_data.stream_list_ht, &iter);
+ assert(!ret);
+ rcu_read_unlock();
+
+ if (stream->out_fd >= 0) {
+ ret = close(stream->out_fd);
+ if (ret) {
+ 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();
+ relayd = consumer_find_relayd(stream->net_seq_idx);
+ if (relayd != NULL) {
+ uatomic_dec(&relayd->refcount);
+ assert(uatomic_read(&relayd->refcount) >= 0);
+
+ /* Closing streams requires to lock the control socket. */
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ ret = relayd_send_close_stream(&relayd->control_sock,
+ stream->relayd_stream_id, stream->next_net_seq_num - 1);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+ if (ret < 0) {
+ DBG("Unable to close stream on the relayd. Continuing");
+ /*
+ * Continue here. There is nothing we can do for the relayd.
+ * Chances are that the relayd has closed the socket so we just
+ * continue cleaning up.
+ */
+ }
+
+ /* Both conditions are met, we destroy the relayd. */
+ if (uatomic_read(&relayd->refcount) == 0 &&
+ uatomic_read(&relayd->destroy_flag)) {
+ 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)
+ && !uatomic_read(&stream->chan->nb_init_streams)) {
+ /* Go for channel deletion! */
+ free_chan = stream->chan;
+ }
+
+end:
+ pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&stream->lock);
+
+ if (free_chan) {
+ consumer_del_channel(free_chan);
+ }
+
+free_stream:
+ call_rcu(&stream->node.head, consumer_free_stream);
+}
+
+/*
+ * 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,
+ struct lttng_ht *ht)
+{
+ int ret = 0;
+ struct consumer_relayd_sock_pair *relayd;
+
+ 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);
+ }
+
+ /* Steal stream identifier to avoid having streams with the same key */
+ consumer_steal_stream_key(stream->key, ht);
+
+ lttng_ht_add_unique_ulong(ht, &stream->node);
+
+ /*
+ * 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);
+
+ rcu_read_unlock();
+
+ pthread_mutex_unlock(&consumer_data.lock);
+ return ret;
+}
+
+/*
+ * Delete data stream that are flagged for deletion (endpoint_status).
+ */
+static void validate_endpoint_status_data_stream(void)
+{
+ struct lttng_ht_iter iter;
+ struct lttng_consumer_stream *stream;
+
+ DBG("Consumer delete flagged data stream");
+
+ rcu_read_lock();
+ cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
+ /* Validate delete flag of the stream */
+ if (stream->endpoint_status != CONSUMER_ENDPOINT_INACTIVE) {
+ continue;
+ }
+ /* Delete it right now */
+ consumer_del_stream(stream, data_ht);
+ }
+ rcu_read_unlock();
+}
+
+/*
+ * Delete metadata stream that are flagged for deletion (endpoint_status).
+ */
+static void validate_endpoint_status_metadata_stream(
+ struct lttng_poll_event *pollset)
+{
+ struct lttng_ht_iter iter;
+ struct lttng_consumer_stream *stream;
+
+ DBG("Consumer delete flagged metadata stream");
+
+ assert(pollset);
+
+ rcu_read_lock();
+ cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
+ /* Validate delete flag of the stream */
+ if (!stream->endpoint_status) {
+ continue;
+ }
+ /*
+ * Remove from pollset so the metadata thread can continue without
+ * blocking on a deleted stream.
+ */
+ lttng_poll_del(pollset, stream->wait_fd);
+
+ /* Delete it right now */
+ consumer_del_metadata_stream(stream, metadata_ht);
+ }
+ rcu_read_unlock();
+}
+
+/*
+ * Thread polls on metadata file descriptor and write them on disk or on the
+ * network.
+ */
+void *consumer_thread_metadata_poll(void *data)
+{
+ int ret, i, pollfd;
+ uint32_t revents, nb_fd;
+ struct lttng_consumer_stream *stream = NULL;
+ struct lttng_ht_iter iter;
+ struct lttng_ht_node_ulong *node;
+ struct lttng_poll_event events;
+ struct lttng_consumer_local_data *ctx = data;
+ ssize_t len;