- rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
- if (rotation_ret == 1) {
- rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
- if (rotation_ret < 0) {
- ERR("Stream rotation error");
- ret = -1;
- goto error;
+ coherent = got_subbuffer && cache_empty && buffer_empty;
+
+ LTTNG_OPTIONAL_SET(&subbuffer->info.metadata.coherent, coherent);
+ status = GET_NEXT_SUBBUFFER_STATUS_OK;
+end:
+ return status;
+}
+
+static int put_next_subbuffer(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuffer)
+{
+ const int ret = lttng_ust_ctl_put_next_subbuf(stream->ustream);
+
+ LTTNG_ASSERT(ret == 0);
+ return ret;
+}
+
+static int signal_metadata(struct lttng_consumer_stream *stream,
+ struct lttng_consumer_local_data *ctx)
+{
+ ASSERT_LOCKED(stream->metadata_rdv_lock);
+ 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;