X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=91725d5d46be64f03935dc4a66f634d1ce5245c1;hp=c6eb71d57807354e729a9f220b02e18d07aaf6ab;hb=82e72193f6fcf9670d2cbd81234ec87639da82e5;hpb=c585b4b94a2f517485cdf1c2f3efe8d26e44b05d

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index c6eb71d57..91725d5d4 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -7,6 +7,7 @@
  *
  */
 
+#include "common/index/ctf-index.h"
 #define _LGPL_SOURCE
 #include <assert.h>
 #include <poll.h>
@@ -568,95 +569,6 @@ void consumer_stream_update_channel_attributes(
 			channel->tracefile_size;
 }
 
-struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
-		uint64_t stream_key,
-		const char *channel_name,
-		uint64_t relayd_id,
-		uint64_t session_id,
-		struct lttng_trace_chunk *trace_chunk,
-		int cpu,
-		int *alloc_ret,
-		enum consumer_channel_type type,
-		unsigned int monitor)
-{
-	int ret;
-	struct lttng_consumer_stream *stream;
-
-	stream = zmalloc(sizeof(*stream));
-	if (stream == NULL) {
-		PERROR("malloc struct lttng_consumer_stream");
-		ret = -ENOMEM;
-		goto end;
-	}
-
-	if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
-		ERR("Failed to acquire trace chunk reference during the creation of a stream");
-		ret = -1;
-		goto error;
-	}
-
-	rcu_read_lock();
-	stream->key = stream_key;
-	stream->trace_chunk = trace_chunk;
-	stream->out_fd = -1;
-	stream->out_fd_offset = 0;
-	stream->output_written = 0;
-	stream->net_seq_idx = relayd_id;
-	stream->session_id = session_id;
-	stream->monitor = monitor;
-	stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
-	stream->index_file = NULL;
-	stream->last_sequence_number = -1ULL;
-	stream->rotate_position = -1ULL;
-	pthread_mutex_init(&stream->lock, NULL);
-	pthread_mutex_init(&stream->metadata_timer_lock, NULL);
-
-	/* If channel is the metadata, flag this stream as metadata. */
-	if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
-		stream->metadata_flag = 1;
-		/* Metadata is flat out. */
-		strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
-		/* Live rendez-vous point. */
-		pthread_cond_init(&stream->metadata_rdv, NULL);
-		pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
-	} else {
-		/* Format stream name to <channel_name>_<cpu_number> */
-		ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
-				channel_name, cpu);
-		if (ret < 0) {
-			PERROR("snprintf stream name");
-			goto error;
-		}
-	}
-
-	/* Key is always the wait_fd for streams. */
-	lttng_ht_node_init_u64(&stream->node, stream->key);
-
-	/* Init node per channel id key */
-	lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
-
-	/* Init session id node with the stream session id */
-	lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
-
-	DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
-			" relayd_id %" PRIu64 ", session_id %" PRIu64,
-			stream->name, stream->key, channel_key,
-			stream->net_seq_idx, stream->session_id);
-
-	rcu_read_unlock();
-	return stream;
-
-error:
-	rcu_read_unlock();
-	lttng_trace_chunk_put(stream->trace_chunk);
-	free(stream);
-end:
-	if (alloc_ret) {
-		*alloc_ret = ret;
-	}
-	return NULL;
-}
-
 /*
  * Add a stream to the global list protected by a mutex.
  */
@@ -1063,6 +975,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 		uint64_t session_id_per_pid,
 		unsigned int monitor,
 		unsigned int live_timer_interval,
+		bool is_in_live_session,
 		const char *root_shm_path,
 		const char *shm_path)
 {
@@ -1094,6 +1007,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 	channel->tracefile_count = tracefile_count;
 	channel->monitor = monitor;
 	channel->live_timer_interval = live_timer_interval;
+	channel->is_live = is_in_live_session;
 	pthread_mutex_init(&channel->lock, NULL);
 	pthread_mutex_init(&channel->timer_lock, NULL);
 
@@ -1462,7 +1376,7 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
 struct lttng_consumer_local_data *lttng_consumer_create(
 		enum lttng_consumer_type type,
 		ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
-			struct lttng_consumer_local_data *ctx),
+			struct lttng_consumer_local_data *ctx, bool locked_by_caller),
 		int (*recv_channel)(struct lttng_consumer_channel *channel),
 		int (*recv_stream)(struct lttng_consumer_stream *stream),
 		int (*update_stream)(uint64_t stream_key, uint32_t state))
