kconsumerd: signals should only set flag and write into pipe (part 1)
[lttng-tools.git] / liblttkconsumerd / liblttkconsumerd.c
index e4f5fe2e48323fa9ba2a8e7429501f65be03bc3d..de942a1983182df755938bfe1b840c3262a22215 100644 (file)
 #include "liblttkconsumerd.h"
 #include "lttngerr.h"
 
-/* Init the list of FDs */
-static struct kconsumerd_fd_list kconsumerd_fd_list = {
-       .head = CDS_LIST_HEAD_INIT(kconsumerd_fd_list.head),
+static
+struct kconsumerd_global_data {
+       /*
+        * kconsumerd_data.lock protects kconsumerd_data.fd_list,
+        * kconsumerd_data.fds_count, and kconsumerd_data.need_update.  It
+        * ensures the count matches the number of items in the fd_list.
+        * It ensures the list updates *always* trigger an fd_array
+        * update (therefore need to make list update vs
+        * kconsumerd_data.need_update flag update atomic, and also flag
+        * read, fd array and flag clear atomic).
+        */
+       pthread_mutex_t lock;
+       /*
+        * Number of element for the list below.  Protected by
+        * kconsumerd_data.lock.
+        */
+       unsigned int fds_count;
+       /*
+        * List of FDs.  Protected by kconsumerd_data.lock.
+        */
+       struct kconsumerd_fd_list fd_list;
+       /*
+        * Flag specifying if the local array of FDs needs update in the
+        * poll function.  Protected by kconsumerd_data.lock.
+        */
+       unsigned int need_update;
+} kconsumerd_data = {
+       .fd_list.head = CDS_LIST_HEAD_INIT(kconsumerd_data.fd_list.head),
 };
 
-/* Number of element for the list below. */
-static unsigned int kconsumerd_fds_count;
-
-/* If the local array of FDs needs update in the poll function */
-static unsigned int kconsumerd_update_fd_array = 1;
-
-/* lock the fd array and structures */
-static pthread_mutex_t kconsumerd_lock_fds;
-
 /* communication with splice */
 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;
 
@@ -62,8 +84,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
@@ -96,15 +123,15 @@ static int kconsumerd_find_session_fd(int fd)
 {
        struct kconsumerd_fd *iter;
 
-       pthread_mutex_lock(&kconsumerd_lock_fds);
-       cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
+       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);
-                       pthread_mutex_unlock(&kconsumerd_lock_fds);
+                       pthread_mutex_unlock(&kconsumerd_data.lock);
                        return 1;
                }
        }
-       pthread_mutex_unlock(&kconsumerd_lock_fds);
+       pthread_mutex_unlock(&kconsumerd_data.lock);
 
        return 0;
 }
