+ written_bytes = stream->read_subbuffer_ops.consume_subbuffer(
+ ctx, stream, &subbuffer);
+ /*
+ * Should write subbuf_size amount of data when network streaming or
+ * the full padded size when we are not streaming.
+ */
+ if ((written_bytes != subbuffer.info.data.subbuf_size &&
+ stream->net_seq_idx != (uint64_t) -1ULL) ||
+ (written_bytes != subbuffer.info.data.padded_subbuf_size &&
+ stream->net_seq_idx ==
+ (uint64_t) -1ULL)) {
+ /*
+ * Display the error but continue processing to try to
+ * release the subbuffer. This is a DBG statement
+ * since this can happen without being a critical
+ * error.
+ */
+ DBG("Failed to write to tracefile (written_bytes: %zd != padded subbuffer size: %lu, subbuffer size: %lu)",
+ written_bytes, subbuffer.info.data.subbuf_size,
+ subbuffer.info.data.padded_subbuf_size);
+ }
+
+ ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+ if (ret) {
+ goto end;
+ }
+
+ if (stream->read_subbuffer_ops.post_consume) {
+ ret = stream->read_subbuffer_ops.post_consume(stream, &subbuffer, ctx);
+ if (ret) {
+ goto end;
+ }
+ }
+
+ /*
+ * After extracting the packet, we check if the stream is now ready to
+ * be rotated and perform the action immediately.
+ *
+ * Don't overwrite `ret` as callers expect the number of bytes
+ * consumed to be returned on success.
+ */
+ 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) {
+ ret = rotation_ret;
+ ERR("Stream rotation error after consuming data");
+ goto end;
+ }
+ } else if (rotation_ret < 0) {
+ ret = rotation_ret;
+ ERR("Failed to check if stream was ready to rotate after consuming data");
+ goto end;
+ }
+
+ if (stream->read_subbuffer_ops.on_sleep) {
+ stream->read_subbuffer_ops.on_sleep(stream, ctx);
+ }
+
+ ret = written_bytes;
+end:
+ if (!locked_by_caller) {
+ stream->read_subbuffer_ops.unlock(stream);
+ }
+
+ return ret;
+error_put_subbuf:
+ (void) stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+ goto end;
+}
+
+int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
+{
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ return lttng_kconsumer_on_recv_stream(stream);
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ return lttng_ustconsumer_on_recv_stream(stream);
+ default:
+ ERR("Unknown consumer_data type");
+ assert(0);
+ return -ENOSYS;
+ }
+}
+
+/*
+ * Allocate and set consumer data hash tables.
+ */
+int lttng_consumer_init(void)
+{
+ consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!consumer_data.channel_ht) {
+ goto error;
+ }
+
+ consumer_data.channels_by_session_id_ht =
+ lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!consumer_data.channels_by_session_id_ht) {
+ goto error;
+ }
+
+ consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!consumer_data.relayd_ht) {