X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=kconsumerd%2Fkconsumerd.c;h=fd0b0b80cd6b19747fe6dcf7378504423e4dd3af;hp=fcd4101a62d0bdbdff1b7ffcd34a86365324afbf;hb=33a2b85433875769e92ca44a680c46b9498f5174;hpb=252fd492b48721806ed8d72b508ea3733bc70845

diff --git a/kconsumerd/kconsumerd.c b/kconsumerd/kconsumerd.c
index fcd4101a6..fd0b0b80c 100644
--- a/kconsumerd/kconsumerd.c
+++ b/kconsumerd/kconsumerd.c
@@ -67,6 +67,12 @@ static int poll_pipe[2];
 /* socket to communicate errors with sessiond */
 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;
@@ -156,7 +162,7 @@ static int add_fd(struct lttcomm_kconsumerd_msg *buf, int consumerd_fd)
 	/* Opening the tracefile in write mode */
 	DBG("Opening %s for writing", tmp_fd->path_name);
 	ret = open(tmp_fd->path_name,
-			O_WRONLY|O_CREAT, S_IRWXU|S_IRWXG|S_IRWXO);
+			O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
 	if (ret < 0) {
 		ERR("Opening %s", tmp_fd->path_name);
 		perror("open");
@@ -185,6 +191,11 @@ end:
  */
 static void sighandler(int sig)
 {
+	if (sig == SIGINT && sigintcount++ == 0) {
+		DBG("ignoring first SIGINT");
+		return;
+	}
+
 	cleanup();
 
 	return;
@@ -304,18 +315,18 @@ static int on_read_subbuffer(struct ltt_kconsumerd_fd *kconsumerd_fd,
 splice_error:
 	/* send the appropriate error description to sessiond */
 	switch(ret) {
-		case EBADF:
-			send_error(KCONSUMERD_SPLICE_EBADF);
-			break;
-		case EINVAL:
-			send_error(KCONSUMERD_SPLICE_EINVAL);
-			break;
-		case ENOMEM:
-			send_error(KCONSUMERD_SPLICE_ENOMEM);
-			break;
-		case ESPIPE:
-			send_error(KCONSUMERD_SPLICE_ESPIPE);
-			break;
+	case EBADF:
+		send_error(KCONSUMERD_SPLICE_EBADF);
+		break;
+	case EINVAL:
+		send_error(KCONSUMERD_SPLICE_EINVAL);
+		break;
+	case ENOMEM:
+		send_error(KCONSUMERD_SPLICE_ENOMEM);
+		break;
+	case ESPIPE:
+		send_error(KCONSUMERD_SPLICE_ESPIPE);
+		break;
 	}
 
 end:
@@ -384,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) {
@@ -403,85 +414,82 @@ 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];
-	int ret, i, tmp2;
+	int ret = 0, i, tmp2;
 	struct cmsghdr *cmsg;
 	int nb_fd;
-	char tmp[CMSG_SPACE(size)];
-	struct lttcomm_kconsumerd_msg *buf;
+	char recv_fd[CMSG_SPACE(sizeof(int))];
+	struct lttcomm_kconsumerd_msg lkm;
+
 	/* the number of fds we are about to receive */
-	nb_fd = size/sizeof(struct lttcomm_kconsumerd_msg);
+	nb_fd = size / sizeof(struct lttcomm_kconsumerd_msg);
 
-	buf = malloc(size);
+	for (i = 0; i < nb_fd; i++) {
+		memset(&msg, 0, sizeof(msg));
 
-	memset(&msg, 0, sizeof(msg));
+		/* Prepare to receive the structures */
+		iov[0].iov_base = &lkm;
+		iov[0].iov_len = sizeof(lkm);
+		msg.msg_iov = iov;
+		msg.msg_iovlen = 1;
 
-	/* Prepare to receive the structures */
-	iov[0].iov_base = buf;
-	iov[0].iov_len = size;
-	msg.msg_iov = iov;
-	msg.msg_iovlen = 1;
+		msg.msg_control = recv_fd;
+		msg.msg_controllen = sizeof(recv_fd);
 
-	msg.msg_control = tmp;
-	msg.msg_controllen = sizeof(tmp);
+		DBG("Waiting to receive fd");
+		if ((ret = recvmsg(sfd, &msg, 0)) < 0) {
+			perror("recvmsg");
+			continue;
+		}
 
-	DBG("Waiting to receive fds");
-	if ((ret = recvmsg(sfd, &msg, 0)) < 0) {
-		perror("recvmsg");
-	}
-	if (ret != size) {
-		ERR("Received only %d, expected %d", ret, size);
-		send_error(KCONSUMERD_ERROR_RECV_FD);
-		goto end;
-	}
+		if (ret != (size / nb_fd)) {
+			ERR("Received only %d, expected %d", ret, size);
+			send_error(KCONSUMERD_ERROR_RECV_FD);
+			goto end;
+		}
 
-	cmsg = CMSG_FIRSTHDR(&msg);
-	if (!cmsg) {
-		ERR("Invalid control message header");
-		ret = -1;
-		send_error(KCONSUMERD_ERROR_RECV_FD);
-		goto end;
-	}
+		cmsg = CMSG_FIRSTHDR(&msg);
+		if (!cmsg) {
+			ERR("Invalid control message header");
+			ret = -1;
+			send_error(KCONSUMERD_ERROR_RECV_FD);
+			goto end;
+		}
 
-	/* if we received fds */
-	if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
-		DBG("Receive : expecting %d fds", nb_fd);
-		for (i = 0; i < nb_fd; i++) {
+		/* if we received fds */
+		if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
 			switch (cmd_type) {
-				case LTTCOMM_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) {
-						send_error(KCONSUMERD_OUTFD_ERROR);
-						goto end;
-					}
-					break;
-				case LTTCOMM_UPDATE_STREAM:
-					change_fd_state(buf[i].fd, buf[i].state);
-					break;
-				default:
-					break;
+			case ADD_STREAM:
+				DBG("add_fd %s (%d)", lkm.path_name, (CMSG_DATA(cmsg)[0]));
+				ret = add_fd(&lkm, (CMSG_DATA(cmsg)[0]));
+				if (ret < 0) {
+					send_error(KCONSUMERD_OUTFD_ERROR);
+					goto end;
+				}
+				break;
+			case UPDATE_STREAM:
+				change_fd_state(lkm.fd, lkm.state);
+				break;
+			default:
+				break;
 			}
+			/* flag to tell the polling thread to update its fd array */
+			update_fd_array = 1;
+			/* signal the poll thread */
+			tmp2 = write(poll_pipe[1], "4", 1);
+		} else {
+			ERR("Didn't received any fd");
+			send_error(KCONSUMERD_ERROR_RECV_FD);
+			ret = -1;
+			goto end;
 		}
-		/* flag to tell the polling thread to update its fd array */
-		update_fd_array = 1;
-		/* signal the poll thread */
-		tmp2 = write(poll_pipe[1], "4", 1);
-	} else {
-		ERR("Didn't received any fd");
-		send_error(KCONSUMERD_ERROR_RECV_FD);
-		ret = -1;
-		goto end;
 	}
 
 end:
-	if (buf != NULL) {
-		free(buf);
-		buf = NULL;
-	}
+	DBG("consumerd_recv_fd thread exiting");
 	return ret;
 }
 
