Add missing open() option to kconsumerd

[lttng-tools.git] / kconsumerd / kconsumerd.c

diff --git a/kconsumerd/kconsumerd.c b/kconsumerd/kconsumerd.c
index 66f3877d020cddae531fd123fc7df4ab31757e18..8e8f1a6188c1b322449992fb4fd904218cb993a9 100644 (file)
--- a/kconsumerd/kconsumerd.c
+++ b/kconsumerd/kconsumerd.c
@@ -61,6 +61,9 @@ static pthread_t threads[2];
 /* communication with splice */
 static int thread_pipe[2];
 
+/* pipe to wake the poll thread when necessary */
+static int poll_pipe[2];
+
 /* socket to communicate errors with sessiond */
 static int error_socket = -1;
 
@@ -153,7 +156,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");
@@ -404,7 +407,7 @@ static int consumerd_recv_fd(int sfd, int size,
 {
        struct msghdr msg;
        struct iovec iov[1];
-       int ret, i;
+       int ret, i, tmp2;
        struct cmsghdr *cmsg;
        int nb_fd;
        char tmp[CMSG_SPACE(size)];
@@ -465,7 +468,8 @@ static int consumerd_recv_fd(int sfd, int size,
                }
                /* flag to tell the polling thread to update its fd array */
                update_fd_array = 1;
-               send_error(KCONSUMERD_SUCCESS_RECV_FD);
+               /* signal the poll thread */
+               tmp2 = write(poll_pipe[1], "4", 1);
        } else {
                ERR("Didn't received any fd");
                send_error(KCONSUMERD_ERROR_RECV_FD);
@@ -515,7 +519,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) {
@@ -551,27 +555,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;
-       }
-
-       *pollfd = malloc(fds_count * sizeof(struct pollfd));
-       if (*pollfd == NULL) {
-               perror("pollfd malloc");
-               goto error_mem;
-       }
-
-       *local_kconsumerd_fd = malloc(fds_count * 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);
@@ -588,12 +571,17 @@ static int update_poll_array(struct pollfd **pollfd,
                        del_fd(iter);
                }
        }
+       /*
+        * insert the poll_pipe at the end of the array and don't increment i
+        * so nb_fd is the number of real FD
+        */
+       (*pollfd)[i].fd = poll_pipe[0];
+       (*pollfd)[i].events = POLLIN;
+
        update_fd_array = 0;
        pthread_mutex_unlock(&kconsumerd_lock_fds);
        return i;
 
-error_mem:
-       return -ENOMEM;
 }
 
 /*
@@ -610,6 +598,8 @@ static void *thread_poll_fds(void *data)
        struct ltt_kconsumerd_fd **local_kconsumerd_fd = NULL;
        /* local view of fds_count */
        int nb_fd = 0;
+       char tmp;
+       int tmp2;
 
        ret = pipe(thread_pipe);
        if (ret < 0) {
@@ -628,6 +618,27 @@ static void *thread_poll_fds(void *data)
                 * local array as well
                 */
                if (update_fd_array) {
+                       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");
@@ -638,8 +649,8 @@ static void *thread_poll_fds(void *data)
                }
 
                /* poll on the array of fds */
-               DBG("polling on %d fd", nb_fd);
-               num_rdy = poll(pollfd, nb_fd, POLL_TIMEOUT);
+               DBG("polling on %d fd", nb_fd + 1);
+               num_rdy = poll(pollfd, nb_fd + 1, -1);
                DBG("poll num_rdy : %d", num_rdy);
                if (num_rdy == -1) {
                        perror("Poll error");
@@ -647,6 +658,16 @@ static void *thread_poll_fds(void *data)
                        goto end;
                }
 
+               /*
+                * if only the poll_pipe triggered poll to return just return to the
+                * beginning of the loop to update the array
+                */
+               if (num_rdy == 1 && pollfd[nb_fd].revents == POLLIN) {
+                       DBG("poll_pipe wake up");
+                       tmp2 = read(poll_pipe[0], &tmp, 1);
+                       continue;
+               }
+
                /* Take care of high priority channels first. */
                for (i = 0; i < nb_fd; i++) {
                        switch(pollfd[i].revents) {
@@ -830,6 +851,13 @@ int main(int argc, char **argv)
                goto error;
        }
 
+       /* create the pipe to wake to polling thread when needed */
+       ret = pipe(poll_pipe);
+       if (ret < 0) {
+               perror("Error creating poll pipe");
+               goto end;
+       }
+
        /* Connect to the socket created by ltt-sessiond to report errors */
        DBG("Connecting to error socket %s", error_sock_path);
        error_socket = lttcomm_connect_unix_sock(error_sock_path);