X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=kconsumerd%2Fkconsumerd.c;h=fcd4101a62d0bdbdff1b7ffcd34a86365324afbf;hp=96051caafdfeba7dd09b4029c16404d28769cf08;hb=252fd492b48721806ed8d72b508ea3733bc70845;hpb=7e8c38c672e816ea3b0d9a7accdcae4d6e01c668 diff --git a/kconsumerd/kconsumerd.c b/kconsumerd/kconsumerd.c index 96051caaf..fcd4101a6 100644 --- 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; @@ -79,6 +82,7 @@ static char error_sock_path[PATH_MAX]; /* Global error path */ */ static void del_fd(struct ltt_kconsumerd_fd *lcf) { + DBG("Removing %d", lcf->consumerd_fd); pthread_mutex_lock(&kconsumerd_lock_fds); cds_list_del(&lcf->list); if (fds_count > 0) { @@ -103,7 +107,6 @@ static void cleanup() { struct ltt_kconsumerd_fd *iter; - /* remove the socket file */ unlink(command_sock_path); @@ -118,7 +121,8 @@ static void cleanup() } } -/* send_error +/* + * send_error * * send return code to ltt-sessiond */ @@ -330,7 +334,7 @@ static int read_subbuffer(struct ltt_kconsumerd_fd *kconsumerd_fd) long ret = 0; int infd = kconsumerd_fd->consumerd_fd; - DBG("In read_subbuffer"); + DBG("In read_subbuffer (infd : %d)", infd); /* Get the next subbuffer */ err = kernctl_get_next_subbuf(infd); if (err != 0) { @@ -403,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)]; @@ -464,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); @@ -518,23 +523,21 @@ static void *thread_receive_fds(void *data) goto error; } 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) { + if (ret <= 0) { ERR("Receiving the lttcomm_kconsumerd_header, exiting"); goto error; } ret = consumerd_recv_fd(sock, tmp.payload_size, tmp.cmd_type); - if (ret < 0) { + if (ret <= 0) { ERR("Receiving the FD, exiting"); goto error; } } error: - cleanup(); return NULL; } @@ -562,13 +565,15 @@ static int update_poll_array(struct pollfd **pollfd, *local_kconsumerd_fd = NULL; } - *pollfd = malloc(fds_count * sizeof(struct pollfd)); + /* 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; } - *local_kconsumerd_fd = malloc(fds_count * sizeof(struct ltt_kconsumerd_fd)); + /* 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; @@ -581,14 +586,21 @@ static int update_poll_array(struct pollfd **pollfd, DBG("Inside for each"); if (iter->state == ACTIVE_FD) { DBG("Active FD %d", iter->consumerd_fd); - pollfd[i]->fd = iter->consumerd_fd; - pollfd[i]->events = POLLIN | POLLPRI; + (*pollfd)[i].fd = iter->consumerd_fd; + (*pollfd)[i].events = POLLIN | POLLPRI; local_kconsumerd_fd[i] = iter; i++; } else if (iter->state == DELETE_FD) { 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; @@ -608,9 +620,11 @@ static void *thread_poll_fds(void *data) int num_rdy, num_hup, high_prio, ret, i; struct pollfd *pollfd = NULL; /* local view of the fds */ - struct ltt_kconsumerd_fd *local_kconsumerd_fd = NULL; + 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) { @@ -618,6 +632,8 @@ static void *thread_poll_fds(void *data) goto end; } + local_kconsumerd_fd = malloc(sizeof(struct ltt_kconsumerd_fd)); + while (1) { high_prio = 0; num_hup = 0; @@ -627,7 +643,7 @@ static void *thread_poll_fds(void *data) * local array as well */ if (update_fd_array) { - ret = update_poll_array(&pollfd, &local_kconsumerd_fd); + ret = update_poll_array(&pollfd, local_kconsumerd_fd); if (ret < 0) { ERR("Error in allocating pollfd or local_outfds"); send_error(KCONSUMERD_POLL_ERROR); @@ -637,8 +653,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"); @@ -646,6 +662,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) { @@ -666,7 +692,7 @@ static void *thread_poll_fds(void *data) case POLLPRI: DBG("Urgent read on fd %d", pollfd[i].fd); high_prio = 1; - ret = read_subbuffer(&local_kconsumerd_fd[i]); + ret = read_subbuffer(local_kconsumerd_fd[i]); /* it's ok to have an unavailable sub-buffer (FIXME : is it ?) */ if (ret == EAGAIN) { ret = 0; @@ -679,7 +705,7 @@ static void *thread_poll_fds(void *data) if (nb_fd > 0 && num_hup == nb_fd) { DBG("every buffer FD has hung up\n"); send_error(KCONSUMERD_POLL_HUP); - continue; + goto end; } /* Take care of low priority channels. */ @@ -688,7 +714,7 @@ static void *thread_poll_fds(void *data) switch(pollfd[i].revents) { case POLLIN: DBG("Normal read on fd %d", pollfd[i].fd); - ret = read_subbuffer(&local_kconsumerd_fd[i]); + ret = read_subbuffer(local_kconsumerd_fd[i]); /* it's ok to have an unavailable subbuffer (FIXME : is it ?) */ if (ret == EAGAIN) { ret = 0; @@ -829,6 +855,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);