src/bin/lttng-relayd/main.c

   1 /*
   2  * Copyright (C) 2012 - Julien Desfossez <jdesfossez@efficios.com>
   3  *                      David Goulet <dgoulet@efficios.com>
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License, version 2 only,
   7  * as published by the Free Software Foundation.
   8  *
   9  * This program is distributed in the hope that it will be useful, but WITHOUT
  10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  12  * more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  17  */
  18
  19 #define _GNU_SOURCE
  20 #include <getopt.h>
  21 #include <grp.h>
  22 #include <limits.h>
  23 #include <pthread.h>
  24 #include <signal.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <sys/mman.h>
  29 #include <sys/mount.h>
  30 #include <sys/resource.h>
  31 #include <sys/socket.h>
  32 #include <sys/stat.h>
  33 #include <sys/types.h>
  34 #include <sys/wait.h>
  35 #include <urcu/futex.h>
  36 #include <urcu/uatomic.h>
  37 #include <unistd.h>
  38 #include <fcntl.h>
  39 #include <config.h>
  40
  41 #include <lttng/lttng.h>
  42 #include <common/common.h>
  43 #include <common/compat/poll.h>
  44 #include <common/compat/socket.h>
  45 #include <common/defaults.h>
  46 #include <common/futex.h>
  47 #include <common/sessiond-comm/sessiond-comm.h>
  48 #include <common/sessiond-comm/inet.h>
  49 #include <common/hashtable/hashtable.h>
  50 #include <common/sessiond-comm/relayd.h>
  51 #include <common/uri.h>
  52
  53 #include "lttng-relayd.h"
  54
  55 /* command line options */
  56 static int opt_daemon;
  57 static char *opt_output_path;
  58 static struct lttng_uri *control_uri = NULL;
  59 static struct lttng_uri *data_uri = NULL;
  60
  61 const char *progname;
  62 static int is_root;                     /* Set to 1 if the daemon is running as root */
  63
  64 /*
  65  * Quit pipe for all threads. This permits a single cancellation point
  66  * for all threads when receiving an event on the pipe.
  67  */
  68 static int thread_quit_pipe[2] = { -1, -1 };
  69
  70 /*
  71  * This pipe is used to inform the worker thread that a command is queued and
  72  * ready to be processed.
  73  */
  74 static int relay_cmd_pipe[2] = { -1, -1 };
  75
  76 static int dispatch_thread_exit;
  77
  78 static pthread_t listener_thread;
  79 static pthread_t dispatcher_thread;
  80 static pthread_t worker_thread;
  81
  82 static uint64_t last_relay_stream_id = 0;
  83 static uint64_t last_relay_session_id = 0;
  84
  85 /*
  86  * Relay command queue.
  87  *
  88  * The relay_thread_listener and relay_thread_dispatcher communicate with this
  89  * queue.
  90  */
  91 static struct relay_cmd_queue relay_cmd_queue;
  92
  93 /* buffer allocated at startup, used to store the trace data */
  94 static char *data_buffer = NULL;
  95 static unsigned int data_buffer_size = 0;
  96
  97 /*
  98  * usage function on stderr
  99  */
 100 static
 101 void usage(void)
 102 {
 103         fprintf(stderr, "Usage: %s OPTIONS\n\nOptions:\n", progname);
 104         fprintf(stderr, "  -h, --help                         Display this usage.\n");
 105         fprintf(stderr, "  -d, --daemonize                    Start as a daemon.\n");
 106         fprintf(stderr, "  -C, --control-port                 Control port listening (URI)\n");
 107         fprintf(stderr, "  -D, --data-port                    Data port listening (URI)\n");
 108         fprintf(stderr, "  -o, --output                       Output path for traces (PATH)\n");
 109         fprintf(stderr, "  -v, --verbose                      Verbose mode. Activate DBG() macro.\n");
 110 }
 111
 112 static
 113 int parse_args(int argc, char **argv)
 114 {
 115         int c;
 116         int ret = 0;
 117         char *default_address;
 118
 119         static struct option long_options[] = {
 120                 { "control-port", 1, 0, 'C' },
 121                 { "data-port", 1, 0, 'D' },
 122                 { "daemonize", 0, 0, 'd' },
 123                 { "help", 0, 0, 'h' },
 124                 { "output", 1, 0, 'o' },
 125                 { "verbose", 0, 0, 'v' },
 126                 { NULL, 0, 0, 0 }
 127         };
 128
 129         while (1) {
 130                 int option_index = 0;
 131                 c = getopt_long(argc, argv, "dhv" "C:D:o:",
 132                                 long_options, &option_index);
 133                 if (c == -1) {
 134                         break;
 135                 }
 136
 137                 switch (c) {
 138                 case 0:
 139                         fprintf(stderr, "option %s", long_options[option_index].name);
 140                         if (optarg) {
 141                                 fprintf(stderr, " with arg %s\n", optarg);
 142                         }
 143                         break;
 144                 case 'C':
 145                         ret = uri_parse(optarg, &control_uri);
 146                         if (ret < 0) {
 147                                 ERR("Invalid control URI specified");
 148                                 goto exit;
 149                         }
 150                         if (control_uri->port == 0) {
 151                                 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 152                         }
 153                         break;
 154                 case 'D':
 155                         ret = uri_parse(optarg, &data_uri);
 156                         if (ret < 0) {
 157                                 ERR("Invalid data URI specified");
 158                                 goto exit;
 159                         }
 160                         if (data_uri->port == 0) {
 161                                 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
 162                         }
 163                         break;
 164                 case 'd':
 165                         opt_daemon = 1;
 166                         break;
 167                 case 'h':
 168                         usage();
 169                         exit(EXIT_FAILURE);
 170                 case 'o':
 171                         ret = asprintf(&opt_output_path, "%s", optarg);
 172                         if (ret < 0) {
 173                                 PERROR("asprintf opt_output_path");
 174                                 goto exit;
 175                         }
 176                         break;
 177                 case 'v':
 178                         /* Verbose level can increase using multiple -v */
 179                         lttng_opt_verbose += 1;
 180                         break;
 181                 default:
 182                         /* Unknown option or other error.
 183                          * Error is printed by getopt, just return */
 184                         ret = -1;
 185                         goto exit;
 186                 }
 187         }
 188
 189         /* assign default values */
 190         if (control_uri == NULL) {
 191                 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
 192                                 DEFAULT_NETWORK_CONTROL_PORT);
 193                 if (ret < 0) {
 194                         PERROR("asprintf default data address");
 195                         goto exit;
 196                 }
 197
 198                 ret = uri_parse(default_address, &control_uri);
 199                 free(default_address);
 200                 if (ret < 0) {
 201                         ERR("Invalid control URI specified");
 202                         goto exit;
 203                 }
 204         }
 205         if (data_uri == NULL) {
 206                 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
 207                                 DEFAULT_NETWORK_DATA_PORT);
 208                 if (ret < 0) {
 209                         PERROR("asprintf default data address");
 210                         goto exit;
 211                 }
 212
 213                 ret = uri_parse(default_address, &data_uri);
 214                 free(default_address);
 215                 if (ret < 0) {
 216                         ERR("Invalid data URI specified");
 217                         goto exit;
 218                 }
 219         }
 220
 221 exit:
 222         return ret;
 223 }
 224
 225 /*
 226  * Cleanup the daemon
 227  */
 228 static
 229 void cleanup(void)
 230 {
 231         int i, ret;
 232
 233         DBG("Cleaning up");
 234
 235         for (i = 0; i < 2; i++) {
 236                 if (thread_quit_pipe[i] >= 0) {
 237                         ret = close(thread_quit_pipe[i]);
 238                         if (ret) {
 239                                 PERROR("close");
 240                         }
 241                 }
 242         }
 243 }
 244
 245 /*
 246  * Write to writable pipe used to notify a thread.
 247  */
 248 static
 249 int notify_thread_pipe(int wpipe)
 250 {
 251         int ret;
 252
 253         ret = write(wpipe, "!", 1);
 254         if (ret < 0) {
 255                 PERROR("write poll pipe");
 256         }
 257
 258         return ret;
 259 }
 260
 261 /*
 262  * Stop all threads by closing the thread quit pipe.
 263  */
 264 static
 265 void stop_threads(void)
 266 {
 267         int ret;
 268
 269         /* Stopping all threads */
 270         DBG("Terminating all threads");
 271         ret = notify_thread_pipe(thread_quit_pipe[1]);
 272         if (ret < 0) {
 273                 ERR("write error on thread quit pipe");
 274         }
 275
 276         /* Dispatch thread */
 277         dispatch_thread_exit = 1;
 278         futex_nto1_wake(&relay_cmd_queue.futex);
 279 }
 280
 281 /*
 282  * Signal handler for the daemon
 283  *
 284  * Simply stop all worker threads, leaving main() return gracefully after
 285  * joining all threads and calling cleanup().
 286  */
 287 static
 288 void sighandler(int sig)
 289 {
 290         switch (sig) {
 291         case SIGPIPE:
 292                 DBG("SIGPIPE caught");
 293                 return;
 294         case SIGINT:
 295                 DBG("SIGINT caught");
 296                 stop_threads();
 297                 break;
 298         case SIGTERM:
 299                 DBG("SIGTERM caught");
 300                 stop_threads();
 301                 break;
 302         default:
 303                 break;
 304         }
 305 }
 306
 307 /*
 308  * Setup signal handler for :
 309  *              SIGINT, SIGTERM, SIGPIPE
 310  */
 311 static
 312 int set_signal_handler(void)
 313 {
 314         int ret = 0;
 315         struct sigaction sa;
 316         sigset_t sigset;
 317
 318         if ((ret = sigemptyset(&sigset)) < 0) {
 319                 PERROR("sigemptyset");
 320                 return ret;
 321         }
 322
 323         sa.sa_handler = sighandler;
 324         sa.sa_mask = sigset;
 325         sa.sa_flags = 0;
 326         if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
 327                 PERROR("sigaction");
 328                 return ret;
 329         }
 330
 331         if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
 332                 PERROR("sigaction");
 333                 return ret;
 334         }
 335
 336         if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
 337                 PERROR("sigaction");
 338                 return ret;
 339         }
 340
 341         DBG("Signal handler set for SIGTERM, SIGPIPE and SIGINT");
 342
 343         return ret;
 344 }
 345
 346 /*
 347  * Init thread quit pipe.
 348  *
 349  * Return -1 on error or 0 if all pipes are created.
 350  */
 351 static
 352 int init_thread_quit_pipe(void)
 353 {
 354         int ret, i;
 355
 356         ret = pipe(thread_quit_pipe);
 357         if (ret < 0) {
 358                 PERROR("thread quit pipe");
 359                 goto error;
 360         }
 361
 362         for (i = 0; i < 2; i++) {
 363                 ret = fcntl(thread_quit_pipe[i], F_SETFD, FD_CLOEXEC);
 364                 if (ret < 0) {
 365                         PERROR("fcntl");
 366                         goto error;
 367                 }
 368         }
 369
 370 error:
 371         return ret;
 372 }
 373
 374 /*
 375  * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
 376  */
 377 static
 378 int create_thread_poll_set(struct lttng_poll_event *events, int size)
 379 {
 380         int ret;
 381
 382         if (events == NULL || size == 0) {
 383                 ret = -1;
 384                 goto error;
 385         }
 386
 387         ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
 388         if (ret < 0) {
 389                 goto error;
 390         }
 391
 392         /* Add quit pipe */
 393         ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN);
 394         if (ret < 0) {
 395                 goto error;
 396         }
 397
 398         return 0;
 399
 400 error:
 401         return ret;
 402 }
 403
 404 /*
 405  * Check if the thread quit pipe was triggered.
 406  *
 407  * Return 1 if it was triggered else 0;
 408  */
 409 static
 410 int check_thread_quit_pipe(int fd, uint32_t events)
 411 {
 412         if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
 413                 return 1;
 414         }
 415
 416         return 0;
 417 }
 418
 419 /*
 420  * Create and init socket from uri.
 421  */
 422 static
 423 struct lttcomm_sock *relay_init_sock(struct lttng_uri *uri)
 424 {
 425         int ret;
 426         struct lttcomm_sock *sock = NULL;
 427
 428         sock = lttcomm_alloc_sock_from_uri(uri);
 429         if (sock == NULL) {
 430                 ERR("Allocating socket");
 431                 goto error;
 432         }
 433
 434         ret = lttcomm_create_sock(sock);
 435         if (ret < 0) {
 436                 goto error;
 437         }
 438         DBG("Listening on sock %d", sock->fd);
 439
 440         ret = sock->ops->bind(sock);
 441         if (ret < 0) {
 442                 goto error;
 443         }
 444
 445         ret = sock->ops->listen(sock, -1);
 446         if (ret < 0) {
 447                 goto error;
 448
 449         }
 450
 451         return sock;
 452
 453 error:
 454         if (sock) {
 455                 lttcomm_destroy_sock(sock);
 456         }
 457         return NULL;
 458 }
 459
 460 /*
 461  * This thread manages the listening for new connections on the network
 462  */
 463 static
 464 void *relay_thread_listener(void *data)
 465 {
 466         int i, ret, pollfd;
 467         int val = 1;
 468         uint32_t revents, nb_fd;
 469         struct lttng_poll_event events;
 470         struct lttcomm_sock *control_sock, *data_sock;
 471
 472         /*
 473          * Get allocated in this thread, enqueued to a global queue, dequeued and
 474          * freed in the worker thread.
 475          */
 476         struct relay_command *relay_cmd = NULL;
 477
 478         DBG("[thread] Relay listener started");
 479
 480         control_sock = relay_init_sock(control_uri);
 481         if (!control_sock) {
 482                 goto error_sock;
 483         }
 484
 485         data_sock = relay_init_sock(data_uri);
 486         if (!data_sock) {
 487                 goto error_sock;
 488         }
 489
 490         /*
 491          * Pass 3 as size here for the thread quit pipe, control and data socket.
 492          */
 493         ret = create_thread_poll_set(&events, 3);
 494         if (ret < 0) {
 495                 goto error_create_poll;
 496         }
 497
 498         /* Add the control socket */
 499         ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
 500         if (ret < 0) {
 501                 goto error_poll_add;
 502         }
 503
 504         /* Add the data socket */
 505         ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
 506         if (ret < 0) {
 507                 goto error_poll_add;
 508         }
 509
 510         while (1) {
 511                 DBG("Listener accepting connections");
 512
 513                 nb_fd = LTTNG_POLL_GETNB(&events);
 514
 515 restart:
 516                 ret = lttng_poll_wait(&events, -1);
 517                 if (ret < 0) {
 518                         /*
 519                          * Restart interrupted system call.
 520                          */
 521                         if (errno == EINTR) {
 522                                 goto restart;
 523                         }
 524                         goto error;
 525                 }
 526
 527                 DBG("Relay new connection received");
 528                 for (i = 0; i < nb_fd; i++) {
 529                         /* Fetch once the poll data */
 530                         revents = LTTNG_POLL_GETEV(&events, i);
 531                         pollfd = LTTNG_POLL_GETFD(&events, i);
 532
 533                         /* Thread quit pipe has been closed. Killing thread. */
 534                         ret = check_thread_quit_pipe(pollfd, revents);
 535                         if (ret) {
 536                                 goto error;
 537                         }
 538
 539                         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 540                                 ERR("socket poll error");
 541                                 goto error;
 542                         } else if (revents & LPOLLIN) {
 543                                 struct lttcomm_sock *newsock = NULL;
 544
 545                                 relay_cmd = zmalloc(sizeof(struct relay_command));
 546                                 if (relay_cmd == NULL) {
 547                                         PERROR("relay command zmalloc");
 548                                         goto error;
 549                                 }
 550
 551                                 if (pollfd == data_sock->fd) {
 552                                         newsock = data_sock->ops->accept(data_sock);
 553                                         if (newsock < 0) {
 554                                                 PERROR("accepting data sock");
 555                                                 goto error;
 556                                         }
 557                                         relay_cmd->type = RELAY_DATA;
 558                                         DBG("Relay data connection accepted, socket %d", newsock->fd);
 559                                 } else if (pollfd == control_sock->fd) {
 560                                         newsock = control_sock->ops->accept(control_sock);
 561                                         if (newsock < 0) {
 562                                                 PERROR("accepting control sock");
 563                                                 goto error;
 564                                         }
 565                                         relay_cmd->type = RELAY_CONTROL;
 566                                         DBG("Relay control connection accepted, socket %d", newsock->fd);
 567                                 }
 568                                 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR,
 569                                                 &val, sizeof(int));
 570                                 if (ret < 0) {
 571                                         PERROR("setsockopt inet");
 572                                         goto error;
 573                                 }
 574                                 relay_cmd->sock = newsock;
 575                                 /*
 576                                  * Lock free enqueue the request.
 577                                  */
 578                                 cds_wfq_enqueue(&relay_cmd_queue.queue, &relay_cmd->node);
 579
 580                                 /*
 581                                  * Wake the dispatch queue futex. Implicit memory
 582                                  * barrier with the exchange in cds_wfq_enqueue.
 583                                  */
 584                                 futex_nto1_wake(&relay_cmd_queue.futex);
 585                         }
 586                 }
 587         }
 588
 589 error:
 590 error_poll_add:
 591         lttng_poll_clean(&events);
 592 error_create_poll:
 593         if (control_sock->fd >= 0) {
 594                 ret = control_sock->ops->close(control_sock);
 595                 if (ret) {
 596                         PERROR("close");
 597                 }
 598                 lttcomm_destroy_sock(control_sock);
 599         }
 600         if (data_sock->fd >= 0) {
 601                 ret = data_sock->ops->close(data_sock);
 602                 if (ret) {
 603                         PERROR("close");
 604                 }
 605                 lttcomm_destroy_sock(data_sock);
 606         }
 607
 608         DBG("Relay listener thread cleanup complete");
 609         stop_threads();
 610 error_sock:
 611         return NULL;
 612 }
 613
 614 /*
 615  * This thread manages the dispatching of the requests to worker threads
 616  */
 617 static
 618 void *relay_thread_dispatcher(void *data)
 619 {
 620         int ret;
 621         struct cds_wfq_node *node;
 622         struct relay_command *relay_cmd = NULL;
 623
 624         DBG("[thread] Relay dispatcher started");
 625
 626         while (!dispatch_thread_exit) {
 627                 /* Atomically prepare the queue futex */
 628                 futex_nto1_prepare(&relay_cmd_queue.futex);
 629
 630                 do {
 631                         /* Dequeue commands */
 632                         node = cds_wfq_dequeue_blocking(&relay_cmd_queue.queue);
 633                         if (node == NULL) {
 634                                 DBG("Woken up but nothing in the relay command queue");
 635                                 /* Continue thread execution */
 636                                 break;
 637                         }
 638
 639                         relay_cmd = caa_container_of(node, struct relay_command, node);
 640                         DBG("Dispatching request waiting on sock %d", relay_cmd->sock->fd);
 641
 642                         /*
 643                          * Inform worker thread of the new request. This
 644                          * call is blocking so we can be assured that the data will be read
 645                          * at some point in time or wait to the end of the world :)
 646                          */
 647                         ret = write(relay_cmd_pipe[1], relay_cmd,
 648                                         sizeof(struct relay_command));
 649                         free(relay_cmd);
 650                         if (ret < 0) {
 651                                 PERROR("write cmd pipe");
 652                                 goto error;
 653                         }
 654                 } while (node != NULL);
 655
 656                 /* Futex wait on queue. Blocking call on futex() */
 657                 futex_nto1_wait(&relay_cmd_queue.futex);
 658         }
 659
 660 error:
 661         DBG("Dispatch thread dying");
 662         stop_threads();
 663         return NULL;
 664 }
 665
 666 /*
 667  * Return the realpath(3) of the path even if the last directory token does not
 668  * exist. For example, with /tmp/test1/test2, if test2/ does not exist but the
 669  * /tmp/test1 does, the real path is returned. In normal time, realpath(3)
 670  * fails if the end point directory does not exist.
 671  */
 672 static
 673 char *expand_full_path(const char *path)
 674 {
 675         const char *end_path = path;
 676         char *next, *cut_path, *expanded_path;
 677
 678         /* Find last token delimited by '/' */
 679         while ((next = strpbrk(end_path + 1, "/"))) {
 680                 end_path = next;
 681         }
 682
 683         /* Cut last token from original path */
 684         cut_path = strndup(path, end_path - path);
 685
 686         expanded_path = malloc(PATH_MAX);
 687         if (expanded_path == NULL) {
 688                 goto error;
 689         }
 690
 691         expanded_path = realpath((char *)cut_path, expanded_path);
 692         if (expanded_path == NULL) {
 693                 switch (errno) {
 694                 case ENOENT:
 695                         ERR("%s: No such file or directory", cut_path);
 696                         break;
 697                 default:
 698                         PERROR("realpath");
 699                         break;
 700                 }
 701                 goto error;
 702         }
 703
 704         /* Add end part to expanded path */
 705         strcat(expanded_path, end_path);
 706
 707         free(cut_path);
 708         return expanded_path;
 709
 710 error:
 711         free(cut_path);
 712         return NULL;
 713 }
 714
 715
 716 /*
 717  *  config_get_default_path
 718  *
 719  *  Returns the HOME directory path. Caller MUST NOT free(3) the return pointer.
 720  */
 721 static
 722 char *config_get_default_path(void)
 723 {
 724         return getenv("HOME");
 725 }
 726
 727 /*
 728  * Create recursively directory using the FULL path.
 729  */
 730 static
 731 int mkdir_recursive(char *path, mode_t mode)
 732 {
 733         char *p, tmp[PATH_MAX];
 734         struct stat statbuf;
 735         size_t len;
 736         int ret;
 737
 738         ret = snprintf(tmp, sizeof(tmp), "%s", path);
 739         if (ret < 0) {
 740                 PERROR("snprintf mkdir");
 741                 goto error;
 742         }
 743
 744         len = ret;
 745         if (tmp[len - 1] == '/') {
 746                 tmp[len - 1] = 0;
 747         }
 748
 749         for (p = tmp + 1; *p; p++) {
 750                 if (*p == '/') {
 751                         *p = 0;
 752                         if (tmp[strlen(tmp) - 1] == '.' &&
 753                                         tmp[strlen(tmp) - 2] == '.' &&
 754                                         tmp[strlen(tmp) - 3] == '/') {
 755                                 ERR("Using '/../' is not permitted in the trace path (%s)",
 756                                                 tmp);
 757                                 ret = -1;
 758                                 goto error;
 759                         }
 760                         ret = stat(tmp, &statbuf);
 761                         if (ret < 0) {
 762                                 ret = mkdir(tmp, mode);
 763                                 if (ret < 0) {
 764                                         if (!(errno == EEXIST)) {
 765                                                 PERROR("mkdir recursive");
 766                                                 ret = -errno;
 767                                                 goto error;
 768                                         }
 769                                 }
 770                         }
 771                         *p = '/';
 772                 }
 773         }
 774
 775         ret = mkdir(tmp, mode);
 776         if (ret < 0) {
 777                 if (!(errno == EEXIST)) {
 778                         PERROR("mkdir recursive last piece");
 779                         ret = -errno;
 780                 } else {
 781                         ret = 0;
 782                 }
 783         }
 784
 785 error:
 786         return ret;
 787 }
 788
 789 /*
 790  * create_output_path: create the output trace directory
 791  */
 792 static
 793 char *create_output_path(char *path_name)
 794 {
 795         int ret = 0;
 796         char *alloc_path = NULL;
 797         char *traces_path = NULL;
 798         char *full_path = NULL;
 799
 800         /* Auto output path */
 801         if (opt_output_path == NULL) {
 802                 alloc_path = strdup(config_get_default_path());
 803                 if (alloc_path == NULL) {
 804                         ERR("Home path not found.\n \
 805                                         Please specify an output path using -o, --output PATH");
 806                         ret = -1;
 807                         goto exit;
 808                 }
 809
 810                 ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
 811                                 "/%s", alloc_path, path_name);
 812                 if (ret < 0) {
 813                         PERROR("asprintf trace dir name");
 814                         goto exit;
 815                 }
 816         } else {
 817                 full_path = expand_full_path(opt_output_path);
 818                 ret = asprintf(&traces_path, "%s/%s", full_path, path_name);
 819                 if (ret < 0) {
 820                         PERROR("asprintf trace dir name");
 821                         goto exit;
 822                 }
 823         }
 824         free(alloc_path);
 825         free(full_path);
 826
 827 exit:
 828         return traces_path;
 829 }
 830
 831 /*
 832  * relay_delete_session: Free all memory associated with a session and
 833  * close all the FDs
 834  */
 835 static
 836 void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht)
 837 {
 838         struct lttng_ht_iter iter;
 839         struct lttng_ht_node_ulong *node;
 840         struct relay_stream *stream;
 841         int ret;
 842
 843         if (!cmd->session)
 844                 return;
 845
 846         DBG("Relay deleting session %lu", cmd->session->id);
 847         free(cmd->session->sock);
 848
 849         cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) {
 850                 node = lttng_ht_iter_get_node_ulong(&iter);
 851                 if (node) {
 852                         stream = caa_container_of(node,
 853                                         struct relay_stream, stream_n);
 854                         if (stream->session == cmd->session) {
 855                                 close(stream->fd);
 856                                 ret = lttng_ht_del(streams_ht, &iter);
 857                                 assert(!ret);
 858                                 free(stream);
 859                         }
 860                 }
 861         }
 862 }
 863
 864 /*
 865  * relay_add_stream: allocate a new stream for a session
 866  */
 867 static
 868 int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
 869                 struct relay_command *cmd, struct lttng_ht *streams_ht)
 870 {
 871         struct relay_session *session = cmd->session;
 872         struct lttcomm_relayd_add_stream stream_info;
 873         struct relay_stream *stream = NULL;
 874         struct lttcomm_relayd_status_stream reply;
 875         char *path = NULL, *root_path = NULL;
 876         int ret, send_ret;
 877
 878         if (!session || session->version_check_done == 0) {
 879                 ERR("Trying to add a stream before version check");
 880                 ret = -1;
 881                 goto end_no_session;
 882         }
 883
 884         /* FIXME : use data_size for something ? */
 885         ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
 886                         sizeof(struct lttcomm_relayd_add_stream), MSG_WAITALL);
 887         if (ret < sizeof(struct lttcomm_relayd_add_stream)) {
 888                 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
 889                 ret = -1;
 890                 goto end_no_session;
 891         }
 892         stream = zmalloc(sizeof(struct relay_stream));
 893         if (stream == NULL) {
 894                 PERROR("relay stream zmalloc");
 895                 ret = -1;
 896                 goto end_no_session;
 897         }
 898
 899         stream->stream_handle = ++last_relay_stream_id;
 900         stream->seq = 0;
 901         stream->session = session;
 902
 903         root_path = create_output_path(stream_info.pathname);
 904         if (!root_path) {
 905                 ret = -1;
 906                 goto end;
 907         }
 908         ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG);
 909         if (ret < 0) {
 910                 free(root_path);
 911                 ERR("relay creating output directory");
 912                 goto end;
 913         }
 914
 915         ret = asprintf(&path, "%s/%s", root_path, stream_info.channel_name);
 916         if (ret < 0) {
 917                 PERROR("asprintf stream path");
 918                 goto end;
 919         }
 920
 921         ret = open(path, O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
 922         if (ret < 0) {
 923                 PERROR("Relay creating trace file");
 924                 goto end;
 925         }
 926
 927         stream->fd = ret;
 928         DBG("Tracefile %s created", path);
 929
 930         lttng_ht_node_init_ulong(&stream->stream_n,
 931                         (unsigned long) stream->stream_handle);
 932         lttng_ht_add_unique_ulong(streams_ht,
 933                         &stream->stream_n);
 934
 935         DBG("Relay new stream added %s", stream_info.channel_name);
 936
 937 end:
 938         free(path);
 939         free(root_path);
 940         /* send the session id to the client or a negative return code on error */
 941         if (ret < 0) {
 942                 reply.ret_code = htobe32(LTTCOMM_ERR);
 943         } else {
 944                 reply.ret_code = htobe32(LTTCOMM_OK);
 945         }
 946         reply.handle = htobe64(stream->stream_handle);
 947         send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
 948                         sizeof(struct lttcomm_relayd_status_stream), 0);
 949         if (send_ret < 0) {
 950                 ERR("Relay sending stream id");
 951         }
 952
 953 end_no_session:
 954         return ret;
 955 }
 956
 957 /*
 958  * relay_unknown_command: send -1 if received unknown command
 959  */
 960 static
 961 void relay_unknown_command(struct relay_command *cmd)
 962 {
 963         struct lttcomm_relayd_generic_reply reply;
 964         int ret;
 965
 966         reply.ret_code = htobe32(LTTCOMM_ERR);
 967         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
 968                         sizeof(struct lttcomm_relayd_generic_reply), 0);
 969         if (ret < 0) {
 970                 ERR("Relay sending unknown command");
 971         }
 972 }
 973
 974 /*
 975  * relay_start: send an acknowledgment to the client to tell if we are
 976  * ready to receive data. We are ready if a session is established.
 977  */
 978 static
 979 int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
 980                 struct relay_command *cmd)
 981 {
 982         int ret = htobe32(LTTCOMM_OK);
 983         struct lttcomm_relayd_generic_reply reply;
 984         struct relay_session *session = cmd->session;
 985
 986         if (!session) {
 987                 DBG("Trying to start the streaming without a session established");
 988                 ret = htobe32(LTTCOMM_ERR);
 989         }
 990
 991         reply.ret_code = ret;
 992         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
 993                         sizeof(struct lttcomm_relayd_generic_reply), 0);
 994         if (ret < 0) {
 995                 ERR("Relay sending start ack");
 996         }
 997
 998         return ret;
 999 }
