Merged! Thanks FIN David
On 11-08-12 10:14 AM, Mathieu Desnoyers wrote: > * Julien Desfossez ([email protected]) wrote: >> The init function of the library now takes a function as argument to >> allow a consumer using the library to control the function to be called >> when data is ready in a buffer. >> The kconsumerd_on_read_subbuffer_mmap and >> kconsumerd_on_read_subbuffer_splice are now exported to allow a consumer >> to use them directly if needed. >> Also the library has now a context, where all local parameters are >> registered instead of static variables. That way, we can have multiple >> callers using the library within the same process. >> Only the flag indicating that all fds are closed remain global to the >> library and shared among callers. > > Acked-by: Mathieu Desnoyers <[email protected]> > > Thanks! > > Mathieu > >> >> Signed-off-by: Julien Desfossez <[email protected]> >> --- >> liblttkconsumerd/lttkconsumerd.c | 281 >> +++++++++++++++----------------------- >> liblttkconsumerd/lttkconsumerd.h | 61 +++++++- >> ltt-kconsumerd/ltt-kconsumerd.c | 110 +++++++++++++-- >> 3 files changed, 263 insertions(+), 189 deletions(-) >> >> diff --git a/liblttkconsumerd/lttkconsumerd.c >> b/liblttkconsumerd/lttkconsumerd.c >> index 5c22d5e..d4908d1 100644 >> --- a/liblttkconsumerd/lttkconsumerd.c >> +++ b/liblttkconsumerd/lttkconsumerd.c >> @@ -62,26 +62,13 @@ struct kconsumerd_global_data { >> unsigned int need_update; >> } kconsumerd_data = { >> .fd_list.head = CDS_LIST_HEAD_INIT(kconsumerd_data.fd_list.head), >> + .fds_count = 0, >> + .need_update = 1, >> }; >> >> -/* communication with splice */ >> -static int kconsumerd_thread_pipe[2]; >> - >> -/* pipe to wake the poll thread when necessary */ >> -static int kconsumerd_poll_pipe[2]; >> - >> -/* to let the signal handler wake up the fd receiver thread */ >> -static int kconsumerd_should_quit[2]; >> - >> /* timeout parameter, to control the polling thread grace period */ >> static int kconsumerd_poll_timeout = -1; >> >> -/* socket to communicate errors with sessiond */ >> -static int kconsumerd_error_socket; >> - >> -/* socket to exchange commands with sessiond */ >> -static char *kconsumerd_command_sock_path; >> - >> /* >> * 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 >> @@ -95,9 +82,9 @@ static volatile int kconsumerd_quit = 0; >> * >> * Set the error socket >> */ >> -void kconsumerd_set_error_socket(int sock) >> +void kconsumerd_set_error_socket(struct kconsumerd_local_data *ctx, int >> sock) >> { >> - kconsumerd_error_socket = sock; >> + ctx->kconsumerd_error_socket = sock; >> } >> >> /* >> @@ -105,9 +92,10 @@ void kconsumerd_set_error_socket(int sock) >> * >> * Set the command socket path >> */ >> -void kconsumerd_set_command_socket_path(char *sock) >> +void kconsumerd_set_command_socket_path(struct kconsumerd_local_data *ctx, >> + char *sock) >> { >> - kconsumerd_command_sock_path = sock; >> + ctx->kconsumerd_command_sock_path = sock; >> } >> >> /* >> @@ -144,7 +132,9 @@ static void kconsumerd_del_fd(struct kconsumerd_fd *lcf) >> if (kconsumerd_data.fds_count > 0) { >> kconsumerd_data.fds_count--; >> if (lcf != NULL) { >> - close(lcf->out_fd); >> + if (lcf->out_fd != 0) { >> + close(lcf->out_fd); >> + } >> close(lcf->consumerd_fd); >> free(lcf); >> lcf = NULL; >> @@ -161,8 +151,8 @@ static void kconsumerd_del_fd(struct kconsumerd_fd *lcf) >> */ >> static int kconsumerd_add_fd(struct lttcomm_kconsumerd_msg *buf, int >> consumerd_fd) >> { >> - int ret; >> struct kconsumerd_fd *tmp_fd; >> + int ret = 0; >> >> pthread_mutex_lock(&kconsumerd_data.lock); >> /* Check if already exist */ >> @@ -176,22 +166,24 @@ static int kconsumerd_add_fd(struct >> lttcomm_kconsumerd_msg *buf, int consumerd_f >> tmp_fd->consumerd_fd = consumerd_fd; >> tmp_fd->state = buf->state; >> tmp_fd->max_sb_size = buf->max_sb_size; >> + tmp_fd->out_fd = 0; >> + tmp_fd->out_fd_offset = 0; >> strncpy(tmp_fd->path_name, buf->path_name, PATH_MAX); >> tmp_fd->path_name[PATH_MAX - 1] = '\0'; >> >> /* Opening the tracefile in write mode */ >> - ret = open(tmp_fd->path_name, >> - O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO); >> - if (ret < 0) { >> - ERR("Opening %s", tmp_fd->path_name); >> - perror("open"); >> - goto end; >> + if (tmp_fd->path_name != NULL) { >> + ret = open(tmp_fd->path_name, >> + O_WRONLY|O_CREAT|O_TRUNC, >> S_IRWXU|S_IRWXG|S_IRWXO); >> + if (ret < 0) { >> + ERR("Opening %s", tmp_fd->path_name); >> + perror("open"); >> + goto end; >> + } >> + tmp_fd->out_fd = ret; >> + DBG("Adding %s (%d, %d, %d)", tmp_fd->path_name, >> + tmp_fd->sessiond_fd, tmp_fd->consumerd_fd, >> tmp_fd->out_fd); >> } >> - tmp_fd->out_fd = ret; >> - tmp_fd->out_fd_offset = 0; >> - >> - DBG("Adding %s (%d, %d, %d)", tmp_fd->path_name, >> - tmp_fd->sessiond_fd, tmp_fd->consumerd_fd, >> tmp_fd->out_fd); >> >> cds_list_add(&tmp_fd->list, &kconsumerd_data.fd_list.head); >> kconsumerd_data.fds_count++; >> @@ -231,16 +223,14 @@ static void kconsumerd_change_fd_state(int sessiond_fd, >> * 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) >> +static int kconsumerd_update_poll_array(struct kconsumerd_local_data *ctx, >> + struct pollfd **pollfd, struct kconsumerd_fd >> **local_kconsumerd_fd) >> { >> struct kconsumerd_fd *iter; >> int i = 0; >> >> DBG("Updating poll fd array"); >> - >> 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); >> (*pollfd)[i].fd = iter->consumerd_fd; >> @@ -254,7 +244,7 @@ static int kconsumerd_update_poll_array(struct pollfd >> **pollfd, >> * insert the kconsumerd_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 = kconsumerd_poll_pipe[0]; >> + (*pollfd)[i].fd = ctx->kconsumerd_poll_pipe[0]; >> (*pollfd)[i].events = POLLIN; >> return i; >> } >> @@ -266,7 +256,7 @@ static int kconsumerd_update_poll_array(struct pollfd >> **pollfd, >> * mmap the ring buffer, read it and write the data to the tracefile. >> * Returns the number of bytes written >> */ >> -static int kconsumerd_on_read_subbuffer_mmap( >> +int kconsumerd_on_read_subbuffer_mmap(struct kconsumerd_local_data *ctx, >> struct kconsumerd_fd *kconsumerd_fd, unsigned long len) >> { >> unsigned long mmap_len, mmap_offset, padded_len, padding_len; >> @@ -379,7 +369,7 @@ end: >> * Splice the data from the ring buffer to the tracefile. >> * Returns the number of bytes spliced >> */ >> -static int kconsumerd_on_read_subbuffer( >> +int kconsumerd_on_read_subbuffer_splice(struct kconsumerd_local_data *ctx, >> struct kconsumerd_fd *kconsumerd_fd, unsigned long len) >> { >> long ret = 0; >> @@ -391,7 +381,7 @@ static int kconsumerd_on_read_subbuffer( >> while (len > 0) { >> DBG("splice chan to pipe offset %lu (fd : %d)", >> (unsigned long)offset, fd); >> - ret = splice(fd, &offset, kconsumerd_thread_pipe[1], NULL, len, >> + ret = splice(fd, &offset, ctx->kconsumerd_thread_pipe[1], NULL, >> len, >> SPLICE_F_MOVE | SPLICE_F_MORE); >> DBG("splice chan to pipe ret %ld", ret); >> if (ret < 0) { >> @@ -400,7 +390,7 @@ static int kconsumerd_on_read_subbuffer( >> goto splice_error; >> } >> >> - ret = splice(kconsumerd_thread_pipe[0], NULL, outfd, NULL, ret, >> + ret = splice(ctx->kconsumerd_thread_pipe[0], NULL, outfd, NULL, >> ret, >> SPLICE_F_MOVE | SPLICE_F_MORE); >> DBG("splice pipe to file %ld", ret); >> if (ret < 0) { >> @@ -452,98 +442,17 @@ splice_error: >> /* send the appropriate error description to sessiond */ >> switch(ret) { >> case EBADF: >> - kconsumerd_send_error(KCONSUMERD_SPLICE_EBADF); >> + kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_EBADF); >> break; >> case EINVAL: >> - kconsumerd_send_error(KCONSUMERD_SPLICE_EINVAL); >> + kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_EINVAL); >> break; >> case ENOMEM: >> - kconsumerd_send_error(KCONSUMERD_SPLICE_ENOMEM); >> + kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_ENOMEM); >> break; >> case ESPIPE: >> - kconsumerd_send_error(KCONSUMERD_SPLICE_ESPIPE); >> - break; >> - } >> - >> -end: >> - return ret; >> -} >> - >> -/* >> - * kconsumerd_read_subbuffer >> - * >> - * Consume data on a file descriptor and write it on a trace file >> - */ >> -static int kconsumerd_read_subbuffer(struct kconsumerd_fd *kconsumerd_fd) >> -{ >> - unsigned long len; >> - int err; >> - long ret = 0; >> - int infd = kconsumerd_fd->consumerd_fd; >> - >> - DBG("In kconsumerd_read_subbuffer (infd : %d)", infd); >> - /* Get the next subbuffer */ >> - err = kernctl_get_next_subbuf(infd); >> - if (err != 0) { >> - ret = errno; >> - perror("Reserving sub buffer failed (everything is normal, " >> - "it is due to concurrency)"); >> - goto end; >> - } >> - >> - switch (DEFAULT_KERNEL_CHANNEL_OUTPUT) { >> - case LTTNG_EVENT_SPLICE: >> - /* read the whole subbuffer */ >> - err = kernctl_get_padded_subbuf_size(infd, &len); >> - if (err != 0) { >> - ret = errno; >> - perror("Getting sub-buffer len failed."); >> - goto end; >> - } >> - >> - /* splice the subbuffer to the tracefile */ >> - ret = kconsumerd_on_read_subbuffer(kconsumerd_fd, len); >> - if (ret < 0) { >> - /* >> - * display the error but continue processing to try >> - * to release the subbuffer >> - */ >> - ERR("Error splicing to tracefile"); >> - } >> - break; >> - case LTTNG_EVENT_MMAP: >> - /* read the used subbuffer size */ >> - err = kernctl_get_subbuf_size(infd, &len); >> - if (err != 0) { >> - ret = errno; >> - perror("Getting sub-buffer len failed."); >> - goto end; >> - } >> - /* write the subbuffer to the tracefile */ >> - ret = kconsumerd_on_read_subbuffer_mmap(kconsumerd_fd, len); >> - if (ret < 0) { >> - /* >> - * display the error but continue processing to try >> - * to release the subbuffer >> - */ >> - ERR("Error writing to tracefile"); >> - } >> + kconsumerd_send_error(ctx, KCONSUMERD_SPLICE_ESPIPE); >> break; >> - default: >> - ERR("Unknown output method"); >> - ret = -1; >> - } >> - >> - err = kernctl_put_next_subbuf(infd); >> - if (err != 0) { >> - ret = errno; >> - if (errno == EFAULT) { >> - perror("Error in unreserving sub buffer\n"); >> - } else if (errno == EIO) { >> - /* Should never happen with newer LTTng versions */ >> - perror("Reader has been pushed by the writer, last >> sub-buffer corrupted."); >> - } >> - goto end; >> } >> >> end: >> @@ -583,8 +492,8 @@ exit: >> * structures describing each fd (path name). >> * Returns the size of received data >> */ >> -static int kconsumerd_consumerd_recv_fd(int sfd, >> - struct pollfd *kconsumerd_sockpoll, int size, >> +static int kconsumerd_consumerd_recv_fd(struct kconsumerd_local_data *ctx, >> + int sfd, struct pollfd *kconsumerd_sockpoll, int size, >> enum kconsumerd_command cmd_type) >> { >> struct iovec iov[1]; >> @@ -624,7 +533,7 @@ static int kconsumerd_consumerd_recv_fd(int sfd, >> >> if (ret != (size / nb_fd)) { >> ERR("Received only %d, expected %d", ret, size); >> - kconsumerd_send_error(KCONSUMERD_ERROR_RECV_FD); >> + kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD); >> goto end; >> } >> >> @@ -632,7 +541,7 @@ static int kconsumerd_consumerd_recv_fd(int sfd, >> if (!cmsg) { >> ERR("Invalid control message header"); >> ret = -1; >> - kconsumerd_send_error(KCONSUMERD_ERROR_RECV_FD); >> + kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD); >> goto end; >> } >> >> @@ -643,7 +552,7 @@ static int kconsumerd_consumerd_recv_fd(int sfd, >> DBG("kconsumerd_add_fd %s (%d)", lkm.path_name, >> ((int *) CMSG_DATA(cmsg))[0]); >> ret = kconsumerd_add_fd(&lkm, ((int *) >> CMSG_DATA(cmsg))[0]); >> if (ret < 0) { >> - >> kconsumerd_send_error(KCONSUMERD_OUTFD_ERROR); >> + kconsumerd_send_error(ctx, >> KCONSUMERD_OUTFD_ERROR); >> goto end; >> } >> break; >> @@ -654,13 +563,13 @@ static int kconsumerd_consumerd_recv_fd(int sfd, >> break; >> } >> /* signal the poll thread */ >> - tmp2 = write(kconsumerd_poll_pipe[1], "4", 1); >> + tmp2 = write(ctx->kconsumerd_poll_pipe[1], "4", 1); >> if (tmp2 < 0) { >> perror("write kconsumerd poll"); >> } >> } else { >> ERR("Didn't received any fd"); >> - kconsumerd_send_error(KCONSUMERD_ERROR_RECV_FD); >> + kconsumerd_send_error(ctx, KCONSUMERD_ERROR_RECV_FD); >> ret = -1; >> goto end; >> } >> @@ -686,12 +595,8 @@ void *kconsumerd_thread_poll_fds(void *data) >> int nb_fd = 0; >> char tmp; >> int tmp2; >> + struct kconsumerd_local_data *ctx = data; >> >> - ret = pipe(kconsumerd_thread_pipe); >> - if (ret < 0) { >> - perror("Error creating pipe"); >> - goto end; >> - } >> >> local_kconsumerd_fd = malloc(sizeof(struct kconsumerd_fd)); >> >> @@ -730,10 +635,10 @@ void *kconsumerd_thread_poll_fds(void *data) >> pthread_mutex_unlock(&kconsumerd_data.lock); >> goto end; >> } >> - ret = kconsumerd_update_poll_array(&pollfd, >> local_kconsumerd_fd); >> + ret = kconsumerd_update_poll_array(ctx, &pollfd, >> local_kconsumerd_fd); >> if (ret < 0) { >> ERR("Error in allocating pollfd or >> local_outfds"); >> - kconsumerd_send_error(KCONSUMERD_POLL_ERROR); >> + kconsumerd_send_error(ctx, >> KCONSUMERD_POLL_ERROR); >> pthread_mutex_unlock(&kconsumerd_data.lock); >> goto end; >> } >> @@ -748,7 +653,7 @@ void *kconsumerd_thread_poll_fds(void *data) >> DBG("poll num_rdy : %d", num_rdy); >> if (num_rdy == -1) { >> perror("Poll error"); >> - kconsumerd_send_error(KCONSUMERD_POLL_ERROR); >> + kconsumerd_send_error(ctx, KCONSUMERD_POLL_ERROR); >> goto end; >> } else if (num_rdy == 0) { >> DBG("Polling thread timed out"); >> @@ -768,7 +673,7 @@ void *kconsumerd_thread_poll_fds(void *data) >> */ >> if (pollfd[nb_fd].revents == POLLIN) { >> DBG("kconsumerd_poll_pipe wake up"); >> - tmp2 = read(kconsumerd_poll_pipe[0], &tmp, 1); >> + tmp2 = read(ctx->kconsumerd_poll_pipe[0], &tmp, 1); >> if (tmp2 < 0) { >> perror("read kconsumerd poll"); >> } >> @@ -796,7 +701,7 @@ void *kconsumerd_thread_poll_fds(void *data) >> case POLLPRI: >> DBG("Urgent read on fd %d", pollfd[i].fd); >> high_prio = 1; >> - ret = >> kconsumerd_read_subbuffer(local_kconsumerd_fd[i]); >> + ret = >> ctx->on_buffer_ready(local_kconsumerd_fd[i]); >> /* it's ok to have an unavailable sub-buffer */ >> if (ret == EAGAIN) { >> ret = 0; >> @@ -819,7 +724,7 @@ void *kconsumerd_thread_poll_fds(void *data) >> for (i = 0; i < nb_fd; i++) { >> if (pollfd[i].revents == POLLIN) { >> DBG("Normal read on fd %d", >> pollfd[i].fd); >> - ret = >> kconsumerd_read_subbuffer(local_kconsumerd_fd[i]); >> + ret = >> ctx->on_buffer_ready(local_kconsumerd_fd[i]); >> /* it's ok to have an unavailable >> subbuffer */ >> if (ret == EAGAIN) { >> ret = 0; >> @@ -842,34 +747,75 @@ end: >> } >> >> /* >> - * kconsumerd_init(void) >> + * kconsumerd_create >> * >> * initialise the necessary environnement : >> - * - inform the polling thread to update the polling array >> + * - create a new context >> * - create the poll_pipe >> * - create the should_quit pipe (for signal handler) >> + * - create the thread pipe (for splice) >> + * Takes a function pointer as argument, this function is called when data >> is >> + * available on a buffer. This function is responsible to do the >> + * kernctl_get_next_subbuf, read the data with mmap or splice depending on >> the >> + * buffer configuration and then kernctl_put_next_subbuf at the end. >> + * Returns a pointer to the new context or NULL on error. >> */ >> -int kconsumerd_init(void) >> +struct kconsumerd_local_data *kconsumerd_create( >> + int (*buffer_ready)(struct kconsumerd_fd *kconsumerd_fd)) >> { >> int ret; >> + struct kconsumerd_local_data *ctx; >> >> - /* need to update the polling array at init time */ >> - kconsumerd_data.need_update = 1; >> + ctx = malloc(sizeof(struct kconsumerd_local_data)); >> + if (ctx == NULL) { >> + perror("allocating context"); >> + goto end; >> + } >> + >> + ctx->on_buffer_ready = buffer_ready; >> >> - ret = pipe(kconsumerd_poll_pipe); >> + ret = pipe(ctx->kconsumerd_poll_pipe); >> if (ret < 0) { >> perror("Error creating poll pipe"); >> + ctx = NULL; >> goto end; >> } >> >> - ret = pipe(kconsumerd_should_quit); >> + ret = pipe(ctx->kconsumerd_should_quit); >> if (ret < 0) { >> perror("Error creating recv pipe"); >> + ctx = NULL; >> + goto end; >> + } >> + >> + ret = pipe(ctx->kconsumerd_thread_pipe); >> + if (ret < 0) { >> + perror("Error creating thread pipe"); >> + ctx = NULL; >> goto end; >> } >> >> end: >> - return ret; >> + return ctx; >> +} >> + >> +/* >> + * kconsumerd_destroy >> + * >> + * Close all fds associated with the instance and free the context >> + */ >> +void kconsumerd_destroy(struct kconsumerd_local_data *ctx) >> +{ >> + close(ctx->kconsumerd_error_socket); >> + close(ctx->kconsumerd_thread_pipe[0]); >> + close(ctx->kconsumerd_thread_pipe[1]); >> + close(ctx->kconsumerd_poll_pipe[0]); >> + close(ctx->kconsumerd_poll_pipe[1]); >> + close(ctx->kconsumerd_should_quit[0]); >> + close(ctx->kconsumerd_should_quit[1]); >> + unlink(ctx->kconsumerd_command_sock_path); >> + free(ctx); >> + ctx = NULL; >> } >> >> /* >> @@ -887,11 +833,12 @@ void *kconsumerd_thread_receive_fds(void *data) >> * avoids making blocking sockets >> */ >> struct pollfd kconsumerd_sockpoll[2]; >> + struct kconsumerd_local_data *ctx = data; >> >> >> - DBG("Creating command socket %s", kconsumerd_command_sock_path); >> - unlink(kconsumerd_command_sock_path); >> - client_socket = lttcomm_create_unix_sock(kconsumerd_command_sock_path); >> + DBG("Creating command socket %s", ctx->kconsumerd_command_sock_path); >> + unlink(ctx->kconsumerd_command_sock_path); >> + client_socket = >> lttcomm_create_unix_sock(ctx->kconsumerd_command_sock_path); >> if (client_socket < 0) { >> ERR("Cannot create command socket"); >> goto end; >> @@ -903,7 +850,7 @@ void *kconsumerd_thread_receive_fds(void *data) >> } >> >> DBG("Sending ready command to ltt-sessiond"); >> - ret = kconsumerd_send_error(KCONSUMERD_COMMAND_SOCK_READY); >> + ret = kconsumerd_send_error(ctx, KCONSUMERD_COMMAND_SOCK_READY); >> if (ret < 0) { >> ERR("Error sending ready command to ltt-sessiond"); >> goto end; >> @@ -916,7 +863,7 @@ void *kconsumerd_thread_receive_fds(void *data) >> } >> >> /* prepare the FDs to poll : to client socket and the should_quit pipe >> */ >> - kconsumerd_sockpoll[0].fd = kconsumerd_should_quit[0]; >> + kconsumerd_sockpoll[0].fd = ctx->kconsumerd_should_quit[0]; >> kconsumerd_sockpoll[0].events = POLLIN | POLLPRI; >> kconsumerd_sockpoll[1].fd = client_socket; >> kconsumerd_sockpoll[1].events = POLLIN | POLLPRI; >> @@ -965,7 +912,7 @@ void *kconsumerd_thread_receive_fds(void *data) >> } >> >> /* we received a command to add or update fds */ >> - ret = kconsumerd_consumerd_recv_fd(sock, kconsumerd_sockpoll, >> + ret = kconsumerd_consumerd_recv_fd(ctx, sock, >> kconsumerd_sockpoll, >> tmp.payload_size, tmp.cmd_type); >> if (ret <= 0) { >> ERR("Receiving the FD, exiting"); >> @@ -991,7 +938,7 @@ end: >> kconsumerd_poll_timeout = KCONSUMERD_POLL_GRACE_PERIOD; >> >> /* wake up the polling thread */ >> - ret = write(kconsumerd_poll_pipe[1], "4", 1); >> + ret = write(ctx->kconsumerd_poll_pipe[1], "4", 1); >> if (ret < 0) { >> perror("poll pipe write"); >> } >> @@ -1001,15 +948,13 @@ end: >> /* >> * kconsumerd_cleanup >> * >> - * Cleanup the daemon's socket on exit >> + * Close all the tracefiles and stream fds, should be called when all >> + * instances are destroyed. >> */ >> void kconsumerd_cleanup(void) >> { >> struct kconsumerd_fd *iter, *tmp; >> >> - /* remove the socket file */ >> - unlink(kconsumerd_command_sock_path); >> - >> /* >> * close all outfd. Called when there are no more threads >> * running (after joining on the threads), no need to protect >> @@ -1025,11 +970,11 @@ void kconsumerd_cleanup(void) >> * >> * Called from signal handler. >> */ >> -void kconsumerd_should_exit(void) >> +void kconsumerd_should_exit(struct kconsumerd_local_data *ctx) >> { >> int ret; >> kconsumerd_quit = 1; >> - ret = write(kconsumerd_should_quit[1], "4", 1); >> + ret = write(ctx->kconsumerd_should_quit[1], "4", 1); >> if (ret < 0) { >> perror("write kconsumerd quit"); >> } >> @@ -1040,10 +985,10 @@ void kconsumerd_should_exit(void) >> * >> * send return code to ltt-sessiond >> */ >> -int kconsumerd_send_error(enum lttcomm_return_code cmd) >> +int kconsumerd_send_error(struct kconsumerd_local_data *ctx, enum >> lttcomm_return_code cmd) >> { >> - if (kconsumerd_error_socket > 0) { >> - return lttcomm_send_unix_sock(kconsumerd_error_socket, &cmd, >> + if (ctx->kconsumerd_error_socket > 0) { >> + return lttcomm_send_unix_sock(ctx->kconsumerd_error_socket, >> &cmd, >> sizeof(enum lttcomm_sessiond_command)); >> } >> >> diff --git a/liblttkconsumerd/lttkconsumerd.h >> b/liblttkconsumerd/lttkconsumerd.h >> index cbdedd2..10e4a55 100644 >> --- a/liblttkconsumerd/lttkconsumerd.h >> +++ b/liblttkconsumerd/lttkconsumerd.h >> @@ -57,15 +57,59 @@ struct kconsumerd_fd { >> unsigned long max_sb_size; /* the subbuffer size for this channel */ >> }; >> >> +struct kconsumerd_local_data { >> + /* function to call when data is available on a buffer */ >> + int (*on_buffer_ready)(struct kconsumerd_fd *kconsumerd_fd); >> + /* socket to communicate errors with sessiond */ >> + int kconsumerd_error_socket; >> + /* socket to exchange commands with sessiond */ >> + char *kconsumerd_command_sock_path; >> + /* communication with splice */ >> + int kconsumerd_thread_pipe[2]; >> + /* pipe to wake the poll thread when necessary */ >> + int kconsumerd_poll_pipe[2]; >> + /* to let the signal handler wake up the fd receiver thread */ >> + int kconsumerd_should_quit[2]; >> +}; >> + >> /* >> - * kconsumerd_init(void) >> + * kconsumerd_create >> * initialise the necessary environnement : >> - * - inform the polling thread to update the polling array >> + * - create a new context >> * - create the poll_pipe >> * - create the should_quit pipe (for signal handler) >> - * returns the return code of pipe, 0 on success, -1 on error >> + * - create the thread pipe (for splice) >> + * Takes a function pointer as argument, this function is called when data >> is >> + * available on a buffer. This function is responsible to do the >> + * kernctl_get_next_subbuf, read the data with mmap or splice depending on >> the >> + * buffer configuration and then kernctl_put_next_subbuf at the end. >> + * Returns a pointer to the new context or NULL on error. >> + */ >> +struct kconsumerd_local_data *kconsumerd_create( >> + int (*buffer_ready)(struct kconsumerd_fd *kconsumerd_fd)); >> + >> +/* >> + * kconsumerd_destroy >> + * Close all fds associated with the instance and free the context >> + */ >> +void kconsumerd_destroy(struct kconsumerd_local_data *ctx); >> + >> +/* >> + * kconsumerd_on_read_subbuffer_mmap >> + * mmap the ring buffer, read it and write the data to the tracefile. >> + * Returns the number of bytes written >> + */ >> +int kconsumerd_on_read_subbuffer_mmap(struct kconsumerd_local_data *ctx, >> + struct kconsumerd_fd *kconsumerd_fd, unsigned long len); >> + >> +/* >> + * kconsumerd_on_read_subbuffer >> + * >> + * Splice the data from the ring buffer to the tracefile. >> + * Returns the number of bytes spliced >> */ >> -int kconsumerd_init(void); >> +int kconsumerd_on_read_subbuffer_splice(struct kconsumerd_local_data *ctx, >> + struct kconsumerd_fd *kconsumerd_fd, unsigned long len); >> >> /* >> * kconsumerd_send_error >> @@ -73,7 +117,8 @@ int kconsumerd_init(void); >> * returns the return code of sendmsg : the number of bytes transmitted >> * or -1 on error. >> */ >> -int kconsumerd_send_error(enum lttcomm_return_code cmd); >> +int kconsumerd_send_error(struct kconsumerd_local_data *ctx, >> + enum lttcomm_return_code cmd); >> >> /* >> * kconsumerd_poll_socket >> @@ -101,7 +146,7 @@ void *kconsumerd_thread_receive_fds(void *data); >> * kconsumerd_should_exit >> * Called from signal handler to ensure a clean exit >> */ >> -void kconsumerd_should_exit(void); >> +void kconsumerd_should_exit(struct kconsumerd_local_data *ctx); >> >> /* >> * kconsumerd_cleanup >> @@ -113,12 +158,12 @@ void kconsumerd_cleanup(void); >> * kconsumerd_set_error_socket >> * Set the error socket for communication with a session daemon >> */ >> -void kconsumerd_set_error_socket(int sock); >> +void kconsumerd_set_error_socket(struct kconsumerd_local_data *ctx, int >> sock); >> >> /* >> * kconsumerd_set_command_socket_path >> * Set the command socket path for communication with a session daemon >> */ >> -void kconsumerd_set_command_socket_path(char *sock); >> +void kconsumerd_set_command_socket_path(struct kconsumerd_local_data *ctx, >> char *sock); >> >> #endif /* _LIBLTTKCONSUMERD_H */ >> diff --git a/ltt-kconsumerd/ltt-kconsumerd.c >> b/ltt-kconsumerd/ltt-kconsumerd.c >> index 1e2841c..64c5ccf 100644 >> --- a/ltt-kconsumerd/ltt-kconsumerd.c >> +++ b/ltt-kconsumerd/ltt-kconsumerd.c >> @@ -55,6 +55,9 @@ static const char *progname; >> char command_sock_path[PATH_MAX]; /* Global command socket path */ >> char error_sock_path[PATH_MAX]; /* Global error path */ >> >> +/* the liblttkconsumerd context */ >> +struct kconsumerd_local_data *ctx; >> + >> /* >> * sighandler >> * >> @@ -67,7 +70,7 @@ static void sighandler(int sig) >> return; >> } >> >> - kconsumerd_should_exit(); >> + kconsumerd_should_exit(ctx); >> } >> >> /* >> @@ -190,6 +193,86 @@ static void parse_args(int argc, char **argv) >> } >> } >> >> +/* >> + * read_subbuffer >> + * >> + * Consume data on a file descriptor and write it on a trace file >> + */ >> +static int read_subbuffer(struct kconsumerd_fd *kconsumerd_fd) >> +{ >> + unsigned long len; >> + int err; >> + long ret = 0; >> + int infd = kconsumerd_fd->consumerd_fd; >> + >> + DBG("In kconsumerd_read_subbuffer (infd : %d)", infd); >> + /* Get the next subbuffer */ >> + err = kernctl_get_next_subbuf(infd); >> + if (err != 0) { >> + ret = errno; >> + perror("Reserving sub buffer failed (everything is normal, " >> + "it is due to concurrency)"); >> + goto end; >> + } >> + >> + switch (DEFAULT_KERNEL_CHANNEL_OUTPUT) { >> + case LTTNG_EVENT_SPLICE: >> + /* read the whole subbuffer */ >> + err = kernctl_get_padded_subbuf_size(infd, &len); >> + if (err != 0) { >> + ret = errno; >> + perror("Getting sub-buffer len failed."); >> + goto end; >> + } >> + >> + /* splice the subbuffer to the tracefile */ >> + ret = kconsumerd_on_read_subbuffer_splice(ctx, >> kconsumerd_fd, len); >> + if (ret < 0) { >> + /* >> + * display the error but continue processing to >> try >> + * to release the subbuffer >> + */ >> + ERR("Error splicing to tracefile"); >> + } >> + break; >> + case LTTNG_EVENT_MMAP: >> + /* read the used subbuffer size */ >> + err = kernctl_get_subbuf_size(infd, &len); >> + if (err != 0) { >> + ret = errno; >> + perror("Getting sub-buffer len failed."); >> + goto end; >> + } >> + /* write the subbuffer to the tracefile */ >> + ret = kconsumerd_on_read_subbuffer_mmap(ctx, >> kconsumerd_fd, len); >> + if (ret < 0) { >> + /* >> + * display the error but continue processing to >> try >> + * to release the subbuffer >> + */ >> + ERR("Error writing to tracefile"); >> + } >> + break; >> + default: >> + ERR("Unknown output method"); >> + ret = -1; >> + } >> + >> + err = kernctl_put_next_subbuf(infd); >> + if (err != 0) { >> + ret = errno; >> + if (errno == EFAULT) { >> + perror("Error in unreserving sub buffer\n"); >> + } else if (errno == EIO) { >> + /* Should never happen with newer LTTng versions */ >> + perror("Reader has been pushed by the writer, last >> sub-buffer corrupted."); >> + } >> + goto end; >> + } >> + >> +end: >> + return ret; >> +} >> >> /* >> * main >> @@ -217,7 +300,13 @@ int main(int argc, char **argv) >> snprintf(command_sock_path, PATH_MAX, >> KCONSUMERD_CMD_SOCK_PATH); >> } >> - kconsumerd_set_command_socket_path(command_sock_path); >> + /* create the pipe to wake to receiving thread when needed */ >> + ctx = kconsumerd_create(read_subbuffer); >> + if (ctx == NULL) { >> + goto error; >> + } >> + >> + kconsumerd_set_command_socket_path(ctx, command_sock_path); >> if (strlen(error_sock_path) == 0) { >> snprintf(error_sock_path, PATH_MAX, >> KCONSUMERD_ERR_SOCK_PATH); >> @@ -227,12 +316,6 @@ int main(int argc, char **argv) >> goto error; >> } >> >> - /* create the pipe to wake to receiving thread when needed */ >> - ret = kconsumerd_init(); >> - if (ret < 0) { >> - goto end; >> - } >> - >> /* Connect to the socket created by ltt-sessiond to report errors */ >> DBG("Connecting to error socket %s", error_sock_path); >> ret = lttcomm_connect_unix_sock(error_sock_path); >> @@ -240,11 +323,11 @@ int main(int argc, char **argv) >> if (ret < 0) { >> WARN("Cannot connect to error socket, is ltt-sessiond started >> ?"); >> } >> - kconsumerd_set_error_socket(ret); >> + kconsumerd_set_error_socket(ctx, ret); >> >> /* Create the thread to manage the receive of fd */ >> ret = pthread_create(&threads[0], NULL, kconsumerd_thread_receive_fds, >> - (void *) NULL); >> + (void *) ctx); >> if (ret != 0) { >> perror("pthread_create"); >> goto error; >> @@ -252,7 +335,7 @@ int main(int argc, char **argv) >> >> /* Create thread to manage the polling/writing of traces */ >> ret = pthread_create(&threads[1], NULL, kconsumerd_thread_poll_fds, >> - (void *) NULL); >> + (void *) ctx); >> if (ret != 0) { >> perror("pthread_create"); >> goto error; >> @@ -266,14 +349,15 @@ int main(int argc, char **argv) >> } >> } >> ret = EXIT_SUCCESS; >> - kconsumerd_send_error(KCONSUMERD_EXIT_SUCCESS); >> + kconsumerd_send_error(ctx, KCONSUMERD_EXIT_SUCCESS); >> goto end; >> >> error: >> ret = EXIT_FAILURE; >> - kconsumerd_send_error(KCONSUMERD_EXIT_FAILURE); >> + kconsumerd_send_error(ctx, KCONSUMERD_EXIT_FAILURE); >> >> end: >> + kconsumerd_destroy(ctx); >> kconsumerd_cleanup(); >> >> return ret; >> -- >> 1.7.4.1 >> > _______________________________________________ ltt-dev mailing list [email protected] http://lists.casi.polymtl.ca/cgi-bin/mailman/listinfo/ltt-dev
