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