X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=008bf8e54904d71158400216393f0fc6c5786e63;hp=363fa4eda6c5ba0dc75296d4c9f7d764c80b21da;hb=32dd26fbc3c69fe677a7917535e10ace066e674c;hpb=7735ef9e674217413a63bd4a09a93ac0958fe58a

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 363fa4eda..008bf8e54 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -201,6 +201,25 @@ void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 	call_rcu(&relayd->node.head, consumer_rcu_free_relayd);
 }
 
+/*
+ * Flag a relayd socket pair for destruction. Destroy it if the refcount
+ * reaches zero.
+ *
+ * RCU read side lock MUST be aquired before calling this function.
+ */
+void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
+{
+	assert(relayd);
+
+	/* Set destroy flag for this object */
+	uatomic_set(&relayd->destroy_flag, 1);
+
+	/* Destroy the relayd if refcount is 0 */
+	if (uatomic_read(&relayd->refcount) == 0) {
+		consumer_destroy_relayd(relayd);
+	}
+}
+
 /*
  * Remove a stream from the global list protected by a mutex. This
  * function is also responsible for freeing its data structures.
@@ -1357,16 +1376,16 @@ splice_error:
 	/* send the appropriate error description to sessiond */
 	switch (ret) {
 	case EBADF:
-		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EBADF);
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EBADF);
 		break;
 	case EINVAL:
-		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EINVAL);
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
 		break;
 	case ENOMEM:
-		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ENOMEM);
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
 		break;
 	case ESPIPE:
-		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ESPIPE);
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
 		break;
 	}
 
@@ -1504,7 +1523,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 					metadata_ht);
 			if (ret < 0) {
 				ERR("Error in allocating pollfd or local_outfds");
-				lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
+				lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
 				pthread_mutex_unlock(&consumer_data.lock);
 				goto end;
 			}
@@ -1530,7 +1549,7 @@ void *lttng_consumer_thread_poll_fds(void *data)
 				goto restart;
 			}
 			perror("Poll error");
-			lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
+			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
 			goto end;
 		} else if (num_rdy == 0) {
 			DBG("Polling thread timed out");
@@ -1707,7 +1726,7 @@ void *lttng_consumer_thread_receive_fds(void *data)
 	}
 
 	DBG("Sending ready command to lttng-sessiond");
-	ret = lttng_consumer_send_error(ctx, CONSUMERD_COMMAND_SOCK_READY);
+	ret = lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY);
 	/* return < 0 on error, but == 0 is not fatal */
 	if (ret < 0) {
 		ERR("Error sending ready command to lttng-sessiond");
@@ -1863,7 +1882,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		/* Not found. Allocate one. */
 		relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
 		if (relayd == NULL) {
-			lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
+			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 			goto error;
 		}
 	}
@@ -1877,7 +1896,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 	/* Get relayd socket from session daemon */
 	ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 	if (ret != sizeof(fd)) {
-		lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
 		ret = -1;
 		goto error;
 	}