X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=b5afc73aa5fb167071145b35e6d8a16a4291cbc6;hp=ab4f604d7a5c525447cdf23453633ef0baa4318a;hb=f22dd89135f6b1749cd75735f04e3cb309da12c1;hpb=84382d49b1e8a17037d0415d84f1c9e250163494 diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index ab4f604d7..b5afc73aa 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -1,6 +1,7 @@ /* * Copyright (C) 2011 - Julien Desfossez * Mathieu Desnoyers + * Copyright (C) 2017 - Jérémie Galarneau * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License, version 2 only, @@ -16,7 +17,7 @@ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE +#define _LGPL_SOURCE #include #include #include @@ -35,12 +36,13 @@ #include #include #include +#include #include #include #include -#include +#include #include -#include +#include #include "kernel-consumer.h" @@ -60,13 +62,25 @@ int lttng_kconsumer_take_snapshot(struct lttng_consumer_stream *stream) ret = kernctl_snapshot(infd); if (ret != 0) { - perror("Getting sub-buffer snapshot."); - ret = -errno; + PERROR("Getting sub-buffer snapshot."); } return ret; } +/* + * Sample consumed and produced positions for a specific fd. + * + * Returns 0 on success, < 0 on error. + */ +int lttng_kconsumer_sample_snapshot_positions( + struct lttng_consumer_stream *stream) +{ + assert(stream); + + return kernctl_snapshot_sample_positions(stream->wait_fd); +} + /* * Get the produced position * @@ -80,8 +94,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"); - ret = -errno; + PERROR("kernctl_snapshot_get_produced"); } return ret; @@ -100,8 +113,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"); - ret = -errno; + PERROR("kernctl_snapshot_get_consumed"); } return ret; @@ -113,11 +125,10 @@ 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; - unsigned long consumed_pos, produced_pos; struct lttng_consumer_channel *channel; struct lttng_consumer_stream *stream; @@ -140,6 +151,9 @@ 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(); @@ -182,12 +196,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; } @@ -214,23 +239,18 @@ 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; } } - /* - * 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,27 +258,33 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, 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; + } 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; } @@ -284,10 +310,19 @@ 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) { @@ -313,7 +348,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: @@ -483,7 +517,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; @@ -524,9 +559,20 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } else { ret = consumer_add_channel(new_channel, ctx); } - if (CONSUMER_CHANNEL_TYPE_DATA) { + if (msg.u.channel.type == CONSUMER_CHANNEL_TYPE_DATA && !ret) { + int monitor_start_ret; + + DBG("Consumer starting monitor timer"); consumer_timer_live_start(new_channel, msg.u.channel.live_timer_interval); + monitor_start_ret = consumer_timer_monitor_start( + new_channel, + msg.u.channel.monitor_timer_interval); + if (monitor_start_ret < 0) { + ERR("Starting channel monitoring timer failed"); + goto end_nosignal; + } + } health_code_update(); @@ -561,7 +607,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(); @@ -641,6 +687,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; @@ -703,6 +753,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. */ @@ -764,7 +827,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(); @@ -773,7 +836,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; } @@ -796,6 +859,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; } @@ -815,7 +879,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; } /* @@ -874,17 +938,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; } } @@ -905,7 +969,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(); @@ -918,6 +982,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. @@ -934,6 +1003,115 @@ 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; + } + case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE: + { + int channel_monitor_pipe; + + ret_code = LTTCOMM_CONSUMERD_SUCCESS; + /* Successfully received the command's type. */ + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + goto error_fatal; + } + + ret = lttcomm_recv_fds_unix_sock(sock, &channel_monitor_pipe, + 1); + if (ret != sizeof(channel_monitor_pipe)) { + ERR("Failed to receive channel monitor pipe"); + goto error_fatal; + } + + DBG("Received channel monitor pipe (%d)", channel_monitor_pipe); + ret = consumer_timer_thread_set_channel_monitor_pipe( + channel_monitor_pipe); + if (!ret) { + int flags; + + ret_code = LTTCOMM_CONSUMERD_SUCCESS; + /* Set the pipe as non-blocking. */ + ret = fcntl(channel_monitor_pipe, F_GETFL, 0); + if (ret == -1) { + PERROR("fcntl get flags of the channel monitoring pipe"); + goto error_fatal; + } + flags = ret; + + ret = fcntl(channel_monitor_pipe, F_SETFL, + flags | O_NONBLOCK); + if (ret == -1) { + PERROR("fcntl set O_NONBLOCK flag of the channel monitoring pipe"); + goto error_fatal; + } + DBG("Channel monitor pipe set as non-blocking"); + } else { + ret_code = LTTCOMM_CONSUMERD_ALREADY_SET; + } + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + goto error_fatal; + } + break; + } default: goto end_nosignal; } @@ -1005,6 +1183,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; } @@ -1031,7 +1235,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; } @@ -1045,6 +1249,106 @@ 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) { + if (ret == -ENOTTY) { + /* Command not implemented by lttng-modules. */ + seq = -1ULL; + ret = 0; + } else { + 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 occurred. We assume only one wrap-around + * has occurred. + */ + 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) { + 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. */ @@ -1070,25 +1374,77 @@ 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; } /* Get the full subbuffer size including padding */ err = kernctl_get_padded_subbuf_size(infd, &len); if (err != 0) { - perror("Getting sub-buffer len failed."); - ret = -errno; + PERROR("Getting sub-buffer len failed."); + 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; + } + ret = err; goto end; } if (!stream->metadata_flag) { ret = get_index_values(&index, infd); 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; + } + 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) { @@ -1122,8 +1478,19 @@ 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."); - ret = -errno; + PERROR("Getting sub-buffer len failed."); + 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; + } + ret = err; goto end; } @@ -1160,13 +1527,13 @@ 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"); - } else if (errno == EIO) { + 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."); + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } - ret = -errno; + ret = err; goto end; } @@ -1179,7 +1546,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; } @@ -1215,14 +1597,18 @@ 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; + assert(!stream->index_file); + stream->index_file = index_file; } } @@ -1233,7 +1619,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;