+ /*
+ * 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 = lttng_ust_ctl_get_subbuf_size(
+ stream->ustream, &subbuf->info.data.subbuf_size);
+ if (ret) {
+ goto end;
+ }
+
+ ret = lttng_ust_ctl_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 = lttng_ust_ctl_get_packet_size(
+ stream->ustream, &subbuf->info.data.packet_size);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer packet size");
+ goto end;
+ }
+
+ ret = lttng_ust_ctl_get_content_size(
+ stream->ustream, &subbuf->info.data.content_size);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer content size");
+ goto end;
+ }
+
+ ret = lttng_ust_ctl_get_timestamp_begin(
+ stream->ustream, &subbuf->info.data.timestamp_begin);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer begin timestamp");
+ goto end;
+ }
+
+ ret = lttng_ust_ctl_get_timestamp_end(
+ stream->ustream, &subbuf->info.data.timestamp_end);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer end timestamp");
+ goto end;
+ }
+
+ ret = lttng_ust_ctl_get_events_discarded(
+ stream->ustream, &subbuf->info.data.events_discarded);
+ if (ret) {
+ PERROR("Failed to get sub-buffer events discarded count");
+ goto end;
+ }
+
+ ret = lttng_ust_ctl_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 = lttng_ust_ctl_get_stream_id(
+ stream->ustream, &subbuf->info.data.stream_id);
+ if (ret < 0) {
+ PERROR("Failed to get stream id");
+ goto end;
+ }
+
+ ret = lttng_ust_ctl_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 enum get_next_subbuffer_status get_next_subbuffer(
+ struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuffer)
+{
+ int ret;
+ enum get_next_subbuffer_status status;
+
+ ret = lttng_ust_ctl_get_next_subbuf(stream->ustream);
+ switch (ret) {
+ case 0:
+ status = GET_NEXT_SUBBUFFER_STATUS_OK;
+ break;
+ case -ENODATA:
+ case -EAGAIN:
+ /*
+ * The caller only expects -ENODATA when there is no data to
+ * read, but the kernel tracer returns -EAGAIN when there is
+ * currently no data for a non-finalized stream, and -ENODATA
+ * when there is no data for a finalized stream. Those can be
+ * combined into a -ENODATA return value.
+ */
+ status = GET_NEXT_SUBBUFFER_STATUS_NO_DATA;
+ goto end;
+ default:
+ status = GET_NEXT_SUBBUFFER_STATUS_ERROR;
+ goto end;
+ }
+
+ ret = get_next_subbuffer_common(stream, subbuffer);
+ if (ret) {
+ status = GET_NEXT_SUBBUFFER_STATUS_ERROR;
+ goto end;
+ }
+end:
+ return status;
+}
+
+static enum get_next_subbuffer_status 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;
+ enum get_next_subbuffer_status status;
+
+ do {
+ ret = lttng_ust_ctl_get_next_subbuf(stream->ustream);
+ if (ret == 0) {
+ got_subbuffer = true;
+ } else {
+ got_subbuffer = false;
+ if (ret != -EAGAIN) {
+ /* Fatal error. */
+ status = GET_NEXT_SUBBUFFER_STATUS_ERROR;