+ struct lttng_consumer_stream *stream;
+
+ if (ht == NULL) {
+ return;
+ }
+
+ rcu_read_lock();
+ cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, waitfd_node.node) {
+ ret = lttng_ht_del(ht, &iter);
+ assert(!ret);
+
+ call_rcu(&stream->waitfd_node.head, consumer_free_metadata_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;
+ }
+
+ rcu_read_lock();
+ iter.iter.node = &stream->waitfd_node.node;
+ ret = lttng_ht_del(ht, &iter);
+ assert(!ret);
+ rcu_read_unlock();
+
+ 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;
+ }
+
+ 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);
+
+ if (free_chan) {
+ consumer_del_channel(free_chan);
+ }
+
+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 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);
+
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ break;
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ ret = lttng_ustconsumer_add_stream(stream);
+ if (ret) {
+ ret = -EINVAL;
+ goto error;
+ }
+
+ /* Steal stream identifier only for UST */
+ consumer_steal_stream_key(stream->wait_fd, ht);
+ break;
+ default:
+ ERR("Unknown consumer_data type");
+ assert(0);
+ ret = -ENOSYS;
+ goto error;
+ }
+
+ /*
+ * 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();
+
+error:
+ 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)
+{
+ 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_ht *metadata_ht = NULL;
+ struct lttng_poll_event events;
+ struct lttng_consumer_local_data *ctx = data;
+ ssize_t len;
+
+ rcu_register_thread();
+
+ DBG("Thread metadata poll started");