X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fbin%2Flttng-consumerd%2Flttng-consumerd.c;h=cf9cb205c35ba73b600bb08125ced84dbb43aada;hb=5c635c724d60fe8e8bfeb00907dfa1e113cc3548;hp=5952334cd47fe1c1d882e9c7880e627218fb044b;hpb=fb3a43a9284f3300e9b66edc2f2c2d2767895423;p=lttng-tools.git diff --git a/src/bin/lttng-consumerd/lttng-consumerd.c b/src/bin/lttng-consumerd/lttng-consumerd.c index 5952334cd..cf9cb205c 100644 --- a/src/bin/lttng-consumerd/lttng-consumerd.c +++ b/src/bin/lttng-consumerd/lttng-consumerd.c @@ -44,15 +44,20 @@ #include #include #include +#include #include #include +#include #include "lttng-consumerd.h" +#include "health-consumerd.h" /* TODO : support UST (all direct kernel-ctl accesses). */ -/* the two threads (receive fd, poll and metadata) */ -static pthread_t threads[3]; +/* threads (channel handling, poll, metadata, sessiond) */ + +static pthread_t channel_thread, data_thread, metadata_thread, + sessiond_thread, metadata_timer_thread, health_thread; /* to count the number of times the user pressed ctrl+c */ static int sigintcount = 0; @@ -69,6 +74,17 @@ static enum lttng_consumer_type opt_type = LTTNG_CONSUMER_KERNEL; /* the liblttngconsumerd context */ static struct lttng_consumer_local_data *ctx; +/* Consumerd health monitoring */ +struct health_app *health_consumerd; + +enum lttng_consumer_type lttng_consumer_get_type(void) +{ + if (!ctx) { + return LTTNG_CONSUMER_UNKNOWN; + } + return ctx->type; +} + /* * Signal handler for the daemon */ @@ -79,6 +95,14 @@ static void sighandler(int sig) return; } + /* + * Ignore SIGPIPE because it should not stop the consumer whenever a + * SIGPIPE is catched through a FD operation. + */ + if (sig == SIGPIPE) { + return; + } + lttng_consumer_should_exit(ctx); } @@ -252,7 +276,6 @@ static void set_ulimit(void) */ int main(int argc, char **argv) { - int i; int ret = 0; void *status; @@ -287,7 +310,7 @@ int main(int argc, char **argv) /* Set up max poll set size */ lttng_poll_set_max_size(); - if (strlen(command_sock_path) == 0) { + if (*command_sock_path == '\0') { switch (opt_type) { case LTTNG_CONSUMER_KERNEL: snprintf(command_sock_path, PATH_MAX, DEFAULT_KCONSUMERD_CMD_SOCK_PATH, @@ -315,6 +338,11 @@ int main(int argc, char **argv) set_ulimit(); } + health_consumerd = health_app_create(NR_HEALTH_CONSUMERD_TYPES); + if (!health_consumerd) { + goto error; + } + /* create the consumer instance with and assign the callbacks */ ctx = lttng_consumer_create(opt_type, lttng_consumer_read_subbuffer, NULL, lttng_consumer_on_recv_stream, NULL); @@ -323,7 +351,7 @@ int main(int argc, char **argv) } lttng_consumer_set_command_sock_path(ctx, command_sock_path); - if (strlen(error_sock_path) == 0) { + if (*error_sock_path == '\0') { switch (opt_type) { case LTTNG_CONSUMER_KERNEL: snprintf(error_sock_path, PATH_MAX, DEFAULT_KCONSUMERD_ERR_SOCK_PATH, @@ -356,40 +384,136 @@ int main(int argc, char **argv) } lttng_consumer_set_error_sock(ctx, ret); - /* Create the thread to manage the receive of fd */ - ret = pthread_create(&threads[0], NULL, lttng_consumer_thread_receive_fds, + /* + * Block RT signals used for UST periodical metadata flush and the live + * timer in main, and create a dedicated thread to handle these signals. + */ + consumer_signal_init(); + + ctx->type = opt_type; + + /* Initialize communication library */ + lttcomm_init(); + + ret = utils_create_pipe(health_quit_pipe); + if (ret < 0) { + goto error_health_pipe; + } + + /* Create thread to manage the client socket */ + ret = pthread_create(&health_thread, NULL, + thread_manage_health, (void *) NULL); + if (ret != 0) { + PERROR("pthread_create health"); + goto health_error; + } + + /* Create thread to manage channels */ + ret = pthread_create(&channel_thread, NULL, consumer_thread_channel_poll, (void *) ctx); if (ret != 0) { perror("pthread_create"); - goto error; + goto channel_error; + } + + /* Create thread to manage the polling/writing of trace metadata */ + ret = pthread_create(&metadata_thread, NULL, consumer_thread_metadata_poll, + (void *) ctx); + if (ret != 0) { + perror("pthread_create"); + goto metadata_error; } - /* Create thread to manage the polling/writing of traces */ - ret = pthread_create(&threads[1], NULL, lttng_consumer_thread_poll_fds, + /* Create thread to manage the polling/writing of trace data */ + ret = pthread_create(&data_thread, NULL, consumer_thread_data_poll, (void *) ctx); if (ret != 0) { perror("pthread_create"); + goto data_error; + } + + /* Create the thread to manage the receive of fd */ + ret = pthread_create(&sessiond_thread, NULL, consumer_thread_sessiond_poll, + (void *) ctx); + if (ret != 0) { + perror("pthread_create"); + goto sessiond_error; + } + + /* + * Create the thread to manage the UST metadata periodic timer and + * live timer. + */ + ret = pthread_create(&metadata_timer_thread, NULL, + consumer_timer_thread, (void *) ctx); + if (ret != 0) { + perror("pthread_create"); + goto metadata_timer_error; + } + + ret = pthread_detach(metadata_timer_thread); + if (ret) { + errno = ret; + perror("pthread_detach"); + } + +metadata_timer_error: + ret = pthread_join(sessiond_thread, &status); + if (ret != 0) { + perror("pthread_join"); goto error; } - for (i = 0; i < 2; i++) { - ret = pthread_join(threads[i], &status); - if (ret != 0) { - perror("pthread_join"); - goto error; - } +sessiond_error: + ret = pthread_join(data_thread, &status); + if (ret != 0) { + perror("pthread_join"); + goto error; + } + +data_error: + ret = pthread_join(metadata_thread, &status); + if (ret != 0) { + perror("pthread_join"); + goto error; + } + +metadata_error: + ret = pthread_join(channel_thread, &status); + if (ret != 0) { + perror("pthread_join"); + goto error; + } + +channel_error: + ret = pthread_join(health_thread, &status); + if (ret != 0) { + PERROR("pthread_join health thread"); + goto error; /* join error, exit without cleanup */ + } + +health_error: + utils_close_pipe(health_quit_pipe); + +error_health_pipe: + if (!ret) { + ret = EXIT_SUCCESS; + lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_EXIT_SUCCESS); + goto end; } - ret = EXIT_SUCCESS; - lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_EXIT_SUCCESS); - goto end; error: ret = EXIT_FAILURE; - lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_EXIT_FAILURE); + if (ctx) { + lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_EXIT_FAILURE); + } end: lttng_consumer_destroy(ctx); lttng_consumer_cleanup(); + if (health_consumerd) { + health_app_destroy(health_consumerd); + } return ret; }