X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=liblttkconsumerd%2Fliblttkconsumerd.c;h=de942a1983182df755938bfe1b840c3262a22215;hp=742fddebe47304ab9def8c339f94c76b11e4d4e9;hb=3dcd27214e19ba1175a7ac4057d3ce2c7b99c599;hpb=1ce86c9ab52be743e3348fd0c90153d9a9f63c49 diff --git a/liblttkconsumerd/liblttkconsumerd.c b/liblttkconsumerd/liblttkconsumerd.c index 742fddebe..de942a198 100644 --- a/liblttkconsumerd/liblttkconsumerd.c +++ b/liblttkconsumerd/liblttkconsumerd.c @@ -33,26 +33,48 @@ #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 @@ -85,6 +112,30 @@ void kconsumerd_set_command_socket_path(char *sock) kconsumerd_command_sock_path = sock; } +/* + * kconsumerd_find_session_fd + * + * Find a session fd in the global list. + * + * Return 1 if found else 0 + */ +static int kconsumerd_find_session_fd(int fd) +{ + struct kconsumerd_fd *iter; + + 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_data.lock); + return 1; + } + } + pthread_mutex_unlock(&kconsumerd_data.lock); + + return 0; +} + /* * kconsumerd_del_fd * @@ -92,10 +143,10 @@ void kconsumerd_set_command_socket_path(char *sock) */ 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); @@ -103,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); } /* @@ -113,8 +165,15 @@ static void kconsumerd_del_fd(struct kconsumerd_fd *lcf) */ static int kconsumerd_add_fd(struct lttcomm_kconsumerd_msg *buf, int consumerd_fd) { - struct kconsumerd_fd *tmp_fd; 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) { + goto end; + } tmp_fd = malloc(sizeof(struct kconsumerd_fd)); tmp_fd->sessiond_fd = buf->fd; @@ -137,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; } @@ -155,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); } /* @@ -170,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) @@ -178,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); @@ -197,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; } @@ -563,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 { @@ -576,7 +633,6 @@ static int kconsumerd_consumerd_recv_fd(int sfd, int size, } end: - DBG("kconsumerd_consumerd_recv_fd thread exiting"); return ret; } @@ -592,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; @@ -613,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; @@ -622,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); @@ -663,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; @@ -678,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: @@ -738,7 +801,6 @@ end: free(local_kconsumerd_fd); local_kconsumerd_fd = NULL; } - kconsumerd_cleanup(); return NULL; } @@ -783,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) { @@ -791,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) { @@ -801,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) { @@ -838,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 *