X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=161bf7e324ba38abcc6976f91f219c9f2257d4af;hp=f4eaf705f0b811230d0a9bb1f2f74458c7f7e6d0;hb=5ab7344e564fad002de7efc879ffbed5bd23d5ef;hpb=c30aaa51f34105a7f20b9ceb39866001843db6e6

diff --git a/src/common/consumer.c b/src/common/consumer.c
index f4eaf705f..161bf7e32 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -62,20 +62,23 @@ volatile int consumer_quit = 0;
  * Find a stream. The consumer_data.lock must be locked during this
  * call.
  */
-static struct lttng_consumer_stream *consumer_find_stream(int key)
+static struct lttng_consumer_stream *consumer_find_stream(int key,
+		struct lttng_ht *ht)
 {
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_ulong *node;
 	struct lttng_consumer_stream *stream = NULL;
 
+	assert(ht);
+
 	/* Negative keys are lookup failures */
-	if (key < 0)
+	if (key < 0) {
 		return NULL;
+	}
 
 	rcu_read_lock();
 
-	lttng_ht_lookup(consumer_data.stream_ht, (void *)((unsigned long) key),
-			&iter);
+	lttng_ht_lookup(ht, (void *)((unsigned long) key), &iter);
 	node = lttng_ht_iter_get_node_ulong(&iter);
 	if (node != NULL) {
 		stream = caa_container_of(node, struct lttng_consumer_stream, node);
@@ -86,12 +89,12 @@ static struct lttng_consumer_stream *consumer_find_stream(int key)
 	return stream;
 }
 
-static void consumer_steal_stream_key(int key)
+static void consumer_steal_stream_key(int key, struct lttng_ht *ht)
 {
 	struct lttng_consumer_stream *stream;
 
 	rcu_read_lock();
-	stream = consumer_find_stream(key);
+	stream = consumer_find_stream(key, ht);
 	if (stream) {
 		stream->key = -1;
 		/*
@@ -111,8 +114,9 @@ static struct lttng_consumer_channel *consumer_find_channel(int key)
 	struct lttng_consumer_channel *channel = NULL;
 
 	/* Negative keys are lookup failures */
-	if (key < 0)
+	if (key < 0) {
 		return NULL;
+	}
 
 	rcu_read_lock();
 
@@ -242,7 +246,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 		if (stream->mmap_base != NULL) {
 			ret = munmap(stream->mmap_base, stream->mmap_len);
 			if (ret != 0) {
-				perror("munmap");
+				PERROR("munmap");
 			}
 		}
 		break;
@@ -345,20 +349,28 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 		uid_t uid,
 		gid_t gid,
 		int net_index,
-		int metadata_flag)
+		int metadata_flag,
+		int *alloc_ret)
 {
 	struct lttng_consumer_stream *stream;
 	int ret;
 
 	stream = zmalloc(sizeof(*stream));
 	if (stream == NULL) {
-		perror("malloc struct lttng_consumer_stream");
+		PERROR("malloc struct lttng_consumer_stream");
+		*alloc_ret = -ENOMEM;
 		goto end;
 	}
+
+	/*
+	 * Get stream's channel reference. Needed when adding the stream to the
+	 * global hash table.
+	 */
 	stream->chan = consumer_find_channel(channel_key);
 	if (!stream->chan) {
-		perror("Unable to find channel key");
-		goto end;
+		*alloc_ret = -ENOENT;
+		ERR("Unable to find channel for stream %d", stream_key);
+		goto error;
 	}
 	stream->chan->refcount++;
 	stream->key = stream_key;
@@ -387,14 +399,14 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 		stream->cpu = stream->chan->cpucount++;
 		ret = lttng_ustconsumer_allocate_stream(stream);
 		if (ret) {
-			free(stream);
-			return NULL;
+			*alloc_ret = -EINVAL;
+			goto error;
 		}
 		break;
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
-		goto end;
+		*alloc_ret = -EINVAL;
+		goto error;
 	}
 
 	/*
@@ -413,9 +425,12 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 			stream->shm_fd, stream->wait_fd,
 			(unsigned long long) stream->mmap_len, stream->out_fd,
 			stream->net_seq_idx);
+	return stream;
 
+error:
+	free(stream);
 end:
-	return stream;
+	return NULL;
 }
 
 /*
@@ -430,7 +445,7 @@ int consumer_add_stream(struct lttng_consumer_stream *stream)
 
 	pthread_mutex_lock(&consumer_data.lock);
 	/* Steal stream identifier, for UST */
-	consumer_steal_stream_key(stream->key);
+	consumer_steal_stream_key(stream->key, consumer_data.stream_ht);
 
 	rcu_read_lock();
 	lttng_ht_lookup(consumer_data.stream_ht,
@@ -607,7 +622,7 @@ void consumer_change_stream_state(int stream_key,
 	struct lttng_consumer_stream *stream;
 
 	pthread_mutex_lock(&consumer_data.lock);
-	stream = consumer_find_stream(stream_key);
+	stream = consumer_find_stream(stream_key, consumer_data.stream_ht);
 	if (stream) {
 		stream->state = state;
 	}
@@ -659,7 +674,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 	if (channel->mmap_base != NULL) {
 		ret = munmap(channel->mmap_base, channel->mmap_len);
 		if (ret != 0) {
-			perror("munmap");
+			PERROR("munmap");
 		}
 	}
 	if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
@@ -692,7 +707,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(
 
 	channel = zmalloc(sizeof(*channel));
 	if (channel == NULL) {
-		perror("malloc struct lttng_consumer_channel");
+		PERROR("malloc struct lttng_consumer_channel");
 		goto end;
 	}
 	channel->key = channel_key;
@@ -816,7 +831,7 @@ restart:
 		if (errno == EINTR) {
 			goto restart;
 		}
-		perror("Poll error");
+		PERROR("Poll error");
 		goto exit;
 	}
 	if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
@@ -908,7 +923,7 @@ void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
 		ret = write(ctx->consumer_should_quit[1], "4", 1);
 	} while (ret < 0 && errno == EINTR);
 	if (ret < 0) {
-		perror("write consumer quit");
+		PERROR("write consumer quit");
 	}
 }
 
@@ -980,7 +995,7 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 
 	ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
 	if (ctx == NULL) {
-		perror("allocating context");
+		PERROR("allocating context");
 		goto error;
 	}
 
@@ -993,33 +1008,33 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 
 	ret = pipe(ctx->consumer_poll_pipe);
 	if (ret < 0) {
-		perror("Error creating poll pipe");
+		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);
 	if (ret < 0) {
-		perror("fcntl O_NONBLOCK");
+		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");
+		PERROR("fcntl O_NONBLOCK");
 		goto error_poll_fcntl;
 	}
 
 	ret = pipe(ctx->consumer_should_quit);
 	if (ret < 0) {
-		perror("Error creating recv pipe");
+		PERROR("Error creating recv pipe");
 		goto error_quit_pipe;
 	}
 
 	ret = pipe(ctx->consumer_thread_pipe);
 	if (ret < 0) {
-		perror("Error creating thread pipe");
+		PERROR("Error creating thread pipe");
 		goto error_thread_pipe;
 	}
 
@@ -1503,9 +1518,7 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 }
 
 /*
- * Iterate over all stream element of the hashtable and free them. This is race
- * free since the hashtable received MUST be in a race free synchronization
- * state. It's the caller responsability to make sure of that.
+ * Iterate over all streams of the hashtable and free them properly.
  */
 static void destroy_stream_ht(struct lttng_ht *ht)
 {
@@ -1522,7 +1535,7 @@ static void destroy_stream_ht(struct lttng_ht *ht)
 		ret = lttng_ht_del(ht, &iter);
 		assert(!ret);
 
-		free(stream);
+		call_rcu(&stream->node.head, consumer_free_stream);
 	}
 	rcu_read_unlock();
 
