+static void stream_complete_rotation(struct relay_stream *stream)
+{
+ DBG("Rotation completed for stream %" PRIu64, stream->stream_handle);
+ lttng_trace_chunk_put(stream->trace_chunk);
+ stream->trace_chunk = stream->ongoing_rotation.value.next_trace_chunk;
+ stream->ongoing_rotation = (typeof(stream->ongoing_rotation)) {};
+}
+
+/*
+ * If too much data has been written in a tracefile before we received the
+ * rotation command, we have to move the excess data to the new tracefile and
+ * perform the rotation. This can happen because the control and data
+ * connections are separate, the indexes as well as the commands arrive from
+ * the control connection and we have no control over the order so we could be
+ * in a situation where too much data has been received on the data connection
+ * before the rotation command on the control connection arrives.
+ */
+static int rotate_truncate_stream(struct relay_stream *stream)
+{
+ int ret, new_fd;
+ off_t lseek_ret;
+ uint64_t diff, pos = 0;
+ char buf[FILE_IO_STACK_BUFFER_SIZE];
+
+ assert(!stream->is_metadata);
+
+ assert(stream->tracefile_size_current >
+ stream->pos_after_last_complete_data_index);
+ diff = stream->tracefile_size_current -
+ stream->pos_after_last_complete_data_index;
+
+ /* Create the new tracefile. */
+ new_fd = utils_create_stream_file(stream->path_name,
+ stream->channel_name,
+ stream->tracefile_size, stream->tracefile_count,
+ /* uid */ -1, /* gid */ -1, /* suffix */ NULL);
+ if (new_fd < 0) {
+ ERR("Failed to create new stream file at path %s for channel %s",
+ stream->path_name, stream->channel_name);
+ ret = -1;
+ goto end;
+ }
+
+ /*
+ * Rewind the current tracefile to the position at which the rotation
+ * should have occurred.
+ */
+ lseek_ret = lseek(stream->stream_fd->fd,
+ stream->pos_after_last_complete_data_index, SEEK_SET);
+ if (lseek_ret < 0) {
+ PERROR("seek truncate stream");
+ ret = -1;
+ 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;
+ }
+
+ /*
+ * 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_complete_rotation(stream);
+
+ ret = 0;
+
+end:
+ return ret;
+}
+
+static int stream_create_data_output_file_from_trace_chunk(
+ struct relay_stream *stream,
+ struct lttng_trace_chunk *trace_chunk,
+ bool force_unlink,
+ struct stream_fd **out_stream_fd)
+{
+ int ret, fd;
+ char stream_path[LTTNG_PATH_MAX];
+ enum lttng_trace_chunk_status status;
+ const int flags = O_RDWR | O_CREAT | O_TRUNC;
+ const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
+
+ ASSERT_LOCKED(stream->lock);
+ assert(stream->trace_chunk);
+
+ ret = utils_stream_file_path(stream->path_name, stream->channel_name,
+ stream->tracefile_size, stream->tracefile_current_index,
+ NULL, stream_path, sizeof(stream_path));
+ if (ret < 0) {
+ goto end;
+ }
+
+ if (stream->tracefile_wrapped_around || force_unlink) {
+ /*
+ * The on-disk ring-buffer has wrapped around.
+ * Newly created stream files will replace existing files. Since
+ * live clients may be consuming existing files, the file about
+ * to be replaced is unlinked in order to not overwrite its
+ * content.
+ */
+ status = lttng_trace_chunk_unlink_file(trace_chunk,
+ stream_path);
+ if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ PERROR("Failed to unlink stream file \"%s\" during trace file rotation",
+ stream_path);
+ /*
+ * Don't abort if the file doesn't exist, it is
+ * unexpected, but should not be a fatal error.
+ */
+ if (errno != ENOENT) {
+ ret = -1;
+ goto end;
+ }
+ }
+ }
+
+ status = lttng_trace_chunk_open_file(
+ trace_chunk, stream_path, flags, mode, &fd);
+ if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ERR("Failed to open stream file \"%s\"", stream->channel_name);
+ ret = -1;
+ goto end;
+ }
+
+ *out_stream_fd = stream_fd_create(fd);
+ if (!*out_stream_fd) {
+ if (close(ret)) {
+ PERROR("Error closing stream file descriptor %d", ret);
+ }
+ ret = -1;
+ goto end;
+ }
+end:
+ return ret;
+}
+
+static int stream_rotate_data_file(struct relay_stream *stream)
+{
+ int ret = 0;
+
+ DBG("Rotating stream %" PRIu64 " data file",
+ stream->stream_handle);
+
+ if (stream->stream_fd) {
+ stream_fd_put(stream->stream_fd);
+ stream->stream_fd = NULL;
+ }
+
+ stream->tracefile_wrapped_around = false;
+ stream->tracefile_current_index = 0;
+
+ if (stream->ongoing_rotation.value.next_trace_chunk) {
+ struct stream_fd *new_stream_fd = NULL;
+ enum lttng_trace_chunk_status chunk_status;
+
+ chunk_status = lttng_trace_chunk_create_subdirectory(
+ stream->ongoing_rotation.value.next_trace_chunk,
+ stream->path_name);
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ret = -1;
+ goto end;
+ }
+
+ /* Rotate the data file. */
+ ret = stream_create_data_output_file_from_trace_chunk(stream,
+ stream->ongoing_rotation.value.next_trace_chunk,
+ false, &new_stream_fd);
+ stream->stream_fd = new_stream_fd;
+ if (ret < 0) {
+ ERR("Failed to rotate stream data file");
+ goto end;
+ }
+ }
+ stream->tracefile_size_current = 0;
+ stream->pos_after_last_complete_data_index = 0;
+ stream->ongoing_rotation.value.data_rotated = true;
+
+ if (stream->ongoing_rotation.value.index_rotated) {
+ /* Rotation completed; reset its state. */
+ stream_complete_rotation(stream);
+ }
+end:
+ return ret;
+}
+
+/*
+ * Check if a stream's data file (as opposed to index) should be rotated
+ * (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_data(struct relay_stream *stream)
+{
+ int ret = 0;
+
+ if (caa_likely(!stream->ongoing_rotation.is_set)) {
+ /* No rotation expected. */
+ goto end;
+ }
+
+ if (stream->ongoing_rotation.value.data_rotated) {
+ /* Rotation of the data file has already occurred. */
+ goto end;
+ }
+
+ if (stream->prev_data_seq == -1ULL ||
+ stream->prev_data_seq + 1 < stream->ongoing_rotation.value.seq_num) {
+ /*
+ * The next packet that will be written is not part of the next
+ * chunk yet.
+ */
+ DBG("Stream %" PRIu64 " not yet ready for rotation (rotate_at_seq_num = %" PRIu64
+ ", prev_data_seq = %" PRIu64 ")",
+ stream->stream_handle,
+ stream->ongoing_rotation.value.seq_num,
+ stream->prev_data_seq);
+ goto end;
+ } else if (stream->prev_data_seq > stream->ongoing_rotation.value.seq_num) {
+ /*
+ * prev_data_seq is checked here since indexes and rotation
+ * commands are serialized with respect to each other.
+ */
+ 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 {
+ ret = stream_rotate_data_file(stream);
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * Close the current index file if it is open, and create a new one.
+ *
+ * Return 0 on success, -1 on error.
+ */
+static int create_index_file(struct relay_stream *stream,
+ struct lttng_trace_chunk *chunk)
+{
+ int ret;
+ uint32_t major, minor;
+ char *index_subpath = NULL;
+
+ ASSERT_LOCKED(stream->lock);
+
+ /* Put ref on previous index_file. */
+ if (stream->index_file) {
+ lttng_index_file_put(stream->index_file);
+ stream->index_file = NULL;
+ }
+ major = stream->trace->session->major;
+ minor = stream->trace->session->minor;
+
+ if (!chunk) {
+ ret = 0;
+ goto end;
+ }
+ ret = asprintf(&index_subpath, "%s/%s", stream->path_name,
+ DEFAULT_INDEX_DIR);
+ if (ret < 0) {
+ goto end;
+ }
+
+ ret = lttng_trace_chunk_create_subdirectory(chunk,
+ index_subpath);
+ free(index_subpath);
+ if (ret) {
+ goto end;
+ }
+ stream->index_file = lttng_index_file_create_from_trace_chunk(
+ chunk, stream->path_name,
+ stream->channel_name, stream->tracefile_size,
+ stream->tracefile_current_index,
+ lttng_to_index_major(major, minor),
+ lttng_to_index_minor(major, minor), true);
+ if (!stream->index_file) {
+ ret = -1;
+ goto end;
+ }
+
+ ret = 0;
+
+end:
+ return ret;
+}
+
+/*
+ * Check if a stream's index file should be rotated (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_index(struct relay_stream *stream)
+{
+ int ret = 0;
+
+ if (!stream->ongoing_rotation.is_set) {
+ /* No rotation expected. */
+ goto end;
+ }
+
+ if (stream->ongoing_rotation.value.index_rotated) {
+ /* Rotation of the index has already occurred. */
+ goto end;
+ }
+
+ if (stream->prev_index_seq == -1ULL ||
+ stream->prev_index_seq + 1 < stream->ongoing_rotation.value.seq_num) {
+ DBG("Stream %" PRIu64 " index not yet ready for rotation (rotate_at_seq_num = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
+ stream->stream_handle,
+ stream->ongoing_rotation.value.seq_num,
+ stream->prev_index_seq);
+ goto end;
+ } else {
+ /* The next index belongs to the new trace chunk; rotate. */
+ assert(stream->prev_index_seq + 1 ==
+ stream->ongoing_rotation.value.seq_num);
+ DBG("Rotating stream %" PRIu64 " index file",
+ stream->stream_handle);
+ ret = create_index_file(stream,
+ stream->ongoing_rotation.value.next_trace_chunk);
+ stream->ongoing_rotation.value.index_rotated = true;
+
+ if (stream->ongoing_rotation.value.data_rotated &&
+ stream->ongoing_rotation.value.index_rotated) {
+ /* Rotation completed; reset its state. */
+ DBG("Rotation completed for stream %" PRIu64,
+ stream->stream_handle);
+ stream_complete_rotation(stream);
+ }
+ }
+
+end:
+ return ret;
+}
+
+static int stream_set_trace_chunk(struct relay_stream *stream,
+ struct lttng_trace_chunk *chunk)
+{
+ int ret = 0;
+ enum lttng_trace_chunk_status status;
+ bool acquired_reference;
+ struct stream_fd *new_stream_fd = NULL;
+
+ status = lttng_trace_chunk_create_subdirectory(chunk,
+ stream->path_name);
+ if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ret = -1;
+ goto end;
+ }
+
+ lttng_trace_chunk_put(stream->trace_chunk);
+ acquired_reference = lttng_trace_chunk_get(chunk);
+ assert(acquired_reference);
+ stream->trace_chunk = chunk;
+
+ if (stream->stream_fd) {
+ stream_fd_put(stream->stream_fd);
+ stream->stream_fd = NULL;
+ }
+ ret = stream_create_data_output_file_from_trace_chunk(stream, chunk,
+ false, &new_stream_fd);
+ stream->stream_fd = new_stream_fd;
+end:
+ return ret;
+}
+