X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=8d1a34025687da37b7c3c1555d80441c93d8761d;hp=0b2f07391009f4a317b8f7c876443addad39d660;hb=ca22feea083301934d1c8511851c86fb008c0697;hpb=43c34bc328e6970b298c9f5cd661e2ca648ebf16

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 0b2f07391..8d1a34025 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -172,17 +172,6 @@ void consumer_free_stream(struct rcu_head *head)
 	free(stream);
 }
 
-static
-void consumer_free_metadata_stream(struct rcu_head *head)
-{
-	struct lttng_ht_node_ulong *node =
-		caa_container_of(head, struct lttng_ht_node_ulong, head);
-	struct lttng_consumer_stream *stream =
-		caa_container_of(node, struct lttng_consumer_stream, waitfd_node);
-
-	free(stream);
-}
-
 /*
  * RCU protected relayd socket pair free.
  */
@@ -293,15 +282,15 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
 	ret = lttng_ht_del(ht, &iter);
 	assert(!ret);
 
+	/* Remove node session id from the consumer_data stream ht */
+	iter.iter.node = &stream->node_session_id.node;
+	ret = lttng_ht_del(consumer_data.stream_list_ht, &iter);
+	assert(!ret);
 	rcu_read_unlock();
 
-	if (consumer_data.stream_count <= 0) {
-		goto end;
-	}
+	assert(consumer_data.stream_count > 0);
 	consumer_data.stream_count--;
