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