@@ -501,43 +509,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, retrying");
-		goto error;
+		WARN("On accept");
+		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;
 }
 
@@ -555,29 +570,6 @@ static int update_poll_array(struct pollfd **pollfd,
 	struct ltt_kconsumerd_fd *iter;
 	int i = 0;
 
-	if (*pollfd != NULL) {
-		free(*pollfd);
-		*pollfd = NULL;
-	}
-
-	if (*local_kconsumerd_fd != NULL) {
-		free(*local_kconsumerd_fd);
-		*local_kconsumerd_fd = NULL;
-	}
-
-	/* allocate for all fds + 1 for the poll_pipe */
-	*pollfd = malloc((fds_count + 1) * sizeof(struct pollfd));
-	if (*pollfd == NULL) {
-		perror("pollfd malloc");
-		goto error_mem;
-	}
-
-	/* allocate for all fds + 1 for the poll_pipe */
-	*local_kconsumerd_fd = malloc((fds_count + 1) * sizeof(struct ltt_kconsumerd_fd));
-	if (*local_kconsumerd_fd == NULL) {
-		perror("local_kconsumerd_fd malloc");
-		goto error_mem;
-	}
 
 	DBG("Updating poll fd array");
 	pthread_mutex_lock(&kconsumerd_lock_fds);
@@ -590,8 +582,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);
 		}
 	}
 	/*
@@ -605,8 +595,6 @@ static int update_poll_array(struct pollfd **pollfd,
 	pthread_mutex_unlock(&kconsumerd_lock_fds);
 	return i;
 
-error_mem:
-	return -ENOMEM;
 }
 
 /*
@@ -642,7 +630,28 @@ static void *thread_poll_fds(void *data)
 		 * the ltt_fd_list has been updated, we need to update our
 		 * local array as well
 		 */
