X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=32169034095be1b9fe8889e3c9aee0906dc1e58f;hp=1b01d414a0248b0207bbf72aba620a8fa62b93e4;hb=e57427575fc6deabed09cb5b4fae029d05e5bfd9;hpb=0c759fc95033a3d6d7cb939f39dd643ce7e127ee diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 1b01d414a..321690340 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -16,7 +16,7 @@ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE +#define _LGPL_SOURCE #include #include #include @@ -35,12 +35,13 @@ #include #include #include +#include #include #include #include -#include +#include #include -#include +#include #include "kernel-consumer.h" @@ -60,7 +61,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 +81,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 +101,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 +114,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; @@ -140,6 +141,8 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, } cds_list_for_each_entry(stream, &channel->streams.head, send_node) { + /* Are we at a position _before_ the first available packet ? */ + bool before_first_packet = true; health_code_update(); @@ -176,6 +179,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) { @@ -212,18 +222,14 @@ 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; unsigned long len, padded_len; + int lost_packet = 0; health_code_update(); @@ -238,6 +244,15 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, } DBG("Kernel consumer get subbuf failed. Skipping it."); consumed_pos += stream->max_sb_size; + + /* + * Start accounting lost packets only when we + * already have extracted packets (to match the + * content of the final snapshot). + */ + if (!before_first_packet) { + lost_packet = 1; + } continue; } @@ -281,6 +296,16 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, goto end_unlock; } consumed_pos += stream->max_sb_size; + + /* + * Only account lost packets located between + * succesfully extracted packets (do not account before + * and after since they are not visible in the + * resulting snapshot). + */ + stream->chan->lost_packets += lost_packet; + lost_packet = 0; + before_first_packet = false; } if (relayd_id == (uint64_t) -1ULL) { @@ -434,17 +459,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(); @@ -485,7 +501,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, NULL); if (new_channel == NULL) { lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR); goto end_nosignal; @@ -563,7 +580,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(); @@ -586,9 +603,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(); @@ -644,6 +660,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; @@ -751,6 +771,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 = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } + + health_code_update(); + + /* + * Send status code to session daemon. + */ + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0 || ret_code != LTTCOMM_CONSUMERD_SUCCESS) { + /* 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(); @@ -767,7 +838,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; } /* @@ -826,17 +897,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; } } @@ -857,7 +928,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(); @@ -870,6 +941,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. @@ -886,6 +962,66 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; } + case LTTNG_CONSUMER_DISCARDED_EVENTS: + { + uint64_t ret; + struct lttng_consumer_channel *channel; + uint64_t id = msg.u.discarded_events.session_id; + uint64_t key = msg.u.discarded_events.channel_key; + + DBG("Kernel consumer discarded events command for session id %" + PRIu64 ", channel key %" PRIu64, id, key); + + channel = consumer_find_channel(key); + if (!channel) { + ERR("Kernel consumer discarded events channel %" + PRIu64 " not found", key); + ret = 0; + } else { + ret = channel->discarded_events; + } + + health_code_update(); + + /* Send back returned value to session daemon */ + ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + if (ret < 0) { + PERROR("send discarded events"); + goto error_fatal; + } + + break; + } + case LTTNG_CONSUMER_LOST_PACKETS: + { + uint64_t ret; + struct lttng_consumer_channel *channel; + uint64_t id = msg.u.lost_packets.session_id; + uint64_t key = msg.u.lost_packets.channel_key; + + DBG("Kernel consumer lost packets command for session id %" + PRIu64 ", channel key %" PRIu64, id, key); + + channel = consumer_find_channel(key); + if (!channel) { + ERR("Kernel consumer lost packets channel %" + PRIu64 " not found", key); + ret = 0; + } else { + ret = channel->lost_packets; + } + + health_code_update(); + + /* Send back returned value to session daemon */ + ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); + if (ret < 0) { + PERROR("send lost packets"); + goto error_fatal; + } + + break; + } default: goto end_nosignal; } @@ -911,7 +1047,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; @@ -957,6 +1093,20 @@ static int get_index_values(struct lttng_packet_index *index, int infd) } index->stream_id = htobe64(index->stream_id); + ret = kernctl_get_instance_id(infd, &index->stream_instance_id); + if (ret < 0) { + PERROR("kernctl_get_instance_id"); + goto error; + } + index->stream_instance_id = htobe64(index->stream_instance_id); + + ret = kernctl_get_sequence_number(infd, &index->packet_seq_num); + if (ret < 0) { + PERROR("kernctl_get_sequence_number"); + goto error; + } + index->packet_seq_num = htobe64(index->packet_seq_num); + error: return ret; } @@ -997,6 +1147,92 @@ end: return ret; } +static +int update_stream_stats(struct lttng_consumer_stream *stream) +{ + int ret; + uint64_t seq, discarded; + + ret = kernctl_get_sequence_number(stream->wait_fd, &seq); + if (ret < 0) { + PERROR("kernctl_get_sequence_number"); + goto end; + } + + /* + * Start the sequence when we extract the first packet in case we don't + * start at 0 (for example if a consumer is not connected to the + * session immediately after the beginning). + */ + if (stream->last_sequence_number == -1ULL) { + stream->last_sequence_number = seq; + } else if (seq > stream->last_sequence_number) { + stream->chan->lost_packets += seq - + stream->last_sequence_number - 1; + } else { + /* seq <= last_sequence_number */ + ERR("Sequence number inconsistent : prev = %" PRIu64 + ", current = %" PRIu64, + stream->last_sequence_number, seq); + ret = -1; + goto end; + } + stream->last_sequence_number = seq; + + ret = kernctl_get_events_discarded(stream->wait_fd, &discarded); + if (ret < 0) { + PERROR("kernctl_get_events_discarded"); + goto end; + } + if (discarded < stream->last_discarded_events) { + /* + * Overflow has occured. We assume only one wrap-around + * has occured. + */ + stream->chan->discarded_events += (1ULL << (CAA_BITS_PER_LONG - 1)) - + stream->last_discarded_events + discarded; + } else { + stream->chan->discarded_events += discarded - + stream->last_discarded_events; + } + stream->last_discarded_events = discarded; + ret = 0; + +end: + return ret; +} + +/* + * Check if the local version of the metadata stream matches with the version + * of the metadata stream in the kernel. If it was updated, set the reset flag + * on the stream. + */ +static +int metadata_stream_check_version(int infd, struct lttng_consumer_stream *stream) +{ + int ret; + uint64_t cur_version; + + ret = kernctl_get_metadata_version(infd, &cur_version); + if (ret < 0) { + ERR("Failed to get the metadata version"); + goto end; + } + + if (stream->metadata_version == cur_version) { + ret = 0; + goto end; + } + + DBG("New metadata version detected"); + stream->metadata_version = cur_version; + stream->reset_metadata_flag = 1; + ret = 0; + +end: + return ret; +} + /* * Consume data on a file descriptor and write it on a trace file. */ @@ -1007,7 +1243,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); @@ -1029,18 +1265,48 @@ 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; } 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; + } + ret = update_stream_stats(stream); if (ret < 0) { goto end; } } else { write_index = 0; + ret = metadata_stream_check_version(infd, stream); + if (ret < 0) { + goto end; + } } switch (stream->chan->output) { @@ -1074,7 +1340,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; } @@ -1096,9 +1373,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; @@ -1112,10 +1390,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; @@ -1130,7 +1408,22 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, /* * In live, block until all the metadata is sent. */ + pthread_mutex_lock(&stream->metadata_timer_lock); + assert(!stream->missed_metadata_flush); + stream->waiting_on_metadata = true; + pthread_mutex_unlock(&stream->metadata_timer_lock); + err = consumer_stream_sync_metadata(ctx, stream->session_id); + + pthread_mutex_lock(&stream->metadata_timer_lock); + stream->waiting_on_metadata = false; + if (stream->missed_metadata_flush) { + stream->missed_metadata_flush = false; + pthread_mutex_unlock(&stream->metadata_timer_lock); + (void) consumer_flush_kernel_index(stream); + } else { + pthread_mutex_unlock(&stream->metadata_timer_lock); + } if (err < 0) { goto end; }