@@ -1622,7 +1635,7 @@ static void consumer_del_metadata_stream(struct lttng_consumer_stream *stream)
 		consumer_del_channel(stream->chan);
 	}
 
-	free(stream);
+	call_rcu(&stream->node.head, consumer_free_stream);
 }
 
 /*
@@ -1719,20 +1732,13 @@ restart:
 					close(ctx->consumer_metadata_pipe[0]);
 					continue;
 				} else if (revents & LPOLLIN) {
-					stream = zmalloc(sizeof(struct lttng_consumer_stream));
-					if (stream == NULL) {
-						PERROR("zmalloc metadata consumer stream");
-						goto error;
-					}
-
 					do {
-						/* Get the stream and add it to the local hash table */
-						ret = read(pollfd, stream,
-								sizeof(struct lttng_consumer_stream));
+						/* Get the stream pointer received */
+						ret = read(pollfd, &stream, sizeof(stream));
 					} while (ret < 0 && errno == EINTR);
-					if (ret < 0 || ret < sizeof(struct lttng_consumer_stream)) {
+					if (ret < 0 ||
+							ret < sizeof(struct lttng_consumer_stream *)) {
 						PERROR("read metadata stream");
-						free(stream);
 						/*
 						 * Let's continue here and hope we can still work
 						 * without stopping the consumer. XXX: Should we?
@@ -1885,7 +1891,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			/* allocate for all fds + 1 for the consumer_poll_pipe */
 			pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
 			if (pollfd == NULL) {
-				perror("pollfd malloc");
+				PERROR("pollfd malloc");
 				pthread_mutex_unlock(&consumer_data.lock);
 				goto end;
 			}
@@ -1894,7 +1900,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			local_stream = zmalloc((consumer_data.stream_count + 1) *
 					sizeof(struct lttng_consumer_stream));
 			if (local_stream == NULL) {
-				perror("local_stream malloc");
+				PERROR("local_stream malloc");
 				pthread_mutex_unlock(&consumer_data.lock);
 				goto end;
 			}
@@ -1926,7 +1932,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			if (errno == EINTR) {
 				goto restart;
 			}
-			perror("Poll error");
+			PERROR("Poll error");
 			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
 			goto end;
 		} else if (num_rdy == 0) {
@@ -2098,7 +2104,7 @@ void *lttng_consumer_thread_receive_fds(void *data)
 
 	ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
 	if (ret < 0) {
-		perror("fcntl O_NONBLOCK");
+		PERROR("fcntl O_NONBLOCK");
 		goto end;
 	}
 
@@ -2121,7 +2127,7 @@ void *lttng_consumer_thread_receive_fds(void *data)
 	}
 	ret = fcntl(sock, F_SETFL, O_NONBLOCK);
 	if (ret < 0) {
-		perror("fcntl O_NONBLOCK");
+		PERROR("fcntl O_NONBLOCK");
 		goto end;
 	}