-	if (!stream) {
-		goto end;
-	}
+
 	if (stream->out_fd >= 0) {
 		ret = close(stream->out_fd);
 		if (ret) {
@@ -380,6 +369,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 		gid_t gid,
 		int net_index,
 		int metadata_flag,
+		uint64_t session_id,
 		int *alloc_ret)
 {
 	struct lttng_consumer_stream *stream;
@@ -415,10 +405,24 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 	stream->gid = gid;
 	stream->net_seq_idx = net_index;
 	stream->metadata_flag = metadata_flag;
+	stream->session_id = session_id;
 	strncpy(stream->path_name, path_name, sizeof(stream->path_name));
 	stream->path_name[sizeof(stream->path_name) - 1] = '\0';
-	lttng_ht_node_init_ulong(&stream->waitfd_node, stream->wait_fd);
-	lttng_ht_node_init_ulong(&stream->node, stream->key);
+	pthread_mutex_init(&stream->lock, NULL);
+
+	/*
+	 * Index differently the metadata node because the thread is using an
+	 * internal hash table to match streams in the metadata_ht to the epoll set
+	 * file descriptor.
+	 */
+	if (metadata_flag) {
+		lttng_ht_node_init_ulong(&stream->node, stream->wait_fd);
+	} else {
+		lttng_ht_node_init_ulong(&stream->node, stream->key);
+	}
+
+	/* Init session id node with the stream session id */
+	lttng_ht_node_init_ulong(&stream->node_session_id, stream->session_id);
 
 	/*
 	 * The cpu number is needed before using any ustctl_* actions. Ignored for
@@ -429,10 +433,10 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 	pthread_mutex_unlock(&consumer_data.lock);
 
 	DBG3("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu,"
-			" out_fd %d, net_seq_idx %d)", stream->path_name, stream->key,
-			stream->shm_fd, stream->wait_fd,
+			" out_fd %d, net_seq_idx %d, session_id %" PRIu64,
+			stream->path_name, stream->key, stream->shm_fd, stream->wait_fd,
 			(unsigned long long) stream->mmap_len, stream->out_fd,
-			stream->net_seq_idx);
+			stream->net_seq_idx, stream->session_id);
 	return stream;
 
 error:
@@ -463,6 +467,13 @@ static int consumer_add_stream(struct lttng_consumer_stream *stream,
 
 	lttng_ht_add_unique_ulong(ht, &stream->node);
 
+	/*
+	 * Add stream to the stream_list_ht of the consumer data. No need to steal
+	 * the key since the HT does not use it and we allow to add redundant keys
+	 * into this table.
+	 */
+	lttng_ht_add_ulong(consumer_data.stream_list_ht, &stream->node_session_id);
+
 	/* Check and cleanup relayd */
 	relayd = consumer_find_relayd(stream->net_seq_idx);
 	if (relayd != NULL) {
@@ -630,23 +641,6 @@ error:
 	return outfd;
 }
 
-/*
- * Update a stream according to what we just received.
- */
-void consumer_change_stream_state(int stream_key,
-		enum lttng_consumer_stream_state state)
-{
-	struct lttng_consumer_stream *stream;
-
-	pthread_mutex_lock(&consumer_data.lock);
-	stream = consumer_find_stream(stream_key, consumer_data.stream_ht);
-	if (stream) {
-		stream->state = state;
-	}
-	consumer_data.need_update = 1;
-	pthread_mutex_unlock(&consumer_data.lock);
-}
-
 static
 void consumer_free_channel(struct rcu_head *head)
 {
@@ -822,10 +816,10 @@ static int consumer_update_poll_array(
 	rcu_read_unlock();
 
 	/*
-	 * Insert the consumer_poll_pipe at the end of the array and don't
+	 * Insert the consumer_data_pipe at the end of the array and don't
 	 * increment i so nb_fd is the number of real FD.
 	 */
-	(*pollfd)[i].fd = ctx->consumer_poll_pipe[0];
+	(*pollfd)[i].fd = ctx->consumer_data_pipe[0];
 	(*pollfd)[i].events = POLLIN | POLLPRI;
 	return i;
 }
@@ -904,17 +898,6 @@ void lttng_consumer_cleanup(void)
 
 	rcu_read_lock();
 
-	/*
-	 * close all outfd. Called when there are no more threads running (after
-	 * joining on the threads), no need to protect list iteration with mutex.
-	 */
-	cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, node,
-			node) {
-		struct lttng_consumer_stream *stream =
-			caa_container_of(node, struct lttng_consumer_stream, node);
-		consumer_del_stream(stream, consumer_data.stream_ht);
-	}
-
 	cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
 			node) {
 		struct lttng_consumer_channel *channel =
@@ -924,7 +907,6 @@ void lttng_consumer_cleanup(void)
 
 	rcu_read_unlock();
 
-	lttng_ht_destroy(consumer_data.stream_ht);
 	lttng_ht_destroy(consumer_data.channel_ht);
 }
 
@@ -1022,21 +1004,21 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 	ctx->on_recv_stream = recv_stream;
 	ctx->on_update_stream = update_stream;
 
-	ret = pipe(ctx->consumer_poll_pipe);
+	ret = pipe(ctx->consumer_data_pipe);
 	if (ret < 0) {
 		PERROR("Error creating poll pipe");
 		goto error_poll_pipe;
 	}
 
 	/* set read end of the pipe to non-blocking */
-	ret = fcntl(ctx->consumer_poll_pipe[0], F_SETFL, O_NONBLOCK);
+	ret = fcntl(ctx->consumer_data_pipe[0], F_SETFL, O_NONBLOCK);
 	if (ret < 0) {
 		PERROR("fcntl O_NONBLOCK");
 		goto error_poll_fcntl;
 	}
 
 	/* set write end of the pipe to non-blocking */
-	ret = fcntl(ctx->consumer_poll_pipe[1], F_SETFL, O_NONBLOCK);
+	ret = fcntl(ctx->consumer_data_pipe[1], F_SETFL, O_NONBLOCK);
 	if (ret < 0) {
 		PERROR("fcntl O_NONBLOCK");
 		goto error_poll_fcntl;
@@ -1084,7 +1066,7 @@ error_quit_pipe:
 	for (i = 0; i < 2; i++) {
 		int err;
 
-		err = close(ctx->consumer_poll_pipe[i]);
+		err = close(ctx->consumer_data_pipe[i]);
 		if (err) {
 			PERROR("close");
 		}
@@ -1114,11 +1096,11 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
 	if (ret) {
 		PERROR("close");
 	}
-	ret = close(ctx->consumer_poll_pipe[0]);
+	ret = close(ctx->consumer_data_pipe[0]);
 	if (ret) {
 		PERROR("close");
 	}
-	ret = close(ctx->consumer_poll_pipe[1]);
+	ret = close(ctx->consumer_data_pipe[1]);
 	if (ret) {
 		PERROR("close");
 	}
@@ -1576,11 +1558,11 @@ static void destroy_stream_ht(struct lttng_ht *ht)
 	}
 
 	rcu_read_lock();
-	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, waitfd_node.node) {
+	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
 		ret = lttng_ht_del(ht, &iter);
 		assert(!ret);
 
-		call_rcu(&stream->waitfd_node.head, consumer_free_metadata_stream);
+		call_rcu(&stream->node.head, consumer_free_stream);
 	}
 	rcu_read_unlock();
 
@@ -1633,9 +1615,14 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	}
 
 	rcu_read_lock();
-	iter.iter.node = &stream->waitfd_node.node;
+	iter.iter.node = &stream->node.node;
 	ret = lttng_ht_del(ht, &iter);
 	assert(!ret);
+
+	/* Remove node session id from the consumer_data stream ht */
+	iter.iter.node = &stream->node_session_id.node;
+	ret = lttng_ht_del(consumer_data.stream_list_ht, &iter);
+	assert(!ret);
 	rcu_read_unlock();
 
 	if (stream->out_fd >= 0) {
@@ -1704,7 +1691,7 @@ end:
 	}
 
 free_stream:
-	call_rcu(&stream->waitfd_node.head, consumer_free_metadata_stream);
+	call_rcu(&stream->node.head, consumer_free_stream);
 }
 
 /*
@@ -1753,7 +1740,15 @@ static int consumer_add_metadata_stream(struct lttng_consumer_stream *stream,
 	/* Steal stream identifier to avoid having streams with the same key */
 	consumer_steal_stream_key(stream->key, ht);
 
