X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=dd8c621f177d56968306fb500ce4546870b5bd2d;hp=e7d51078e64b9b5209d4e8208a61a25d9e0bef38;hb=5c827ce0dce140b121032837510f89cb70d1650d;hpb=e71aad1fa4b06a5f91ddceace42366f3d79bd77e

diff --git a/src/common/consumer.c b/src/common/consumer.c
index e7d51078e..dd8c621f1 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -2,19 +2,18 @@
  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
  *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License
- * as published by the Free Software Foundation; only version 2
- * of the License.
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License, version 2 only,
+ * as published by the Free Software Foundation.
  *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
  *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software
- * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 
 #define _GNU_SOURCE
@@ -85,9 +84,18 @@ static void consumer_steal_stream_key(int key)
 {
 	struct lttng_consumer_stream *stream;
 
+	rcu_read_lock();
 	stream = consumer_find_stream(key);
-	if (stream)
+	if (stream) {
 		stream->key = -1;
+		/*
+		 * We don't want the lookup to match, but we still need
+		 * to iterate on this stream when iterating over the hash table. Just
+		 * change the node key.
+		 */
+		stream->node.key = -1;
+	}
+	rcu_read_unlock();
 }
 
 static struct lttng_consumer_channel *consumer_find_channel(int key)
@@ -118,9 +126,29 @@ static void consumer_steal_channel_key(int key)
 {
 	struct lttng_consumer_channel *channel;
 
+	rcu_read_lock();
 	channel = consumer_find_channel(key);
-	if (channel)
+	if (channel) {
 		channel->key = -1;
+		/*
+		 * We don't want the lookup to match, but we still need
+		 * to iterate on this channel when iterating over the hash table. Just
+		 * change the node key.
+		 */
+		channel->node.key = -1;
+	}
+	rcu_read_unlock();
+}
+
+static
+void consumer_free_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, node);
+
+	free(stream);
 }
 
 /*
@@ -155,11 +183,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 	}
 
 	rcu_read_lock();
-
-	/* Get stream node from hash table */
-	lttng_ht_lookup(consumer_data.stream_ht,
-			(void *)((unsigned long) stream->key), &iter);
-	/* Remove stream node from hash table */
+	iter.iter.node = &stream->node.node;
 	ret = lttng_ht_del(consumer_data.stream_ht, &iter);
 	assert(!ret);
 
@@ -192,7 +216,8 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 	}
 	if (!--stream->chan->refcount)
 		free_chan = stream->chan;
-	free(stream);
+
+	call_rcu(&stream->node.head, consumer_free_stream);
 end:
 	consumer_data.need_update = 1;
 	pthread_mutex_unlock(&consumer_data.lock);
@@ -201,16 +226,6 @@ end:
 		consumer_del_channel(free_chan);
 }
 
-static void consumer_del_stream_rcu(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, node);
-
-	consumer_del_stream(stream);
-}
-
 struct lttng_consumer_stream *consumer_allocate_stream(
 		int channel_key, int stream_key,
 		int shm_fd, int wait_fd,
@@ -283,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++;
@@ -309,6 +336,7 @@ int consumer_add_stream(struct lttng_consumer_stream *stream)
 
 end:
 	pthread_mutex_unlock(&consumer_data.lock);
+
 	return ret;
 }
 
@@ -329,6 +357,17 @@ void consumer_change_stream_state(int stream_key,
 	pthread_mutex_unlock(&consumer_data.lock);
 }
 
