X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=f7704b46564a63273dabe03b9e181b4846ed4292;hb=d73bf3d793ee0b0c5b56cb47cb50c27d1789d3bd;hp=7bcb86a36e21a816b50e00216e978a6ffe296688;hpb=2527bf859786192dd530b408717960495f48d208;p=lttng-tools.git diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 7bcb86a36..f7704b465 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -60,7 +60,11 @@ int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream) 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."); } @@ -187,14 +191,6 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, 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) { @@ -662,6 +658,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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; @@ -748,26 +746,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, /* 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(); @@ -983,7 +969,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } case LTTNG_CONSUMER_DISCARDED_EVENTS: { - uint64_t ret; + ssize_t ret; + uint64_t count; struct lttng_consumer_channel *channel; uint64_t id = msg.u.discarded_events.session_id; uint64_t key = msg.u.discarded_events.channel_key; @@ -995,15 +982,15 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel) { ERR("Kernel consumer discarded events channel %" PRIu64 " not found", key); - ret = 0; + count = 0; } else { - ret = channel->discarded_events; + count = channel->discarded_events; } health_code_update(); /* Send back returned value to session daemon */ - ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + ret = lttcomm_send_unix_sock(sock, &count, sizeof(count)); if (ret < 0) { PERROR("send discarded events"); goto error_fatal; @@ -1013,7 +1000,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } case LTTNG_CONSUMER_LOST_PACKETS: { - uint64_t ret; + ssize_t ret; + uint64_t count; struct lttng_consumer_channel *channel; uint64_t id = msg.u.lost_packets.session_id; uint64_t key = msg.u.lost_packets.channel_key; @@ -1025,15 +1013,15 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel) { ERR("Kernel consumer lost packets channel %" PRIu64 " not found", key); - ret = 0; + count = 0; } else { - ret = channel->lost_packets; + count = channel->lost_packets; } health_code_update(); /* Send back returned value to session daemon */ - ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + ret = lttcomm_send_unix_sock(sock, &count, sizeof(count)); if (ret < 0) { PERROR("send lost packets"); goto error_fatal; @@ -1090,6 +1078,95 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } 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; }