-		if (update_fd_array) {
+		if (update_fd_array == 1) {
+			if (pollfd != NULL) {
+				free(pollfd);
+				pollfd = NULL;
+			}
+			if (local_kconsumerd_fd != NULL) {
+				free(local_kconsumerd_fd);
+				local_kconsumerd_fd = NULL;
+			}
+			/* allocate for all fds + 1 for the poll_pipe */
+			pollfd = malloc((fds_count + 1) * sizeof(struct pollfd));
+			if (pollfd == NULL) {
+				perror("pollfd malloc");
+				goto end;
+			}
+			/* allocate for all fds + 1 for the poll_pipe */
+			local_kconsumerd_fd = malloc((fds_count + 1) * sizeof(struct ltt_kconsumerd_fd));
+			if (local_kconsumerd_fd == NULL) {
+				perror("local_kconsumerd_fd malloc");
+				goto end;
+			}
+
 			ret = update_poll_array(&pollfd, local_kconsumerd_fd);
 			if (ret < 0) {
 				ERR("Error in allocating pollfd or local_outfds");
@@ -675,56 +684,61 @@ static void *thread_poll_fds(void *data)
 		/* Take care of high priority channels first. */
 		for (i = 0; i < nb_fd; i++) {
 			switch(pollfd[i].revents) {
-				case POLLERR:
-					ERR("Error returned in polling fd %d.", pollfd[i].fd);
-					num_hup++;
-					send_error(KCONSUMERD_POLL_ERROR);
-					break;
-				case POLLHUP:
-					ERR("Polling fd %d tells it has hung up.", pollfd[i].fd);
-					num_hup++;
-					break;
-				case POLLNVAL:
-					ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
-					send_error(KCONSUMERD_POLL_NVAL);
-					num_hup++;
-					break;
-				case POLLPRI:
-					DBG("Urgent read on fd %d", pollfd[i].fd);
-					high_prio = 1;
-					ret = read_subbuffer(local_kconsumerd_fd[i]);
-					/* it's ok to have an unavailable sub-buffer (FIXME : is it ?) */
-					if (ret == EAGAIN) {
-						ret = 0;
-					}
-					break;
+			case POLLERR:
+				ERR("Error returned in polling fd %d.", pollfd[i].fd);
+				del_fd(local_kconsumerd_fd[i]);
+				update_fd_array = 1;
+				num_hup++;
+				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);
+				del_fd(local_kconsumerd_fd[i]);
+				update_fd_array = 1;
+				num_hup++;
+				break;
+			case POLLPRI:
+				DBG("Urgent read on fd %d", pollfd[i].fd);
+				high_prio = 1;
+				ret = read_subbuffer(local_kconsumerd_fd[i]);
+				/* it's ok to have an unavailable sub-buffer (FIXME : is it ?) */
+				if (ret == EAGAIN) {
+					ret = 0;
+				}
+				break;
 			}
 		}
 
 		/* 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. */
-		if (!high_prio) {
+		if (high_prio == 0) {
 			for (i = 0; i < nb_fd; i++) {
-				switch(pollfd[i].revents) {
-					case POLLIN:
-						DBG("Normal read on fd %d", pollfd[i].fd);
-						ret = read_subbuffer(local_kconsumerd_fd[i]);
-						/* it's ok to have an unavailable subbuffer (FIXME : is it ?) */
-						if (ret == EAGAIN) {
-							ret = 0;
-						}
-						break;
+				if (pollfd[i].revents == POLLIN) {
+					DBG("Normal read on fd %d", pollfd[i].fd);
+					ret = read_subbuffer(local_kconsumerd_fd[i]);
+					/* it's ok to have an unavailable subbuffer (FIXME : is it ?) */
+					if (ret == EAGAIN) {
+						ret = 0;
+					}
 				}
 			}
 		}
 	}
 end:
+	DBG("polling thread exiting");
 	if (pollfd != NULL) {
 		free(pollfd);
 		pollfd = NULL;
@@ -785,36 +799,36 @@ static void parse_args(int argc, char **argv)
 		}
 
 		switch (c) {
-			case 0:
-				fprintf(stderr, "option %s", long_options[option_index].name);
-				if (optarg) {
-					fprintf(stderr, " with arg %s\n", optarg);
-				}
-				break;
-			case 'c':
-				snprintf(command_sock_path, PATH_MAX, "%s", optarg);
-				break;
-			case 'e':
-				snprintf(error_sock_path, PATH_MAX, "%s", optarg);
-				break;
-			case 'd':
-				opt_daemon = 1;
-				break;
-			case 'h':
-				usage();
-				exit(EXIT_FAILURE);
-			case 'q':
-				opt_quiet = 1;
-				break;
-			case 'v':
-				opt_verbose = 1;
-				break;
-			case 'V':
-				fprintf(stdout, "%s\n", VERSION);
-				exit(EXIT_SUCCESS);
-			default:
-				usage();
-				exit(EXIT_FAILURE);
+		case 0:
+			fprintf(stderr, "option %s", long_options[option_index].name);
+			if (optarg) {
+				fprintf(stderr, " with arg %s\n", optarg);
+			}
+			break;
+		case 'c':
+			snprintf(command_sock_path, PATH_MAX, "%s", optarg);
+			break;
+		case 'e':
+			snprintf(error_sock_path, PATH_MAX, "%s", optarg);
+			break;
+		case 'd':
+			opt_daemon = 1;
+			break;
+		case 'h':
+			usage();
+			exit(EXIT_FAILURE);
+		case 'q':
+			opt_quiet = 1;
+			break;
+		case 'v':
+			opt_verbose = 1;
+			break;
+		case 'V':
+			fprintf(stdout, "%s\n", VERSION);
+			exit(EXIT_SUCCESS);
+		default:
+			usage();
+			exit(EXIT_FAILURE);
 		}
 	}
 }