+int stream_init_packet(struct relay_stream *stream, size_t packet_size,
+ bool *file_rotated)
+{
+ int ret = 0;
+
+ ASSERT_LOCKED(stream->lock);
+
+ if (!stream->file || !stream->trace_chunk) {
+ ERR("Protocol error: received a packet for a stream that doesn't have a current trace chunk: stream_id = %" PRIu64 ", channel_name = %s",
+ stream->stream_handle, stream->channel_name);
+ ret = -1;
+ goto end;
+ }
+
+ if (caa_likely(stream->tracefile_size == 0)) {
+ /* No size limit set; nothing to check. */
+ goto end;
+ }
+
+ /*
+ * Check if writing the new packet would exceed the maximal file size.
+ */
+ if (caa_unlikely((stream->tracefile_size_current + packet_size) >
+ stream->tracefile_size)) {
+ const uint64_t new_file_index =
+ (stream->tracefile_current_index + 1) %
+ stream->tracefile_count;
+
+ if (new_file_index < stream->tracefile_current_index) {
+ stream->tracefile_wrapped_around = true;
+ }
+ DBG("New stream packet causes stream file rotation: stream_id = %" PRIu64
+ ", current_file_size = %" PRIu64
+ ", packet_size = %zu, current_file_index = %" PRIu64
+ " new_file_index = %" PRIu64,
+ stream->stream_handle,
+ stream->tracefile_size_current, packet_size,
+ stream->tracefile_current_index, new_file_index);
+ tracefile_array_file_rotate(stream->tfa, TRACEFILE_ROTATE_WRITE);
+ stream->tracefile_current_index = new_file_index;
+
+ if (stream->file) {
+ fs_handle_close(stream->file);
+ stream->file = NULL;
+ }
+ ret = stream_create_data_output_file_from_trace_chunk(stream,
+ stream->trace_chunk, false, &stream->file);
+ if (ret) {
+ ERR("Failed to perform trace file rotation of stream %" PRIu64,
+ stream->stream_handle);
+ goto end;
+ }
+
+ /*
+ * Reset current size because we just performed a stream
+ * rotation.
+ */
+ DBG("%s: reset tracefile_size_current for stream %" PRIu64 " was %" PRIu64,
+ __func__, stream->stream_handle, stream->tracefile_size_current);
+ stream->tracefile_size_current = 0;
+ *file_rotated = true;
+ } else {
+ *file_rotated = false;
+ }
+end:
+ return ret;
+}
+
+/* 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 (!stream->file || !stream->trace_chunk) {
+ ERR("Protocol error: received a packet for a stream that doesn't have a current trace chunk: stream_id = %" PRIu64 ", channel_name = %s",
+ stream->stream_handle, stream->channel_name);
+ ret = -1;
+ goto end;
+ }
+ if (packet) {
+ write_ret = fs_handle_write(
+ stream->file, 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 = fs_handle_write(stream->file, 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) {
+ size_t recv_len;
+
+ recv_len = packet ? packet->size : 0;
+ recv_len += padding_len;
+ stream->metadata_received += recv_len;
+ if (recv_len) {
+ stream->no_new_metadata_notified = false;
+ }
+ }
+
+ 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(stream->trace_chunk);
+ 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_file_rotate(stream->tfa, TRACEFILE_ROTATE_READ);
+ tracefile_array_commit_seq(stream->tfa, stream->index_received_seqcount);
+ stream->index_received_seqcount++;
+ LTTNG_OPTIONAL_SET(&stream->received_packet_seq_num,
+ be64toh(index->index_data.packet_seq_num));
+ *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);
+
+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);
+
+ DBG("stream_add_index for stream %" PRIu64, stream->stream_handle);
+
+ /* 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_file_rotate(stream->tfa, TRACEFILE_ROTATE_READ);
+ tracefile_array_commit_seq(stream->tfa, stream->index_received_seqcount);
+ stream->index_received_seqcount++;
+ stream->pos_after_last_complete_data_index += index->total_size;
+ stream->prev_index_seq = index_info->net_seq_num;
+ LTTNG_OPTIONAL_SET(&stream->received_packet_seq_num,
+ index_info->packet_seq_num);
+
+ ret = try_rotate_stream_index(stream);
+ if (ret < 0) {
+ goto end;
+ }
+ ret = try_rotate_stream_data(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;
+}
+