+/* Note that the packet is not necessarily complete. */
+int stream_write(struct relay_stream *stream,
+ const struct lttng_buffer_view *packet, size_t padding_len)
+{
+ int ret = 0;
+ ssize_t write_ret;
+ size_t padding_to_write = padding_len;
+ char padding_buffer[FILE_IO_STACK_BUFFER_SIZE];
+
+ ASSERT_LOCKED(stream->lock);
+ memset(padding_buffer, 0,
+ min(sizeof(padding_buffer), padding_to_write));
+
+ if (packet) {
+ write_ret = lttng_write(stream->stream_fd->fd,
+ packet->data, packet->size);
+ if (write_ret != packet->size) {
+ PERROR("Failed to write to stream file of %sstream %" PRIu64,
+ stream->is_metadata ? "metadata " : "",
+ stream->stream_handle);
+ ret = -1;
+ goto end;
+ }
+ }
+
+ while (padding_to_write > 0) {
+ const size_t padding_to_write_this_pass =
+ min(padding_to_write, sizeof(padding_buffer));
+
+ write_ret = lttng_write(stream->stream_fd->fd,
+ padding_buffer, padding_to_write_this_pass);
+ if (write_ret != padding_to_write_this_pass) {
+ PERROR("Failed to write padding to file of %sstream %" PRIu64,
+ stream->is_metadata ? "metadata " : "",
+ stream->stream_handle);
+ ret = -1;
+ goto end;
+ }
+ padding_to_write -= padding_to_write_this_pass;
+ }
+
+ if (stream->is_metadata) {
+ stream->metadata_received += packet ? packet->size : 0;
+ stream->metadata_received += padding_len;
+ }
+
+ DBG("Wrote to %sstream %" PRIu64 ": data_length = %zu, padding_length = %zu",
+ stream->is_metadata ? "metadata " : "",
+ stream->stream_handle,
+ packet ? packet->size : (size_t) 0, padding_len);
+end:
+ return ret;
+}
+
+/*
+ * Update index after receiving a packet for a data stream.
+ *
+ * Called with the stream lock held.
+ *
+ * Return 0 on success else a negative value.
+ */
+int stream_update_index(struct relay_stream *stream, uint64_t net_seq_num,
+ bool rotate_index, bool *flushed, uint64_t total_size)
+{
+ int ret = 0;
+ uint64_t data_offset;
+ struct relay_index *index;
+
+ ASSERT_LOCKED(stream->lock);
+ /* Get data offset because we are about to update the index. */
+ data_offset = htobe64(stream->tracefile_size_current);
+
+ DBG("handle_index_data: stream %" PRIu64 " net_seq_num %" PRIu64 " data offset %" PRIu64,
+ stream->stream_handle, net_seq_num, stream->tracefile_size_current);
+
+ /*
+ * Lookup for an existing index for that stream id/sequence
+ * number. If it exists, the control thread has already received the
+ * data for it, thus we need to write it to disk.
+ */
+ index = relay_index_get_by_id_or_create(stream, net_seq_num);
+ if (!index) {
+ ret = -1;
+ goto end;
+ }
+
+ if (rotate_index || !stream->index_file) {
+ ret = create_index_file(stream, stream->trace_chunk);
+ if (ret) {
+ ERR("Failed to create index file for stream %" PRIu64,
+ stream->stream_handle);
+ /* Put self-ref for this index due to error. */
+ relay_index_put(index);
+ index = NULL;
+ goto end;
+ }
+ }
+
+ if (relay_index_set_file(index, stream->index_file, data_offset)) {
+ ret = -1;
+ /* Put self-ref for this index due to error. */
+ relay_index_put(index);
+ index = NULL;
+ goto end;
+ }
+
+ ret = relay_index_try_flush(index);
+ if (ret == 0) {
+ tracefile_array_commit_seq(stream->tfa);
+ stream->index_received_seqcount++;
+ *flushed = true;
+ } else if (ret > 0) {
+ index->total_size = total_size;
+ /* No flush. */
+ ret = 0;
+ } else {
+ /*
+ * ret < 0
+ *
+ * relay_index_try_flush is responsible for the self-reference
+ * put of the index object on error.
+ */
+ ERR("relay_index_try_flush error %d", ret);
+ ret = -1;
+ }
+end:
+ return ret;
+}
+
+int stream_complete_packet(struct relay_stream *stream, size_t packet_total_size,
+ uint64_t sequence_number, bool index_flushed)
+{
+ int ret = 0;
+
+ ASSERT_LOCKED(stream->lock);
+
+ stream->tracefile_size_current += packet_total_size;
+ if (index_flushed) {
+ stream->pos_after_last_complete_data_index =
+ stream->tracefile_size_current;
+ stream->prev_index_seq = sequence_number;
+ ret = try_rotate_stream_index(stream);
+ if (ret < 0) {
+ goto end;
+ }
+ }
+
+ stream->prev_data_seq = sequence_number;
+ ret = try_rotate_stream_data(stream);
+ if (ret < 0) {
+ goto end;
+ }
+end:
+ return ret;
+}
+
+int stream_add_index(struct relay_stream *stream,
+ const struct lttcomm_relayd_index *index_info)
+{
+ int ret = 0;
+ struct relay_index *index;
+
+ ASSERT_LOCKED(stream->lock);
+
+ /* Live beacon handling */
+ if (index_info->packet_size == 0) {
+ DBG("Received live beacon for stream %" PRIu64,
+ stream->stream_handle);
+
+ /*
+ * Only flag a stream inactive when it has already
+ * received data and no indexes are in flight.
+ */
+ if (stream->index_received_seqcount > 0
+ && stream->indexes_in_flight == 0) {
+ stream->beacon_ts_end = index_info->timestamp_end;
+ }
+ ret = 0;
+ goto end;
+ } else {
+ stream->beacon_ts_end = -1ULL;
+ }
+
+ if (stream->ctf_stream_id == -1ULL) {
+ stream->ctf_stream_id = index_info->stream_id;
+ }
+
+ index = relay_index_get_by_id_or_create(stream, index_info->net_seq_num);
+ if (!index) {
+ ret = -1;
+ ERR("Failed to get or create index %" PRIu64,
+ index_info->net_seq_num);
+ goto end;
+ }
+ if (relay_index_set_control_data(index, index_info,
+ stream->trace->session->minor)) {
+ ERR("set_index_control_data error");
+ relay_index_put(index);
+ ret = -1;
+ goto end;
+ }
+ ret = relay_index_try_flush(index);
+ if (ret == 0) {
+ tracefile_array_commit_seq(stream->tfa);
+ stream->index_received_seqcount++;
+ stream->pos_after_last_complete_data_index += index->total_size;
+ stream->prev_index_seq = index_info->net_seq_num;
+
+ ret = try_rotate_stream_index(stream);
+ if (ret < 0) {
+ goto end;
+ }
+ } else if (ret > 0) {
+ /* no flush. */
+ ret = 0;
+ } else {
+ /*
+ * ret < 0
+ *
+ * relay_index_try_flush is responsible for the self-reference
+ * put of the index object on error.
+ */
+ ERR("relay_index_try_flush error %d", ret);
+ ret = -1;
+ }
+end:
+ return ret;
+}
+
+static void print_stream_indexes(struct relay_stream *stream)