@@ -1669,19 +1583,18 @@ end:
  * Returns the number of bytes written
  */
 ssize_t lttng_consumer_on_read_subbuffer_mmap(
-		struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *stream, unsigned long len,
-		unsigned long padding,
-		struct ctf_packet_index *index)
+		struct lttng_consumer_stream *stream,
+		const struct lttng_buffer_view *buffer,
+		unsigned long padding)
 {
-	unsigned long mmap_offset;
-	void *mmap_base;
 	ssize_t ret = 0;
 	off_t orig_offset = stream->out_fd_offset;
 	/* Default is on the disk */
 	int outfd = stream->out_fd;
 	struct consumer_relayd_sock_pair *relayd = NULL;
 	unsigned int relayd_hang_up = 0;
+	const size_t subbuf_content_size = buffer->size - padding;
+	size_t write_len;
 
 	/* RCU lock for the relayd pointer */
 	rcu_read_lock();
@@ -1697,39 +1610,9 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		}
 	}
 
-	/* get the offset inside the fd to mmap */
-	switch (consumer_data.type) {
-	case LTTNG_CONSUMER_KERNEL:
-		mmap_base = stream->mmap_base;
-		ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
-		if (ret < 0) {
-			PERROR("tracer ctl get_mmap_read_offset");
-			goto end;
-		}
-		break;
-	case LTTNG_CONSUMER32_UST:
-	case LTTNG_CONSUMER64_UST:
-		mmap_base = lttng_ustctl_get_mmap_base(stream);
-		if (!mmap_base) {
-			ERR("read mmap get mmap base for stream %s", stream->name);
-			ret = -EPERM;
-			goto end;
-		}
-		ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
-		if (ret != 0) {
-			PERROR("tracer ctl get_mmap_read_offset");
-			ret = -EINVAL;
-			goto end;
-		}
-		break;
-	default:
-		ERR("Unknown consumer_data type");
-		assert(0);
-	}
-
 	/* Handle stream on the relayd if the output is on the network */
 	if (relayd) {
-		unsigned long netlen = len;
+		unsigned long netlen = subbuf_content_size;
 
 		/*
 		 * Lock the control socket for the complete duration of the function
@@ -1767,10 +1650,10 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 				goto write_error;
 			}
 		}
-	} else {
-		/* No streaming, we have to set the len with the full padding */
-		len += padding;
 
+		write_len = subbuf_content_size;
+	} else {
+		/* No streaming; we have to write the full padding. */
 		if (stream->metadata_flag && stream->reset_metadata_flag) {
 			ret = utils_truncate_stream_file(stream->out_fd, 0);
 			if (ret < 0) {
@@ -1784,7 +1667,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		 * Check if we need to change the tracefile before writing the packet.
 		 */
 		if (stream->chan->tracefile_size > 0 &&
-				(stream->tracefile_size_current + len) >
+				(stream->tracefile_size_current + buffer->size) >
 				stream->chan->tracefile_size) {
 			ret = consumer_stream_rotate_output_files(stream);
 			if (ret) {
@@ -1793,19 +1676,17 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 			outfd = stream->out_fd;
 			orig_offset = 0;
 		}
-		stream->tracefile_size_current += len;
-		if (index) {
-			index->offset = htobe64(stream->out_fd_offset);
-		}
+		stream->tracefile_size_current += buffer->size;
+		write_len = buffer->size;
 	}
 
 	/*
 	 * This call guarantee that len or less is returned. It's impossible to
 	 * receive a ret value that is bigger than len.
 	 */
-	ret = lttng_write(outfd, mmap_base + mmap_offset, len);
-	DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
-	if (ret < 0 || ((size_t) ret != len)) {
+	ret = lttng_write(outfd, buffer->data, write_len);
+	DBG("Consumer mmap write() ret %zd (len %zu)", ret, write_len);
+	if (ret < 0 || ((size_t) ret != write_len)) {
 		/*
 		 * Report error to caller if nothing was written else at least send the
 		 * amount written.
@@ -1826,7 +1707,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 			DBG("Consumer mmap write detected relayd hang up");
 		} else {
 			/* Unhandled error, print it and stop function right now. */
-			PERROR("Error in write mmap (ret %zd != len %lu)", ret, len);
+			PERROR("Error in write mmap (ret %zd != write_len %zu)", ret,
+					write_len);
 		}
 		goto write_error;
 	}
@@ -1835,9 +1717,9 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 	/* This call is useless on a socket so better save a syscall. */
 	if (!relayd) {
 		/* This won't block, but will start writeout asynchronously */
-		lttng_sync_file_range(outfd, stream->out_fd_offset, len,
+		lttng_sync_file_range(outfd, stream->out_fd_offset, write_len,
 				SYNC_FILE_RANGE_WRITE);
-		stream->out_fd_offset += len;
+		stream->out_fd_offset += write_len;
 		lttng_consumer_sync_trace_file(stream, orig_offset);
 	}
 
@@ -1871,8 +1753,7 @@ end:
 ssize_t lttng_consumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len,
-		unsigned long padding,
-		struct ctf_packet_index *index)
+		unsigned long padding)
 {
 	ssize_t ret = 0, written = 0, ret_splice = 0;
 	loff_t offset = 0;
@@ -1976,7 +1857,6 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 			orig_offset = 0;
 		}
 		stream->tracefile_size_current += len;
-		index->offset = htobe64(stream->out_fd_offset);
 	}
 
 	while (len > 0) {
@@ -2543,7 +2423,7 @@ restart:
 				do {
 					health_code_update();
 
-					len = ctx->on_buffer_ready(stream, ctx);
+					len = ctx->on_buffer_ready(stream, ctx, false);
 					/*
 					 * We don't check the return value here since if we get
 					 * a negative len, it means an error occurred thus we
@@ -2570,7 +2450,7 @@ restart:
 					do {
 						health_code_update();
 
-						len = ctx->on_buffer_ready(stream, ctx);
+						len = ctx->on_buffer_ready(stream, ctx, false);
 						/*
 						 * We don't check the return value here since if we get
 						 * a negative len, it means an error occurred thus we
@@ -2786,7 +2666,7 @@ void *consumer_thread_data_poll(void *data)
 			if (pollfd[i].revents & POLLPRI) {
 				DBG("Urgent read on fd %d", pollfd[i].fd);
 				high_prio = 1;
-				len = ctx->on_buffer_ready(local_stream[i], ctx);
+				len = ctx->on_buffer_ready(local_stream[i], ctx, false);
 				/* it's ok to have an unavailable sub-buffer */
 				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					/* Clean the stream and free it. */
@@ -2817,7 +2697,7 @@ void *consumer_thread_data_poll(void *data)
 					local_stream[i]->hangup_flush_done ||
 					local_stream[i]->has_data) {
 				DBG("Normal read on fd %d", pollfd[i].fd);
-				len = ctx->on_buffer_ready(local_stream[i], ctx);
+				len = ctx->on_buffer_ready(local_stream[i], ctx, false);
 				/* it's ok to have an unavailable sub-buffer */
 				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					/* Clean the stream and free it. */
@@ -3431,39 +3311,123 @@ error_testpoint:
 }
 
 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
-		struct lttng_consumer_local_data *ctx)
+		struct lttng_consumer_local_data *ctx,
+		bool locked_by_caller)
 {
-	ssize_t ret;
+	ssize_t ret, written_bytes;
+	int rotation_ret;
+	struct stream_subbuffer subbuffer = {};
 
-	pthread_mutex_lock(&stream->chan->lock);
-	pthread_mutex_lock(&stream->lock);
-	if (stream->metadata_flag) {
-		pthread_mutex_lock(&stream->metadata_rdv_lock);
+	if (!locked_by_caller) {
+		stream->read_subbuffer_ops.lock(stream);
 	}
 
-	switch (consumer_data.type) {
-	case LTTNG_CONSUMER_KERNEL:
-		ret = lttng_kconsumer_read_subbuffer(stream, ctx);
-		break;
-	case LTTNG_CONSUMER32_UST:
-	case LTTNG_CONSUMER64_UST:
-		ret = lttng_ustconsumer_read_subbuffer(stream, ctx);
-		break;
-	default:
-		ERR("Unknown consumer_data type");
-		assert(0);
-		ret = -ENOSYS;
-		break;
+	if (stream->read_subbuffer_ops.on_wake_up) {
+		ret = stream->read_subbuffer_ops.on_wake_up(stream);
+		if (ret) {
+			goto end;
+		}
 	}
 
-	if (stream->metadata_flag) {
-		pthread_cond_broadcast(&stream->metadata_rdv);
-		pthread_mutex_unlock(&stream->metadata_rdv_lock);
+	/*
+	 * If the stream was flagged to be ready for rotation before we extract
+	 * the next packet, rotate it now.
+	 */
+	if (stream->rotate_ready) {
+		DBG("Rotate stream before consuming data");
+		ret = lttng_consumer_rotate_stream(ctx, stream);
+		if (ret < 0) {
+			ERR("Stream rotation error before consuming data");
+			goto end;
+		}
+	}
+
+	ret = stream->read_subbuffer_ops.get_next_subbuffer(stream, &subbuffer);
+	if (ret) {
+		if (ret == -ENODATA) {
+			/* Not an error. */
+			ret = 0;
+			goto sleep_stream;
+		}
+		goto end;
+	}
+
+	ret = stream->read_subbuffer_ops.pre_consume_subbuffer(
+			stream, &subbuffer);
+	if (ret) {
+		goto error_put_subbuf;
+	}
+
+	written_bytes = stream->read_subbuffer_ops.consume_subbuffer(
+			ctx, stream, &subbuffer);
+	/*
+	 * Should write subbuf_size amount of data when network streaming or
+	 * the full padded size when we are not streaming.
+	 */
+	if ((written_bytes != subbuffer.info.data.subbuf_size &&
+			    stream->net_seq_idx != (uint64_t) -1ULL) ||
+			(written_bytes != subbuffer.info.data.padded_subbuf_size &&
+					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 can happen without being a critical
+		 * error.
+		 */
+		DBG("Failed to write to tracefile (written_bytes: %zd != padded subbuffer size: %lu, subbuffer size: %lu)",
+				written_bytes, subbuffer.info.data.subbuf_size,
+				subbuffer.info.data.padded_subbuf_size);
+	}
+
+	ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+	if (ret) {
+		goto end;
+	}
+
+	if (stream->read_subbuffer_ops.post_consume) {
+		ret = stream->read_subbuffer_ops.post_consume(stream, &subbuffer, ctx);
+		if (ret) {
+			goto end;
+		}
+	}
+
+	/*
+	 * After extracting the packet, we check if the stream is now ready to
+	 * be rotated and perform the action immediately.
+	 *
+	 * Don't overwrite `ret` as callers expect the number of bytes
+	 * consumed to be returned on success.
+	 */
+	rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
+	if (rotation_ret == 1) {
+		rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
+		if (rotation_ret < 0) {
+			ret = rotation_ret;
+			ERR("Stream rotation error after consuming data");
+			goto end;
+		}
+	} else if (rotation_ret < 0) {
+		ret = rotation_ret;
+		ERR("Failed to check if stream was ready to rotate after consuming data");
+		goto end;
+	}
+
+sleep_stream:
+	if (stream->read_subbuffer_ops.on_sleep) {
+		stream->read_subbuffer_ops.on_sleep(stream, ctx);
+	}
+
+	ret = written_bytes;
+end:
+	if (!locked_by_caller) {
+		stream->read_subbuffer_ops.unlock(stream);
 	}
-	pthread_mutex_unlock(&stream->lock);
-	pthread_mutex_unlock(&stream->chan->lock);
 
 	return ret;
+error_put_subbuf:
+	(void) stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+	goto end;
 }
 
 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
@@ -3953,8 +3917,18 @@ int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_act
 		} else {
 			ret = kernctl_buffer_flush_empty(stream->wait_fd);
 			if (ret < 0) {
-				ERR("Failed to flush kernel stream");
-				goto end;
+				/*
+				 * Doing a buffer flush which does not take into
+				 * account empty packets. This is not perfect,
+				 * 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;
+				}
 			}
 		}
 		break;
@@ -4193,7 +4167,7 @@ int consumer_clear_buffer(struct lttng_consumer_stream *stream)
 	case LTTNG_CONSUMER_KERNEL:
 		ret = kernctl_buffer_clear(stream->wait_fd);
 		if (ret < 0) {
-			ERR("Failed to flush kernel stream");
+			ERR("Failed to clear kernel stream (ret = %d)", ret);
 			goto end;
 		}
 		break;