+ channel = consumer_find_channel(key);
+ if (!channel) {
+ DBG("Channel %" PRIu64 " not found", key);
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ } else {
+ /*
+ * Sample the rotate position of all the streams in
+ * this channel.
+ */
+ ret = lttng_consumer_rotate_channel(channel, key,
+ msg.u.rotate_channel.relayd_id,
+ msg.u.rotate_channel.metadata,
+ ctx);
+ if (ret < 0) {
+ ERR("Rotate channel failed");
+ ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL;
+ }
+
+ health_code_update();
+ }
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ /* Somehow, the session daemon is not responding anymore. */
+ goto end_rotate_channel_nosignal;
+ }
+
+ /*
+ * Rotate the streams that are ready right now.
+ * FIXME: this is a second consecutive iteration over the
+ * streams in a channel, there is probably a better way to
+ * handle this, but it needs to be after the
+ * consumer_send_status_msg() call.
+ */
+ if (channel) {
+ ret = lttng_consumer_rotate_ready_streams(
+ channel, key, ctx);
+ if (ret < 0) {
+ ERR("Rotate channel failed");
+ }
+ }
+ break;
+end_rotate_channel_nosignal:
+ goto end_nosignal;
+ }
+ case LTTNG_CONSUMER_INIT:
+ {
+ ret_code = lttng_consumer_init_command(ctx,
+ msg.u.init.sessiond_uuid);
+ health_code_update();
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ /* Somehow, the session daemon is not responding anymore. */
+ goto end_nosignal;
+ }
+ break;
+ }
+ case LTTNG_CONSUMER_CREATE_TRACE_CHUNK:
+ {
+ const struct lttng_credentials credentials = {
+ .uid = msg.u.create_trace_chunk.credentials.value.uid,
+ .gid = msg.u.create_trace_chunk.credentials.value.gid,
+ };
+ const bool is_local_trace =
+ !msg.u.create_trace_chunk.relayd_id.is_set;
+ const uint64_t relayd_id =
+ msg.u.create_trace_chunk.relayd_id.value;
+ const char *chunk_override_name =
+ *msg.u.create_trace_chunk.override_name ?
+ msg.u.create_trace_chunk.override_name :
+ NULL;
+ LTTNG_OPTIONAL(struct lttng_directory_handle) chunk_directory_handle =
+ LTTNG_OPTIONAL_INIT;
+
+ /*
+ * The session daemon will only provide a chunk directory file
+ * descriptor for local traces.
+ */
+ if (is_local_trace) {
+ int chunk_dirfd;
+
+ /* Acnowledge the reception of the command. */
+ ret = consumer_send_status_msg(sock,
+ LTTCOMM_CONSUMERD_SUCCESS);
+ if (ret < 0) {
+ /* Somehow, the session daemon is not responding anymore. */
+ goto end_nosignal;
+ }
+
+ ret = lttcomm_recv_fds_unix_sock(sock, &chunk_dirfd, 1);
+ if (ret != sizeof(chunk_dirfd)) {
+ ERR("Failed to receive trace chunk directory file descriptor");
+ goto error_fatal;
+ }
+
+ DBG("Received trace chunk directory fd (%d)",
+ chunk_dirfd);
+ ret = lttng_directory_handle_init_from_dirfd(
+ &chunk_directory_handle.value,
+ chunk_dirfd);
+ if (ret) {
+ ERR("Failed to initialize chunk directory handle from directory file descriptor");
+ if (close(chunk_dirfd)) {
+ PERROR("Failed to close chunk directory file descriptor");
+ }
+ goto error_fatal;
+ }
+ chunk_directory_handle.is_set = true;
+ }
+
+ ret_code = lttng_consumer_create_trace_chunk(
+ !is_local_trace ? &relayd_id : NULL,
+ msg.u.create_trace_chunk.session_id,
+ msg.u.create_trace_chunk.chunk_id,
+ (time_t) msg.u.create_trace_chunk
+ .creation_timestamp,
+ chunk_override_name,
+ msg.u.create_trace_chunk.credentials.is_set ?
+ &credentials :
+ NULL,
+ chunk_directory_handle.is_set ?
+ &chunk_directory_handle.value :
+ NULL);
+
+ if (chunk_directory_handle.is_set) {
+ lttng_directory_handle_fini(
+ &chunk_directory_handle.value);
+ }
+ goto end_msg_sessiond;
+ }
+ case LTTNG_CONSUMER_CLOSE_TRACE_CHUNK:
+ {
+ enum lttng_trace_chunk_command_type close_command =
+ msg.u.close_trace_chunk.close_command.value;
+ const uint64_t relayd_id =
+ msg.u.close_trace_chunk.relayd_id.value;
+ struct lttcomm_consumer_close_trace_chunk_reply reply;
+ char closed_trace_chunk_path[LTTNG_PATH_MAX];
+ int ret;
+
+ ret_code = lttng_consumer_close_trace_chunk(
+ msg.u.close_trace_chunk.relayd_id.is_set ?
+ &relayd_id :
+ NULL,
+ msg.u.close_trace_chunk.session_id,
+ msg.u.close_trace_chunk.chunk_id,
+ (time_t) msg.u.close_trace_chunk.close_timestamp,
+ msg.u.close_trace_chunk.close_command.is_set ?
+ &close_command :
+ NULL, closed_trace_chunk_path);
+ reply.ret_code = ret_code;
+ reply.path_length = strlen(closed_trace_chunk_path) + 1;
+ ret = lttcomm_send_unix_sock(sock, &reply, sizeof(reply));
+ if (ret != sizeof(reply)) {
+ goto error_fatal;
+ }
+ ret = lttcomm_send_unix_sock(sock, closed_trace_chunk_path,
+ reply.path_length);
+ if (ret != reply.path_length) {
+ goto error_fatal;
+ }
+ goto end_nosignal;
+ }
+ case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS:
+ {
+ const uint64_t relayd_id =
+ msg.u.trace_chunk_exists.relayd_id.value;
+
+ ret_code = lttng_consumer_trace_chunk_exists(
+ msg.u.trace_chunk_exists.relayd_id.is_set ?
+ &relayd_id : NULL,
+ msg.u.trace_chunk_exists.session_id,
+ msg.u.trace_chunk_exists.chunk_id);
+ goto end_msg_sessiond;
+ }
+ default:
+ break;
+ }
+
+end_nosignal: