X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=63a2ab288c865e338582485199a7a268064bb4af;hb=9276e5c88e693249bd31197baecf58310df8167e;hp=46b961e8980d8f30e68062a73dd2979c6ca6bf92;hpb=ec135c61effc83fff0695e0e36a4182a8be83c5c;p=lttng-tools.git

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index 46b961e89..63a2ab288 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -323,6 +323,7 @@ static void free_relayd_rcu(struct rcu_head *head)
 	(void) relayd_close(&relayd->control_sock);
 	(void) relayd_close(&relayd->data_sock);
 
+	pthread_mutex_destroy(&relayd->ctrl_sock_mutex);
 	free(relayd);
 }
 
@@ -464,14 +465,13 @@ static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
  * If a local data context is available, notify the threads that the streams'
  * state have changed.
  */
-static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
-		struct lttng_consumer_local_data *ctx)
+void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd)
 {
 	uint64_t netidx;
 
 	assert(relayd);
 
-	DBG("Cleaning up relayd sockets");
+	DBG("Cleaning up relayd object ID %"PRIu64, relayd->net_seq_idx);
 
 	/* Save the net sequence index before destroying the object */
 	netidx = relayd->net_seq_idx;
@@ -491,10 +491,8 @@ static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
 	 * memory barrier ordering the updates of the end point status from the
 	 * read of this status which happens AFTER receiving this notify.
 	 */
-	if (ctx) {
-		notify_thread_lttng_pipe(ctx->consumer_data_pipe);
-		notify_thread_lttng_pipe(ctx->consumer_metadata_pipe);
-	}
+	notify_thread_lttng_pipe(relayd->ctx->consumer_data_pipe);
+	notify_thread_lttng_pipe(relayd->ctx->consumer_metadata_pipe);
 }
 
 /*
@@ -812,6 +810,8 @@ int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
 				stream->trace_archive_id);
 		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 		if (ret < 0) {
+			ERR("Relayd add stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+			lttng_consumer_cleanup_relayd(relayd);
 			goto end;
 		}
 
@@ -853,6 +853,8 @@ int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
 		ret = relayd_streams_sent(&relayd->control_sock);
 		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 		if (ret < 0) {
+			ERR("Relayd streams sent failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+			lttng_consumer_cleanup_relayd(relayd);
 			goto end;
 		}
 	} else {
@@ -1720,7 +1722,8 @@ write_error:
 	 * cleanup the relayd object and all associated streams.
 	 */
 	if (relayd && relayd_hang_up) {
-		cleanup_relayd(relayd, ctx);
+		ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+		lttng_consumer_cleanup_relayd(relayd);
 	}
 
 end:
@@ -1946,7 +1949,8 @@ write_error:
 	 * cleanup the relayd object and all associated streams.
 	 */
 	if (relayd && relayd_hang_up) {
-		cleanup_relayd(relayd, ctx);
+		ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+		lttng_consumer_cleanup_relayd(relayd);
 		/* Skip splice error so the consumer does not fail */
 		goto end;
 	}
@@ -2576,7 +2580,9 @@ void *consumer_thread_data_poll(void *data)
 	/* local view of the streams */
 	struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
 	/* local view of consumer_data.fds_count */
-	int nb_fd = 0, nb_pipes_fd;
+	int nb_fd = 0;
+	/* 2 for the consumer_data_pipe and wake up pipe */
+	const int nb_pipes_fd = 2;
 	/* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
 	int nb_inactive_fd = 0;
 	struct lttng_consumer_local_data *ctx = data;
@@ -2616,12 +2622,7 @@ void *consumer_thread_data_poll(void *data)
 			free(local_stream);
 			local_stream = NULL;
 
-			/*
-			 * Allocate for all fds + 2:
-			 *   +1 for the consumer_data_pipe
-			 *   +1 for wake up pipe
-			 */
-			nb_pipes_fd = 2;
+			/* Allocate for all fds */
 			pollfd = zmalloc((consumer_data.stream_count + nb_pipes_fd) * sizeof(struct pollfd));
 			if (pollfd == NULL) {
 				PERROR("pollfd malloc");
@@ -3648,6 +3649,7 @@ error:
 	 * Add relayd socket pair to consumer data hashtable. If object already
 	 * exists or on error, the function gracefully returns.
 	 */
+	relayd->ctx = ctx;
 	add_relayd(relayd);
 
 	/* All good! */
@@ -3774,6 +3776,8 @@ int consumer_data_pending(uint64_t id)
 		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 		if (ret < 0) {
 			/* Communication error thus the relayd so no data pending. */
+			ERR("Relayd begin data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+			lttng_consumer_cleanup_relayd(relayd);
 			goto data_not_pending;
 		}
 	}
@@ -3815,6 +3819,13 @@ int consumer_data_pending(uint64_t id)
 						stream->relayd_stream_id,
 						stream->next_net_seq_num - 1);
 			}
+			if (ret < 0) {
+				ERR("Relayd data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+				lttng_consumer_cleanup_relayd(relayd);
+				pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+				pthread_mutex_unlock(&stream->lock);
+				goto data_not_pending;
+			}
 			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 			if (ret == 1) {
 				pthread_mutex_unlock(&stream->lock);
@@ -3833,6 +3844,8 @@ int consumer_data_pending(uint64_t id)
 				relayd->relayd_session_id, &is_data_inflight);
 		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 		if (ret < 0) {
+			ERR("Relayd end data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
+			lttng_consumer_cleanup_relayd(relayd);
 			goto data_not_pending;
 		}
 		if (is_data_inflight) {