X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=b55fd153d7830e1d817ed4144dbd65fdf776b00b;hp=a9e4dee661b1d1ba13da01a2db4d74ee16f68e8e;hb=f724d81e402a1eb319ce0392c5b6770787a28edc;hpb=c80048c6378b9ee7796c4b833a2c07f6050cc6d7

diff --git a/src/common/consumer.c b/src/common/consumer.c
index a9e4dee66..b55fd153d 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -69,8 +69,9 @@ static struct lttng_consumer_stream *consumer_find_stream(int key)
 	struct lttng_consumer_stream *stream = NULL;
 
 	/* Negative keys are lookup failures */
-	if (key < 0)
+	if (key < 0) {
 		return NULL;
+	}
 
 	rcu_read_lock();
 
@@ -111,8 +112,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 +244,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;
@@ -353,13 +355,19 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 
 	stream = zmalloc(sizeof(*stream));
 	if (stream == NULL) {
-		perror("malloc struct lttng_consumer_stream");
+		PERROR("malloc struct lttng_consumer_stream");
 		*alloc_ret = -ENOMEM;
-		return NULL;
+		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) {
 		*alloc_ret = -ENOENT;
+		ERR("Unable to find channel for stream %d", stream_key);
 		goto error;
 	}
 	stream->chan->refcount++;
@@ -419,6 +427,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(
 
 error:
 	free(stream);
+end:
 	return NULL;
 }
 
@@ -663,7 +672,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) {
@@ -696,7 +705,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;
@@ -820,7 +829,7 @@ restart:
 		if (errno == EINTR) {
 			goto restart;
 		}
-		perror("Poll error");
+		PERROR("Poll error");
 		goto exit;
 	}
 	if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
@@ -912,7 +921,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");
 	}
 }
 
@@ -984,7 +993,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;
 	}
 
@@ -997,33 +1006,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;
 	}
 
@@ -1507,9 +1516,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)
 {
@@ -1526,7 +1533,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();
 
@@ -1626,7 +1633,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);
 }
 
 /*
@@ -1723,20 +1730,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?
@@ -1889,7 +1889,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;
 			}
@@ -1898,7 +1898,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;
 			}
@@ -1930,7 +1930,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) {
@@ -2102,7 +2102,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;
 	}
 
@@ -2125,7 +2125,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;
 	}