X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=liblttkconsumerd%2Fliblttkconsumerd.c;h=25e00132538781d15f1b350f668883d01e67e24f;hp=c1bad734dd4bbe1d616d777fb6c9e773e3885783;hb=b58c70aeb7e3742eb09863475722468fab12044b;hpb=242cd1874f3d8f678e795eea4d35552eff683ed8

diff --git a/liblttkconsumerd/liblttkconsumerd.c b/liblttkconsumerd/liblttkconsumerd.c
index c1bad734d..25e001325 100644
--- a/liblttkconsumerd/liblttkconsumerd.c
+++ b/liblttkconsumerd/liblttkconsumerd.c
@@ -61,6 +61,7 @@ struct kconsumerd_global_data {
 	unsigned int need_update;
 } kconsumerd_data = {
 	.fd_list.head = CDS_LIST_HEAD_INIT(kconsumerd_data.fd_list.head),
+	.need_update = 1,
 };
 
 /* communication with splice */
@@ -69,6 +70,13 @@ static int kconsumerd_thread_pipe[2];
 /* pipe to wake the poll thread when necessary */
 static int kconsumerd_poll_pipe[2];
 
+/*
+ * TODO: create a should_quit pipe to let the signal handler wake up the
+ * fd receiver thread. It should be initialized before any signal can be
+ * received by the library.
+ */
+
+
 /* timeout parameter, to control the polling thread grace period */
 static int kconsumerd_poll_timeout = -1;
 
@@ -78,8 +86,13 @@ static int kconsumerd_error_socket;
 /* socket to exchange commands with sessiond */
 static char *kconsumerd_command_sock_path;
 
-/* flag to inform the polling thread to kconsumerd_quit when all fd hung up */
-static int kconsumerd_quit = 0;
+/*
+ * flag to inform the polling thread to quit when all fd hung up.
+ * Updated by the kconsumerd_thread_receive_fds when it notices that all
+ * fds has hung up. Also updated by the signal handler
+ * (kconsumerd_should_exit()). Read by the polling threads.
+ */
+static volatile int kconsumerd_quit = 0;
 
 /*
  * kconsumerd_set_error_socket
@@ -105,6 +118,7 @@ void kconsumerd_set_command_socket_path(char *sock)
  * kconsumerd_find_session_fd
  *
  * Find a session fd in the global list.
+ * The kconsumerd_data.lock must be locked during this call
  *
  * Return 1 if found else 0
  */
@@ -112,7 +126,6 @@ static int kconsumerd_find_session_fd(int fd)
 {
 	struct kconsumerd_fd *iter;
 
-	pthread_mutex_lock(&kconsumerd_data.lock);
 	cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
 		if (iter->sessiond_fd == fd) {
 			DBG("Duplicate session fd %d", fd);
@@ -120,7 +133,6 @@ static int kconsumerd_find_session_fd(int fd)
 			return 1;
 		}
 	}
-	pthread_mutex_unlock(&kconsumerd_data.lock);
 
 	return 0;
 }
@@ -790,7 +802,6 @@ end:
 		free(local_kconsumerd_fd);
 		local_kconsumerd_fd = NULL;
 	}
-	kconsumerd_cleanup();
 	return NULL;
 }
 
@@ -835,6 +846,8 @@ void *kconsumerd_thread_receive_fds(void *data)
 		goto end;
 	}
 
+	/* TODO: poll on socket and "should_quit" fd pipe */
+	/* TODO: change blocking call into non-blocking call */
 	/* Blocking call, waiting for transmission */
 	sock = lttcomm_accept_unix_sock(client_socket);
 	if (sock <= 0) {
@@ -843,6 +856,8 @@ void *kconsumerd_thread_receive_fds(void *data)
 	}
 	while (1) {
 		/* We first get the number of fd we are about to receive */
+		/* TODO: poll on sock and "should_quit" fd pipe */
+		/* TODO: change recv into a non-blocking call */
 		ret = lttcomm_recv_unix_sock(sock, &tmp,
 				sizeof(struct lttcomm_kconsumerd_header));
 		if (ret <= 0) {
@@ -853,6 +868,10 @@ void *kconsumerd_thread_receive_fds(void *data)
 			DBG("Received STOP command");
 			goto end;
 		}
+		if (kconsumerd_quit) {
+			DBG("kconsumerd_thread_receive_fds received quit from signal");
+			goto end;
+		}
 		/* we received a command to add or update fds */
 		ret = kconsumerd_consumerd_recv_fd(sock, tmp.payload_size, tmp.cmd_type);
 		if (ret <= 0) {
@@ -890,19 +909,35 @@ end:
  *
  *  Cleanup the daemon's socket on exit
  */
-void kconsumerd_cleanup()
+void kconsumerd_cleanup(void)
 {
 	struct kconsumerd_fd *iter;
 
 	/* remove the socket file */
 	unlink(kconsumerd_command_sock_path);
 
-	/* close all outfd */
+	/*
+	 * close all outfd. Called when there are no more threads
+	 * running (after joining on the threads), no need to protect
+	 * list iteration with mutex.
+	 */
 	cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
 		kconsumerd_del_fd(iter);
 	}
 }
 
+/*
+ * Called from signal handler.
+ */
+void kconsumerd_should_exit(void)
+{
+	kconsumerd_quit = 1;
+	/*
+	 * TODO: write into a should_quit pipe to wake up the fd
+	 * receiver thread.
+	 */
+}
+
 /*
  * kconsumerd_send_error
  *