X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=553d44269454058233ee9c048d4ed3d46a071609;hb=ab1f27f2803120be93a7c148022d5c9c412e2b7b;hp=04d7efe5494443ab18dfea64d5e67561616fd66b;hpb=c69b5db6ca0fdd83e11aadd4d4227d74a72df8c6;p=lttng-tools.git

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 04d7efe54..553d44269 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -17,6 +17,7 @@
  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
+#include <stdint.h>
 #define _LGPL_SOURCE
 #include <assert.h>
 #include <lttng/ust-ctl.h>
@@ -120,26 +121,6 @@ error:
 	return ret;
 }
 
-/*
- * Allocate and return a consumer channel object.
- */
-static struct lttng_consumer_channel *allocate_channel(uint64_t session_id,
-		const uint64_t *chunk_id, const char *pathname, const char *name,
-		uint64_t relayd_id, uint64_t key, enum lttng_event_output output,
-		uint64_t tracefile_size, uint64_t tracefile_count,
-		uint64_t session_id_per_pid, unsigned int monitor,
-		unsigned int live_timer_interval,
-		const char *root_shm_path, const char *shm_path)
-{
-	assert(pathname);
-	assert(name);
-
-	return consumer_allocate_channel(key, session_id, chunk_id, pathname,
-			name, relayd_id, output, tracefile_size,
-			tracefile_count, session_id_per_pid, monitor,
-			live_timer_interval, root_shm_path, shm_path);
-}
-
 /*
  * Allocate and return a consumer stream object. If _alloc_ret is not NULL, the
  * error value if applicable is set in it else it is kept untouched.
@@ -156,7 +137,9 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
 	assert(channel);
 	assert(ctx);
 
-	stream = consumer_allocate_stream(channel->key,
+	stream = consumer_allocate_stream(
+			channel,
+			channel->key,
 			key,
 			channel->name,
 			channel->relayd_id,
@@ -185,7 +168,6 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
 	}
 
 	consumer_stream_update_channel_attributes(stream, channel);
-	stream->chan = channel;
 
 error:
 	if (_alloc_ret) {
@@ -775,10 +757,19 @@ static int flush_channel(uint64_t chan_key)
 		health_code_update();
 
 		pthread_mutex_lock(&stream->lock);
+
+		/*
+		 * Protect against concurrent teardown of a stream.
+		 */
+		if (cds_lfht_is_node_deleted(&stream->node.node)) {
+			goto next;
+		}
+
 		if (!stream->quiescent) {
 			ustctl_flush_buffer(stream->ustream, 0);
 			stream->quiescent = true;
 		}
+next:
 		pthread_mutex_unlock(&stream->lock);
 	}
 error:
@@ -1070,7 +1061,6 @@ error_stream:
 	 * Clean up the stream completly because the next snapshot will use a new
 	 * metadata stream.
 	 */
-	pthread_mutex_lock(&metadata_stream->lock);
 	consumer_stream_destroy(metadata_stream, NULL);
 	cds_list_del(&metadata_stream->send_node);
 	metadata_channel->metadata_stream = NULL;
@@ -1080,6 +1070,35 @@ error:
 	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;
+
+	mmap_base = ustctl_get_mmap_base(stream->ustream);
+	if (!mmap_base) {
+		ERR("Failed to get mmap base for stream `%s`",
+				stream->name);
+		ret = -EPERM;
+		goto error;
+	}
+
+	ret = ustctl_get_mmap_read_offset(stream->ustream, &mmap_offset);
+	if (ret != 0) {
+		ERR("Failed to get mmap offset for stream `%s`", stream->name);
+		ret = -EINVAL;
+		goto error;
+	}
+
+	*addr = mmap_base + mmap_offset;
+error:
+	return ret;
+
+}
+
 /*
  * Take a snapshot of all the stream of a channel.
  * RCU read-side lock and the channel lock must be held by the caller.
@@ -1182,6 +1201,8 @@ static int snapshot_channel(struct lttng_consumer_channel *channel,
 		while ((long) (consumed_pos - produced_pos) < 0) {
 			ssize_t read_len;
 			unsigned long len, padded_len;
+			const char *subbuf_addr;
+			struct lttng_buffer_view subbuf_view;
 
 			health_code_update();
 
@@ -1211,8 +1232,16 @@ static int snapshot_channel(struct lttng_consumer_channel *channel,
 				goto error_put_subbuf;
 			}
 
-			read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len,
-					padded_len - len, NULL);
+			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);
 			if (use_relayd) {
 				if (read_len != len) {
 					ret = -EPERM;
@@ -1441,19 +1470,21 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		};
 
 		/* Create a plain object and reserve a channel key. */
