X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=ltt-kconsumerd%2Fltt-kconsumerd.c;h=7eab42e522d0f45bcb126e101f90d4d543a604e4;hp=4180f890117257135f60436f57e9b6249882cf79;hb=8e0dbb653e6872725775d640e2a64d4b08cceb3a;hpb=4de84ad900e5b6e739f3c36aa1b86746081333d7 diff --git a/ltt-kconsumerd/ltt-kconsumerd.c b/ltt-kconsumerd/ltt-kconsumerd.c index 4180f8901..7eab42e52 100644 --- a/ltt-kconsumerd/ltt-kconsumerd.c +++ b/ltt-kconsumerd/ltt-kconsumerd.c @@ -36,13 +36,17 @@ #include #include #include +#include + +#include #include "lttngerr.h" -#include "libkernelctl.h" -#include "liblttkconsumerd.h" +#include "kernelctl.h" +#include "ltt-kconsumerd.h" +#include "lttng-sessiond-comm.h" /* the two threads (receive fd and poll) */ -pthread_t threads[2]; +static pthread_t threads[2]; /* to count the number of time the user pressed ctrl+c */ static int sigintcount = 0; @@ -52,13 +56,14 @@ int opt_quiet; int opt_verbose; static int opt_daemon; static const char *progname; -char command_sock_path[PATH_MAX]; /* Global command socket path */ -char error_sock_path[PATH_MAX]; /* Global error path */ +static char command_sock_path[PATH_MAX]; /* Global command socket path */ +static char error_sock_path[PATH_MAX]; /* Global error path */ + +/* the liblttngkconsumerd context */ +static struct lttng_kconsumerd_local_data *ctx; /* - * sighandler - * - * Signal handler for the daemon + * Signal handler for the daemon */ static void sighandler(int sig) { @@ -67,13 +72,11 @@ static void sighandler(int sig) return; } - kconsumerd_should_exit(); + lttng_kconsumerd_should_exit(ctx); } /* - * set_signal_handler - * - * Setup signal handler for : + * Setup signal handler for : * SIGINT, SIGTERM, SIGPIPE */ static int set_signal_handler(void) @@ -190,6 +193,138 @@ static void parse_args(int argc, char **argv) } } +/* + * Consume data on a file descriptor and write it on a trace file. + */ +static int read_subbuffer(struct lttng_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; + /* + * This is a debug message even for single-threaded consumer, + * because poll() have more relaxed criterions than get subbuf, + * so get_subbuf may fail for short race windows where poll() + * would issue wakeups. + */ + DBG("Reserving sub buffer failed (everything is normal, " + "it is due to concurrency)"); + goto end; + } + + switch (kconsumerd_fd->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 = lttng_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_padded_subbuf_size(infd, &len); + if (err != 0) { + ret = errno; + perror("Getting sub-buffer len failed."); + goto end; + } + /* write the subbuffer to the tracefile */ + ret = lttng_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; +} + +static int on_recv_fd(struct lttng_kconsumerd_fd *kconsumerd_fd) +{ + int ret; + + /* Opening the tracefile in write mode */ + if (kconsumerd_fd->path_name != NULL) { + ret = open(kconsumerd_fd->path_name, + O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO); + if (ret < 0) { + ERR("Opening %s", kconsumerd_fd->path_name); + perror("open"); + goto error; + } + kconsumerd_fd->out_fd = ret; + } + + if (kconsumerd_fd->output == LTTNG_EVENT_MMAP) { + /* get the len of the mmap region */ + ret = kernctl_get_mmap_len(kconsumerd_fd->consumerd_fd, &kconsumerd_fd->mmap_len); + if (ret != 0) { + ret = errno; + perror("kernctl_get_mmap_len"); + goto error_close_fd; + } + + kconsumerd_fd->mmap_base = mmap(NULL, kconsumerd_fd->mmap_len, + PROT_READ, MAP_PRIVATE, kconsumerd_fd->consumerd_fd, 0); + if (kconsumerd_fd->mmap_base == MAP_FAILED) { + perror("Error mmaping"); + ret = -1; + goto error_close_fd; + } + } + + /* we return 0 to let the library handle the FD internally */ + return 0; + +error_close_fd: + { + int err; + + err = close(kconsumerd_fd->out_fd); + assert(!err); + } +error: + return ret; +} /* * main @@ -217,7 +352,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 consumer instance with and assign the callbacks */ + ctx = lttng_kconsumerd_create(read_subbuffer, on_recv_fd, NULL); + if (ctx == NULL) { + goto error; + } + + lttng_kconsumerd_set_command_sock_path(ctx, command_sock_path); if (strlen(error_sock_path) == 0) { snprintf(error_sock_path, PATH_MAX, KCONSUMERD_ERR_SOCK_PATH); @@ -227,12 +368,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,19 +375,19 @@ int main(int argc, char **argv) if (ret < 0) { WARN("Cannot connect to error socket, is ltt-sessiond started ?"); } - kconsumerd_set_error_socket(ret); + lttng_kconsumerd_set_error_sock(ctx, ret); /* Create the thread to manage the receive of fd */ - ret = pthread_create(&threads[0], NULL, kconsumerd_thread_receive_fds, - (void *) NULL); + ret = pthread_create(&threads[0], NULL, lttng_kconsumerd_thread_receive_fds, + (void *) ctx); if (ret != 0) { perror("pthread_create"); goto error; } /* Create thread to manage the polling/writing of traces */ - ret = pthread_create(&threads[1], NULL, kconsumerd_thread_poll_fds, - (void *) NULL); + ret = pthread_create(&threads[1], NULL, lttng_kconsumerd_thread_poll_fds, + (void *) ctx); if (ret != 0) { perror("pthread_create"); goto error; @@ -266,15 +401,16 @@ int main(int argc, char **argv) } } ret = EXIT_SUCCESS; - kconsumerd_send_error(KCONSUMERD_EXIT_SUCCESS); + lttng_kconsumerd_send_error(ctx, KCONSUMERD_EXIT_SUCCESS); goto end; error: ret = EXIT_FAILURE; - kconsumerd_send_error(KCONSUMERD_EXIT_FAILURE); + lttng_kconsumerd_send_error(ctx, KCONSUMERD_EXIT_FAILURE); end: - kconsumerd_cleanup(); + lttng_kconsumerd_destroy(ctx); + lttng_kconsumerd_cleanup(); return ret; }