X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=57dc2ba6a76767d623c62fc552cb413fd0a0e71f;hp=a506737ab4170880b23c28a959c546e65ed9f81c;hb=f263b7fd113e51d0737554e8232b8669e142a260;hpb=9ce5646a7ef9b8d7936c46649a21ee546fadd538 diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index a506737ab..57dc2ba6a 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -29,11 +29,13 @@ #include #include +#include #include #include #include #include #include +#include #include #include #include @@ -42,7 +44,6 @@ #include #include "kernel-consumer.h" -#include "../../bin/lttng-consumerd/health-consumerd.h" extern struct lttng_consumer_global_data consumer_data; extern int consumer_poll_timeout; @@ -176,6 +177,13 @@ 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; + } + } ret = kernctl_buffer_flush(stream->wait_fd); if (ret < 0) { @@ -418,7 +426,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll) { ssize_t ret; - enum lttng_error_code ret_code = LTTNG_OK; + enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS; struct lttcomm_consumer_msg msg; health_code_update(); @@ -434,17 +442,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); - if (msg.cmd_type == LTTNG_CONSUMER_STOP) { - /* - * Notify the session daemon that the command is completed. - * - * On transport layer error, the function call will print an error - * message so handling the returned code is a bit useless since we - * return an error code anyway. - */ - (void) consumer_send_status_msg(sock, ret_code); - return -ENOENT; - } + /* Deprecated command */ + assert(msg.cmd_type != LTTNG_CONSUMER_STOP); health_code_update(); @@ -577,7 +576,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); - if (ret_code != LTTNG_OK) { + if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) { /* Channel was not found. */ goto end_nosignal; } @@ -586,9 +585,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_poll_entry(); ret = lttng_consumer_poll_socket(consumer_sockpoll); health_poll_exit(); - if (ret < 0) { - rcu_read_unlock(); - return -EINTR; + if (ret) { + goto error_fatal; } health_code_update(); @@ -751,6 +749,57 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, new_stream->name, fd, new_stream->relayd_stream_id); break; } + case LTTNG_CONSUMER_STREAMS_SENT: + { + struct lttng_consumer_channel *channel; + + /* + * Get stream's channel reference. Needed when adding the stream to the + * global hash table. + */ + channel = consumer_find_channel(msg.u.sent_streams.channel_key); + if (!channel) { + /* + * We could not find the channel. Can happen if cpu hotplug + * happens while tearing down. + */ + ERR("Unable to find channel key %" PRIu64, + msg.u.sent_streams.channel_key); + ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND; + } + + health_code_update(); + + /* + * Send status code to session daemon. + */ + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + + health_code_update(); + + /* + * We should not send this message if we don't monitor the + * streams in this channel. + */ + if (!channel->monitor) { + break; + } + + health_code_update(); + /* Send stream to relayd if the stream has an ID. */ + if (msg.u.sent_streams.net_seq_idx != (uint64_t) -1ULL) { + ret = consumer_send_relayd_streams_sent( + msg.u.sent_streams.net_seq_idx); + if (ret < 0) { + goto end_nosignal; + } + } + break; + } case LTTNG_CONSUMER_UPDATE_STREAM: { rcu_read_unlock(); @@ -870,6 +919,11 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); + /* Stop right now if no channel was found. */ + if (!channel) { + goto end_nosignal; + } + /* * This command should ONLY be issued for channel with streams set in * no monitor mode. @@ -911,7 +965,7 @@ error_fatal: * * Return 0 on success or else a negative value. */ -static int get_index_values(struct lttng_packet_index *index, int infd) +static int get_index_values(struct ctf_packet_index *index, int infd) { int ret; @@ -1007,7 +1061,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, int err, write_index = 1; ssize_t ret = 0; int infd = stream->wait_fd; - struct lttng_packet_index index; + struct ctf_packet_index index; DBG("In read_subbuffer (infd : %d)", infd); @@ -1096,9 +1150,10 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) { /* * Display the error but continue processing to try to release the - * subbuffer + * subbuffer. This is a DBG statement since this is possible to + * happen without being a critical error. */ - ERR("Error writing to tracefile " + DBG("Error writing to tracefile " "(ret: %zd != len: %lu != subbuf_size: %lu)", ret, len, subbuf_size); write_index = 0;