X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;ds=sidebyside;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=0af417b9cdb701f657bb34ddcf0d65a1487bb455;hb=d7ba13889c8692b14f99238ddf2721ed78df89d2;hp=061123d61160d32016249aa9856b8f0e4a43b5c6;hpb=2336629e66d2d2a7bdf85523b32f87aa49d05238;p=lttng-tools.git diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 061123d61..0af417b9c 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -17,6 +17,7 @@ */ #define _GNU_SOURCE +#define _LGPL_SOURCE #include #include #include @@ -35,6 +36,7 @@ #include #include #include +#include #include #include #include @@ -60,7 +62,7 @@ int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream) ret = kernctl_snapshot(infd); if (ret != 0) { - perror("Getting sub-buffer snapshot."); + PERROR("Getting sub-buffer snapshot."); ret = -errno; } @@ -80,7 +82,7 @@ int lttng_kconsumer_get_produced_snapshot(struct lttng_consumer_stream *stream, ret = kernctl_snapshot_get_produced(infd, pos); if (ret != 0) { - perror("kernctl_snapshot_get_produced"); + PERROR("kernctl_snapshot_get_produced"); ret = -errno; } @@ -100,7 +102,7 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream, ret = kernctl_snapshot_get_consumed(infd, pos); if (ret != 0) { - perror("kernctl_snapshot_get_consumed"); + PERROR("kernctl_snapshot_get_consumed"); ret = -errno; } @@ -113,7 +115,7 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream, * Returns 0 on success, < 0 on error */ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, - uint64_t relayd_id, uint64_t max_stream_size, + uint64_t relayd_id, uint64_t nb_packets_per_stream, struct lttng_consumer_local_data *ctx) { int ret; @@ -219,14 +221,9 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, } } - /* - * The original value is sent back if max stream size is larger than - * the possible size of the snapshot. Also, we asume that the session - * daemon should never send a maximum stream size that is lower than - * subbuffer size. - */ - consumed_pos = consumer_get_consumed_maxsize(consumed_pos, - produced_pos, max_stream_size); + consumed_pos = consumer_get_consume_start_pos(consumed_pos, + produced_pos, nb_packets_per_stream, + stream->max_sb_size); while (consumed_pos < produced_pos) { ssize_t read_len; @@ -441,17 +438,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(); @@ -492,7 +480,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.channel.tracefile_size, msg.u.channel.tracefile_count, 0, msg.u.channel.monitor, - msg.u.channel.live_timer_interval); + msg.u.channel.live_timer_interval, + NULL); if (new_channel == NULL) { lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR); goto end_nosignal; @@ -570,7 +559,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, * happens while tearing down. */ ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key); - ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND; + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; } health_code_update(); @@ -593,9 +582,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(); @@ -651,6 +639,10 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, switch (channel->output) { case CONSUMER_CHANNEL_SPLICE: new_stream->output = LTTNG_EVENT_SPLICE; + ret = utils_create_pipe(new_stream->splice_pipe); + if (ret < 0) { + goto end_nosignal; + } break; case CONSUMER_CHANNEL_MMAP: new_stream->output = LTTNG_EVENT_MMAP; @@ -774,7 +766,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, */ ERR("Unable to find channel key %" PRIu64, msg.u.sent_streams.channel_key); - ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND; + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; } health_code_update(); @@ -783,7 +775,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, * Send status code to session daemon. */ ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { + if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) { /* Somehow, the session daemon is not responding anymore. */ goto end_nosignal; } @@ -825,7 +817,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, relayd = consumer_find_relayd(index); if (relayd == NULL) { DBG("Unable to find relayd %" PRIu64, index); - ret_code = LTTNG_ERR_NO_CONSUMER; + ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL; } /* @@ -884,17 +876,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.snapshot_channel.relayd_id, ctx); if (ret < 0) { ERR("Snapshot metadata failed"); - ret_code = LTTNG_ERR_KERN_META_FAIL; + ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA; } } else { ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key, msg.u.snapshot_channel.pathname, msg.u.snapshot_channel.relayd_id, - msg.u.snapshot_channel.max_stream_size, + msg.u.snapshot_channel.nb_packets_per_stream, ctx); if (ret < 0) { ERR("Snapshot channel failed"); - ret_code = LTTNG_ERR_KERN_CHAN_FAIL; + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; } } @@ -915,7 +907,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, channel = consumer_find_channel(key); if (!channel) { ERR("Kernel consumer destroy channel %" PRIu64 " not found", key); - ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND; + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; } health_code_update(); @@ -928,6 +920,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. @@ -1087,7 +1084,18 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, /* Get the full subbuffer size including padding */ err = kernctl_get_padded_subbuf_size(infd, &len); if (err != 0) { - perror("Getting sub-buffer len failed."); + PERROR("Getting sub-buffer len failed."); + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (errno == EFAULT) { + PERROR("Error in unreserving sub buffer\n"); + } else if (errno == EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = -errno; + goto end; + } ret = -errno; goto end; } @@ -1095,6 +1103,17 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, if (!stream->metadata_flag) { ret = get_index_values(&index, infd); if (ret < 0) { + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (errno == EFAULT) { + PERROR("Error in unreserving sub buffer\n"); + } else if (errno == EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = -errno; + goto end; + } goto end; } } else { @@ -1132,7 +1151,18 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, /* Get subbuffer size without padding */ err = kernctl_get_subbuf_size(infd, &subbuf_size); if (err != 0) { - perror("Getting sub-buffer len failed."); + PERROR("Getting sub-buffer len failed."); + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (errno == EFAULT) { + PERROR("Error in unreserving sub buffer\n"); + } else if (errno == EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = -errno; + goto end; + } ret = -errno; goto end; } @@ -1171,10 +1201,10 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, err = kernctl_put_next_subbuf(infd); if (err != 0) { if (errno == EFAULT) { - perror("Error in unreserving sub buffer\n"); + PERROR("Error in unreserving sub buffer\n"); } else if (errno == EIO) { /* Should never happen with newer LTTng versions */ - perror("Reader has been pushed by the writer, last sub-buffer corrupted."); + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } ret = -errno; goto end;