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

diff --git a/src/common/consumer.c b/src/common/consumer.c
index e7d51078e..05bf85b3d 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -123,6 +123,17 @@ static void consumer_steal_channel_key(int key)
 		channel->key = -1;
 }
 
+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);
+}
+
 /*
  * Remove a stream from the global list protected by a mutex. This
  * function is also responsible for freeing its data structures.
@@ -159,9 +170,11 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 	/* 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 */
-	ret = lttng_ht_del(consumer_data.stream_ht, &iter);
-	assert(!ret);
+	/*
+	 * Remove stream node from hash table. It can fail if it's been
+	 * replaced due to key reuse.
+	 */
+	(void) lttng_ht_del(consumer_data.stream_ht, &iter);
 
 	rcu_read_unlock();
 
@@ -192,7 +205,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 +215,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,
@@ -288,7 +292,12 @@ int consumer_add_stream(struct lttng_consumer_stream *stream)
 	/* Steal stream identifier, for UST */
 	consumer_steal_stream_key(stream->key);
 	rcu_read_lock();
-	lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
+	/*
+	 * We simply remove the old channel from the hash table. It's
+	 * ok, since we know for sure the sessiond wants to replace it
+	 * with the new version, because the key has been reused.
+	 */
+	(void) lttng_ht_add_replace_ulong(consumer_data.stream_ht, &stream->node);
 	rcu_read_unlock();
 	consumer_data.stream_count++;
 	consumer_data.need_update = 1;
@@ -309,6 +318,7 @@ int consumer_add_stream(struct lttng_consumer_stream *stream)
 
 end:
 	pthread_mutex_unlock(&consumer_data.lock);
+
 	return ret;
 }
 
@@ -329,6 +339,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.
@@ -357,8 +378,12 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 
 	lttng_ht_lookup(consumer_data.channel_ht,
 			(void *)((unsigned long) channel->key), &iter);
-	ret = lttng_ht_del(consumer_data.channel_ht, &iter);
-	assert(!ret);
+
+	/*
+	 * Remove channel node from hash table. It can fail if it's been
+	 * replaced due to key reuse.
+	 */
+	(void) lttng_ht_del(consumer_data.channel_ht, &iter);
 
 	rcu_read_unlock();
 
@@ -380,21 +405,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,
@@ -455,9 +471,15 @@ int consumer_add_channel(struct lttng_consumer_channel *channel)
 	/* Steal channel identifier, for UST */
 	consumer_steal_channel_key(channel->key);
 	rcu_read_lock();
-	lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
+	/*
+	 * We simply remove the old channel from the hash table. It's
+	 * ok, since we know for sure the sessiond wants to replace it
+	 * with the new version, because the key has been reused.
+	 */
+	(void) lttng_ht_add_replace_ulong(consumer_data.channel_ht, &channel->node);
 	rcu_read_unlock();
 	pthread_mutex_unlock(&consumer_data.lock);
+
 	return 0;
 }
 
@@ -568,7 +590,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,16 +601,16 @@ 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();
@@ -1048,25 +1069,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++;
 				}
 			}