*
*/
-#include "common/buffer-view.h"
-#include <stdint.h>
#define _LGPL_SOURCE
#include <assert.h>
#include <poll.h>
#include <common/index/index.h>
#include <common/consumer/consumer-timer.h>
#include <common/optional.h>
+#include <common/buffer-view.h>
+#include <common/consumer/consumer.h>
+#include <stdint.h>
#include "kernel-consumer.h"
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
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)",
health_code_update();
pthread_mutex_lock(&channel->lock);
- new_stream = consumer_allocate_stream(
+ new_stream = consumer_stream_create(
channel,
channel->key,
fd,
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
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.
}
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)
}
}
+ lttng_kconsumer_set_stream_ops(stream);
+
/* we return 0 to let the library handle the FD internally */
return 0;
projects / lttng-tools.git / blobdiff