@@ -116,10 +143,10 @@ static int kconsumerd_find_session_fd(int fd)
  */
 static void kconsumerd_del_fd(struct kconsumerd_fd *lcf)
 {
-       pthread_mutex_lock(&kconsumerd_lock_fds);
+       pthread_mutex_lock(&kconsumerd_data.lock);
        cds_list_del(&lcf->list);
-       if (kconsumerd_fds_count > 0) {
-               kconsumerd_fds_count--;
+       if (kconsumerd_data.fds_count > 0) {
+               kconsumerd_data.fds_count--;
                if (lcf != NULL) {
                        close(lcf->out_fd);
                        close(lcf->consumerd_fd);
@@ -127,7 +154,8 @@ static void kconsumerd_del_fd(struct kconsumerd_fd *lcf)
                        lcf = NULL;
                }
        }
-       pthread_mutex_unlock(&kconsumerd_lock_fds);
+       kconsumerd_data.need_update = 1;
+       pthread_mutex_unlock(&kconsumerd_data.lock);
 }
 
 /*
@@ -140,6 +168,7 @@ static int kconsumerd_add_fd(struct lttcomm_kconsumerd_msg *buf, int consumerd_f
        int ret;
        struct kconsumerd_fd *tmp_fd;
 
+       pthread_mutex_lock(&kconsumerd_data.lock);
        /* Check if already exist */
        ret = kconsumerd_find_session_fd(buf->fd);
        if (ret == 1) {
@@ -167,12 +196,11 @@ static int kconsumerd_add_fd(struct lttcomm_kconsumerd_msg *buf, int consumerd_f
        DBG("Adding %s (%d, %d, %d)", tmp_fd->path_name,
                        tmp_fd->sessiond_fd, tmp_fd->consumerd_fd, tmp_fd->out_fd);
 
-       pthread_mutex_lock(&kconsumerd_lock_fds);
-       cds_list_add(&tmp_fd->list, &kconsumerd_fd_list.head);
-       kconsumerd_fds_count++;
-       pthread_mutex_unlock(&kconsumerd_lock_fds);
-
+       cds_list_add(&tmp_fd->list, &kconsumerd_data.fd_list.head);
+       kconsumerd_data.fds_count++;
+       kconsumerd_data.need_update = 1;
 end:
+       pthread_mutex_unlock(&kconsumerd_data.lock);
        return ret;
 }
 
@@ -185,12 +213,16 @@ static void kconsumerd_change_fd_state(int sessiond_fd,
                enum kconsumerd_fd_state state)
 {
        struct kconsumerd_fd *iter;
-       cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
+
+       pthread_mutex_lock(&kconsumerd_data.lock);
+       cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
                if (iter->sessiond_fd == sessiond_fd) {
                        iter->state = state;
                        break;
                }
        }
+       kconsumerd_data.need_update = 1;
+       pthread_mutex_unlock(&kconsumerd_data.lock);
 }
 
 /*
@@ -200,6 +232,7 @@ static void kconsumerd_change_fd_state(int sessiond_fd,
  * to avoid doing a lookup in the linked list and concurrency issues
  * when writing is needed.
  * Returns the number of fds in the structures
+ * Called with kconsumerd_data.lock held.
  */
 static int kconsumerd_update_poll_array(struct pollfd **pollfd,
                struct kconsumerd_fd **local_kconsumerd_fd)
@@ -208,9 +241,8 @@ static int kconsumerd_update_poll_array(struct pollfd **pollfd,
        int i = 0;
 
        DBG("Updating poll fd array");
-       pthread_mutex_lock(&kconsumerd_lock_fds);
 
-       cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
+       cds_list_for_each_entry(iter, &kconsumerd_data.fd_list.head, list) {
                DBG("Inside for each");
                if (iter->state == ACTIVE_FD) {
                        DBG("Active FD %d", iter->consumerd_fd);
@@ -227,9 +259,6 @@ static int kconsumerd_update_poll_array(struct pollfd **pollfd,
         */
        (*pollfd)[i].fd = kconsumerd_poll_pipe[0];
        (*pollfd)[i].events = POLLIN;
-
-       kconsumerd_update_fd_array = 0;
-       pthread_mutex_unlock(&kconsumerd_lock_fds);
        return i;
 }
 
@@ -593,8 +622,6 @@ static int kconsumerd_consumerd_recv_fd(int sfd, int size,
                        default:
                                break;
                        }
-                       /* flag to tell the polling thread to update its fd array */
-                       kconsumerd_update_fd_array = 1;
                        /* signal the poll thread */
                        tmp2 = write(kconsumerd_poll_pipe[1], "4", 1);
                } else {
@@ -621,7 +648,7 @@ void *kconsumerd_thread_poll_fds(void *data)
        struct pollfd *pollfd = NULL;
        /* local view of the fds */
        struct kconsumerd_fd **local_kconsumerd_fd = NULL;
-       /* local view of kconsumerd_fds_count */
+       /* local view of kconsumerd_data.fds_count */
        int nb_fd = 0;
        char tmp;
        int tmp2;
@@ -642,7 +669,8 @@ void *kconsumerd_thread_poll_fds(void *data)
                 * the ltt_fd_list has been updated, we need to update our
                 * local array as well
                 */
-               if (kconsumerd_update_fd_array == 1) {
+               pthread_mutex_lock(&kconsumerd_data.lock);
+               if (kconsumerd_data.need_update) {
                        if (pollfd != NULL) {
                                free(pollfd);
                                pollfd = NULL;
@@ -651,27 +679,34 @@ void *kconsumerd_thread_poll_fds(void *data)
                                free(local_kconsumerd_fd);
                                local_kconsumerd_fd = NULL;
                        }
+
                        /* allocate for all fds + 1 for the kconsumerd_poll_pipe */
-                       pollfd = malloc((kconsumerd_fds_count + 1) * sizeof(struct pollfd));
+                       pollfd = malloc((kconsumerd_data.fds_count + 1) * sizeof(struct pollfd));
                        if (pollfd == NULL) {
                                perror("pollfd malloc");
+                               pthread_mutex_unlock(&kconsumerd_data.lock);
                                goto end;
                        }
+
                        /* allocate for all fds + 1 for the kconsumerd_poll_pipe */
-                       local_kconsumerd_fd = malloc((kconsumerd_fds_count + 1) *
+                       local_kconsumerd_fd = malloc((kconsumerd_data.fds_count + 1) *
                                        sizeof(struct kconsumerd_fd));
                        if (local_kconsumerd_fd == NULL) {
                                perror("local_kconsumerd_fd malloc");
+                               pthread_mutex_unlock(&kconsumerd_data.lock);
                                goto end;
                        }
                        ret = kconsumerd_update_poll_array(&pollfd, local_kconsumerd_fd);
                        if (ret < 0) {
                                ERR("Error in allocating pollfd or local_outfds");
                                kconsumerd_send_error(KCONSUMERD_POLL_ERROR);
+                               pthread_mutex_unlock(&kconsumerd_data.lock);
                                goto end;
                        }
                        nb_fd = ret;
+                       kconsumerd_data.need_update = 0;
                }
+               pthread_mutex_unlock(&kconsumerd_data.lock);
 
                /* poll on the array of fds */
                DBG("polling on %d fd", nb_fd + 1);
@@ -692,10 +727,12 @@ void *kconsumerd_thread_poll_fds(void *data)
                }
 
                /*
-                * if only the kconsumerd_poll_pipe triggered poll to return just
-                * return to the beginning of the loop to update the array
+                * If the kconsumerd_poll_pipe triggered poll go
+                * directly to the beginning of the loop to update the
+                * array. We want to prioritize array update over
+                * low-priority reads.
                 */
-               if (num_rdy == 1 && pollfd[nb_fd].revents == POLLIN) {
+               if (pollfd[nb_fd].revents == POLLIN) {
                        DBG("kconsumerd_poll_pipe wake up");
                        tmp2 = read(kconsumerd_poll_pipe[0], &tmp, 1);
                        continue;
@@ -707,19 +744,16 @@ void *kconsumerd_thread_poll_fds(void *data)
                        case POLLERR:
                                ERR("Error returned in polling fd %d.", pollfd[i].fd);
                                kconsumerd_del_fd(local_kconsumerd_fd[i]);
-                               kconsumerd_update_fd_array = 1;
                                num_hup++;
                                break;
                        case POLLHUP:
                                DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
                                kconsumerd_del_fd(local_kconsumerd_fd[i]);
-                               kconsumerd_update_fd_array = 1;
                                num_hup++;
                                break;
                        case POLLNVAL:
                                ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
                                kconsumerd_del_fd(local_kconsumerd_fd[i]);
-                               kconsumerd_update_fd_array = 1;
                                num_hup++;
                                break;
                        case POLLPRI:
@@ -767,7 +801,6 @@ end:
                free(local_kconsumerd_fd);
                local_kconsumerd_fd = NULL;
        }
-       kconsumerd_cleanup();
        return NULL;
 }
 
@@ -812,6 +845,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) {
@@ -820,6 +855,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) {
@@ -830,6 +867,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) {
@@ -867,19 +908,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 */
-       cds_list_for_each_entry(iter, &kconsumerd_fd_list.head, list) {
+       /*
+        * 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
  *
This page took 0.039969 seconds and 4 git commands to generate.