+ if (is_local_trace) {
+ int chunk_dirfd;
+ int ret_send_status;
+ ssize_t ret_recv;
+
+ /* Acnowledge the reception of the command. */
+ ret_send_status = consumer_send_status_msg(
+ sock, LTTCOMM_CONSUMERD_SUCCESS);
+ if (ret_send_status < 0) {
+ /* Somehow, the session daemon is not responding anymore. */
+ goto end_nosignal;
+ }
+
+ ret_recv = lttcomm_recv_fds_unix_sock(
+ sock, &chunk_dirfd, 1);
+ if (ret_recv != sizeof(chunk_dirfd)) {
+ ERR("Failed to receive trace chunk directory file descriptor");
+ goto error_fatal;
+ }
+
+ DBG("Received trace chunk directory fd (%d)",
+ chunk_dirfd);
+ chunk_directory_handle = lttng_directory_handle_create_from_dirfd(
+ chunk_dirfd);
+ if (!chunk_directory_handle) {
+ 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;
+ }
+ }
+
+ 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);
+ lttng_directory_handle_put(chunk_directory_handle);
+ 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 path[LTTNG_PATH_MAX];
+ ssize_t ret_send;
+
+ 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, path);
+ reply.ret_code = ret_code;
+ reply.path_length = strlen(path) + 1;
+ ret_send = lttcomm_send_unix_sock(sock, &reply, sizeof(reply));
+ if (ret_send != sizeof(reply)) {
+ goto error_fatal;
+ }
+ ret_send = lttcomm_send_unix_sock(
+ sock, path, reply.path_length);
+ if (ret_send != 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;
+ }
+ case LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS:
+ {
+ const uint64_t key = msg.u.open_channel_packets.key;
+ struct lttng_consumer_channel *channel =
+ consumer_find_channel(key);
+
+ if (channel) {
+ pthread_mutex_lock(&channel->lock);
+ ret_code = lttng_consumer_open_channel_packets(channel);
+ pthread_mutex_unlock(&channel->lock);
+ } else {
+ WARN("Channel %" PRIu64 " not found", key);
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ }
+
+ health_code_update();
+ goto end_msg_sessiond;
+ }
+ default:
+ goto end_nosignal;
+ }
+
+end_nosignal:
+ /*
+ * Return 1 to indicate success since the 0 value can be a socket
+ * shutdown during the recv() or send() call.
+ */
+ ret_func = 1;
+ goto end;
+error_fatal:
+ /* This will issue a consumer stop. */
+ ret_func = -1;
+ goto end;
+end_msg_sessiond:
+ /*
+ * The returned value here is not useful since either way we'll return 1 to
+ * the caller because the session daemon socket management is done
+ * elsewhere. Returning a negative code or 0 will shutdown the consumer.
+ */
+ {
+ int ret_send_status;
+
+ ret_send_status = consumer_send_status_msg(sock, ret_code);
+ if (ret_send_status < 0) {
+ goto error_fatal;
+ }
+ }
+
+ ret_func = 1;
+
+end:
+ health_code_update();
+ rcu_read_unlock();
+ return ret_func;
+}
+
+/*
+ * Sync metadata meaning request them to the session daemon and snapshot to the
+ * metadata thread can consumer them.
+ *
+ * Metadata stream lock MUST be acquired.
+ */
+enum sync_metadata_status lttng_kconsumer_sync_metadata(
+ struct lttng_consumer_stream *metadata)
+{
+ int ret;
+ enum sync_metadata_status status;
+
+ LTTNG_ASSERT(metadata);
+
+ ret = kernctl_buffer_flush(metadata->wait_fd);
+ if (ret < 0) {
+ ERR("Failed to flush kernel stream");
+ status = SYNC_METADATA_STATUS_ERROR;
+ goto end;
+ }
+
+ ret = kernctl_snapshot(metadata->wait_fd);
+ if (ret < 0) {
+ if (errno == EAGAIN) {
+ /* No new metadata, exit. */
+ DBG("Sync metadata, no new kernel metadata");
+ status = SYNC_METADATA_STATUS_NO_DATA;
+ } else {
+ ERR("Sync metadata, taking kernel snapshot failed.");
+ status = SYNC_METADATA_STATUS_ERROR;
+ }
+ } else {
+ status = SYNC_METADATA_STATUS_NEW_DATA;
+ }
+
+end:
+ return status;
+}
+
+static
+int extract_common_subbuffer_info(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuf)
+{
+ int ret;
+
+ ret = kernctl_get_subbuf_size(
+ stream->wait_fd, &subbuf->info.data.subbuf_size);
+ if (ret) {
+ goto end;
+ }
+
+ ret = kernctl_get_padded_subbuf_size(
+ stream->wait_fd, &subbuf->info.data.padded_subbuf_size);
+ if (ret) {
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+static
+int extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuf)
+{
+ int ret;
+
+ ret = extract_common_subbuffer_info(stream, subbuf);
+ if (ret) {
+ goto end;
+ }
+
+ ret = kernctl_get_metadata_version(
+ stream->wait_fd, &subbuf->info.metadata.version);
+ if (ret) {
+ goto end;
+ }
+
+end:
+ return ret;
+}
+
+static
+int extract_data_subbuffer_info(struct lttng_consumer_stream *stream,
+ struct stream_subbuffer *subbuf)
+{
+ int ret;
+
+ ret = extract_common_subbuffer_info(stream, subbuf);
+ if (ret) {
+ goto end;
+ }
+
+ ret = kernctl_get_packet_size(
+ stream->wait_fd, &subbuf->info.data.packet_size);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer packet size");
+ goto end;
+ }
+
+ ret = kernctl_get_content_size(
+ stream->wait_fd, &subbuf->info.data.content_size);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer content size");
+ goto end;
+ }
+
+ ret = kernctl_get_timestamp_begin(
+ stream->wait_fd, &subbuf->info.data.timestamp_begin);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer begin timestamp");
+ goto end;
+ }
+
+ ret = kernctl_get_timestamp_end(
+ stream->wait_fd, &subbuf->info.data.timestamp_end);
+ if (ret < 0) {
+ PERROR("Failed to get sub-buffer end timestamp");
+ goto end;
+ }
+
+ ret = kernctl_get_events_discarded(
+ stream->wait_fd, &subbuf->info.data.events_discarded);
+ if (ret) {
+ PERROR("Failed to get sub-buffer events discarded count");
+ goto end;
+ }
+
+ ret = kernctl_get_sequence_number(stream->wait_fd,
+ &subbuf->info.data.sequence_number.value);
+ if (ret) {
+ /* May not be supported by older LTTng-modules. */
+ if (ret != -ENOTTY) {
+ PERROR("Failed to get sub-buffer sequence number");
+ goto end;
+ }
+ } else {
+ subbuf->info.data.sequence_number.is_set = true;
+ }
+
+ ret = kernctl_get_stream_id(
+ stream->wait_fd, &subbuf->info.data.stream_id);
+ if (ret < 0) {
+ PERROR("Failed to get stream id");
+ goto end;
+ }
+
+ ret = kernctl_get_instance_id(stream->wait_fd,
+ &subbuf->info.data.stream_instance_id.value);
+ if (ret) {
+ /* May not be supported by older LTTng-modules. */
+ if (ret != -ENOTTY) {
+ PERROR("Failed to get stream instance id");
+ goto end;