1000
1001 /*
1002  * Get stream from stream id.
1003  */
1004 static
1005 struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id,
1006                 struct lttng_ht *streams_ht)
1007 {
1008         struct lttng_ht_node_ulong *node;
1009         struct lttng_ht_iter iter;
1010         struct relay_stream *ret;
1011
1012         lttng_ht_lookup(streams_ht,
1013                         (void *)((unsigned long) stream_id),
1014                         &iter);
1015         node = lttng_ht_iter_get_node_ulong(&iter);
1016         if (node == NULL) {
1017                 DBG("Relay stream %lu not found", stream_id);
1018                 ret = NULL;
1019                 goto end;
1020         }
1021
1022         ret = caa_container_of(node, struct relay_stream, stream_n);
1023
1024 end:
1025         return ret;
1026 }
1027
1028 /*
1029  * relay_recv_metadata: receive the metada for the session.
1030  */
1031 static
1032 int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1033                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1034 {
1035         int ret = htobe32(LTTCOMM_OK);
1036         struct relay_session *session = cmd->session;
1037         struct lttcomm_relayd_metadata_payload *metadata_struct;
1038         struct relay_stream *metadata_stream;
1039         uint64_t data_size, payload_size;
1040
1041         if (!session) {
1042                 ERR("Metadata sent before version check");
1043                 ret = -1;
1044                 goto end;
1045         }
1046
1047         data_size = be64toh(recv_hdr->data_size);
1048         payload_size = data_size - sizeof(uint64_t);
1049         if (data_buffer_size < data_size) {
1050                 data_buffer = realloc(data_buffer, data_size);
1051                 if (!data_buffer) {
1052                         ERR("Allocating data buffer");
1053                         ret = -1;
1054                         goto end;
1055                 }
1056                 data_buffer_size = data_size;
1057         }
1058         memset(data_buffer, 0, data_size);
1059         DBG2("Relay receiving metadata, waiting for %lu bytes", data_size);
1060         ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
1061         if (ret < 0 || ret != data_size) {
1062                 ret = -1;
1063                 ERR("Relay didn't receive the whole metadata");
1064                 goto end;
1065         }
1066         metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
1067         metadata_stream = relay_stream_from_stream_id(
1068                         be64toh(metadata_struct->stream_id), streams_ht);
1069         if (!metadata_stream) {
1070                 ret = -1;
1071                 goto end;
1072         }
1073
1074         ret = write(metadata_stream->fd, metadata_struct->payload,
1075                         payload_size);
1076         if (ret < (payload_size)) {
1077                 ERR("Relay error writing metadata on file");
1078                 ret = -1;
1079                 goto end;
1080         }
1081         DBG2("Relay metadata written");
1082
1083 end:
1084         return ret;
1085 }
1086
1087 /*
1088  * relay_send_version: send relayd version number
1089  */
1090 static
1091 int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1092                 struct relay_command *cmd)
1093 {
1094         int ret = htobe32(LTTCOMM_OK);
1095         struct lttcomm_relayd_version reply;
1096         struct relay_session *session = NULL;
1097
1098         if (cmd->session == NULL) {
1099                 session = zmalloc(sizeof(struct relay_session));
1100                 if (session == NULL) {
1101                         PERROR("relay session zmalloc");
1102                         ret = -1;
1103                         goto end;
1104                 }
1105                 session->id = ++last_relay_session_id;
1106                 DBG("Created session %lu", session->id);
1107                 cmd->session = session;
1108         }
1109         session->version_check_done = 1;
1110
1111         sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
1112         reply.major = htobe32(reply.major);
1113         reply.minor = htobe32(reply.minor);
1114         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1115                         sizeof(struct lttcomm_relayd_version), 0);
1116         if (ret < 0) {
1117                 ERR("Relay sending version");
1118         }
1119         DBG("Version check done");
1120
1121 end:
1122         return ret;
1123 }
1124
1125 /*
1126  * relay_process_control: Process the commands received on the control socket
1127  */
1128 static
1129 int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1130                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1131 {
1132         int ret = 0;
1133
1134         switch (be32toh(recv_hdr->cmd)) {
1135                 /*
1136         case RELAYD_CREATE_SESSION:
1137                 ret = relay_create_session(recv_hdr, cmd);
1138                 break;
1139                 */
1140         case RELAYD_ADD_STREAM:
1141                 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1142                 break;
1143         case RELAYD_START_DATA:
1144                 ret = relay_start(recv_hdr, cmd);
1145                 break;
1146         case RELAYD_SEND_METADATA:
1147                 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1148                 break;
1149         case RELAYD_VERSION:
1150                 ret = relay_send_version(recv_hdr, cmd);
1151                 break;
1152         case RELAYD_UPDATE_SYNC_INFO:
1153         default:
1154                 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1155                 relay_unknown_command(cmd);
1156                 ret = -1;
1157                 goto end;
1158         }
1159
1160 end:
1161         return ret;
1162 }
1163
1164 /*
1165  * relay_process_data: Process the data received on the data socket
1166  */
1167 static
1168 int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1169 {
1170         int ret = 0;
1171         struct relay_stream *stream;
1172         struct lttcomm_relayd_data_hdr data_hdr;
1173         uint64_t stream_id;
1174         uint32_t data_size;
1175
1176         ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1177                         sizeof(struct lttcomm_relayd_data_hdr), MSG_WAITALL);
1178         if (ret <= 0) {
1179                 ERR("Connections seems to be closed");
1180                 ret = -1;
1181                 goto end;
1182         }
1183
1184         stream_id = be64toh(data_hdr.stream_id);
1185         stream = relay_stream_from_stream_id(stream_id, streams_ht);
1186         if (!stream) {
1187                 ret = -1;
1188                 goto end;
1189         }
1190
1191         data_size = be32toh(data_hdr.data_size);
1192         if (data_buffer_size < data_size) {
1193                 data_buffer = realloc(data_buffer, data_size);
1194                 if (!data_buffer) {
1195                         ERR("Allocating data buffer");
1196                         ret = -1;
1197                         goto end;
1198                 }
1199                 data_buffer_size = data_size;
1200         }
1201         memset(data_buffer, 0, data_size);
1202
1203         ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
1204         if (ret <= 0) {
1205                 ret = -1;
1206                 goto end;
1207         }
1208
1209         ret = write(stream->fd, data_buffer, data_size);
1210         if (ret < data_size) {
1211                 ERR("Relay error writing data to file");
1212                 ret = -1;
1213                 goto end;
1214         }
1215         DBG2("Relay wrote %d bytes to tracefile for stream id %lu", ret, stream->stream_handle);
1216
1217 end:
1218         return ret;
1219 }
1220
1221 static
1222 void relay_cleanup_connection(struct lttng_ht *relay_connections_ht, struct lttng_poll_event *events,
1223                 struct lttng_ht *streams_ht, int pollfd, struct lttng_ht_iter *iter)
1224 {
1225         int ret;
1226
1227         ret = lttng_ht_del(relay_connections_ht, iter);
1228         assert(!ret);
1229         lttng_poll_del(events, pollfd);
1230
1231         ret = close(pollfd);
1232         if (ret < 0) {
1233                 ERR("Closing pollfd %d", pollfd);
1234         }
1235 }
1236
1237 static
1238 int relay_add_connection(int fd, struct lttng_poll_event *events,
1239                 struct lttng_ht *relay_connections_ht)
1240 {
1241         int ret;
1242         struct relay_command *relay_connection;
1243
1244         relay_connection = zmalloc(sizeof(struct relay_command));
1245         if (relay_connection == NULL) {
1246                 PERROR("Relay command zmalloc");
1247                 ret = -1;
1248                 goto end;
1249         }
1250         ret = read(fd, relay_connection, sizeof(struct relay_command));
1251         if (ret < 0 || ret < sizeof(relay_connection)) {
1252                 PERROR("read relay cmd pipe");
1253                 ret = -1;
1254                 goto end;
1255         }
1256
1257         lttng_ht_node_init_ulong(&relay_connection->sock_n,
1258                         (unsigned long) relay_connection->sock->fd);
1259         lttng_ht_add_unique_ulong(relay_connections_ht,
1260                         &relay_connection->sock_n);
1261         ret = lttng_poll_add(events,
1262                         relay_connection->sock->fd,
1263                         LPOLLIN | LPOLLRDHUP);
1264
1265 end:
1266         return ret;
1267 }
1268
1269 /*
1270  * This thread does the actual work
1271  */
1272 static
1273 void *relay_thread_worker(void *data)
1274 {
1275         int i, ret, pollfd;
1276         uint32_t revents, nb_fd;
1277         struct relay_command *relay_connection;
1278         struct lttng_poll_event events;
1279         struct lttng_ht *relay_connections_ht;
1280         struct lttng_ht_node_ulong *node;
1281         struct lttng_ht_iter iter;
1282         struct lttng_ht *streams_ht;
1283         struct lttcomm_relayd_hdr recv_hdr;
1284
1285         DBG("[thread] Relay worker started");
1286
1287         /* table of connections indexed on socket */
1288         relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1289
1290         /* tables of streams indexed by stream ID */
1291         streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1292
1293         ret = create_thread_poll_set(&events, 2);
1294         if (ret < 0) {
1295                 goto error_poll_create;
1296         }
1297
1298         ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1299         if (ret < 0) {
1300                 goto error;
1301         }
1302
1303         while (1) {
1304                 /* Zeroed the events structure */
1305                 lttng_poll_reset(&events);
1306
1307                 nb_fd = LTTNG_POLL_GETNB(&events);
1308
1309                 /* Infinite blocking call, waiting for transmission */
1310         restart:
1311                 ret = lttng_poll_wait(&events, -1);
1312                 if (ret < 0) {
1313                         /*
1314                          * Restart interrupted system call.
1315                          */
1316                         if (errno == EINTR) {
1317                                 goto restart;
1318                         }
1319                         goto error;
1320                 }
1321
1322                 for (i = 0; i < nb_fd; i++) {
1323                         /* Fetch once the poll data */
1324                         revents = LTTNG_POLL_GETEV(&events, i);
1325                         pollfd = LTTNG_POLL_GETFD(&events, i);
1326
1327                         /* Thread quit pipe has been closed. Killing thread. */
1328                         ret = check_thread_quit_pipe(pollfd, revents);
1329                         if (ret) {
1330                                 goto error;
1331                         }
1332
1333                         /* Inspect the relay cmd pipe for new connection */
1334                         if (pollfd == relay_cmd_pipe[0]) {
1335                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1336                                         ERR("Relay pipe error");
1337                                         goto error;
1338                                 } else if (revents & LPOLLIN) {
1339                                         DBG("Relay command received");
1340                                         ret = relay_add_connection(relay_cmd_pipe[0],
1341                                                         &events, relay_connections_ht);
1342                                         if (ret < 0) {
1343                                                 goto error;
1344                                         }
1345                                 }
1346                         } else if (revents > 0) {
1347                                 lttng_ht_lookup(relay_connections_ht,
1348                                                 (void *)((unsigned long) pollfd),
1349                                                 &iter);
1350                                 node = lttng_ht_iter_get_node_ulong(&iter);
1351                                 if (node == NULL) {
1352                                         DBG2("Relay sock %d not found", pollfd);
1353                                         goto error;
1354                                 }
1355                                 relay_connection = caa_container_of(node,
1356                                                 struct relay_command, sock_n);
1357
1358                                 if (revents & (LPOLLERR)) {
1359                                         ERR("POLL ERROR");
1360                                         relay_cleanup_connection(relay_connections_ht,
1361                                                         &events, streams_ht, pollfd, &iter);
1362                                         free(relay_connection);
1363                                 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1364                                         DBG("Socket %d hung up", pollfd);
1365                                         relay_cleanup_connection(relay_connections_ht,
1366                                                         &events, streams_ht, pollfd, &iter);
1367                                         if (relay_connection->type == RELAY_CONTROL) {
1368                                                 relay_delete_session(relay_connection, streams_ht);
1369                                         }
1370                                         free(relay_connection);
1371                                 } else if (revents & LPOLLIN) {
1372                                         /* control socket */
1373                                         if (relay_connection->type == RELAY_CONTROL) {
1374                                                 ret = relay_connection->sock->ops->recvmsg(
1375                                                                 relay_connection->sock, &recv_hdr,
1376                                                                 sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL);
1377                                                 /* connection closed */
1378                                                 if (ret <= 0) {
1379                                                         relay_cleanup_connection(relay_connections_ht,
1380                                                                         &events, streams_ht, pollfd, &iter);
1381                                                         relay_delete_session(relay_connection, streams_ht);
1382                                                         free(relay_connection);
1383                                                         DBG("Control connection closed with %d", pollfd);
1384                                                 } else {
1385                                                         if (relay_connection->session) {
1386                                                                 DBG2("Relay worker receiving data for session : %lu",
1387                                                                                 relay_connection->session->id);
1388                                                         }
1389                                                         ret = relay_process_control(&recv_hdr,
1390                                                                         relay_connection,
1391                                                                         streams_ht);
1392                                                         /*
1393                                                          * there was an error in processing a control
1394                                                          * command: clear the session
1395                                                          * */
1396                                                         if (ret < 0) {
1397                                                                 relay_cleanup_connection(relay_connections_ht,
1398                                                                                 &events, streams_ht, pollfd, &iter);
1399                                                                 free(relay_connection);
1400                                                                 DBG("Connection closed with %d", pollfd);
1401                                                         }
1402                                                 }
1403                                                 /* data socket */
1404                                         } else if (relay_connection->type == RELAY_DATA) {
1405                                                 ret = relay_process_data(relay_connection, streams_ht);
1406                                                 /* connection closed */
1407                                                 if (ret < 0) {
1408                                                         relay_cleanup_connection(relay_connections_ht,
1409                                                                         &events, streams_ht, pollfd, &iter);
1410                                                         relay_delete_session(relay_connection, streams_ht);
1411                                                         DBG("Data connection closed with %d", pollfd);
1412                                                 }
1413                                         }
1414                                 }
1415                         }
1416                 }
1417         }
1418
1419 error:
1420         lttng_poll_clean(&events);
1421
1422         /* empty the hash table and free the memory */
1423         cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1424                 node = lttng_ht_iter_get_node_ulong(&iter);
1425                 if (node) {
1426                         relay_connection = caa_container_of(node,
1427                                         struct relay_command, sock_n);
1428                         free(relay_connection);
1429                 }
1430                 ret = lttng_ht_del(relay_connections_ht, &iter);
1431                 assert(!ret);
1432         }
1433 error_poll_create:
1434         free(data_buffer);
1435         lttng_ht_destroy(relay_connections_ht);
1436         DBG("Worker thread cleanup complete");
1437         stop_threads();
1438         return NULL;
1439 }
1440
1441 /*
1442  * Create the relay command pipe to wake thread_manage_apps.
1443  * Closed in cleanup().
1444  */
1445 static int create_relay_cmd_pipe(void)
1446 {
1447         int ret, i;
1448
1449         ret = pipe(relay_cmd_pipe);
1450         if (ret < 0) {
1451                 PERROR("relay cmd pipe");
1452                 goto error;
1453         }
1454
1455         for (i = 0; i < 2; i++) {
1456                 ret = fcntl(relay_cmd_pipe[i], F_SETFD, FD_CLOEXEC);
1457                 if (ret < 0) {
1458                         PERROR("fcntl relay_cmd_pipe");
1459                         goto error;
1460                 }
1461         }
1462
1463 error:
1464         return ret;
1465 }
1466
1467 /*
1468  * main
1469  */
1470 int main(int argc, char **argv)
1471 {
1472         int ret = 0;
1473         void *status;
1474
1475         /* Create thread quit pipe */
1476         if ((ret = init_thread_quit_pipe()) < 0) {
1477                 goto error;
1478         }
1479
1480         /* Parse arguments */
1481         progname = argv[0];
1482         if ((ret = parse_args(argc, argv) < 0)) {
1483                 goto error;
1484         }
1485
1486         if ((ret = set_signal_handler()) < 0) {
1487                 goto exit;
1488         }
1489
1490         /* Daemonize */
1491         if (opt_daemon) {
1492                 ret = daemon(0, 0);
1493                 if (ret < 0) {
1494                         PERROR("daemon");
1495                         goto error;
1496                 }
1497         }
1498
1499         /* Check if daemon is UID = 0 */
1500         is_root = !getuid();
1501
1502         if (!is_root) {
1503                 if (control_uri->port < 1024 || data_uri->port < 1024) {
1504                         ERR("Need to be root to use ports < 1024");
1505                         ret = -1;
1506                         goto error;
1507                 }
1508         }
1509
1510         /* Setup the thread apps communication pipe. */
1511         if ((ret = create_relay_cmd_pipe()) < 0) {
1512                 goto exit;
1513         }
1514
1515         /* Init relay command queue. */
1516         cds_wfq_init(&relay_cmd_queue.queue);
1517
1518         /* Set up max poll set size */
1519         lttng_poll_set_max_size();
1520
1521         /* Setup the dispatcher thread */
1522         ret = pthread_create(&dispatcher_thread, NULL,
1523                         relay_thread_dispatcher, (void *) NULL);
1524         if (ret != 0) {
1525                 PERROR("pthread_create dispatcher");
1526                 goto exit_dispatcher;
1527         }
1528
1529         /* Setup the worker thread */
1530         ret = pthread_create(&worker_thread, NULL,
1531                         relay_thread_worker, (void *) NULL);
1532         if (ret != 0) {
1533                 PERROR("pthread_create worker");
1534                 goto exit_worker;
1535         }
1536
1537         /* Setup the listener thread */
1538         ret = pthread_create(&listener_thread, NULL,
1539                         relay_thread_listener, (void *) NULL);
1540         if (ret != 0) {
1541                 PERROR("pthread_create listener");
1542                 goto exit_listener;
1543         }
1544
1545 exit_listener:
1546         ret = pthread_join(listener_thread, &status);
1547         if (ret != 0) {
1548                 PERROR("pthread_join");
1549                 goto error;     /* join error, exit without cleanup */
1550         }
1551
1552 exit_worker:
1553         ret = pthread_join(worker_thread, &status);
1554         if (ret != 0) {
1555                 PERROR("pthread_join");
1556                 goto error;     /* join error, exit without cleanup */
1557         }
1558
1559 exit_dispatcher:
1560         ret = pthread_join(dispatcher_thread, &status);
1561         if (ret != 0) {
1562                 PERROR("pthread_join");
1563                 goto error;     /* join error, exit without cleanup */
1564         }
1565
1566 exit:
1567         cleanup();
1568         if (!ret) {
1569                 exit(EXIT_SUCCESS);
1570         }
1571 error:
1572         exit(EXIT_FAILURE);
1573 }