X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=bcede1ef5d8f3857515baeaf709ad9b78294a684;hb=27babd3a0a164f71d4dc02884ef20099ddfc6755;hp=3aafb519380b334099d584b83ca17d9b201ed6e1;hpb=4891ece8d4eeb2645efdf1467680b037c63ab425;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 3aafb5193..bcede1ef5 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -540,7 +540,6 @@ static int add_stream(struct lttng_consumer_stream *stream,
 		struct lttng_ht *ht)
 {
 	int ret = 0;
-	struct consumer_relayd_sock_pair *relayd;
 
 	assert(stream);
 	assert(ht);
@@ -566,12 +565,6 @@ static int add_stream(struct lttng_consumer_stream *stream,
 	 */
 	lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
 
-	/* Check and cleanup relayd */
-	relayd = consumer_find_relayd(stream->net_seq_idx);
-	if (relayd != NULL) {
-		uatomic_inc(&relayd->refcount);
-	}
-
 	/*
 	 * When nb_init_stream_left reaches 0, we don't need to trigger any action
 	 * in terms of destroying the associated channel, because the action that
@@ -709,6 +702,7 @@ int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
 			goto end;
 		}
 		uatomic_inc(&relayd->refcount);
+		stream->sent_to_relayd = 1;
 	} else {
 		ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
 				stream->key, stream->net_seq_idx);
@@ -1134,7 +1128,7 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 			struct lttng_consumer_local_data *ctx),
 		int (*recv_channel)(struct lttng_consumer_channel *channel),
 		int (*recv_stream)(struct lttng_consumer_stream *stream),
-		int (*update_stream)(int stream_key, uint32_t state))
+		int (*update_stream)(uint64_t stream_key, uint32_t state))
 {
 	int ret;
 	struct lttng_consumer_local_data *ctx;
@@ -1969,7 +1963,6 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 		struct lttng_ht *ht)
 {
 	int ret = 0;
-	struct consumer_relayd_sock_pair *relayd;
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_u64 *node;
 
@@ -1996,12 +1989,6 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
 	node = lttng_ht_iter_get_node_u64(&iter);
 	assert(!node);
 
-	/* Find relayd and, if one is found, increment refcount. */
-	relayd = consumer_find_relayd(stream->net_seq_idx);
-	if (relayd != NULL) {
-		uatomic_inc(&relayd->refcount);
-	}
-
 	/*
 	 * When nb_init_stream_left reaches 0, we don't need to trigger any action
 	 * in terms of destroying the associated channel, because the action that
@@ -2304,7 +2291,11 @@ void *consumer_thread_data_poll(void *data)
 		goto end;
 	}
 
-	local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
+	local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
+	if (local_stream == NULL) {
+		PERROR("local_stream malloc");
+		goto end;
+	}
 
 	while (1) {
 		high_prio = 0;
@@ -3044,7 +3035,7 @@ void lttng_consumer_init(void)
 int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
 		struct lttng_consumer_local_data *ctx, int sock,
 		struct pollfd *consumer_sockpoll,
-		struct lttcomm_relayd_sock *relayd_sock, unsigned int sessiond_id)
+		struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id)
 {
 	int fd = -1, ret = -1, relayd_created = 0;
 	enum lttng_error_code ret_code = LTTNG_OK;
@@ -3066,7 +3057,7 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
 			ret_code = LTTCOMM_CONSUMERD_ENOMEM;
 			goto error;
 		} else {
-			relayd->sessiond_session_id = (uint64_t) sessiond_id;
+			relayd->sessiond_session_id = sessiond_id;
 			relayd_created = 1;
 		}