Merge remote-tracking branch 'cbab-github/tests-cleanup' into cbab

[lttng-tools.git] / src / bin / lttng-consumerd / lttng-consumerd.c

diff --git a/src/bin/lttng-consumerd/lttng-consumerd.c b/src/bin/lttng-consumerd/lttng-consumerd.c
index 3bc700dc948594a937c2ef543a8cbe4c3d4212f7..84868077c39e5e85624a973d70bfca270670bce2 100644 (file)
--- a/src/bin/lttng-consumerd/lttng-consumerd.c
+++ b/src/bin/lttng-consumerd/lttng-consumerd.c
@@ -43,17 +43,16 @@
 
 #include <common/defaults.h>
 #include <common/common.h>
-#include <common/kernel-consumer/kernel-consumer.h>
-#include <common/kernel-ctl/kernel-ctl.h>
+#include <common/consumer.h>
+#include <common/compat/poll.h>
 #include <common/sessiond-comm/sessiond-comm.h>
-#include <common/ust-consumer/ust-consumer.h>
 
 #include "lttng-consumerd.h"
 
 /* TODO : support UST (all direct kernel-ctl accesses). */
 
-/* the two threads (receive fd and poll) */
-static pthread_t threads[2];
+/* threads (channel handling, poll, metadata, sessiond) */
+static pthread_t channel_thread, data_thread, metadata_thread, sessiond_thread;
 
 /* to count the number of times the user pressed ctrl+c */
 static int sigintcount = 0;
@@ -80,6 +79,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);
 }
 
@@ -253,7 +260,6 @@ static void set_ulimit(void)
  */
 int main(int argc, char **argv)
 {
-       int i;
        int ret = 0;
        void *status;
 
@@ -285,7 +291,10 @@ int main(int argc, char **argv)
                }
        }
 
-       if (strlen(command_sock_path) == 0) {
+       /* Set up max poll set size */
+       lttng_poll_set_max_size();
+
+       if (*command_sock_path == '\0') {
                switch (opt_type) {
                case LTTNG_CONSUMER_KERNEL:
                        snprintf(command_sock_path, PATH_MAX, DEFAULT_KCONSUMERD_CMD_SOCK_PATH,
@@ -321,7 +330,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,
@@ -354,36 +363,74 @@ 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,
+       /* 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;
        }
 
-       /* 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 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 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;
+       }
+
+       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;
+       }
+
+       if (!ret) {
+               ret = EXIT_SUCCESS;
+               lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_EXIT_SUCCESS);
+               goto end;
        }
-       ret = EXIT_SUCCESS;
-       lttng_consumer_send_error(ctx, CONSUMERD_EXIT_SUCCESS);
-       goto end;
 
 error:
        ret = EXIT_FAILURE;
-       lttng_consumer_send_error(ctx, CONSUMERD_EXIT_FAILURE);
+       lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_EXIT_FAILURE);
 
 end:
        lttng_consumer_destroy(ctx);