+static
+void consumer_free_channel(struct rcu_head *head)
+{
+	struct lttng_ht_node_ulong *node =
+		caa_container_of(head, struct lttng_ht_node_ulong, head);
+	struct lttng_consumer_channel *channel =
+		caa_container_of(node, struct lttng_consumer_channel, node);
+
+	free(channel);
+}
+
 /*
  * Remove a channel from the global list protected by a mutex. This
  * function is also responsible for freeing its data structures.
@@ -354,12 +393,9 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 	}
 
 	rcu_read_lock();
-
-	lttng_ht_lookup(consumer_data.channel_ht,
-			(void *)((unsigned long) channel->key), &iter);
+	iter.iter.node = &channel->node.node;
 	ret = lttng_ht_del(consumer_data.channel_ht, &iter);
 	assert(!ret);
-
 	rcu_read_unlock();
 
 	if (channel->mmap_base != NULL) {
@@ -380,21 +416,12 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 			PERROR("close");
 		}
 	}
-	free(channel);
+
+	call_rcu(&channel->node.head, consumer_free_channel);
 end:
 	pthread_mutex_unlock(&consumer_data.lock);
 }
 
-static void consumer_del_channel_rcu(struct rcu_head *head)
-{
-	struct lttng_ht_node_ulong *node =
-		caa_container_of(head, struct lttng_ht_node_ulong, head);
-	struct lttng_consumer_channel *channel=
-		caa_container_of(node, struct lttng_consumer_channel, node);
-
-	consumer_del_channel(channel);
-}
-
 struct lttng_consumer_channel *consumer_allocate_channel(
 		int channel_key,
 		int shm_fd, int wait_fd,
@@ -451,13 +478,28 @@ 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);
+
 	return 0;
 }
 
@@ -477,6 +519,7 @@ int consumer_update_poll_array(
 	struct lttng_consumer_stream *stream;
 
 	DBG("Updating poll fd array");
+	rcu_read_lock();
 	cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, stream,
 			node.node) {
 		if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM) {
@@ -488,13 +531,14 @@ int consumer_update_poll_array(
 		local_stream[i] = stream;
 		i++;
 	}
+	rcu_read_unlock();
 
 	/*
 	 * Insert the consumer_poll_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].events = POLLIN;
+	(*pollfd)[i].events = POLLIN | POLLPRI;
 	return i;
 }
 
@@ -518,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;
 	}
@@ -568,7 +612,6 @@ int lttng_consumer_send_error(
  */
 void lttng_consumer_cleanup(void)
 {
-	int ret;
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_ulong *node;
 
@@ -580,19 +623,22 @@ void lttng_consumer_cleanup(void)
 	 */
 	cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, node,
 			node) {
-		ret = lttng_ht_del(consumer_data.stream_ht, &iter);
-		assert(!ret);
-		call_rcu(&node->head, consumer_del_stream_rcu);
+		struct lttng_consumer_stream *stream =
+			caa_container_of(node, struct lttng_consumer_stream, node);
+		consumer_del_stream(stream);
 	}
 
 	cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
 			node) {
-		ret = lttng_ht_del(consumer_data.channel_ht, &iter);
-		assert(!ret);
-		call_rcu(&node->head, consumer_del_channel_rcu);
+		struct lttng_consumer_channel *channel =
+			caa_container_of(node, struct lttng_consumer_channel, node);
+		consumer_del_channel(channel);
 	}
 
 	rcu_read_unlock();
+
+	lttng_ht_destroy(consumer_data.stream_ht);
+	lttng_ht_destroy(consumer_data.channel_ht);
 }
 
 /*
@@ -693,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");
@@ -717,6 +777,7 @@ error_thread_pipe:
 			PERROR("close");
 		}
 	}
+error_poll_fcntl:
 error_quit_pipe:
 	for (i = 0; i < 2; i++) {
 		int err;
@@ -890,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();
@@ -975,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;
 		}
 
@@ -1015,8 +1077,6 @@ void *lttng_consumer_thread_poll_fds(void *data)
 					local_stream[i]->hangup_flush_done) {
 				ssize_t len;
 
-				assert(!(pollfd[i].revents & POLLERR));
-				assert(!(pollfd[i].revents & POLLNVAL));
 				DBG("Normal read on fd %d", pollfd[i].fd);
 				len = ctx->on_buffer_ready(local_stream[i], ctx);
 				/* it's ok to have an unavailable sub-buffer */
@@ -1048,25 +1108,19 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			if ((pollfd[i].revents & POLLHUP)) {
 				DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
-					rcu_read_lock();
-					consumer_del_stream_rcu(&local_stream[i]->node.head);
-					rcu_read_unlock();
+					consumer_del_stream(local_stream[i]);
 					num_hup++;
 				}
 			} else if (pollfd[i].revents & POLLERR) {
 				ERR("Error returned in polling fd %d.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
-					rcu_read_lock();
-					consumer_del_stream_rcu(&local_stream[i]->node.head);
-					rcu_read_unlock();
+					consumer_del_stream(local_stream[i]);
 					num_hup++;
 				}
 			} else if (pollfd[i].revents & POLLNVAL) {
 				ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
-					rcu_read_lock();
-					consumer_del_stream_rcu(&local_stream[i]->node.head);
-					rcu_read_unlock();
+					consumer_del_stream(local_stream[i]);
 					num_hup++;
 				}
 			}
@@ -1193,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;
 }