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