From: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Date: Wed, 1 May 2013 20:50:55 +0000 (-0400)
Subject: Fix: consumerd channel destroy race
X-Git-Tag: v2.2.0-rc2~18
X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=commitdiff_plain;h=a0cbdd2ea782afd7c263ff748dcca597095b61a0

Fix: consumerd channel destroy race

The LTTNG_CONSUMER_DESTROY_CHANNEL command is used to handle errors. It
should not call destroy_channel(), because at that point the channel
management thread has ownership of the channel. Send a notification to
the channel management thread asking it to destroy the channel instead.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
---

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 0e9c52e83..c4518cc03 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -50,12 +50,14 @@ struct lttng_consumer_global_data consumer_data = {
 
 enum consumer_channel_action {
 	CONSUMER_CHANNEL_ADD,
+	CONSUMER_CHANNEL_DEL,
 	CONSUMER_CHANNEL_QUIT,
 };
 
 struct consumer_channel_msg {
 	enum consumer_channel_action action;
-	struct lttng_consumer_channel *chan;
+	struct lttng_consumer_channel *chan;	/* add */
+	uint64_t key;				/* del */
 };
 
 /*
@@ -91,6 +93,7 @@ static void notify_thread_pipe(int wpipe)
 
 static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_channel *chan,
+		uint64_t key,
 		enum consumer_channel_action action)
 {
 	struct consumer_channel_msg msg;
@@ -103,8 +106,15 @@ static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
 	} while (ret < 0 && errno == EINTR);
 }
 
+void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
+		uint64_t key)
+{
+	notify_channel_pipe(ctx, NULL, key, CONSUMER_CHANNEL_DEL);
+}
+
 static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_channel **chan,
+		uint64_t *key,
 		enum consumer_channel_action *action)
 {
 	struct consumer_channel_msg msg;
@@ -116,6 +126,7 @@ static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
 	if (ret > 0) {
 		*action = msg.action;
 		*chan = msg.chan;
+		*key = msg.key;
 	}
 	return ret;
 }
@@ -902,7 +913,7 @@ end:
 
 	if (!ret && channel->wait_fd != -1 &&
 			channel->metadata_stream == NULL) {
-		notify_channel_pipe(ctx, channel, CONSUMER_CHANNEL_ADD);
+		notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
 	}
 	return ret;
 }
@@ -2693,8 +2704,9 @@ restart:
 					continue;
 				} else if (revents & LPOLLIN) {
 					enum consumer_channel_action action;
+					uint64_t key;
 
-					ret = read_channel_pipe(ctx, &chan, &action);
+					ret = read_channel_pipe(ctx, &chan, &key, &action);
 					if (ret <= 0) {
 						ERR("Error reading channel pipe");
 						continue;
@@ -2713,6 +2725,27 @@ restart:
 						lttng_poll_add(&events, chan->wait_fd,
 								LPOLLIN | LPOLLPRI);
 						break;
+					case CONSUMER_CHANNEL_DEL:
+					{
+						chan = consumer_find_channel(key);
+						if (!chan) {
+							ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
+							break;
+						}
+						lttng_poll_del(&events, chan->wait_fd);
+						ret = lttng_ht_del(channel_ht, &iter);
+						assert(ret == 0);
+						consumer_close_channel_streams(chan);
+
+						/*
+						 * Release our own refcount. Force channel deletion even if
+						 * streams were not initialized.
+						 */
+						if (!uatomic_sub_return(&chan->refcount, 1)) {
+							consumer_del_channel(chan);
+						}
+						goto restart;
+					}
 					case CONSUMER_CHANNEL_QUIT:
 						/*
 						 * Remove the pipe from the poll set and continue the loop
@@ -2933,7 +2966,7 @@ end:
 	 */
 	notify_thread_pipe(ctx->consumer_data_pipe[1]);
 
-	notify_channel_pipe(ctx, NULL, CONSUMER_CHANNEL_QUIT);
+	notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
 
 	/* Cleaning up possibly open sockets. */
 	if (sock >= 0) {
diff --git a/src/common/consumer.h b/src/common/consumer.h
index 844310681..43989e4c5 100644
--- a/src/common/consumer.h
+++ b/src/common/consumer.h
@@ -527,5 +527,7 @@ int consumer_data_pending(uint64_t id);
 int consumer_send_status_msg(int sock, int ret_code);
 int consumer_send_status_channel(int sock,
 		struct lttng_consumer_channel *channel);
+void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
+		uint64_t key);
 
 #endif /* LIB_CONSUMER_H */
diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 8c15c7e13..7d9444cd0 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -998,17 +998,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	case LTTNG_CONSUMER_DESTROY_CHANNEL:
 	{
 		uint64_t key = msg.u.destroy_channel.key;
-		struct lttng_consumer_channel *channel;
-
-		channel = consumer_find_channel(key);
-		if (!channel) {
-			ERR("UST consumer get channel key %" PRIu64 " not found", key);
-			ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND;
-			goto end_msg_sessiond;
-		}
-
-		destroy_channel(channel);
 
+		/*
+		 * Only called if streams have not been sent to stream
+		 * manager thread. However, channel has been sent to
+		 * channel manager thread.
+		 */
+		notify_thread_del_channel(ctx, key);
 		goto end_msg_sessiond;
 	}
 	case LTTNG_CONSUMER_CLOSE_METADATA: