X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=dd8c621f177d56968306fb500ce4546870b5bd2d;hp=d68bc5c2713eab333b80f9c7b196fa0b43de936c;hb=15d3d5d89e2ca0656e653323c053d29e72674d32;hpb=d14d33bf091e72b23b1f90ea18a0a01bed098b76

diff --git a/src/common/consumer.c b/src/common/consumer.c
index d68bc5c27..dd8c621f1 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -298,11 +298,23 @@ end:
 int consumer_add_stream(struct lttng_consumer_stream *stream)
 {
 	int ret = 0;
+	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_iter iter;
 
 	pthread_mutex_lock(&consumer_data.lock);
 	/* Steal stream identifier, for UST */
 	consumer_steal_stream_key(stream->key);
 	rcu_read_lock();
+
+	lttng_ht_lookup(consumer_data.stream_ht,
+			(void *)((unsigned long) stream->key), &iter);
+	node = lttng_ht_iter_get_node_ulong(&iter);
+	if (node != NULL) {
+		rcu_read_unlock();
+		/* Stream already exist. Ignore the insertion */
+		goto end;
+	}
+
 	lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
 	rcu_read_unlock();
 	consumer_data.stream_count++;
@@ -466,11 +478,25 @@ end:
  */
 int consumer_add_channel(struct lttng_consumer_channel *channel)
 {
+	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_iter iter;
+
 	pthread_mutex_lock(&consumer_data.lock);
 	/* Steal channel identifier, for UST */
 	consumer_steal_channel_key(channel->key);
 	rcu_read_lock();
+
+	lttng_ht_lookup(consumer_data.channel_ht,
+			(void *)((unsigned long) channel->key), &iter);
+	node = lttng_ht_iter_get_node_ulong(&iter);
+	if (node != NULL) {
+		/* Channel already exist. Ignore the insertion */
+		goto end;
+	}
+
 	lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
+
+end:
 	rcu_read_unlock();
 	pthread_mutex_unlock(&consumer_data.lock);
 
@@ -512,7 +538,7 @@ int consumer_update_poll_array(
 	 * increment i so nb_fd is the number of real FD.
 	 */
 	(*pollfd)[i].fd = ctx->consumer_poll_pipe[0];
-	(*pollfd)[i].events = POLLIN;
+	(*pollfd)[i].events = POLLIN | POLLPRI;
 	return i;
 }
 
@@ -536,7 +562,7 @@ restart:
 		perror("Poll error");
 		goto exit;
 	}
-	if (consumer_sockpoll[0].revents == POLLIN) {
+	if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
 		DBG("consumer_should_quit wake up");
 		goto exit;
 	}
@@ -610,6 +636,9 @@ void lttng_consumer_cleanup(void)
 	}
 
 	rcu_read_unlock();
+
+	lttng_ht_destroy(consumer_data.stream_ht);
+	lttng_ht_destroy(consumer_data.channel_ht);
 }
 
 /*
@@ -710,6 +739,20 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 		goto error_poll_pipe;
 	}
 
+	/* set read end of the pipe to non-blocking */
+	ret = fcntl(ctx->consumer_poll_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);
+	if (ret < 0) {
+		perror("fcntl O_NONBLOCK");
+		goto error_poll_fcntl;
+	}
+
 	ret = pipe(ctx->consumer_should_quit);
 	if (ret < 0) {
 		perror("Error creating recv pipe");
@@ -734,6 +777,7 @@ error_thread_pipe:
 			PERROR("close");
 		}
 	}
+error_poll_fcntl:
 error_quit_pipe:
 	for (i = 0; i < 2; i++) {
 		int err;
@@ -907,8 +951,6 @@ void *lttng_consumer_thread_poll_fds(void *data)
 	struct lttng_consumer_stream **local_stream = NULL;
 	/* local view of consumer_data.fds_count */
 	int nb_fd = 0;
-	char tmp;
-	int tmp2;
 	struct lttng_consumer_local_data *ctx = data;
 
 	rcu_register_thread();
@@ -992,12 +1034,15 @@ void *lttng_consumer_thread_poll_fds(void *data)
 		 * array. We want to prioritize array update over
 		 * low-priority reads.
 		 */
-		if (pollfd[nb_fd].revents & POLLIN) {
+		if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
+			size_t pipe_readlen;
+			char tmp;
+
 			DBG("consumer_poll_pipe wake up");
-			tmp2 = read(ctx->consumer_poll_pipe[0], &tmp, 1);
-			if (tmp2 < 0) {
-				perror("read consumer poll");
-			}
+			/* Consume 1 byte of pipe data */
+			do {
+				pipe_readlen = read(ctx->consumer_poll_pipe[0], &tmp, 1);
+			} while (pipe_readlen == -1 && errno == EINTR);
 			continue;
 		}
 
@@ -1202,11 +1247,20 @@ end:
 	 */
 	consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
 
-	/* wake up the polling thread */
-	ret = write(ctx->consumer_poll_pipe[1], "4", 1);
-	if (ret < 0) {
-		perror("poll pipe write");
-	}
+	/*
+	 * Wake-up the other end by writing a null byte in the pipe
+	 * (non-blocking). Important note: Because writing into the
+	 * pipe is non-blocking (and therefore we allow dropping wakeup
+	 * data, as long as there is wakeup data present in the pipe
+	 * buffer to wake up the other end), the other end should
+	 * perform the following sequence for waiting:
+	 * 1) empty the pipe (reads).
+	 * 2) perform update operation.
+	 * 3) wait on the pipe (poll).
+	 */
+	do {
+		ret = write(ctx->consumer_poll_pipe[1], "", 1);
+	} while (ret == -1UL && errno == EINTR);
 	rcu_unregister_thread();
 	return NULL;
 }