-		channel = allocate_channel(msg.u.ask_channel.session_id,
+		channel = consumer_allocate_channel(
+				msg.u.ask_channel.key,
+				msg.u.ask_channel.session_id,
 				msg.u.ask_channel.chunk_id.is_set ?
 						&chunk_id : NULL,
 				msg.u.ask_channel.pathname,
 				msg.u.ask_channel.name,
 				msg.u.ask_channel.relayd_id,
-				msg.u.ask_channel.key,
 				(enum lttng_event_output) msg.u.ask_channel.output,
 				msg.u.ask_channel.tracefile_size,
 				msg.u.ask_channel.tracefile_count,
 				msg.u.ask_channel.session_id_per_pid,
 				msg.u.ask_channel.monitor,
 				msg.u.ask_channel.live_timer_interval,
+				msg.u.ask_channel.is_live,
 				msg.u.ask_channel.root_shm_path,
 				msg.u.ask_channel.shm_path);
 		if (!channel) {
@@ -2096,6 +2127,9 @@ end_rotate_channel_nosignal:
 				msg.u.close_trace_chunk.close_command.value;
 		const uint64_t relayd_id =
 				msg.u.close_trace_chunk.relayd_id.value;
+		struct lttcomm_consumer_close_trace_chunk_reply reply;
+		char closed_trace_chunk_path[LTTNG_PATH_MAX];
+		int ret;
 
 		ret_code = lttng_consumer_close_trace_chunk(
 				msg.u.close_trace_chunk.relayd_id.is_set ?
@@ -2106,8 +2140,19 @@ end_rotate_channel_nosignal:
 				(time_t) msg.u.close_trace_chunk.close_timestamp,
 				msg.u.close_trace_chunk.close_command.is_set ?
 						&close_command :
-						NULL);
-		goto end_msg_sessiond;
+						NULL, closed_trace_chunk_path);
+		reply.ret_code = ret_code;
+		reply.path_length = strlen(closed_trace_chunk_path) + 1;
+		ret = lttcomm_send_unix_sock(sock, &reply, sizeof(reply));
+		if (ret != sizeof(reply)) {
+			goto error_fatal;
+		}
+		ret = lttcomm_send_unix_sock(sock, closed_trace_chunk_path,
+				reply.path_length);
+		if (ret != reply.path_length) {
+			goto error_fatal;
+		}
+		goto end_nosignal;
 	}
 	case LTTNG_CONSUMER_TRACE_CHUNK_EXISTS:
 	{
@@ -2148,7 +2193,6 @@ end_msg_sessiond:
 
 end_channel_error:
 	if (channel) {
-		pthread_mutex_unlock(&channel->lock);
 		/*
 		 * Free channel here since no one has a reference to it. We don't
 		 * free after that because a stream can store this pointer.
@@ -2175,31 +2219,6 @@ end:
 	return ret;
 }
 
-/*
- * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
- * compiled out, we isolate it in this library.
- */
-int lttng_ustctl_get_mmap_read_offset(struct lttng_consumer_stream *stream,
-		unsigned long *off)
-{
-	assert(stream);
-	assert(stream->ustream);
-
-	return ustctl_get_mmap_read_offset(stream->ustream, off);
-}
-
-/*
- * Wrapper over the mmap() read offset from ust-ctl library. Since this can be
- * compiled out, we isolate it in this library.
- */
-void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
-{
-	assert(stream);
-	assert(stream->ustream);
-
-	return ustctl_get_mmap_base(stream->ustream);
-}
-
 void lttng_ustctl_flush_buffer(struct lttng_consumer_stream *stream,
 		int producer_active)
 {
@@ -2401,62 +2420,69 @@ static int get_index_values(struct ctf_packet_index *index,
 		struct ustctl_consumer_stream *ustream)
 {
 	int ret;
+	uint64_t packet_size, content_size, timestamp_begin, timestamp_end,
+			events_discarded, stream_id, stream_instance_id,
+			packet_seq_num;
 
-	ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin);
+	ret = ustctl_get_timestamp_begin(ustream, &timestamp_begin);
 	if (ret < 0) {
 		PERROR("ustctl_get_timestamp_begin");
 		goto error;
 	}
-	index->timestamp_begin = htobe64(index->timestamp_begin);
 
-	ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end);
+	ret = ustctl_get_timestamp_end(ustream, &timestamp_end);
 	if (ret < 0) {
 		PERROR("ustctl_get_timestamp_end");
 		goto error;
 	}
-	index->timestamp_end = htobe64(index->timestamp_end);
 
-	ret = ustctl_get_events_discarded(ustream, &index->events_discarded);
+	ret = ustctl_get_events_discarded(ustream, &events_discarded);
 	if (ret < 0) {
 		PERROR("ustctl_get_events_discarded");
 		goto error;
 	}
-	index->events_discarded = htobe64(index->events_discarded);
 
-	ret = ustctl_get_content_size(ustream, &index->content_size);
+	ret = ustctl_get_content_size(ustream, &content_size);
 	if (ret < 0) {
 		PERROR("ustctl_get_content_size");
 		goto error;
 	}
-	index->content_size = htobe64(index->content_size);
 
-	ret = ustctl_get_packet_size(ustream, &index->packet_size);
+	ret = ustctl_get_packet_size(ustream, &packet_size);
 	if (ret < 0) {
 		PERROR("ustctl_get_packet_size");
 		goto error;
 	}
-	index->packet_size = htobe64(index->packet_size);
 
-	ret = ustctl_get_stream_id(ustream, &index->stream_id);
+	ret = ustctl_get_stream_id(ustream, &stream_id);
 	if (ret < 0) {
 		PERROR("ustctl_get_stream_id");
 		goto error;
 	}
-	index->stream_id = htobe64(index->stream_id);
 
-	ret = ustctl_get_instance_id(ustream, &index->stream_instance_id);
+	ret = ustctl_get_instance_id(ustream, &stream_instance_id);
 	if (ret < 0) {
 		PERROR("ustctl_get_instance_id");
 		goto error;
 	}
-	index->stream_instance_id = htobe64(index->stream_instance_id);
 
-	ret = ustctl_get_sequence_number(ustream, &index->packet_seq_num);
+	ret = ustctl_get_sequence_number(ustream, &packet_seq_num);
 	if (ret < 0) {
 		PERROR("ustctl_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;
@@ -2554,37 +2580,59 @@ end:
  * interacting with sessiond, else we cause a deadlock with live
  * awaiting on metadata to be pushed out.
  *
+ * The RCU read side lock must be held by the caller.
+ *
  * Return 0 if new metadatda is available, EAGAIN if the metadata stream
  * is empty or a negative value on error.
  */
 int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *metadata)
+		struct lttng_consumer_stream *metadata_stream)
 {
 	int ret;
 	int retry = 0;
+	struct lttng_consumer_channel *metadata_channel;
 
 	assert(ctx);
-	assert(metadata);
+	assert(metadata_stream);
 
-	pthread_mutex_unlock(&metadata->lock);
+	metadata_channel = metadata_stream->chan;
+	pthread_mutex_unlock(&metadata_stream->lock);
 	/*
 	 * Request metadata from the sessiond, but don't wait for the flush
 	 * because we locked the metadata thread.
 	 */
-	ret = lttng_ustconsumer_request_metadata(ctx, metadata->chan, 0, 0);
-	pthread_mutex_lock(&metadata->lock);
+	ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 0);
+	pthread_mutex_lock(&metadata_stream->lock);
 	if (ret < 0) {
 		goto end;
 	}
 
-	ret = commit_one_metadata_packet(metadata);
+	/*
+	 * The metadata stream and channel can be deleted while the
+	 * metadata stream lock was released. The streamed is checked
+	 * for deletion before we use it further.
+	 *
+	 * Note that it is safe to access a logically-deleted stream since its
+	 * existence is still guaranteed by the RCU read side lock. However,
+	 * it should no longer be used. The close/deletion of the metadata
+	 * channel and stream already guarantees that all metadata has been
+	 * consumed. Therefore, there is nothing left to do in this function.
+	 */
+	if (consumer_stream_is_deleted(metadata_stream)) {
+		DBG("Metadata stream %" PRIu64 " was deleted during the metadata synchronization",
+				metadata_stream->key);
+		ret = 0;
+		goto end;
+	}
+
+	ret = commit_one_metadata_packet(metadata_stream);
 	if (ret <= 0) {
 		goto end;
 	} else if (ret > 0) {
 		retry = 1;
 	}
 
-	ret = ustctl_snapshot(metadata->ustream);
+	ret = ustctl_snapshot(metadata_stream->ustream);
 	if (ret < 0) {
 		if (errno != EAGAIN) {
 			ERR("Sync metadata, taking UST snapshot");
@@ -2724,10 +2772,12 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx)
 {
 	unsigned long len, subbuf_size, padding;
-	int err, write_index = 1, rotation_ret;
+	int err, write_index = 1;
 	long ret = 0;
 	struct ustctl_consumer_stream *ustream;
 	struct ctf_packet_index index;
+	const char *subbuf_addr;
+	struct lttng_buffer_view subbuf_view;
 
 	assert(stream);
 	assert(stream->ustream);
@@ -2760,20 +2810,6 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		}
 	}
 
-	/*
-	 * 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 extracting data");
-		rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
-		if (rotation_ret < 0) {
-			ERR("Stream rotation error");
-			ret = -1;
-			goto error;
-		}
-	}
-
 retry:
 	/* Get the next subbuffer */
 	err = ustctl_get_next_subbuf(ustream);
@@ -2837,11 +2873,21 @@ retry:
 
 	padding = len - subbuf_size;
 
+	ret = get_current_subbuf_addr(stream, &subbuf_addr);
+	if (ret) {
+		write_index = 0;
+		goto error_put_subbuf;
+	}
+
+	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)) {
@@ -2858,6 +2904,7 @@ retry:
 				ret, len, subbuf_size);
 		write_index = 0;
 	}
