+
+ channel->wait_fd = ustctl_channel_get_wait_fd(channel->uchan);
+
+ /*
+ * For the snapshots (no monitor), we create the metadata streams
+ * on demand, not during the channel creation.
+ */
+ if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && !channel->monitor) {
+ ret = 0;
+ goto end;
+ }
+
+ /* Open all streams for this channel. */
+ ret = create_ust_streams(channel, ctx);
+ if (ret < 0) {
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * Send all stream of a channel to the right thread handling it.
+ *
+ * On error, return a negative value else 0 on success.
+ */
+static int send_streams_to_thread(struct lttng_consumer_channel *channel,
+ struct lttng_consumer_local_data *ctx)
+{
+ int ret = 0;
+ struct lttng_consumer_stream *stream, *stmp;
+
+ assert(channel);
+ assert(ctx);
+
+ /* Send streams to the corresponding thread. */
+ cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
+ send_node) {
+ /* Sending the stream to the thread. */
+ ret = send_stream_to_thread(stream, ctx);
+ if (ret < 0) {
+ /*
+ * If we are unable to send the stream to the thread, there is
+ * a big problem so just stop everything.
+ */
+ goto error;
+ }
+
+ /* Remove node from the channel stream list. */
+ cds_list_del(&stream->send_node);
+ }
+
+error:
+ return ret;
+}
+
+/*
+ * Write metadata to the given channel using ustctl to convert the string to
+ * the ringbuffer.
+ * Called only from consumer_metadata_cache_write.
+ * The metadata cache lock MUST be acquired to write in the cache.
+ *
+ * Return 0 on success else a negative value.
+ */
+int lttng_ustconsumer_push_metadata(struct lttng_consumer_channel *metadata,
+ const char *metadata_str, uint64_t target_offset, uint64_t len)
+{
+ int ret;
+
+ assert(metadata);
+ assert(metadata_str);
+
+ DBG("UST consumer writing metadata to channel %s", metadata->name);
+
+ if (!metadata->metadata_stream) {
+ ret = 0;
+ goto error;
+ }
+
+ assert(target_offset <= metadata->metadata_cache->max_offset);
+ ret = ustctl_write_metadata_to_channel(metadata->uchan,
+ metadata_str + target_offset, len);
+ if (ret < 0) {
+ ERR("ustctl write metadata fail with ret %d, len %" PRIu64, ret, len);
+ goto error;
+ }
+
+ ustctl_flush_buffer(metadata->metadata_stream->ustream, 1);
+
+error:
+ return ret;
+}
+
+/*
+ * Flush channel's streams using the given key to retrieve the channel.
+ *
+ * Return 0 on success else an LTTng error code.
+ */
+static int flush_channel(uint64_t chan_key)
+{
+ int ret = 0;
+ struct lttng_consumer_channel *channel;
+ struct lttng_consumer_stream *stream;
+ struct lttng_ht *ht;
+ struct lttng_ht_iter iter;
+
+ DBG("UST consumer flush channel key %" PRIu64, chan_key);
+
+ rcu_read_lock();
+ channel = consumer_find_channel(chan_key);
+ if (!channel) {
+ ERR("UST consumer flush channel %" PRIu64 " not found", chan_key);
+ ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+ goto error;
+ }
+
+ ht = consumer_data.stream_per_chan_id_ht;
+
+ /* For each stream of the channel id, flush it. */
+ 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) {
+ ustctl_flush_buffer(stream->ustream, 1);
+ }
+error:
+ rcu_read_unlock();
+ return ret;
+}
+
+/*
+ * Close metadata stream wakeup_fd using the given key to retrieve the channel.
+ * RCU read side lock MUST be acquired before calling this function.
+ *
+ * Return 0 on success else an LTTng error code.
+ */
+static int close_metadata(uint64_t chan_key)
+{
+ int ret = 0;
+ struct lttng_consumer_channel *channel;
+
+ DBG("UST consumer close metadata key %" PRIu64, chan_key);
+
+ channel = consumer_find_channel(chan_key);
+ if (!channel) {
+ /*
+ * This is possible if the metadata thread has issue a delete because
+ * the endpoint point of the stream hung up. There is no way the
+ * session daemon can know about it thus use a DBG instead of an actual
+ * error.
+ */
+ DBG("UST consumer close metadata %" PRIu64 " not found", chan_key);
+ ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+ goto error;
+ }
+
+ pthread_mutex_lock(&consumer_data.lock);
+
+ if (cds_lfht_is_node_deleted(&channel->node.node)) {
+ goto error_unlock;
+ }
+
+ if (channel->switch_timer_enabled == 1) {
+ DBG("Deleting timer on metadata channel");
+ consumer_timer_switch_stop(channel);
+ }
+
+ if (channel->metadata_stream) {
+ ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream);
+ if (ret < 0) {
+ ERR("UST consumer unable to close fd of metadata (ret: %d)", ret);
+ ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
+ goto error_unlock;
+ }
+ }
+
+error_unlock:
+ pthread_mutex_unlock(&consumer_data.lock);
+error:
+ return ret;
+}
+
+/*
+ * RCU read side lock MUST be acquired before calling this function.
+ *
+ * Return 0 on success else an LTTng error code.
+ */
+static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
+{
+ int ret;
+ struct lttng_consumer_channel *metadata;
+
+ DBG("UST consumer setup metadata key %" PRIu64, key);
+
+ metadata = consumer_find_channel(key);
+ if (!metadata) {
+ ERR("UST consumer push metadata %" PRIu64 " not found", key);
+ ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+ goto end;
+ }
+
+ /*
+ * In no monitor mode, the metadata channel has no stream(s) so skip the
+ * ownership transfer to the metadata thread.
+ */
+ if (!metadata->monitor) {
+ DBG("Metadata channel in no monitor");
+ ret = 0;
+ goto end;
+ }
+
+ /*
+ * Send metadata stream to relayd if one available. Availability is
+ * known if the stream is still in the list of the channel.
+ */
+ if (cds_list_empty(&metadata->streams.head)) {
+ ERR("Metadata channel key %" PRIu64 ", no stream available.", key);
+ ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
+ goto error;
+ }
+
+ /* Send metadata stream to relayd if needed. */
+ ret = send_stream_to_relayd(metadata->metadata_stream);
+ if (ret < 0) {
+ ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
+ goto error;
+ }
+
+ ret = send_streams_to_thread(metadata, ctx);
+ if (ret < 0) {
+ /*
+ * If we are unable to send the stream to the thread, there is
+ * a big problem so just stop everything.
+ */
+ ret = LTTCOMM_CONSUMERD_FATAL;
+ goto error;
+ }
+ /* List MUST be empty after or else it could be reused. */
+ assert(cds_list_empty(&metadata->streams.head));
+
+ ret = 0;
+ goto end;
+
+error:
+ /*
+ * Delete metadata channel on error. At this point, the metadata stream can
+ * NOT be monitored by the metadata thread thus having the guarantee that
+ * the stream is still in the local stream list of the channel. This call
+ * will make sure to clean that list.
+ */
+ consumer_del_channel(metadata);
+end:
+ return ret;
+}
+
+/*
+ * Snapshot the whole metadata.
+ *
+ * Returns 0 on success, < 0 on error
+ */
+static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
+ struct lttng_consumer_local_data *ctx)
+{
+ int ret = 0;
+ ssize_t write_len;
+ uint64_t total_len = 0;
+ struct lttng_consumer_channel *metadata_channel;
+ struct lttng_consumer_stream *metadata_stream;
+
+ assert(path);
+ assert(ctx);
+
+ DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s",
+ key, path);
+
+ rcu_read_lock();
+
+ metadata_channel = consumer_find_channel(key);
+ if (!metadata_channel) {
+ ERR("UST snapshot metadata channel not found for key %lu", key);
+ ret = -1;
+ goto error;
+ }
+ assert(!metadata_channel->monitor);
+
+ /*
+ * Ask the sessiond if we have new metadata waiting and update the
+ * consumer metadata cache.
+ */
+ ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel);
+ if (ret < 0) {
+ goto error;
+ }
+
+ /*
+ * The metadata stream is NOT created in no monitor mode when the channel
+ * is created on a sessiond ask channel command.
+ */
+ ret = create_ust_streams(metadata_channel, ctx);
+ if (ret < 0) {
+ goto error;
+ }
+
+ metadata_stream = metadata_channel->metadata_stream;
+ assert(metadata_stream);
+
+ if (relayd_id != (uint64_t) -1ULL) {
+ metadata_stream->net_seq_idx = relayd_id;
+ ret = consumer_send_relayd_stream(metadata_stream, path);
+ if (ret < 0) {
+ goto error_stream;
+ }
+ } else {
+ ret = utils_create_stream_file(path, metadata_stream->name,
+ metadata_stream->chan->tracefile_size,
+ metadata_stream->tracefile_count_current,
+ metadata_stream->uid, metadata_stream->gid);
+ if (ret < 0) {
+ goto error_stream;
+ }
+ metadata_stream->out_fd = ret;
+ metadata_stream->tracefile_size_current = 0;
+ }
+
+ pthread_mutex_lock(&metadata_channel->metadata_cache->lock);
+ while (total_len < metadata_channel->metadata_cache->total_bytes_written) {
+ /*
+ * Write at most one packet of metadata into the channel
+ * to avoid blocking here.
+ */
+ write_len = ustctl_write_one_packet_to_channel(metadata_channel->uchan,
+ metadata_channel->metadata_cache->data,
+ metadata_channel->metadata_cache->total_bytes_written);
+ if (write_len < 0) {
+ ERR("UST consumer snapshot writing metadata packet");
+ ret = -1;
+ goto error_unlock;
+ }
+ total_len += write_len;
+
+ DBG("Written %" PRIu64 " bytes to metadata (left: %" PRIu64 ")",
+ write_len,
+ metadata_channel->metadata_cache->total_bytes_written - write_len);
+ ustctl_flush_buffer(metadata_stream->ustream, 1);
+ ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
+ if (ret < 0) {
+ goto error_unlock;
+ }
+ }
+
+error_unlock:
+ pthread_mutex_unlock(&metadata_channel->metadata_cache->lock);
+
+error_stream:
+ /*
+ * Clean up the stream completly because the next snapshot will use a new
+ * metadata stream.
+ */
+ cds_list_del(&metadata_stream->send_node);
+ consumer_stream_destroy(metadata_stream, NULL);
+ metadata_channel->metadata_stream = NULL;
+
+error:
+ rcu_read_unlock();
+ return ret;
+}
+
+/*
+ * Take a snapshot of all the stream of a channel.
+ *
+ * Returns 0 on success, < 0 on error
+ */
+static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
+ struct lttng_consumer_local_data *ctx)
+{
+ int ret;
+ unsigned use_relayd = 0;
+ unsigned long consumed_pos, produced_pos;
+ struct lttng_consumer_channel *channel;
+ struct lttng_consumer_stream *stream;
+
+ assert(path);
+ assert(ctx);
+
+ rcu_read_lock();
+
+ if (relayd_id != (uint64_t) -1ULL) {
+ use_relayd = 1;
+ }
+
+ channel = consumer_find_channel(key);
+ if (!channel) {
+ ERR("UST snapshot channel not found for key %lu", key);
+ ret = -1;
+ goto error;
+ }
+ assert(!channel->monitor);
+ DBG("UST consumer snapshot channel %lu", key);
+
+ cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+ /* Lock stream because we are about to change its state. */
+ pthread_mutex_lock(&stream->lock);
+ stream->net_seq_idx = relayd_id;
+
+ if (use_relayd) {
+ ret = consumer_send_relayd_stream(stream, path);
+ if (ret < 0) {
+ goto error_unlock;
+ }
+ } else {
+ ret = utils_create_stream_file(path, stream->name,
+ stream->chan->tracefile_size,
+ stream->tracefile_count_current,
+ stream->uid, stream->gid);
+ if (ret < 0) {
+ goto error_unlock;
+ }
+ stream->out_fd = ret;
+ stream->tracefile_size_current = 0;
+
+ DBG("UST consumer snapshot stream %s/%s (%" PRIu64 ")", path,
+ stream->name, stream->key);
+ }
+
+ ustctl_flush_buffer(stream->ustream, 1);
+
+ ret = lttng_ustconsumer_take_snapshot(stream);
+ if (ret < 0) {
+ ERR("Taking UST snapshot");
+ goto error_unlock;