X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=ltt-kconsumerd%2Fltt-kconsumerd.c;h=a81be55e7c9dbdaed5daf63c2c68d4c6280ac1a8;hp=ade19399da97931fb06c32b99ddc70bd02481305;hb=558e70e9829c97ecd228516f6e312768be0536d7;hpb=82a3637f639486c07ff937ab03e1e9532379d26a diff --git a/ltt-kconsumerd/ltt-kconsumerd.c b/ltt-kconsumerd/ltt-kconsumerd.c index ade19399d..a81be55e7 100644 --- a/ltt-kconsumerd/ltt-kconsumerd.c +++ b/ltt-kconsumerd/ltt-kconsumerd.c @@ -36,13 +36,16 @@ #include #include #include +#include -#include "lttngerr.h" -#include "libkernelctl.h" -#include "liblttkconsumerd.h" +#include +#include +#include +#include +#include /* 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 +55,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 +71,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 +192,141 @@ 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 */ + unsigned long mmap_len; + + ret = kernctl_get_mmap_len(kconsumerd_fd->consumerd_fd, &mmap_len); + if (ret != 0) { + ret = errno; + perror("kernctl_get_mmap_len"); + goto error_close_fd; + } + kconsumerd_fd->mmap_len = (size_t) mmap_len; + + 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 +354,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,13 +370,6 @@ int main(int argc, char **argv) goto error; } - /* create the pipe to wake to polling thread when needed */ - ret = kconsumerd_create_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); ret = lttcomm_connect_unix_sock(error_sock_path); @@ -241,19 +377,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; @@ -267,15 +403,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; }