X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=d9c4d2685f6cc7e1da9a6c0ccac0b0a56757ace3;hb=7881b1c495675c9b0f9df93fbaeff6cf49310f2a;hp=e5c0c2e83a3930e1079ef96d056767b64893a7c7;hpb=234cd6367843a2106a4cb10f8fb99443208516df;p=lttng-tools.git diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index e5c0c2e83..d9c4d2685 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -118,7 +118,6 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, struct lttng_consumer_local_data *ctx) { int ret; - unsigned long consumed_pos, produced_pos; struct lttng_consumer_channel *channel; struct lttng_consumer_stream *stream; @@ -143,6 +142,7 @@ 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; + unsigned long consumed_pos, produced_pos; health_code_update(); @@ -185,6 +185,7 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, ERR("sending streams sent to relayd"); goto end_unlock; } + channel->streams_sent_to_relayd = true; } ret = kernctl_buffer_flush(stream->wait_fd); @@ -726,6 +727,19 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, consumer_stream_free(new_stream); goto end_nosignal; } + + /* + * If adding an extra stream to an already + * existing channel (e.g. cpu hotplug), we need + * to send the "streams_sent" command to relayd. + */ + if (channel->streams_sent_to_relayd) { + ret = consumer_send_relayd_streams_sent( + new_stream->net_seq_idx); + if (ret < 0) { + goto end_nosignal; + } + } } /* Get the right pipe where the stream will be sent. */ @@ -819,6 +833,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (ret < 0) { goto end_nosignal; } + channel->streams_sent_to_relayd = true; } break; } @@ -969,18 +984,18 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + ret = 0; + } else { + ret = channel->discarded_events; } - DBG("Kernel consumer discarded events command for session id %" - PRIu64 ", channel key %" PRIu64, id, key); - - ret = channel->discarded_events; - health_code_update(); /* Send back returned value to session daemon */ @@ -999,18 +1014,18 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + ret = 0; + } else { + ret = channel->lost_packets; } - DBG("Kernel consumer lost packets command for session id %" - PRIu64 ", channel key %" PRIu64, id, key); - - ret = channel->lost_packets; - health_code_update(); /* Send back returned value to session daemon */ @@ -1095,15 +1110,27 @@ static int get_index_values(struct ctf_packet_index *index, int infd) ret = kernctl_get_instance_id(infd, &index->stream_instance_id); if (ret < 0) { - PERROR("kernctl_get_instance_id"); - goto error; + if (ret == -ENOTTY) { + /* Command not implemented by lttng-modules. */ + index->stream_instance_id = -1ULL; + ret = 0; + } else { + 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; + if (ret == -ENOTTY) { + /* Command not implemented by lttng-modules. */ + index->packet_seq_num = -1ULL; + ret = 0; + } else { + PERROR("kernctl_get_sequence_number"); + goto error; + } } index->packet_seq_num = htobe64(index->packet_seq_num); @@ -1155,8 +1182,14 @@ int update_stream_stats(struct lttng_consumer_stream *stream) ret = kernctl_get_sequence_number(stream->wait_fd, &seq); if (ret < 0) { - PERROR("kernctl_get_sequence_number"); - goto end; + if (ret == -ENOTTY) { + /* Command not implemented by lttng-modules. */ + seq = -1ULL; + ret = 0; + } else { + PERROR("kernctl_get_sequence_number"); + goto end; + } } /* @@ -1186,8 +1219,8 @@ int update_stream_stats(struct lttng_consumer_stream *stream) } if (discarded < stream->last_discarded_events) { /* - * Overflow has occured. We assume only one wrap-around - * has occured. + * Overflow has occurred. We assume only one wrap-around + * has occurred. */ stream->chan->discarded_events += (1ULL << (CAA_BITS_PER_LONG - 1)) - stream->last_discarded_events + discarded; @@ -1202,6 +1235,45 @@ 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) { + if (ret == -ENOTTY) { + /* + * LTTng-modules does not implement this + * command. + */ + ret = 0; + goto end; + } + 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. */ @@ -1268,10 +1340,36 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, } ret = update_stream_stats(stream); if (ret < 0) { + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (err == -EFAULT) { + PERROR("Error in unreserving sub buffer\n"); + } else if (err == -EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = err; + goto end; + } goto end; } } else { write_index = 0; + ret = metadata_stream_check_version(infd, stream); + if (ret < 0) { + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (err == -EFAULT) { + PERROR("Error in unreserving sub buffer\n"); + } else if (err == -EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = err; + goto end; + } + goto end; + } } switch (stream->chan->output) { @@ -1424,14 +1522,17 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) stream->tracefile_size_current = 0; if (!stream->metadata_flag) { - ret = index_create_file(stream->chan->pathname, + struct lttng_index_file *index_file; + + index_file = lttng_index_file_create(stream->chan->pathname, stream->name, stream->uid, stream->gid, stream->chan->tracefile_size, - stream->tracefile_count_current); - if (ret < 0) { + stream->tracefile_count_current, + CTF_INDEX_MAJOR, CTF_INDEX_MINOR); + if (!index_file) { goto error; } - stream->index_fd = ret; + stream->index_file = index_file; } }