X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=49b5070f76171b4613922bd84c1f4b7f7b941193;hb=ace0e591600a76d76d54edef87ffb24afff6d209;hp=42d471d1a9a040b10e6ee48b56d66f34544f8091;hpb=c13082a921ed95e1039d1e28da35941930b15fb6;p=lttng-tools.git diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 42d471d1a..49b5070f7 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -16,6 +16,8 @@ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ +#include "common/buffer-view.h" +#include #define _LGPL_SOURCE #include #include @@ -60,9 +62,12 @@ 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."); - ret = -errno; } return ret; @@ -82,7 +87,6 @@ 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"); - ret = -errno; } return ret; @@ -102,12 +106,30 @@ 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"); - ret = -errno; } return ret; } +static +int get_current_subbuf_addr(struct lttng_consumer_stream *stream, + const char **addr) +{ + int ret; + unsigned long mmap_offset; + const char *mmap_base = stream->mmap_base; + + ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset); + if (ret < 0) { + PERROR("Failed to get mmap read offset"); + goto error; + } + + *addr = mmap_base + mmap_offset; +error: + return ret; +} + /* * Take a snapshot of all the stream of a channel * @@ -118,7 +140,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; @@ -141,8 +162,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(); @@ -155,7 +175,7 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, * Assign the received relayd ID so we can use it for streaming. The streams * are not visible to anyone so this is OK to change it. */ - stream->net_seq_idx = relayd_id; + stream->relayd_id = relayd_id; channel->relayd_id = relayd_id; if (relayd_id != (uint64_t) -1ULL) { ret = consumer_send_relayd_stream(stream, path); @@ -185,12 +205,23 @@ 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); + ret = kernctl_buffer_flush_empty(stream->wait_fd); if (ret < 0) { - ERR("Failed to flush kernel stream"); - ret = -errno; + /* + * Doing a buffer flush which does not take into + * account empty packets. This is not perfect + * for stream intersection, but required as a + * fall-back when "flush_empty" is not + * implemented by lttng-modules. + */ + ret = kernctl_buffer_flush(stream->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end_unlock; + } goto end_unlock; } @@ -217,7 +248,6 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, &stream->max_sb_size); if (ret < 0) { ERR("Getting kernel max_sb_size"); - ret = -errno; goto end_unlock; } } @@ -229,48 +259,45 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, while (consumed_pos < produced_pos) { ssize_t read_len; unsigned long len, padded_len; - int lost_packet = 0; + const char *subbuf_addr; + struct lttng_buffer_view subbuf_view; health_code_update(); - DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos); ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos); if (ret < 0) { - if (errno != EAGAIN) { + if (ret != -EAGAIN) { PERROR("kernctl_get_subbuf snapshot"); - ret = -errno; goto end_unlock; } 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; - } + stream->chan->lost_packets++; continue; } ret = kernctl_get_subbuf_size(stream->wait_fd, &len); if (ret < 0) { ERR("Snapshot kernctl_get_subbuf_size"); - ret = -errno; goto error_put_subbuf; } ret = kernctl_get_padded_subbuf_size(stream->wait_fd, &padded_len); if (ret < 0) { ERR("Snapshot kernctl_get_padded_subbuf_size"); - ret = -errno; goto error_put_subbuf; } - read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len, + ret = get_current_subbuf_addr(stream, &subbuf_addr); + if (ret) { + goto error_put_subbuf; + } + + subbuf_view = lttng_buffer_view_init( + subbuf_addr, 0, padded_len); + read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, + stream, &subbuf_view, padded_len - len, NULL); /* * We write the padded len in local tracefiles but the data len @@ -292,20 +319,9 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, ret = kernctl_put_subbuf(stream->wait_fd); if (ret < 0) { ERR("Snapshot kernctl_put_subbuf"); - ret = -errno; 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) { @@ -319,7 +335,7 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, } } else { close_relayd_stream(stream); - stream->net_seq_idx = (uint64_t) -1ULL; + stream->relayd_id = (uint64_t) -1ULL; } pthread_mutex_unlock(&stream->lock); } @@ -331,7 +347,6 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, error_put_subbuf: ret = kernctl_put_subbuf(stream->wait_fd); if (ret < 0) { - ret = -errno; ERR("Snapshot kernctl_put_subbuf error path"); } end_unlock: @@ -346,7 +361,7 @@ end: * * Returns 0 on success, < 0 on error */ -int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, +static int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, struct lttng_consumer_local_data *ctx) { int ret, use_relayd = 0; @@ -365,11 +380,12 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (!metadata_channel) { ERR("Kernel snapshot metadata not found for key %" PRIu64, key); ret = -1; - goto error; + goto error_no_channel; } metadata_stream = metadata_channel->metadata_stream; assert(metadata_stream); + pthread_mutex_lock(&metadata_stream->lock); /* Flag once that we have a valid relayd for the stream. */ if (relayd_id != (uint64_t) -1ULL) { @@ -379,7 +395,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (use_relayd) { ret = consumer_send_relayd_stream(metadata_stream, path); if (ret < 0) { - goto error; + goto error_snapshot; } } else { ret = utils_create_stream_file(path, metadata_stream->name, @@ -387,7 +403,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, metadata_stream->tracefile_count_current, metadata_stream->uid, metadata_stream->gid, NULL); if (ret < 0) { - goto error; + goto error_snapshot; } metadata_stream->out_fd = ret; } @@ -400,7 +416,8 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (ret_read != -EAGAIN) { ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)", ret_read); - goto error; + ret = ret_read; + goto error_snapshot; } /* ret_read is negative at this point so we will exit the loop. */ continue; @@ -409,7 +426,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (use_relayd) { close_relayd_stream(metadata_stream); - metadata_stream->net_seq_idx = (uint64_t) -1ULL; + metadata_stream->relayd_id = (uint64_t) -1ULL; } else { if (metadata_stream->out_fd >= 0) { ret = close(metadata_stream->out_fd); @@ -425,11 +442,12 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, } ret = 0; - +error_snapshot: + pthread_mutex_unlock(&metadata_stream->lock); cds_list_del(&metadata_stream->send_node); consumer_stream_destroy(metadata_stream, NULL); metadata_channel->metadata_stream = NULL; -error: +error_no_channel: rcu_read_unlock(); return ret; } @@ -471,10 +489,10 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, case LTTNG_CONSUMER_ADD_RELAYD_SOCKET: { /* Session daemon status message are handled in the following call. */ - ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index, + consumer_add_relayd_socket(msg.u.relayd_sock.net_index, msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll, &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id, - msg.u.relayd_sock.relayd_session_id); + msg.u.relayd_sock.relayd_session_id); goto end_nosignal; } case LTTNG_CONSUMER_ADD_CHANNEL: @@ -713,19 +731,32 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel->monitor) { DBG("Kernel consumer add stream %s in no monitor mode with " "relayd id %" PRIu64, new_stream->name, - new_stream->net_seq_idx); + new_stream->relayd_id); cds_list_add(&new_stream->send_node, &channel->streams.head); break; } /* Send stream to relayd if the stream has an ID. */ - if (new_stream->net_seq_idx != (uint64_t) -1ULL) { + if (new_stream->relayd_id != (uint64_t) -1ULL) { ret = consumer_send_relayd_stream(new_stream, new_stream->chan->pathname); if (ret < 0) { 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->relayd_id); + if (ret < 0) { + goto end_nosignal; + } + } } /* Get the right pipe where the stream will be sent. */ @@ -819,6 +850,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 +1001,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 +1031,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 */ @@ -1093,6 +1125,32 @@ static int get_index_values(struct ctf_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) { + 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) { + 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); + error: return ret; } @@ -1119,7 +1177,7 @@ int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata) ret = kernctl_snapshot(metadata->wait_fd); if (ret < 0) { - if (errno != EAGAIN) { + if (ret != -EAGAIN) { ERR("Sync metadata, taking kernel snapshot failed."); goto end; } @@ -1141,8 +1199,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; + } } /* @@ -1172,8 +1236,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; @@ -1188,6 +1252,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. */ @@ -1213,7 +1316,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, */ DBG("Reserving sub buffer failed (everything is normal, " "it is due to concurrency)"); - ret = -errno; + ret = err; goto end; } @@ -1223,16 +1326,16 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, PERROR("Getting sub-buffer len failed."); err = kernctl_put_subbuf(infd); if (err != 0) { - if (errno == EFAULT) { + if (err == -EFAULT) { PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { + } else if (err == -EIO) { /* Should never happen with newer LTTng versions */ PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; + ret = err; goto end; } - ret = -errno; + ret = err; goto end; } @@ -1241,23 +1344,49 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, if (ret < 0) { err = kernctl_put_subbuf(infd); if (err != 0) { - if (errno == EFAULT) { + if (err == -EFAULT) { PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { + } else if (err == -EIO) { /* Should never happen with newer LTTng versions */ PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; + ret = err; goto end; } goto end; } 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) { @@ -1288,40 +1417,51 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, } break; case CONSUMER_CHANNEL_MMAP: + { + const char *subbuf_addr; + struct lttng_buffer_view subbuf_view; + /* Get subbuffer size without padding */ err = kernctl_get_subbuf_size(infd, &subbuf_size); if (err != 0) { PERROR("Getting sub-buffer len failed."); err = kernctl_put_subbuf(infd); if (err != 0) { - if (errno == EFAULT) { + if (err == -EFAULT) { PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { + } else if (err == -EIO) { /* Should never happen with newer LTTng versions */ PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; + ret = err; goto end; } - ret = -errno; + ret = err; goto end; } + ret = get_current_subbuf_addr(stream, &subbuf_addr); + if (ret) { + goto error_put_subbuf; + } + /* Make sure the tracer is not gone mad on us! */ assert(len >= subbuf_size); padding = len - subbuf_size; + subbuf_view = lttng_buffer_view_init(subbuf_addr, 0, len); + /* write the subbuffer to the tracefile */ - ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, - padding, &index); + ret = lttng_consumer_on_read_subbuffer_mmap( + ctx, stream, &subbuf_view, padding, &index); /* - * The mmap operation should write subbuf_size amount of data when - * network streaming or the full padding (len) size when we are _not_ - * streaming. + * The mmap operation should write subbuf_size amount of data + * when network streaming or the full padding (len) size when we + * are _not_ streaming. */ - if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) || - (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) { + if ((ret != subbuf_size && stream->relayd_id != (uint64_t) -1ULL) || + (ret != len && stream->relayd_id == (uint64_t) -1ULL)) { /* * Display the error but continue processing to try to release the * subbuffer. This is a DBG statement since this is possible to @@ -1333,20 +1473,21 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, write_index = 0; } break; + } default: ERR("Unknown output method"); ret = -EPERM; } - +error_put_subbuf: err = kernctl_put_next_subbuf(infd); if (err != 0) { - if (errno == EFAULT) { + if (err == -EFAULT) { PERROR("Error in unreserving sub buffer\n"); - } else if (errno == EIO) { + } else if (err == -EIO) { /* Should never happen with newer LTTng versions */ PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; + ret = err; goto end; } @@ -1399,7 +1540,7 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) * Don't create anything if this is set for streaming or should not be * monitored. */ - if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) { + if (stream->relayd_id == (uint64_t) -1ULL && stream->chan->monitor) { ret = utils_create_stream_file(stream->chan->pathname, stream->name, stream->chan->tracefile_size, stream->tracefile_count_current, stream->uid, stream->gid, NULL); @@ -1410,14 +1551,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; } } @@ -1428,7 +1572,6 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) ret = kernctl_get_mmap_len(stream->wait_fd, &mmap_len); if (ret != 0) { PERROR("kernctl_get_mmap_len"); - ret = -errno; goto error_close_fd; } stream->mmap_len = (size_t) mmap_len;