+error_put_subbuf:
 	err = ustctl_put_next_subbuf(ustream);
 	assert(err == 0);
 
@@ -2874,7 +2921,7 @@ retry:
 
 	/* Write index if needed. */
 	if (!write_index) {
-		goto rotate;
+		goto end;
 	}
 
 	if (stream->chan->live_timer_interval && !stream->metadata_flag) {
@@ -2909,24 +2956,7 @@ retry:
 		goto error;
 	}
 
-rotate:
-	/*
-	 * After extracting the packet, we check if the stream is now ready to be
-	 * rotated and perform the action immediately.
-	 */
-	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) {
-			ERR("Stream rotation error");
-			ret = -1;
-			goto error;
-		}
-	} else if (rotation_ret < 0) {
-		ERR("Checking if stream is ready to rotate");
-		ret = -1;
-		goto error;
-	}
+end:
 error:
 	return ret;
 }
@@ -3031,7 +3061,7 @@ end:
  * Stop a given metadata channel timer if enabled and close the wait fd which
  * is the poll pipe of the metadata stream.
  *
- * This MUST be called with the metadata channel acquired.
+ * This MUST be called with the metadata channel lock acquired.
  */
 void lttng_ustconsumer_close_metadata(struct lttng_consumer_channel *metadata)
 {