X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=kconsumerd%2Fkconsumerd.c;h=69aa47a6e5f0b175113b407eb91429c451b9af5b;hb=5dc1855009c9b215a108bc56e7df64d9862c8144;hp=fcd4101a62d0bdbdff1b7ffcd34a86365324afbf;hpb=252fd492b48721806ed8d72b508ea3733bc70845;p=lttng-tools.git

diff --git a/kconsumerd/kconsumerd.c b/kconsumerd/kconsumerd.c
index fcd4101a6..69aa47a6e 100644
--- a/kconsumerd/kconsumerd.c
+++ b/kconsumerd/kconsumerd.c
@@ -67,6 +67,9 @@ 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;
+
 /* Argument variables */
 int opt_quiet;
 int opt_verbose;
@@ -156,7 +159,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 +188,11 @@ end:
  */
 static void sighandler(int sig)
 {
+	if (sig == SIGINT && sigintcount++ == 0) {
+		DBG("ignoring first SIGINT");
+		return;
+	}
+
 	cleanup();
 
 	return;
@@ -304,18 +312,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 +392,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,7 +411,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];
@@ -451,19 +459,19 @@ 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:
-					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)", 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 UPDATE_STREAM:
+				change_fd_state(buf[i].fd, buf[i].state);
+				break;
+			default:
+				break;
 			}
 		}
 		/* flag to tell the polling thread to update its fd array */
@@ -519,7 +527,7 @@ static void *thread_receive_fds(void *data)
 	/* Blocking call, waiting for transmission */
 	sock = lttcomm_accept_unix_sock(client_socket);
 	if (sock <= 0) {
-		WARN("On accept, retrying");
+		WARN("On accept");
 		goto error;
 	}
 	while (1) {
@@ -555,29 +563,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);
@@ -605,8 +590,6 @@ static int update_poll_array(struct pollfd **pollfd,
 	pthread_mutex_unlock(&kconsumerd_lock_fds);
 	return i;
 
-error_mem:
-	return -ENOMEM;
 }
 
 /*
@@ -642,7 +625,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,29 +679,29 @@ 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);
+				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;
 			}
 		}
 
@@ -709,17 +713,15 @@ static void *thread_poll_fds(void *data)
 		}
 
 		/* 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;
+					}
 				}
 			}
 		}
@@ -785,36 +787,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);
 		}
 	}
 }