+ /* We have to send a status channel message indicating an error. */
+ ret = consumer_send_status_channel(sock, NULL);
+ if (ret < 0) {
+ /* Stop everything if session daemon can not be notified. */
+ goto error_fatal;
+ }
+ ret = 1;
+ goto end;
+
+error_fatal:
+ /* This will issue a consumer stop. */
+ ret = -1;
+ goto end;
+
+end:
+ rcu_read_unlock();
+ health_code_update();
+ return ret;
+}
+
+void lttng_ustctl_flush_buffer(struct lttng_consumer_stream *stream,
+ int producer_active)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ ustctl_flush_buffer(stream->ustream, producer_active);
+}
+
+/*
+ * Take a snapshot for a specific stream.
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ return ustctl_snapshot(stream->ustream);
+}
+
+/*
+ * Sample consumed and produced positions for a specific stream.
+ *
+ * Returns 0 on success, < 0 on error.
+ */
+int lttng_ustconsumer_sample_snapshot_positions(
+ struct lttng_consumer_stream *stream)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ return ustctl_snapshot_sample_positions(stream->ustream);
+}
+
+/*
+ * Get the produced position
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int lttng_ustconsumer_get_produced_snapshot(
+ struct lttng_consumer_stream *stream, unsigned long *pos)
+{
+ assert(stream);
+ assert(stream->ustream);
+ assert(pos);
+
+ return ustctl_snapshot_get_produced(stream->ustream, pos);
+}
+
+/*
+ * Get the consumed position
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int lttng_ustconsumer_get_consumed_snapshot(
+ struct lttng_consumer_stream *stream, unsigned long *pos)
+{
+ assert(stream);
+ assert(stream->ustream);
+ assert(pos);
+
+ return ustctl_snapshot_get_consumed(stream->ustream, pos);
+}
+
+void lttng_ustconsumer_flush_buffer(struct lttng_consumer_stream *stream,
+ int producer)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ ustctl_flush_buffer(stream->ustream, producer);
+}
+
+void lttng_ustconsumer_clear_buffer(struct lttng_consumer_stream *stream)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ ustctl_clear_buffer(stream->ustream);
+}
+
+int lttng_ustconsumer_get_current_timestamp(
+ struct lttng_consumer_stream *stream, uint64_t *ts)
+{
+ assert(stream);
+ assert(stream->ustream);
+ assert(ts);
+
+ return ustctl_get_current_timestamp(stream->ustream, ts);
+}
+
+int lttng_ustconsumer_get_sequence_number(
+ struct lttng_consumer_stream *stream, uint64_t *seq)
+{
+ assert(stream);
+ assert(stream->ustream);
+ assert(seq);
+
+ return ustctl_get_sequence_number(stream->ustream, seq);
+}
+
+/*
+ * Called when the stream signals the consumer that it has hung up.
+ */
+void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ pthread_mutex_lock(&stream->lock);
+ if (!stream->quiescent) {
+ ustctl_flush_buffer(stream->ustream, 0);
+ stream->quiescent = true;
+ }
+ pthread_mutex_unlock(&stream->lock);
+ stream->hangup_flush_done = 1;
+}
+
+void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
+{
+ int i;
+
+ assert(chan);
+ assert(chan->uchan);
+ assert(chan->buffer_credentials.is_set);
+
+ if (chan->switch_timer_enabled == 1) {
+ consumer_timer_switch_stop(chan);
+ }
+ for (i = 0; i < chan->nr_stream_fds; i++) {
+ int ret;
+
+ ret = close(chan->stream_fds[i]);
+ if (ret) {
+ PERROR("close");
+ }
+ if (chan->shm_path[0]) {
+ char shm_path[PATH_MAX];
+
+ ret = get_stream_shm_path(shm_path, chan->shm_path, i);
+ if (ret) {
+ ERR("Cannot get stream shm path");
+ }
+ ret = run_as_unlink(shm_path,
+ chan->buffer_credentials.value.uid,
+ chan->buffer_credentials.value.gid);
+ if (ret) {
+ PERROR("unlink %s", shm_path);
+ }
+ }
+ }
+}
+
+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);
+}
+
+/*
+ * Populate index values of a UST stream. Values are set in big endian order.
+ *
+ * Return 0 on success or else a negative value.
+ */
+static int get_index_values(struct ctf_packet_index *index,
+ struct ustctl_consumer_stream *ustream)
+{
+ int ret;
+ uint64_t packet_size, content_size, timestamp_begin, timestamp_end,
+ events_discarded, stream_id, stream_instance_id,
+ packet_seq_num;
+
+ ret = ustctl_get_timestamp_begin(ustream, ×tamp_begin);
+ if (ret < 0) {
+ PERROR("ustctl_get_timestamp_begin");
+ goto error;
+ }
+
+ ret = ustctl_get_timestamp_end(ustream, ×tamp_end);
+ if (ret < 0) {
+ PERROR("ustctl_get_timestamp_end");
+ goto error;
+ }
+
+ ret = ustctl_get_events_discarded(ustream, &events_discarded);
+ if (ret < 0) {
+ PERROR("ustctl_get_events_discarded");
+ goto error;
+ }
+
+ ret = ustctl_get_content_size(ustream, &content_size);
+ if (ret < 0) {
+ PERROR("ustctl_get_content_size");
+ goto error;
+ }
+
+ ret = ustctl_get_packet_size(ustream, &packet_size);
+ if (ret < 0) {
+ PERROR("ustctl_get_packet_size");
+ goto error;
+ }
+
+ ret = ustctl_get_stream_id(ustream, &stream_id);
+ if (ret < 0) {
+ PERROR("ustctl_get_stream_id");
+ goto error;
+ }
+
+ ret = ustctl_get_instance_id(ustream, &stream_instance_id);
+ if (ret < 0) {
+ PERROR("ustctl_get_instance_id");
+ goto error;
+ }
+
+ ret = ustctl_get_sequence_number(ustream, &packet_seq_num);
+ if (ret < 0) {
+ PERROR("ustctl_get_sequence_number");
+ goto error;
+ }
+
+ *index = (typeof(*index)) {
+ .offset = index->offset,
+ .packet_size = htobe64(packet_size),
+ .content_size = htobe64(content_size),
+ .timestamp_begin = htobe64(timestamp_begin),
+ .timestamp_end = htobe64(timestamp_end),
+ .events_discarded = htobe64(events_discarded),
+ .stream_id = htobe64(stream_id),
+ .stream_instance_id = htobe64(stream_instance_id),
+ .packet_seq_num = htobe64(packet_seq_num),
+ };
+
+error:
+ return ret;
+}
+
+static
+void metadata_stream_reset_cache(struct lttng_consumer_stream *stream,
+ struct consumer_metadata_cache *cache)
+{
+ DBG("Metadata stream update to version %" PRIu64,
+ cache->version);
+ stream->ust_metadata_pushed = 0;
+ stream->metadata_version = cache->version;
+ stream->reset_metadata_flag = 1;
+}
+
+/*
+ * Check if the version of the metadata stream and metadata cache match.
+ * If the cache got updated, reset the metadata stream.
+ * The stream lock and metadata cache lock MUST be held.
+ * Return 0 on success, a negative value on error.
+ */
+static
+int metadata_stream_check_version(struct lttng_consumer_stream *stream)
+{
+ int ret = 0;
+ struct consumer_metadata_cache *cache = stream->chan->metadata_cache;
+
+ if (cache->version == stream->metadata_version) {
+ goto end;
+ }
+ metadata_stream_reset_cache(stream, cache);
+
+end:
+ return ret;
+}
+
+/*
+ * Write up to one packet from the metadata cache to the channel.
+ *
+ * Returns the number of bytes pushed in the cache, 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);
+ ret = metadata_stream_check_version(stream);
+ if (ret < 0) {
+ goto end;
+ }
+ if (stream->chan->metadata_cache->max_offset
+ == stream->ust_metadata_pushed) {
+ 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 = -1;
+ 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.
+ *
+ * Return 0 if new metadatda is available, EAGAIN if the metadata stream
+ * is empty or a negative value on error.
+ */
+int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *metadata_stream)
+{
+ int ret;
+ int retry = 0;
+ 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) {
+ 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);
+ ret = 0;
+ goto end;
+ }
+
+ ret = commit_one_metadata_packet(metadata_stream);
+ if (ret <= 0) {
+ goto end;
+ } else if (ret > 0) {
+ retry = 1;
+ }
+
+ ret = ustctl_snapshot(metadata_stream->ustream);