+ health_code_update();
+
+ pthread_mutex_lock(&stream->lock);
+ stream->quiescent = false;
+ pthread_mutex_unlock(&stream->lock);
+ }
+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;
+ unsigned int channel_monitor;
+
+ 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);
+ pthread_mutex_lock(&channel->lock);
+ channel_monitor = channel->monitor;
+ if (cds_lfht_is_node_deleted(&channel->node.node)) {
+ goto error_unlock;
+ }
+
+ lttng_ustconsumer_close_metadata(channel);
+ pthread_mutex_unlock(&channel->lock);
+ pthread_mutex_unlock(&consumer_data.lock);
+
+ /*
+ * The ownership of a metadata channel depends on the type of
+ * session to which it belongs. In effect, the monitor flag is checked
+ * to determine if this metadata channel is in "snapshot" mode or not.
+ *
+ * In the non-snapshot case, the metadata channel is created along with
+ * a single stream which will remain present until the metadata channel
+ * is destroyed (on the destruction of its session). In this case, the
+ * metadata stream in "monitored" by the metadata poll thread and holds
+ * the ownership of its channel.
+ *
+ * Closing the metadata will cause the metadata stream's "metadata poll
+ * pipe" to be closed. Closing this pipe will wake-up the metadata poll
+ * thread which will teardown the metadata stream which, in return,
+ * deletes the metadata channel.
+ *
+ * In the snapshot case, the metadata stream is created and destroyed
+ * on every snapshot record. Since the channel doesn't have an owner
+ * other than the session daemon, it is safe to destroy it immediately
+ * on reception of the CLOSE_METADATA command.
+ */
+ if (!channel_monitor) {
+ /*
+ * The channel and consumer_data locks must be
+ * released before this call since consumer_del_channel
+ * re-acquires the channel and consumer_data locks to teardown
+ * the channel and queue its reclamation by the "call_rcu"
+ * worker thread.
+ */
+ consumer_del_channel(channel);
+ }
+
+ return ret;
+error_unlock:
+ pthread_mutex_unlock(&channel->lock);
+ 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_no_stream;
+ }
+
+ /* Send metadata stream to relayd if needed. */
+ if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) {
+ ret = consumer_send_relayd_stream(metadata->metadata_stream,
+ metadata->pathname);
+ if (ret < 0) {
+ ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
+ goto error;
+ }
+ ret = consumer_send_relayd_streams_sent(
+ metadata->metadata_stream->net_seq_idx);
+ if (ret < 0) {
+ ret = LTTCOMM_CONSUMERD_RELAYD_FAIL;
+ goto error;
+ }
+ }
+
+ /*
+ * Ownership of metadata stream is passed along. Freeing is handled by
+ * the callee.
+ */
+ 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 send_streams_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_stream_destroy(metadata->metadata_stream, NULL);
+ cds_list_del(&metadata->metadata_stream->send_node);
+ metadata->metadata_stream = NULL;
+send_streams_error:
+error_no_stream:
+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;
+ 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 %" PRIu64,
+ key);
+ ret = -1;
+ goto error;
+ }
+ assert(!metadata_channel->monitor);
+
+ health_code_update();
+
+ /*
+ * Ask the sessiond if we have new metadata waiting and update the
+ * consumer metadata cache.
+ */
+ ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 1);
+ if (ret < 0) {
+ goto error;
+ }
+
+ health_code_update();
+
+ /*
+ * 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, NULL);
+ if (ret < 0) {
+ goto error_stream;
+ }
+ metadata_stream->out_fd = ret;
+ metadata_stream->tracefile_size_current = 0;
+ }
+
+ do {
+ health_code_update();
+
+ ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
+ if (ret < 0) {
+ goto error_stream;
+ }
+ } while (ret > 0);
+
+error_stream:
+ /*
+ * Clean up the stream completly because the next snapshot will use a new
+ * metadata stream.
+ */
+ consumer_stream_destroy(metadata_stream, NULL);
+ cds_list_del(&metadata_stream->send_node);
+ 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,
+ uint64_t nb_packets_per_stream, 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 %" PRIu64, key);
+ ret = -1;
+ goto error;
+ }
+ assert(!channel->monitor);
+ DBG("UST consumer snapshot channel %" PRIu64, key);
+
+ cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+ health_code_update();
+
+ /* 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, NULL);
+ 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);
+ }
+
+ /*
+ * If tracing is active, we want to perform a "full" buffer flush.
+ * Else, if quiescent, it has already been done by the prior stop.
+ */
+ if (!stream->quiescent) {
+ ustctl_flush_buffer(stream->ustream, 0);
+ }
+
+ ret = lttng_ustconsumer_take_snapshot(stream);
+ if (ret < 0) {
+ ERR("Taking UST snapshot");
+ goto error_unlock;
+ }
+
+ ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
+ if (ret < 0) {
+ ERR("Produced UST snapshot position");
+ goto error_unlock;
+ }
+
+ ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
+ if (ret < 0) {
+ ERR("Consumerd UST snapshot position");
+ goto error_unlock;
+ }
+
+ /*
+ * The original value is sent back if max stream size is larger than
+ * the possible size of the snapshot. Also, we assume that the session
+ * daemon should never send a maximum stream size that is lower than
+ * subbuffer size.
+ */
+ consumed_pos = consumer_get_consume_start_pos(consumed_pos,
+ produced_pos, nb_packets_per_stream,
+ stream->max_sb_size);
+
+ while (consumed_pos < produced_pos) {
+ ssize_t read_len;
+ unsigned long len, padded_len;
+
+ health_code_update();
+
+ DBG("UST consumer taking snapshot at pos %lu", consumed_pos);
+
+ ret = ustctl_get_subbuf(stream->ustream, &consumed_pos);
+ if (ret < 0) {
+ if (ret != -EAGAIN) {
+ PERROR("ustctl_get_subbuf snapshot");
+ goto error_close_stream;
+ }
+ DBG("UST consumer get subbuf failed. Skipping it.");
+ consumed_pos += stream->max_sb_size;
+ stream->chan->lost_packets++;
+ continue;
+ }
+
+ ret = ustctl_get_subbuf_size(stream->ustream, &len);
+ if (ret < 0) {
+ ERR("Snapshot ustctl_get_subbuf_size");
+ goto error_put_subbuf;
+ }
+
+ ret = ustctl_get_padded_subbuf_size(stream->ustream, &padded_len);
+ if (ret < 0) {
+ ERR("Snapshot ustctl_get_padded_subbuf_size");
+ goto error_put_subbuf;
+ }
+
+ read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
+ padded_len - len, NULL);
+ if (use_relayd) {
+ if (read_len != len) {
+ ret = -EPERM;
+ goto error_put_subbuf;
+ }
+ } else {
+ if (read_len != padded_len) {
+ ret = -EPERM;
+ goto error_put_subbuf;
+ }
+ }
+
+ ret = ustctl_put_subbuf(stream->ustream);
+ if (ret < 0) {
+ ERR("Snapshot ustctl_put_subbuf");
+ goto error_close_stream;
+ }
+ consumed_pos += stream->max_sb_size;
+ }
+
+ /* Simply close the stream so we can use it on the next snapshot. */
+ consumer_stream_close(stream);
+ pthread_mutex_unlock(&stream->lock);
+ }
+
+ rcu_read_unlock();
+ return 0;
+
+error_put_subbuf:
+ if (ustctl_put_subbuf(stream->ustream) < 0) {
+ ERR("Snapshot ustctl_put_subbuf");
+ }
+error_close_stream:
+ consumer_stream_close(stream);
+error_unlock:
+ pthread_mutex_unlock(&stream->lock);
+error:
+ rcu_read_unlock();
+ return ret;
+}
+
+/*
+ * Receive the metadata updates from the sessiond. Supports receiving
+ * overlapping metadata, but is needs to always belong to a contiguous
+ * range starting from 0.
+ * Be careful about the locks held when calling this function: it needs
+ * the metadata cache flush to concurrently progress in order to
+ * complete.
+ */
+int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
+ uint64_t len, uint64_t version,
+ struct lttng_consumer_channel *channel, int timer, int wait)
+{
+ int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ char *metadata_str;
+
+ DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len);
+
+ metadata_str = zmalloc(len * sizeof(char));
+ if (!metadata_str) {
+ PERROR("zmalloc metadata string");
+ ret_code = LTTCOMM_CONSUMERD_ENOMEM;
+ goto end;
+ }
+
+ health_code_update();
+
+ /* Receive metadata string. */
+ ret = lttcomm_recv_unix_sock(sock, metadata_str, len);
+ if (ret < 0) {
+ /* Session daemon is dead so return gracefully. */
+ ret_code = ret;
+ goto end_free;
+ }
+
+ health_code_update();
+
+ pthread_mutex_lock(&channel->metadata_cache->lock);
+ ret = consumer_metadata_cache_write(channel, offset, len, version,
+ metadata_str);
+ if (ret < 0) {
+ /* Unable to handle metadata. Notify session daemon. */
+ ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
+ /*
+ * Skip metadata flush on write error since the offset and len might
+ * not have been updated which could create an infinite loop below when
+ * waiting for the metadata cache to be flushed.
+ */
+ pthread_mutex_unlock(&channel->metadata_cache->lock);
+ goto end_free;
+ }
+ pthread_mutex_unlock(&channel->metadata_cache->lock);
+
+ if (!wait) {
+ goto end_free;
+ }
+ while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
+ DBG("Waiting for metadata to be flushed");
+
+ health_code_update();
+
+ usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
+ }
+
+end_free:
+ free(metadata_str);
+end:
+ return ret_code;
+}
+
+/*
+ * Receive command from session daemon and process it.
+ *
+ * Return 1 on success else a negative value or 0.
+ */
+int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
+ int sock, struct pollfd *consumer_sockpoll)
+{
+ ssize_t ret;
+ enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ struct lttcomm_consumer_msg msg;
+ struct lttng_consumer_channel *channel = NULL;
+
+ health_code_update();
+
+ ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
+ if (ret != sizeof(msg)) {
+ DBG("Consumer received unexpected message size %zd (expects %zu)",
+ ret, sizeof(msg));
+ /*
+ * The ret value might 0 meaning an orderly shutdown but this is ok
+ * since the caller handles this.
+ */
+ if (ret > 0) {
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
+ ret = -1;
+ }
+ return ret;
+ }
+
+ health_code_update();
+
+ /* deprecated */
+ assert(msg.cmd_type != LTTNG_CONSUMER_STOP);
+
+ health_code_update();
+
+ /* relayd needs RCU read-side lock */
+ rcu_read_lock();
+
+ switch (msg.cmd_type) {
+ case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
+ {
+ /* Session daemon status message are handled in the following call. */
+ consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
+ msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
+ &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
+ msg.u.relayd_sock.relayd_session_id);
+ goto end_nosignal;
+ }
+ case LTTNG_CONSUMER_DESTROY_RELAYD:
+ {
+ uint64_t index = msg.u.destroy_relayd.net_seq_idx;
+ struct consumer_relayd_sock_pair *relayd;
+
+ DBG("UST consumer destroying relayd %" PRIu64, index);
+
+ /* Get relayd reference if exists. */
+ relayd = consumer_find_relayd(index);
+ if (relayd == NULL) {
+ DBG("Unable to find relayd %" PRIu64, index);
+ ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
+ }
+
+ /*
+ * Each relayd socket pair has a refcount of stream attached to it
+ * which tells if the relayd is still active or not depending on the
+ * refcount value.
+ *
+ * This will set the destroy flag of the relayd object and destroy it
+ * if the refcount reaches zero when called.
+ *
+ * The destroy can happen either here or when a stream fd hangs up.
+ */
+ if (relayd) {
+ consumer_flag_relayd_for_destroy(relayd);
+ }
+
+ goto end_msg_sessiond;
+ }
+ case LTTNG_CONSUMER_UPDATE_STREAM:
+ {
+ rcu_read_unlock();
+ return -ENOSYS;
+ }
+ case LTTNG_CONSUMER_DATA_PENDING:
+ {
+ int ret, is_data_pending;
+ uint64_t id = msg.u.data_pending.session_id;
+
+ DBG("UST consumer data pending command for id %" PRIu64, id);
+
+ is_data_pending = consumer_data_pending(id);
+
+ /* Send back returned value to session daemon */
+ ret = lttcomm_send_unix_sock(sock, &is_data_pending,
+ sizeof(is_data_pending));
+ if (ret < 0) {
+ DBG("Error when sending the data pending ret code: %d", ret);
+ goto error_fatal;
+ }