- /*
- * Rewind the current tracefile to the position at which the rotation
- * should have occured.
- */
- 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->rotate_at_seq_num = -1ULL;
-
- 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->rotate_at_seq_num == -1ULL) {
- /* No rotation expected. */
- goto end;
- }
-
- if (stream->index_rotated) {
- /* Rotation of the index has already occurred. */
- goto end;
- }
-
- if (stream->prev_index_seq == -1ULL ||
- stream->prev_index_seq < stream->rotate_at_seq_num) {
- DBG("Stream %" PRIu64 " index not yet ready for rotation (rotate_at_seq_num = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
- stream->stream_handle,
- stream->rotate_at_seq_num,
- stream->prev_index_seq);
- goto end;
- } else if (stream->prev_index_seq != stream->rotate_at_seq_num) {
- /*
- * Unexpected, protocol error/bug.
- * It could mean that we received a rotation position
- * that is in the past.
- */
- ERR("Stream %" PRIu64 " index is in an inconsistent state (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
- stream->stream_handle,
- stream->rotate_at_seq_num,
- stream->prev_data_seq,
- stream->prev_index_seq);
- ret = -1;
- goto end;
- } else {
- DBG("Rotating stream %" PRIu64 " index file",
- stream->stream_handle);
- ret = create_rotate_index_file(stream, stream->path_name);
- stream->index_rotated = true;
-
- if (stream->data_rotated && stream->index_rotated) {
- /* Rotation completed; reset its state. */
- DBG("Rotation completed for stream %" PRIu64,
- stream->stream_handle);
- stream->rotate_at_seq_num = -1ULL;
- stream->data_rotated = false;
- stream->index_rotated = false;
- }
- }
-
-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 (stream->rotate_at_seq_num == -1ULL) {
- /* No rotation expected. */
- goto end;
- }
-
- if (stream->data_rotated) {
- /* Rotation of the data file has already occurred. */
- goto end;
- }
-
- if (stream->prev_data_seq == -1ULL ||
- stream->prev_data_seq < stream->rotate_at_seq_num) {
- DBG("Stream %" PRIu64 " not yet ready for rotation (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ")",
- stream->stream_handle,
- stream->rotate_at_seq_num,
- stream->prev_data_seq);
- goto end;
- } else if (stream->prev_data_seq > stream->rotate_at_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 if (stream->prev_data_seq != stream->rotate_at_seq_num) {
- /*
- * Unexpected, protocol error/bug.
- * It could mean that we received a rotation position
- * that is in the past.
- */
- ERR("Stream %" PRIu64 " data is in an inconsistent state (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ")",
- stream->stream_handle,
- stream->rotate_at_seq_num,
- stream->prev_data_seq);
- ret = -1;
- goto end;
- } else {
- ret = do_rotate_stream_data(stream);
- }
-
-end:
- return ret;
-}
-
-/*
- * relay_recv_metadata: receive the metadata for the session.
- */
-static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
- struct relay_connection *conn,
- const struct lttng_buffer_view *payload)
-{
- int ret = 0;
- ssize_t size_ret;
- struct relay_session *session = conn->session;
- struct lttcomm_relayd_metadata_payload metadata_payload_header;
- struct relay_stream *metadata_stream;
- uint64_t metadata_payload_size;
-
- if (!session) {
- ERR("Metadata sent before version check");
- ret = -1;
- goto end;
- }
-
- if (recv_hdr->data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
- ERR("Incorrect data size");
- ret = -1;
- goto end;
- }
- metadata_payload_size = recv_hdr->data_size -
- sizeof(struct lttcomm_relayd_metadata_payload);
-
- memcpy(&metadata_payload_header, payload->data,
- sizeof(metadata_payload_header));
- metadata_payload_header.stream_id = be64toh(
- metadata_payload_header.stream_id);
- metadata_payload_header.padding_size = be32toh(
- metadata_payload_header.padding_size);
-
- metadata_stream = stream_get_by_id(metadata_payload_header.stream_id);
- if (!metadata_stream) {
- ret = -1;
- goto end;
- }
-
- pthread_mutex_lock(&metadata_stream->lock);
-
- size_ret = lttng_write(metadata_stream->stream_fd->fd,
- payload->data + sizeof(metadata_payload_header),
- metadata_payload_size);
- if (size_ret < metadata_payload_size) {
- ERR("Relay error writing metadata on file");
- ret = -1;
- goto end_put;
- }
-
- size_ret = write_padding_to_file(metadata_stream->stream_fd->fd,
- metadata_payload_header.padding_size);
- if (size_ret < (int64_t) metadata_payload_header.padding_size) {
- ret = -1;
- goto end_put;
- }
-
- metadata_stream->metadata_received +=
- metadata_payload_size + metadata_payload_header.padding_size;
- DBG2("Relay metadata written. Updated metadata_received %" PRIu64,
- metadata_stream->metadata_received);
-
- ret = try_rotate_stream_data(metadata_stream);
- if (ret < 0) {
- goto end_put;
- }
-
-end_put:
- pthread_mutex_unlock(&metadata_stream->lock);
- stream_put(metadata_stream);
-end:
- return ret;
-}
-
-/*
- * relay_send_version: send relayd version number
- */
-static int relay_send_version(const struct lttcomm_relayd_hdr *recv_hdr,
- struct relay_connection *conn,
- const struct lttng_buffer_view *payload)
-{
- int ret;
- ssize_t send_ret;
- struct lttcomm_relayd_version reply, msg;
- bool compatible = true;
-
- conn->version_check_done = true;
-
- /* Get version from the other side. */
- if (payload->size < sizeof(msg)) {
- ERR("Unexpected payload size in \"relay_send_version\": expected >= %zu bytes, got %zu bytes",
- sizeof(msg), payload->size);