+ header.old_path_length = be32toh(header.old_path_length);
+ header.new_path_length = be32toh(header.new_path_length);
+ received_paths_size = header.old_path_length + header.new_path_length;
+
+ /* Ensure the paths don't exceed their allowed size. */
+ ret = validate_rotate_rename_path_length("old", header.old_path_length);
+ if (ret) {
+ goto end;
+ }
+ ret = validate_rotate_rename_path_length("new", header.new_path_length);
+ if (ret) {
+ goto end;
+ }
+
+ received_paths = zmalloc(received_paths_size);
+ if (!received_paths) {
+ PERROR("Could not allocate rotate commands paths reception buffer");
+ ret = -1;
+ goto end;
+ }
+
+ network_ret = conn->sock->ops->recvmsg(conn->sock, received_paths,
+ received_paths_size, 0);
+ if (network_ret < (ssize_t) received_paths_size) {
+ if (network_ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown",
+ conn->sock->fd);
+ } else {
+ ERR("Relay failed to received rename command paths (%zu bytes): recvmsg() returned %zi",
+ received_paths_size, network_ret);
+ }
+ ret = -1;
+ goto end_no_reply;
+ }
+
+ /* Validate that both paths received are NULL terminated. */
+ if (received_paths[header.old_path_length - 1] != '\0') {
+ ERR("relay_rotate_rename command's \"old\" path is invalid (not NULL terminated)");
+ ret = -1;
+ goto end;
+ }
+ if (received_paths[received_paths_size - 1] != '\0') {
+ ERR("relay_rotate_rename command's \"new\" path is invalid (not NULL terminated)");
+ ret = -1;
+ goto end;
+ }
+
+ received_old_path = received_paths;
+ received_new_path = received_paths + header.old_path_length;
+
+ complete_old_path = create_output_path(received_old_path);
+ if (!complete_old_path) {
+ ERR("Failed to build old output path in rotate_rename command");
+ ret = -1;
+ goto end;
+ }
+
+ complete_new_path = create_output_path(received_new_path);
+ if (!complete_new_path) {
+ ERR("Failed to build new output path in rotate_rename command");
+ ret = -1;
+ goto end;
+ }
+
+ ret = utils_mkdir_recursive(complete_new_path, S_IRWXU | S_IRWXG,
+ -1, -1);
+ if (ret < 0) {
+ ERR("Failed to mkdir() rotate_rename's \"new\" output directory at \"%s\"",
+ complete_new_path);
+ goto end;
+ }
+
+ /*
+ * If a domain has not yet created its channel, the domain-specific
+ * folder might not exist, but this is not an error.
+ */
+ ret = rename(complete_old_path, complete_new_path);
+ if (ret < 0 && errno != ENOENT) {
+ PERROR("Renaming chunk in rotate_rename command from \"%s\" to \"%s\"",
+ complete_old_path, complete_new_path);
+ goto end;
+ }
+ ret = 0;
+
+end:
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ network_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
+ sizeof(struct lttcomm_relayd_generic_reply), 0);
+ if (network_ret < sizeof(struct lttcomm_relayd_generic_reply)) {
+ ERR("Relay sending stream id");
+ ret = -1;
+ }
+
+end_no_reply:
+ free(received_paths);
+ free(complete_old_path);
+ free(complete_new_path);
+ return ret;
+}
+
+/*
+ * Process the commands received on the control socket
+ */
+static int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret = 0;
+
+ switch (be32toh(recv_hdr->cmd)) {
+ case RELAYD_CREATE_SESSION:
+ ret = relay_create_session(recv_hdr, conn);
+ break;
+ case RELAYD_ADD_STREAM:
+ ret = relay_add_stream(recv_hdr, conn);
+ break;
+ case RELAYD_START_DATA:
+ ret = relay_start(recv_hdr, conn);
+ break;
+ case RELAYD_SEND_METADATA:
+ ret = relay_recv_metadata(recv_hdr, conn);
+ break;
+ case RELAYD_VERSION:
+ ret = relay_send_version(recv_hdr, conn);
+ break;
+ case RELAYD_CLOSE_STREAM:
+ ret = relay_close_stream(recv_hdr, conn);
+ break;
+ case RELAYD_DATA_PENDING:
+ ret = relay_data_pending(recv_hdr, conn);
+ break;
+ case RELAYD_QUIESCENT_CONTROL:
+ ret = relay_quiescent_control(recv_hdr, conn);
+ break;
+ case RELAYD_BEGIN_DATA_PENDING:
+ ret = relay_begin_data_pending(recv_hdr, conn);
+ break;
+ case RELAYD_END_DATA_PENDING:
+ ret = relay_end_data_pending(recv_hdr, conn);
+ break;
+ case RELAYD_SEND_INDEX:
+ ret = relay_recv_index(recv_hdr, conn);
+ break;
+ case RELAYD_STREAMS_SENT:
+ ret = relay_streams_sent(recv_hdr, conn);
+ break;
+ case RELAYD_RESET_METADATA:
+ ret = relay_reset_metadata(recv_hdr, conn);
+ break;
+ case RELAYD_ROTATE_RENAME:
+ ret = relay_rotate_rename(recv_hdr, conn);
+ break;
+ case RELAYD_MKDIR:
+ ret = relay_mkdir(recv_hdr, conn);
+ break;
+ case RELAYD_UPDATE_SYNC_INFO:
+ default:
+ ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
+ relay_unknown_command(conn);
+ ret = -1;
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+/*
+ * Handle index for a data stream.
+ *
+ * Called with the stream lock held.
+ *
+ * Return 0 on success else a negative value.
+ */
+static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
+ int rotate_index)
+{
+ int ret = 0;
+ uint64_t data_offset;
+ struct relay_index *index;
+
+ /* 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) {
+ uint32_t major, minor;
+
+ /* 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;
+ stream->index_file = lttng_index_file_create(stream->path_name,
+ stream->channel_name,
+ -1, -1, stream->tracefile_size,
+ tracefile_array_get_file_index_head(stream->tfa),
+ lttng_to_index_major(major, minor),
+ lttng_to_index_minor(major, minor));
+ if (!stream->index_file) {
+ ret = -1;
+ /* 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++;
+ } else if (ret > 0) {
+ /* No flush. */
+ ret = 0;
+ } else {
+ /* Put self-ref for this index due to error. */
+ relay_index_put(index);
+ index = NULL;
+ ret = -1;
+ }
+end: