int infd = stream->wait_fd;
ret = kernctl_snapshot(infd);
- if (ret != 0) {
+ /*
+ * -EAGAIN is not an error, it just means that there is no data to
+ * be read.
+ */
+ if (ret != 0 && ret != -EAGAIN) {
PERROR("Getting sub-buffer snapshot.");
}
DBG("Kernel consumer snapshot stream %s/%s (%" PRIu64 ")",
path, stream->name, stream->key);
}
- if (relayd_id != -1ULL) {
- ret = consumer_send_relayd_streams_sent(relayd_id);
- if (ret < 0) {
- ERR("sending streams sent to relayd");
- goto end_unlock;
- }
- channel->streams_sent_to_relayd = true;
- }
ret = kernctl_buffer_flush_empty(stream->wait_fd);
if (ret < 0) {
new_stream->chan = channel;
new_stream->wait_fd = fd;
+ consumer_stream_update_channel_attributes(new_stream,
+ channel);
switch (channel->output) {
case CONSUMER_CHANNEL_SPLICE:
new_stream->output = LTTNG_EVENT_SPLICE;
/* Get the right pipe where the stream will be sent. */
if (new_stream->metadata_flag) {
- ret = consumer_add_metadata_stream(new_stream);
- if (ret) {
- ERR("Consumer add metadata stream %" PRIu64 " failed. Continuing",
- new_stream->key);
- consumer_stream_free(new_stream);
- goto end_nosignal;
- }
+ consumer_add_metadata_stream(new_stream);
stream_pipe = ctx->consumer_metadata_pipe;
} else {
- ret = consumer_add_data_stream(new_stream);
- if (ret) {
- ERR("Consumer add stream %" PRIu64 " failed. Continuing",
- new_stream->key);
- consumer_stream_free(new_stream);
- goto end_nosignal;
- }
+ consumer_add_data_stream(new_stream);
stream_pipe = ctx->consumer_data_pipe;
}
- /* Vitible to other threads */
+ /* Visible to other threads */
new_stream->globally_visible = 1;
health_code_update();
}
break;
}
+ case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE:
+ {
+ int channel_rotate_pipe;
+ int flags;
+
+ ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ /* Successfully received the command's type. */
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ goto error_fatal;
+ }
+
+ ret = lttcomm_recv_fds_unix_sock(sock, &channel_rotate_pipe, 1);
+ if (ret != (ssize_t) sizeof(channel_rotate_pipe)) {
+ ERR("Failed to receive channel rotate pipe");
+ goto error_fatal;
+ }
+
+ DBG("Received channel rotate pipe (%d)", channel_rotate_pipe);
+ ctx->channel_rotate_pipe = channel_rotate_pipe;
+ /* Set the pipe as non-blocking. */
+ ret = fcntl(channel_rotate_pipe, F_GETFL, 0);
+ if (ret == -1) {
+ PERROR("fcntl get flags of the channel rotate pipe");
+ goto error_fatal;
+ }
+ flags = ret;
+
+ ret = fcntl(channel_rotate_pipe, F_SETFL, flags | O_NONBLOCK);
+ if (ret == -1) {
+ PERROR("fcntl set O_NONBLOCK flag of the channel rotate pipe");
+ goto error_fatal;
+ }
+ DBG("Channel rotate pipe set as non-blocking");
+ ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ goto error_fatal;
+ }
+ break;
+ }
+ case LTTNG_CONSUMER_ROTATE_RENAME:
+ {
+ DBG("Consumer rename session %" PRIu64 " after rotation, old path = \"%s\", new path = \"%s\"",
+ msg.u.rotate_rename.session_id,
+ msg.u.rotate_rename.old_path,
+ msg.u.rotate_rename.new_path);
+ ret = lttng_consumer_rotate_rename(msg.u.rotate_rename.old_path,
+ msg.u.rotate_rename.new_path,
+ msg.u.rotate_rename.uid,
+ msg.u.rotate_rename.gid,
+ msg.u.rotate_rename.relayd_id);
+ if (ret < 0) {
+ ERR("Rotate rename failed");
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ }
+
+ 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_MKDIR:
+ {
+ DBG("Consumer mkdir %s in session %" PRIu64,
+ msg.u.mkdir.path,
+ msg.u.mkdir.session_id);
+ ret = lttng_consumer_mkdir(msg.u.mkdir.path,
+ msg.u.mkdir.uid,
+ msg.u.mkdir.gid,
+ msg.u.mkdir.relayd_id);
+ if (ret < 0) {
+ ERR("consumer mkdir failed");
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ }
+
+ 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;
+ }
default:
goto end_nosignal;
}