X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;fp=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=e06f3b3e079aae44f494ba6d4409f02120018af4;hp=4a2e4e07d7c0b8e5eac8e34927afad5f8762da4a;hb=6f9449c22eef59294cf1e1dc3610a5cbf14baec0;hpb=23d565989350c270c68e9a6c8edfbe2dd6a6895d diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 4a2e4e07d..e06f3b3e0 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -7,8 +7,6 @@ * */ -#include "common/buffer-view.h" -#include #define _LGPL_SOURCE #include #include @@ -36,6 +34,9 @@ #include #include #include +#include +#include +#include #include "kernel-consumer.h" @@ -288,7 +289,7 @@ static int lttng_kconsumer_snapshot_channel( subbuf_addr, 0, padded_len); read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, &subbuf_view, - padded_len - len, NULL); + padded_len - len); /* * We write the padded len in local tracefiles but the data len * when using a relay. Display the error but continue processing @@ -399,7 +400,7 @@ static int lttng_kconsumer_snapshot_metadata( do { health_code_update(); - ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx); + ret_read = lttng_consumer_read_subbuffer(metadata_stream, ctx, true); if (ret_read < 0) { if (ret_read != -EAGAIN) { ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)", @@ -655,7 +656,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); pthread_mutex_lock(&channel->lock); - new_stream = consumer_allocate_stream( + new_stream = consumer_stream_create( channel, channel->key, fd, @@ -690,23 +691,6 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, consumer_stream_update_channel_attributes(new_stream, channel); - switch (channel->output) { - case CONSUMER_CHANNEL_SPLICE: - new_stream->output = LTTNG_EVENT_SPLICE; - ret = utils_create_pipe(new_stream->splice_pipe); - if (ret < 0) { - pthread_mutex_unlock(&channel->lock); - goto error_add_stream_nosignal; - } - break; - case CONSUMER_CHANNEL_MMAP: - new_stream->output = LTTNG_EVENT_MMAP; - break; - default: - ERR("Stream output unknown %d", channel->output); - pthread_mutex_unlock(&channel->lock); - goto error_add_stream_nosignal; - } /* * We've just assigned the channel to the stream so increment the @@ -1358,93 +1342,6 @@ end: return ret; } -/* - * Populate index values of a kernel stream. Values are set in big endian order. - * - * Return 0 on success or else a negative value. - */ -static int get_index_values(struct ctf_packet_index *index, int infd) -{ - int ret; - uint64_t packet_size, content_size, timestamp_begin, timestamp_end, - events_discarded, stream_id, stream_instance_id, - packet_seq_num; - - ret = kernctl_get_timestamp_begin(infd, ×tamp_begin); - if (ret < 0) { - PERROR("kernctl_get_timestamp_begin"); - goto error; - } - - ret = kernctl_get_timestamp_end(infd, ×tamp_end); - if (ret < 0) { - PERROR("kernctl_get_timestamp_end"); - goto error; - } - - ret = kernctl_get_events_discarded(infd, &events_discarded); - if (ret < 0) { - PERROR("kernctl_get_events_discarded"); - goto error; - } - - ret = kernctl_get_content_size(infd, &content_size); - if (ret < 0) { - PERROR("kernctl_get_content_size"); - goto error; - } - - ret = kernctl_get_packet_size(infd, &packet_size); - if (ret < 0) { - PERROR("kernctl_get_packet_size"); - goto error; - } - - ret = kernctl_get_stream_id(infd, &stream_id); - if (ret < 0) { - PERROR("kernctl_get_stream_id"); - goto error; - } - - ret = kernctl_get_instance_id(infd, &stream_instance_id); - if (ret < 0) { - if (ret == -ENOTTY) { - /* Command not implemented by lttng-modules. */ - stream_instance_id = -1ULL; - } else { - PERROR("kernctl_get_instance_id"); - goto error; - } - } - - ret = kernctl_get_sequence_number(infd, &packet_seq_num); - if (ret < 0) { - if (ret == -ENOTTY) { - /* Command not implemented by lttng-modules. */ - packet_seq_num = -1ULL; - ret = 0; - } else { - PERROR("kernctl_get_sequence_number"); - goto error; - } - } - index->packet_seq_num = htobe64(index->packet_seq_num); - - *index = (typeof(*index)) { - .offset = index->offset, - .packet_size = htobe64(packet_size), - .content_size = htobe64(content_size), - .timestamp_begin = htobe64(timestamp_begin), - .timestamp_end = htobe64(timestamp_end), - .events_discarded = htobe64(events_discarded), - .stream_id = htobe64(stream_id), - .stream_instance_id = htobe64(stream_instance_id), - .packet_seq_num = htobe64(packet_seq_num), - }; - -error: - return ret; -} /* * Sync metadata meaning request them to the session daemon and snapshot to the * metadata thread can consumer them. @@ -1483,348 +1380,226 @@ end: } static -int update_stream_stats(struct lttng_consumer_stream *stream) +int extract_common_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) { 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; - stream->sequence_number_unavailable = true; - } 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; + ret = kernctl_get_subbuf_size( + stream->wait_fd, &subbuf->info.data.subbuf_size); + if (ret) { goto end; } - stream->last_sequence_number = seq; - ret = kernctl_get_events_discarded(stream->wait_fd, &discarded); - if (ret < 0) { - PERROR("kernctl_get_events_discarded"); + ret = kernctl_get_padded_subbuf_size( + stream->wait_fd, &subbuf->info.data.padded_subbuf_size); + if (ret) { 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 extract_metadata_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) { 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"); + ret = extract_common_subbuffer_info(stream, subbuf); + if (ret) { goto end; } - if (stream->metadata_version == cur_version) { - ret = 0; + ret = kernctl_get_metadata_version( + stream->wait_fd, &subbuf->info.metadata.version); + if (ret) { 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. - * The stream and channel locks must be held by the caller. - */ -ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx) +static +int extract_data_subbuffer_info(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuf) { - unsigned long len, subbuf_size, padding; - int err, write_index = 1; - ssize_t ret = 0; - int infd = stream->wait_fd; - struct ctf_packet_index index = {}; - bool in_error_state = false; + int ret; - DBG("In read_subbuffer (infd : %d)", infd); + ret = extract_common_subbuffer_info(stream, subbuf); + if (ret) { + goto end; + } + ret = kernctl_get_packet_size( + stream->wait_fd, &subbuf->info.data.packet_size); + if (ret < 0) { + PERROR("Failed to get sub-buffer packet size"); + goto end; + } - /* Get the next subbuffer */ - err = kernctl_get_next_subbuf(infd); - if (err != 0) { - /* - * This is a debug message even for single-threaded consumer, - * because poll() have more relaxed criterions than get subbuf, - * so get_subbuf may fail for short race windows where poll() - * would issue wakeups. - */ - DBG("Reserving sub buffer failed (everything is normal, " - "it is due to concurrency)"); - ret = err; - goto error; + ret = kernctl_get_content_size( + stream->wait_fd, &subbuf->info.data.content_size); + if (ret < 0) { + PERROR("Failed to get sub-buffer content size"); + 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."); - 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 error; - } - ret = err; - goto error; + ret = kernctl_get_timestamp_begin( + stream->wait_fd, &subbuf->info.data.timestamp_begin); + if (ret < 0) { + PERROR("Failed to get sub-buffer begin timestamp"); + 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 error; - } - goto error; - } - 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 error; - } - goto error; + ret = kernctl_get_timestamp_end( + stream->wait_fd, &subbuf->info.data.timestamp_end); + if (ret < 0) { + PERROR("Failed to get sub-buffer end timestamp"); + goto end; + } + + ret = kernctl_get_events_discarded( + stream->wait_fd, &subbuf->info.data.events_discarded); + if (ret) { + PERROR("Failed to get sub-buffer events discarded count"); + goto end; + } + + ret = kernctl_get_sequence_number(stream->wait_fd, + &subbuf->info.data.sequence_number.value); + if (ret) { + /* May not be supported by older LTTng-modules. */ + if (ret != -ENOTTY) { + PERROR("Failed to get sub-buffer sequence number"); + 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 error; - } - goto error; - } + subbuf->info.data.sequence_number.is_set = true; } - switch (stream->chan->output) { - case CONSUMER_CHANNEL_SPLICE: - /* - * XXX: The lttng-modules splice "actor" does not handle copying - * partial pages hence only using the subbuffer size without the - * padding makes the splice fail. - */ - subbuf_size = len; - padding = 0; + ret = kernctl_get_stream_id( + stream->wait_fd, &subbuf->info.data.stream_id); + if (ret < 0) { + PERROR("Failed to get stream id"); + goto end; + } - /* splice the subbuffer to the tracefile */ - ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size, - padding, &index); - /* - * XXX: Splice does not support network streaming so the return value - * is simply checked against subbuf_size and not like the mmap() op. - */ - if (ret != subbuf_size) { - /* - * display the error but continue processing to try - * to release the subbuffer - */ - ERR("Error splicing to tracefile (ret: %zd != len: %lu)", - ret, subbuf_size); - write_index = 0; - } - 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 (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 error; - } - ret = err; - goto error; + ret = kernctl_get_instance_id(stream->wait_fd, + &subbuf->info.data.stream_instance_id.value); + if (ret) { + /* May not be supported by older LTTng-modules. */ + if (ret != -ENOTTY) { + PERROR("Failed to get stream instance id"); + goto end; } + } else { + subbuf->info.data.stream_instance_id.is_set = true; + } +end: + return ret; +} - ret = get_current_subbuf_addr(stream, &subbuf_addr); - if (ret) { - goto error_put_subbuf; - } +static +int get_subbuffer_common(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + int ret; - /* Make sure the tracer is not gone mad on us! */ - assert(len >= subbuf_size); + ret = kernctl_get_next_subbuf(stream->wait_fd); + if (ret) { + goto end; + } - padding = len - subbuf_size; + ret = stream->read_subbuffer_ops.extract_subbuffer_info( + stream, subbuffer); +end: + return ret; +} - subbuf_view = lttng_buffer_view_init(subbuf_addr, 0, len); +static +int get_next_subbuffer_splice(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + int ret; - /* write the subbuffer to the tracefile */ - 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. - */ - if ((ret != subbuf_size && stream->net_seq_idx != (uint64_t) -1ULL) || - (ret != len && stream->net_seq_idx == (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 - * happen without being a critical error. - */ - DBG("Error writing to tracefile " - "(ret: %zd != len: %lu != subbuf_size: %lu)", - ret, len, subbuf_size); - write_index = 0; - } - break; - } - default: - ERR("Unknown output method"); - ret = -EPERM; + ret = get_subbuffer_common(stream, subbuffer); + if (ret) { + goto end; } -error_put_subbuf: - err = kernctl_put_next_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 error; - } else if (in_error_state) { - goto error; + + subbuffer->buffer.fd = stream->wait_fd; +end: + return ret; +} + +static +int get_next_subbuffer_mmap(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + int ret; + const char *addr; + + ret = get_subbuffer_common(stream, subbuffer); + if (ret) { + goto end; } - /* Write index if needed. */ - if (!write_index) { + ret = get_current_subbuf_addr(stream, &addr); + if (ret) { goto end; } - if (stream->chan->live_timer_interval && !stream->metadata_flag) { - /* - * 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 error; + subbuffer->buffer.buffer = lttng_buffer_view_init( + addr, 0, subbuffer->info.data.padded_subbuf_size); +end: + return ret; +} + +static +int put_next_subbuffer(struct lttng_consumer_stream *stream, + struct stream_subbuffer *subbuffer) +{ + const int ret = kernctl_put_next_subbuf(stream->wait_fd); + + if (ret) { + if (ret == -EFAULT) { + PERROR("Error in unreserving sub buffer"); + } else if (ret == -EIO) { + /* Should never happen with newer LTTng versions */ + PERROR("Reader has been pushed by the writer, last sub-buffer corrupted"); } } - err = consumer_stream_write_index(stream, &index); - if (err < 0) { - goto error; + return ret; +} + +static void lttng_kconsumer_set_stream_ops( + struct lttng_consumer_stream *stream) +{ + if (stream->chan->output == CONSUMER_CHANNEL_MMAP) { + stream->read_subbuffer_ops.get_next_subbuffer = + get_next_subbuffer_mmap; + } else { + stream->read_subbuffer_ops.get_next_subbuffer = + get_next_subbuffer_splice; } -end: -error: - return ret; + if (stream->metadata_flag) { + stream->read_subbuffer_ops.extract_subbuffer_info = + extract_metadata_subbuffer_info; + } else { + stream->read_subbuffer_ops.extract_subbuffer_info = + extract_data_subbuffer_info; + if (stream->chan->is_live) { + stream->read_subbuffer_ops.send_live_beacon = + consumer_flush_kernel_index; + } + } + + stream->read_subbuffer_ops.put_next_subbuffer = put_next_subbuffer; } int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) @@ -1865,6 +1640,8 @@ int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream) } } + lttng_kconsumer_set_stream_ops(stream); + /* we return 0 to let the library handle the FD internally */ return 0;