+void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan)
+{
+ assert(chan);
+ assert(chan->uchan);
+ assert(chan->buffer_credentials.is_set);
+
+ consumer_metadata_cache_destroy(chan);
+ ustctl_destroy_channel(chan->uchan);
+ /* Try to rmdir all directories under shm_path root. */
+ if (chan->root_shm_path[0]) {
+ (void) run_as_rmdir_recursive(chan->root_shm_path,
+ chan->buffer_credentials.value.uid,
+ chan->buffer_credentials.value.gid,
+ LTTNG_DIRECTORY_HANDLE_SKIP_NON_EMPTY_FLAG);
+ }
+ free(chan->stream_fds);
+}
+
+void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ if (stream->chan->switch_timer_enabled == 1) {
+ consumer_timer_switch_stop(stream->chan);
+ }
+ ustctl_destroy_stream(stream->ustream);
+}
+
+int lttng_ustconsumer_get_wakeup_fd(struct lttng_consumer_stream *stream)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ return ustctl_stream_get_wakeup_fd(stream->ustream);
+}
+
+int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ return ustctl_stream_close_wakeup_fd(stream->ustream);
+}
+
+static
+void metadata_stream_reset_cache_consumed_position(
+ struct lttng_consumer_stream *stream)
+{
+ DBG("Reset metadata cache of session %" PRIu64,
+ stream->chan->session_id);
+ stream->ust_metadata_pushed = 0;
+}
+
+/*
+ * Write up to one packet from the metadata cache to the channel.
+ *
+ * Returns the number of bytes pushed from the cache into the ring buffer, or a
+ * negative value on error.
+ */
+static
+int commit_one_metadata_packet(struct lttng_consumer_stream *stream)
+{
+ ssize_t write_len;
+ int ret;
+
+ pthread_mutex_lock(&stream->chan->metadata_cache->lock);
+ if (stream->chan->metadata_cache->max_offset ==
+ stream->ust_metadata_pushed) {
+ /*
+ * In the context of a user space metadata channel, a
+ * change in version can be detected in two ways:
+ * 1) During the pre-consume of the `read_subbuffer` loop,
+ * 2) When populating the metadata ring buffer (i.e. here).
+ *
+ * This function is invoked when there is no metadata
+ * available in the ring-buffer. If all data was consumed
+ * up to the size of the metadata cache, there is no metadata
+ * to insert in the ring-buffer.
+ *
+ * However, the metadata version could still have changed (a
+ * regeneration without any new data will yield the same cache
+ * size).
+ *
+ * The cache's version is checked for a version change and the
+ * consumed position is reset if one occurred.
+ *
+ * This check is only necessary for the user space domain as
+ * it has to manage the cache explicitly. If this reset was not
+ * performed, no metadata would be consumed (and no reset would
+ * occur as part of the pre-consume) until the metadata size
+ * exceeded the cache size.
+ */
+ if (stream->metadata_version !=
+ stream->chan->metadata_cache->version) {
+ metadata_stream_reset_cache_consumed_position(stream);
+ consumer_stream_metadata_set_version(stream,
+ stream->chan->metadata_cache->version);
+ } else {
+ ret = 0;
+ goto end;
+ }
+ }
+
+ write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan,
+ &stream->chan->metadata_cache->data[stream->ust_metadata_pushed],
+ stream->chan->metadata_cache->max_offset
+ - stream->ust_metadata_pushed);
+ assert(write_len != 0);
+ if (write_len < 0) {
+ ERR("Writing one metadata packet");
+ ret = write_len;
+ goto end;
+ }
+ stream->ust_metadata_pushed += write_len;
+
+ assert(stream->chan->metadata_cache->max_offset >=
+ stream->ust_metadata_pushed);
+ ret = write_len;
+
+ /*
+ * Switch packet (but don't open the next one) on every commit of
+ * a metadata packet. Since the subbuffer is fully filled (with padding,
+ * if needed), the stream is "quiescent" after this commit.
+ */
+ ustctl_flush_buffer(stream->ustream, 1);
+ stream->quiescent = true;
+end:
+ pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
+ return ret;
+}
+
+
+/*
+ * Sync metadata meaning request them to the session daemon and snapshot to the
+ * metadata thread can consumer them.
+ *
+ * Metadata stream lock is held here, but we need to release it when
+ * interacting with sessiond, else we cause a deadlock with live
+ * awaiting on metadata to be pushed out.
+ *
+ * The RCU read side lock must be held by the caller.
+ */
+enum sync_metadata_status lttng_ustconsumer_sync_metadata(
+ struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *metadata_stream)
+{
+ int ret;
+ enum sync_metadata_status status;
+ struct lttng_consumer_channel *metadata_channel;
+
+ assert(ctx);
+ assert(metadata_stream);
+
+ metadata_channel = metadata_stream->chan;
+ pthread_mutex_unlock(&metadata_stream->lock);
+ /*
+ * Request metadata from the sessiond, but don't wait for the flush
+ * because we locked the metadata thread.
+ */
+ ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 0);
+ pthread_mutex_lock(&metadata_stream->lock);
+ if (ret < 0) {
+ status = SYNC_METADATA_STATUS_ERROR;
+ goto end;
+ }
+
+ /*
+ * The metadata stream and channel can be deleted while the
+ * metadata stream lock was released. The streamed is checked
+ * for deletion before we use it further.
+ *
+ * Note that it is safe to access a logically-deleted stream since its
+ * existence is still guaranteed by the RCU read side lock. However,
+ * it should no longer be used. The close/deletion of the metadata
+ * channel and stream already guarantees that all metadata has been
+ * consumed. Therefore, there is nothing left to do in this function.
+ */
+ if (consumer_stream_is_deleted(metadata_stream)) {
+ DBG("Metadata stream %" PRIu64 " was deleted during the metadata synchronization",
+ metadata_stream->key);
+ status = SYNC_METADATA_STATUS_NO_DATA;
+ goto end;
+ }
+
+ ret = commit_one_metadata_packet(metadata_stream);
+ if (ret < 0) {
+ status = SYNC_METADATA_STATUS_ERROR;
+ goto end;
+ } else if (ret > 0) {
+ status = SYNC_METADATA_STATUS_NEW_DATA;
+ } else /* ret == 0 */ {
+ status = SYNC_METADATA_STATUS_NO_DATA;
+ goto end;
+ }
+
+ ret = ustctl_snapshot(metadata_stream->ustream);
+ if (ret < 0) {
+ ERR("Failed to take a snapshot of the metadata ring-buffer positions, ret = %d", ret);
+ status = SYNC_METADATA_STATUS_ERROR;
+ goto end;
+ }
+
+end:
+ return status;
+}
+
+/*
+ * Return 0 on success else a negative value.
+ */
+static int notify_if_more_data(struct lttng_consumer_stream *stream,
+ struct lttng_consumer_local_data *ctx)
+{
+ int ret;
+ struct ustctl_consumer_stream *ustream;
+
+ assert(stream);
+ assert(ctx);
+
+ ustream = stream->ustream;
+
+ /*
+ * First, we are going to check if there is a new subbuffer available
+ * before reading the stream wait_fd.
+ */
+ /* Get the next subbuffer */
+ ret = ustctl_get_next_subbuf(ustream);
+ if (ret) {
+ /* No more data found, flag the stream. */
+ stream->has_data = 0;
+ ret = 0;
+ goto end;
+ }
+
+ ret = ustctl_put_subbuf(ustream);
+ assert(!ret);
+
+ /* This stream still has data. Flag it and wake up the data thread. */
+ stream->has_data = 1;
+
+ if (stream->monitor && !stream->hangup_flush_done && !ctx->has_wakeup) {
+ ssize_t writelen;
+
+ writelen = lttng_pipe_write(ctx->consumer_wakeup_pipe, "!", 1);
+ if (writelen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
+ ret = writelen;
+ goto end;
+ }
+
+ /* The wake up pipe has been notified. */
+ ctx->has_wakeup = 1;
+ }
+ ret = 0;
+
+end:
+ return ret;
+}
+
+static int consumer_stream_ust_on_wake_up(struct lttng_consumer_stream *stream)
+{
+ int ret = 0;
+
+ /*
+ * We can consume the 1 byte written into the wait_fd by
+ * UST. Don't trigger error if we cannot read this one byte
+ * (read returns 0), or if the error is EAGAIN or EWOULDBLOCK.
+ *
+ * This is only done when the stream is monitored by a thread,
+ * before the flush is done after a hangup and if the stream
+ * is not flagged with data since there might be nothing to
+ * consume in the wait fd but still have data available
+ * flagged by the consumer wake up pipe.
+ */
+ if (stream->monitor && !stream->hangup_flush_done && !stream->has_data) {
+ char dummy;
+ ssize_t readlen;
+
+ readlen = lttng_read(stream->wait_fd, &dummy, 1);
+ if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
+ ret = readlen;
+ }
+ }
+
+ return ret;
+}
+
+static int extract_common_subbuffer_info(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuf)
+{
+ int ret;
+
+ ret = ustctl_get_subbuf_size(
+ stream->ustream, &subbuf->info.data.subbuf_size);
+ if (ret) {
+ goto end;
+ }
+
+ ret = ustctl_get_padded_subbuf_size(
+ stream->ustream, &subbuf->info.data.padded_subbuf_size);
+ if (ret) {
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+static int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuf)
+{
+ int ret;
+
+ ret = extract_common_subbuffer_info(stream, subbuf);
+ if (ret) {
+ goto end;
+ }
+
+ subbuf->info.metadata.version = stream->metadata_version;
+
+end:
+ return ret;
+}
+
+static int extract_data_subbuffer_info(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuf)
+{
+ int ret;
+
+ ret = extract_common_subbuffer_info(stream, subbuf);
+ if (ret) {
+ goto end;
+ }
+
+ ret = ustctl_get_packet_size(
+ stream->ustream, &subbuf->info.data.packet_size);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer packet size");
+ goto end;
+ }
+
+ ret = ustctl_get_content_size(
+ stream->ustream, &subbuf->info.data.content_size);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer content size");
+ goto end;
+ }
+
+ ret = ustctl_get_timestamp_begin(
+ stream->ustream, &subbuf->info.data.timestamp_begin);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer begin timestamp");
+ goto end;
+ }
+
+ ret = ustctl_get_timestamp_end(
+ stream->ustream, &subbuf->info.data.timestamp_end);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer end timestamp");
+ goto end;
+ }
+
+ ret = ustctl_get_events_discarded(
+ stream->ustream, &subbuf->info.data.events_discarded);
+ if (ret) {
+ PERROR("Failed to get sub-buffer events discarded count");
+ goto end;
+ }
+
+ ret = ustctl_get_sequence_number(stream->ustream,
+ &subbuf->info.data.sequence_number.value);
+ if (ret) {
+ /* May not be supported by older LTTng-modules. */
+ if (ret != -ENOTTY) {
+ PERROR("Failed to get sub-buffer sequence number");
+ goto end;
+ }
+ } else {
+ subbuf->info.data.sequence_number.is_set = true;
+ }
+
+ ret = ustctl_get_stream_id(
+ stream->ustream, &subbuf->info.data.stream_id);
+ if (ret < 0) {
+ PERROR("Failed to get stream id");
+ goto end;
+ }
+
+ ret = ustctl_get_instance_id(stream->ustream,
+ &subbuf->info.data.stream_instance_id.value);
+ if (ret) {
+ /* May not be supported by older LTTng-modules. */
+ if (ret != -ENOTTY) {
+ PERROR("Failed to get stream instance id");
+ goto end;
+ }
+ } else {
+ subbuf->info.data.stream_instance_id.is_set = true;
+ }
+end:
+ return ret;
+}
+
+static int get_next_subbuffer_common(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuffer)
+{
+ int ret;
+ const char *addr;
+
+ ret = stream->read_subbuffer_ops.extract_subbuffer_info(
+ stream, subbuffer);
+ if (ret) {
+ goto end;
+ }
+
+ ret = get_current_subbuf_addr(stream, &addr);
+ if (ret) {
+ goto end;
+ }
+
+ subbuffer->buffer.buffer = lttng_buffer_view_init(
+ addr, 0, subbuffer->info.data.padded_subbuf_size);
+ assert(subbuffer->buffer.buffer.data != NULL);
+end:
+ return ret;
+}
+
+static int get_next_subbuffer(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuffer)
+{
+ int ret;
+
+ ret = ustctl_get_next_subbuf(stream->ustream);
+ if (ret) {
+ goto end;
+ }
+
+ ret = get_next_subbuffer_common(stream, subbuffer);
+ if (ret) {
+ goto end;
+ }
+end:
+ return ret;
+}
+
+static int get_next_subbuffer_metadata(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuffer)
+{
+ int ret;
+ bool cache_empty;
+ bool got_subbuffer;
+ bool coherent;
+ bool buffer_empty;
+ unsigned long consumed_pos, produced_pos;
+
+ do {
+ ret = ustctl_get_next_subbuf(stream->ustream);
+ if (ret == 0) {
+ got_subbuffer = true;
+ } else {
+ got_subbuffer = false;
+ if (ret != -EAGAIN) {
+ /* Fatal error. */
+ goto end;
+ }
+ }
+
+ /*
+ * Determine if the cache is empty and ensure that a sub-buffer
+ * is made available if the cache is not empty.
+ */
+ if (!got_subbuffer) {
+ ret = commit_one_metadata_packet(stream);
+ if (ret < 0 && ret != -ENOBUFS) {
+ goto end;
+ } else if (ret == 0) {
+ /* Not an error, the cache is empty. */
+ cache_empty = true;
+ ret = -ENODATA;
+ goto end;
+ } else {
+ cache_empty = false;
+ }
+ } else {
+ pthread_mutex_lock(&stream->chan->metadata_cache->lock);
+ cache_empty = stream->chan->metadata_cache->max_offset ==
+ stream->ust_metadata_pushed;
+ pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
+ }
+ } while (!got_subbuffer);
+
+ /* Populate sub-buffer infos and view. */
+ ret = get_next_subbuffer_common(stream, subbuffer);
+ if (ret) {
+ goto end;
+ }
+
+ ret = lttng_ustconsumer_sample_snapshot_positions(stream);
+ if (ret < 0) {
+ /*
+ * -EAGAIN is not expected since we got a sub-buffer and haven't
+ * pushed the consumption position yet (on put_next).
+ */
+ PERROR("Failed to take a snapshot of metadata buffer positions");
+ goto end;
+ }
+
+ ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
+ if (ret) {
+ PERROR("Failed to get metadata consumed position");
+ goto end;
+ }
+
+ ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
+ if (ret) {
+ PERROR("Failed to get metadata produced position");
+ goto end;
+ }
+
+ /* Last sub-buffer of the ring buffer ? */
+ buffer_empty = (consumed_pos + stream->max_sb_size) == produced_pos;
+
+ /*
+ * The sessiond registry lock ensures that coherent units of metadata
+ * are pushed to the consumer daemon at once. Hence, if a sub-buffer is
+ * acquired, the cache is empty, and it is the only available sub-buffer
+ * available, it is safe to assume that it is "coherent".
+ */
+ coherent = got_subbuffer && cache_empty && buffer_empty;
+
+ LTTNG_OPTIONAL_SET(&subbuffer->info.metadata.coherent, coherent);
+end:
+ return ret;
+}
+
+static int put_next_subbuffer(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuffer)
+{
+ const int ret = ustctl_put_next_subbuf(stream->ustream);
+
+ assert(ret == 0);
+ return ret;
+}
+
+static int signal_metadata(struct lttng_consumer_stream *stream,
+ struct lttng_consumer_local_data *ctx)
+{
+ return pthread_cond_broadcast(&stream->metadata_rdv) ? -errno : 0;
+}
+
+static int lttng_ustconsumer_set_stream_ops(
+ struct lttng_consumer_stream *stream)
+{
+ int ret = 0;
+
+ stream->read_subbuffer_ops.on_wake_up = consumer_stream_ust_on_wake_up;
+ if (stream->metadata_flag) {
+ stream->read_subbuffer_ops.get_next_subbuffer =
+ get_next_subbuffer_metadata;
+ stream->read_subbuffer_ops.extract_subbuffer_info =
+ extract_metadata_subbuffer_info;
+ stream->read_subbuffer_ops.reset_metadata =
+ metadata_stream_reset_cache_consumed_position;
+ if (stream->chan->is_live) {
+ stream->read_subbuffer_ops.on_sleep = signal_metadata;
+ ret = consumer_stream_enable_metadata_bucketization(
+ stream);
+ if (ret) {
+ goto end;
+ }
+ }
+ } else {
+ stream->read_subbuffer_ops.get_next_subbuffer =
+ get_next_subbuffer;
+ stream->read_subbuffer_ops.extract_subbuffer_info =
+ extract_data_subbuffer_info;
+ stream->read_subbuffer_ops.on_sleep = notify_if_more_data;
+ if (stream->chan->is_live) {
+ stream->read_subbuffer_ops.send_live_beacon =
+ consumer_flush_ust_index;
+ }
+ }
+
+ stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer;
+end:
+ return ret;
+}
+
+/*
+ * Called when a stream is created.
+ *
+ * Return 0 on success or else a negative value.
+ */
+int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
+{
+ int ret;
+
+ assert(stream);
+
+ /*
+ * Don't create anything if this is set for streaming or if there is
+ * no current trace chunk on the parent channel.
+ */
+ if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor &&
+ stream->chan->trace_chunk) {
+ ret = consumer_stream_create_output_files(stream, true);
+ if (ret) {
+ goto error;
+ }
+ }
+
+ lttng_ustconsumer_set_stream_ops(stream);
+ ret = 0;
+
+error:
+ return ret;
+}
+
+/*
+ * Check if data is still being extracted from the buffers for a specific
+ * stream. Consumer data lock MUST be acquired before calling this function