+ /*
+ * Rewind the current tracefile to the position at which the rotation
+ * should have occured.
+ */
+ ret = lseek(stream->stream_fd->fd,
+ stream->pos_after_last_complete_data_index, SEEK_SET);
+ if (ret < 0) {
+ PERROR("seek truncate stream");
+ goto end;
+ }
+
+ /* Move data from the old file to the new file. */
+ while (pos < diff) {
+ uint64_t count, bytes_left;
+ ssize_t io_ret;
+
+ bytes_left = diff - pos;
+ count = bytes_left > sizeof(buf) ? sizeof(buf) : bytes_left;
+ assert(count <= SIZE_MAX);
+
+ io_ret = lttng_read(stream->stream_fd->fd, buf, count);
+ if (io_ret < (ssize_t) count) {
+ char error_string[256];
+
+ snprintf(error_string, sizeof(error_string),
+ "Failed to read %" PRIu64 " bytes from fd %i in rotate_truncate_stream(), returned %zi",
+ count, stream->stream_fd->fd, io_ret);
+ if (io_ret == -1) {
+ PERROR("%s", error_string);
+ } else {
+ ERR("%s", error_string);
+ }
+ ret = -1;
+ goto end;
+ }
+
+ io_ret = lttng_write(new_fd, buf, count);
+ if (io_ret < (ssize_t) count) {
+ char error_string[256];
+
+ snprintf(error_string, sizeof(error_string),
+ "Failed to write %" PRIu64 " bytes from fd %i in rotate_truncate_stream(), returned %zi",
+ count, new_fd, io_ret);
+ if (io_ret == -1) {
+ PERROR("%s", error_string);
+ } else {
+ ERR("%s", error_string);
+ }
+ ret = -1;
+ goto end;
+ }
+
+ pos += count;
+ }
+
+ /* Truncate the file to get rid of the excess data. */
+ ret = ftruncate(stream->stream_fd->fd,
+ stream->pos_after_last_complete_data_index);
+ if (ret) {
+ PERROR("ftruncate");
+ goto end;
+ }
+
+ ret = close(stream->stream_fd->fd);
+ if (ret < 0) {
+ PERROR("Closing tracefile");
+ goto end;
+ }
+
+ ret = create_rotate_index_file(stream);
+ if (ret < 0) {
+ ERR("Rotate stream index file");
+ goto end;
+ }
+
+ /*
+ * Update the offset and FD of all the eventual indexes created by the
+ * data connection before the rotation command arrived.
+ */
+ ret = relay_index_switch_all_files(stream);
+ if (ret < 0) {
+ ERR("Failed to rotate index file");
+ goto end;
+ }
+
+ stream->stream_fd->fd = new_fd;
+ stream->tracefile_size_current = diff;
+ stream->pos_after_last_complete_data_index = 0;
+ stream->rotate_at_seq_num = -1ULL;
+
+ ret = 0;
+
+end:
+ return ret;
+}
+
+/*
+ * Check if a stream should perform a rotation (for session rotation).
+ * Must be called with the stream lock held.
+ *
+ * Return 0 on success, a negative value on error.
+ */
+static
+int try_rotate_stream(struct relay_stream *stream)
+{
+ int ret = 0;
+
+ /* No rotation expected. */
+ if (stream->rotate_at_seq_num == -1ULL) {
+ goto end;
+ }
+
+ if (stream->prev_seq < stream->rotate_at_seq_num) {
+ DBG("Stream %" PRIu64 " no yet ready for rotation",
+ stream->stream_handle);
+ goto end;
+ } else if (stream->prev_seq > stream->rotate_at_seq_num) {
+ DBG("Rotation after too much data has been written in tracefile "
+ "for stream %" PRIu64 ", need to truncate before "
+ "rotating", stream->stream_handle);
+ ret = rotate_truncate_stream(stream);
+ if (ret) {
+ ERR("Failed to truncate stream");
+ goto end;
+ }
+ } else {
+ /* stream->prev_seq == stream->rotate_at_seq_num */
+ DBG("Stream %" PRIu64 " ready for rotation",
+ stream->stream_handle);
+ ret = do_rotate_stream(stream);
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * relay_recv_metadata: receive the metadata for the session.
+ */
+static int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret = 0;
+ ssize_t size_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_metadata_payload *metadata_struct;
+ struct relay_stream *metadata_stream;
+ uint64_t data_size, payload_size;