X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=kconsumerd%2Fkconsumerd.c;h=5a1fe89a77cbd6772722cf45dbbab5f67732ef1e;hb=b1f11e69d7782029847845f0a5d018733c15a2f0;hp=592bc0842aa084683c512fe5fafcdadbfe405044;hpb=914a571be28d564da39d537a639b83a8d564376c;p=lttng-tools.git

diff --git a/kconsumerd/kconsumerd.c b/kconsumerd/kconsumerd.c
index 592bc0842..5a1fe89a7 100644
--- a/kconsumerd/kconsumerd.c
+++ b/kconsumerd/kconsumerd.c
@@ -70,6 +70,9 @@ static int error_socket = -1;
 /* to count the number of time the user pressed ctrl+c */
 static int sigintcount = 0;
 
+/* flag to inform the polling thread to quit when all fd hung up */
+static int quit = 0;
+
 /* Argument variables */
 int opt_quiet;
 int opt_verbose;
@@ -392,7 +395,7 @@ end:
  * Update a fd according to what we just received
  */
 static void change_fd_state(int sessiond_fd,
-		enum lttcomm_kconsumerd_fd_state state)
+		enum kconsumerd_fd_state state)
 {
 	struct ltt_kconsumerd_fd *iter;
 	cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
@@ -411,7 +414,7 @@ static void change_fd_state(int sessiond_fd,
  * Returns the size of received data
  */
 static int consumerd_recv_fd(int sfd, int size,
-		enum lttcomm_consumerd_command cmd_type)
+		enum kconsumerd_command cmd_type)
 {
 	struct msghdr msg;
 	struct iovec iov[1];
@@ -459,7 +462,7 @@ static int consumerd_recv_fd(int sfd, int size,
 		DBG("Receive : expecting %d fds", nb_fd);
 		for (i = 0; i < nb_fd; i++) {
 			switch (cmd_type) {
-			case LTTCOMM_ADD_STREAM:
+			case ADD_STREAM:
 				DBG("add_fd %s (%d)", buf[i].path_name, ((int *)CMSG_DATA(cmsg))[i]);
 				ret = add_fd(&buf[i], ((int *)CMSG_DATA(cmsg))[i]);
 				if (ret < 0) {
@@ -467,7 +470,7 @@ static int consumerd_recv_fd(int sfd, int size,
 					goto end;
 				}
 				break;
-			case LTTCOMM_UPDATE_STREAM:
+			case UPDATE_STREAM:
 				change_fd_state(buf[i].fd, buf[i].state);
 				break;
 			default:
@@ -486,6 +489,7 @@ static int consumerd_recv_fd(int sfd, int size,
 	}
 
 end:
+	DBG("consumerd_recv_fd thread exiting");
 	if (buf != NULL) {
 		free(buf);
 		buf = NULL;
@@ -509,43 +513,50 @@ static void *thread_receive_fds(void *data)
 	client_socket = lttcomm_create_unix_sock(command_sock_path);
 	if (client_socket < 0) {
 		ERR("Cannot create command socket");
-		goto error;
+		goto end;
 	}
 
 	ret = lttcomm_listen_unix_sock(client_socket);
 	if (ret < 0) {
-		goto error;
+		goto end;
 	}
 
 	DBG("Sending ready command to ltt-sessiond");
 	ret = send_error(KCONSUMERD_COMMAND_SOCK_READY);
 	if (ret < 0) {
 		ERR("Error sending ready command to ltt-sessiond");
-		goto error;
+		goto end;
 	}
 
 	/* Blocking call, waiting for transmission */
 	sock = lttcomm_accept_unix_sock(client_socket);
 	if (sock <= 0) {
 		WARN("On accept");
-		goto error;
+		goto end;
 	}
 	while (1) {
 		/* We first get the number of fd we are about to receive */
 		ret = lttcomm_recv_unix_sock(sock, &tmp,
 				sizeof(struct lttcomm_kconsumerd_header));
 		if (ret <= 0) {
-			ERR("Receiving the lttcomm_kconsumerd_header, exiting");
-			goto error;
+			ERR("Communication interrupted on command socket");
+			goto end;
 		}
+		if (tmp.cmd_type == STOP) {
+			DBG("Received STOP command");
+			quit = 1;
+			goto end;
+		}
+		/* we received a command to add or update fds */
 		ret = consumerd_recv_fd(sock, tmp.payload_size, tmp.cmd_type);
 		if (ret <= 0) {
 			ERR("Receiving the FD, exiting");
-			goto error;
+			goto end;
 		}
 	}
 
-error:
+end:
+	DBG("thread_receive_fds exiting");
 	return NULL;
 }
 
@@ -575,8 +586,6 @@ static int update_poll_array(struct pollfd **pollfd,
 			(*pollfd)[i].events = POLLIN | POLLPRI;
 			local_kconsumerd_fd[i] = iter;
 			i++;
-		} else if (iter->state == DELETE_FD) {
-			del_fd(iter);
 		}
 	}
 	/*
@@ -681,16 +690,20 @@ static void *thread_poll_fds(void *data)
 			switch(pollfd[i].revents) {
 			case POLLERR:
 				ERR("Error returned in polling fd %d.", pollfd[i].fd);
+				del_fd(local_kconsumerd_fd[i]);
+				update_fd_array = 1;
 				num_hup++;
-				send_error(KCONSUMERD_POLL_ERROR);
 				break;
 			case POLLHUP:
 				ERR("Polling fd %d tells it has hung up.", pollfd[i].fd);
+				del_fd(local_kconsumerd_fd[i]);
+				update_fd_array = 1;
 				num_hup++;
 				break;
 			case POLLNVAL:
 				ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
-				send_error(KCONSUMERD_POLL_NVAL);
+				del_fd(local_kconsumerd_fd[i]);
+				update_fd_array = 1;
 				num_hup++;
 				break;
 			case POLLPRI:
@@ -708,8 +721,10 @@ static void *thread_poll_fds(void *data)
 		/* If every buffer FD has hung up, we end the read loop here */
 		if (nb_fd > 0 && num_hup == nb_fd) {
 			DBG("every buffer FD has hung up\n");
-			send_error(KCONSUMERD_POLL_HUP);
-			goto end;
+			if (quit == 1) {
+				goto end;
+			}
+			continue;
 		}
 
 		/* Take care of low priority channels. */
@@ -727,6 +742,7 @@ static void *thread_poll_fds(void *data)
 		}
 	}
 end:
+	DBG("polling thread exiting");
 	if (pollfd != NULL) {
 		free(pollfd);
 		pollfd = NULL;