-	lttng_ht_add_unique_ulong(ht, &stream->waitfd_node);
+	lttng_ht_add_unique_ulong(ht, &stream->node);
+
+	/*
+	 * Add stream to the stream_list_ht of the consumer data. No need to steal
+	 * the key since the HT does not use it and we allow to add redundant keys
+	 * into this table.
+	 */
+	lttng_ht_add_ulong(consumer_data.stream_list_ht, &stream->node_session_id);
+
 	rcu_read_unlock();
 
 	pthread_mutex_unlock(&consumer_data.lock);
@@ -1878,7 +1873,7 @@ restart:
 			assert(node);
 
 			stream = caa_container_of(node, struct lttng_consumer_stream,
-					waitfd_node);
+					node);
 
 			/* Check for error event */
 			if (revents & (LPOLLERR | LPOLLHUP)) {
@@ -1983,7 +1978,7 @@ void *consumer_thread_data_poll(void *data)
 				local_stream = NULL;
 			}
 
-			/* allocate for all fds + 1 for the consumer_poll_pipe */
+			/* allocate for all fds + 1 for the consumer_data_pipe */
 			pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
 			if (pollfd == NULL) {
 				PERROR("pollfd malloc");
@@ -1991,7 +1986,7 @@ void *consumer_thread_data_poll(void *data)
 				goto end;
 			}
 
-			/* allocate for all fds + 1 for the consumer_poll_pipe */
+			/* allocate for all fds + 1 for the consumer_data_pipe */
 			local_stream = zmalloc((consumer_data.stream_count + 1) *
 					sizeof(struct lttng_consumer_stream));
 			if (local_stream == NULL) {
@@ -2037,17 +2032,17 @@ void *consumer_thread_data_poll(void *data)
 		}
 
 		/*
-		 * If the consumer_poll_pipe triggered poll go directly to the
+		 * If the consumer_data_pipe triggered poll go directly to the
 		 * beginning of the loop to update the array. We want to prioritize
 		 * array update over low-priority reads.
 		 */
 		if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
 			size_t pipe_readlen;
 
-			DBG("consumer_poll_pipe wake up");
+			DBG("consumer_data_pipe wake up");
 			/* Consume 1 byte of pipe data */
 			do {
-				pipe_readlen = read(ctx->consumer_poll_pipe[0], &new_stream,
+				pipe_readlen = read(ctx->consumer_data_pipe[0], &new_stream,
 						sizeof(new_stream));
 			} while (pipe_readlen == -1 && errno == EINTR);
 
@@ -2298,7 +2293,7 @@ end:
 	do {
 		struct lttng_consumer_stream *null_stream = NULL;
 
-		ret = write(ctx->consumer_poll_pipe[1], &null_stream,
+		ret = write(ctx->consumer_data_pipe[1], &null_stream,
 				sizeof(null_stream));
 	} while (ret < 0 && errno == EINTR);
 
@@ -2342,9 +2337,9 @@ int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
  */
 void lttng_consumer_init(void)
 {
-	consumer_data.stream_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 	consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 	consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
+	consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 
 	metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 	assert(metadata_ht);
@@ -2443,3 +2438,66 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 error:
 	return ret;
 }
+
+/*
+ * Check if for a given session id there is still data needed to be extract
+ * from the buffers.
+ *
+ * Return 1 if data is in fact available to be read or else 0.
+ */
+int consumer_data_available(uint64_t id)
+{
+	int ret;
+	struct lttng_ht_iter iter;
+	struct lttng_ht *ht;
+	struct lttng_consumer_stream *stream;
+	int (*data_available)(struct lttng_consumer_stream *);
+
+	DBG("Consumer data available command on session id %" PRIu64, id);
+
+	pthread_mutex_lock(&consumer_data.lock);
+
+	switch (consumer_data.type) {
+	case LTTNG_CONSUMER_KERNEL:
+		data_available = lttng_kconsumer_data_available;
+		break;
+	case LTTNG_CONSUMER32_UST:
+	case LTTNG_CONSUMER64_UST:
+		data_available = lttng_ustconsumer_data_available;
+		break;
+	default:
+		ERR("Unknown consumer data type");
+		assert(0);
+	}
+
+	/* Ease our life a bit */
+	ht = consumer_data.stream_list_ht;
+
+	cds_lfht_for_each_entry_duplicate(ht->ht, (long unsigned int) ht->hash_fct,
+			ht->match_fct, (void *)((unsigned long) id),
+			&iter.iter, stream, node_session_id.node) {
+		/* Check the stream for data. */
+		ret = data_available(stream);
+		if (ret == 0) {
+			goto data_not_available;
+		}
+	}
+
+	/* TODO: Support to ask the relayd if the streams are remote */
+
+	/*
+	 * Finding _no_ node in the hash table means that the stream(s) have been
+	 * removed thus data is guaranteed to be available for analysis from the
+	 * trace files. This is *only* true for local consumer and not network
+	 * streaming.
+	 */
+
+	/* Data is available to be read by a viewer. */
+	pthread_mutex_unlock(&consumer_data.lock);
+	return 1;
+
+data_not_available:
+	/* Data is still being extracted from buffers. */
+	pthread_mutex_unlock(&consumer_data.lock);
+	return 0;
+}