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         uri_free(control_uri);
 243         uri_free(data_uri);
 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         DBG("[thread] Relay listener started");
 476
 477         control_sock = relay_init_sock(control_uri);
 478         if (!control_sock) {
 479                 goto error_sock_control;
 480         }
 481
 482         data_sock = relay_init_sock(data_uri);
 483         if (!data_sock) {
 484                 goto error_sock_relay;
 485         }
 486
 487         /*
 488          * Pass 3 as size here for the thread quit pipe, control and data socket.
 489          */
 490         ret = create_thread_poll_set(&events, 3);
 491         if (ret < 0) {
 492                 goto error_create_poll;
 493         }
 494
 495         /* Add the control socket */
 496         ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
 497         if (ret < 0) {
 498                 goto error_poll_add;
 499         }
 500
 501         /* Add the data socket */
 502         ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
 503         if (ret < 0) {
 504                 goto error_poll_add;
 505         }
 506
 507         while (1) {
 508                 DBG("Listener accepting connections");
 509
 510                 nb_fd = LTTNG_POLL_GETNB(&events);
 511
 512 restart:
 513                 ret = lttng_poll_wait(&events, -1);
 514                 if (ret < 0) {
 515                         /*
 516                          * Restart interrupted system call.
 517                          */
 518                         if (errno == EINTR) {
 519                                 goto restart;
 520                         }
 521                         goto error;
 522                 }
 523
 524                 DBG("Relay new connection received");
 525                 for (i = 0; i < nb_fd; i++) {
 526                         /* Fetch once the poll data */
 527                         revents = LTTNG_POLL_GETEV(&events, i);
 528                         pollfd = LTTNG_POLL_GETFD(&events, i);
 529
 530                         /* Thread quit pipe has been closed. Killing thread. */
 531                         ret = check_thread_quit_pipe(pollfd, revents);
 532                         if (ret) {
 533                                 err = 0;
 534                                 goto exit;
 535                         }
 536
 537                         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 538                                 ERR("socket poll error");
 539                                 goto error;
 540                         } else if (revents & LPOLLIN) {
 541                                 /*
 542                                  * Get allocated in this thread,
 543                                  * enqueued to a global queue, dequeued
 544                                  * and freed in the worker thread.
 545                                  */
 546                                 struct relay_command *relay_cmd;
 547                                 struct lttcomm_sock *newsock;
 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) {
 558                                                 PERROR("accepting data sock");
 559                                                 free(relay_cmd);
 560                                                 goto error;
 561                                         }
 562                                         relay_cmd->type = RELAY_DATA;
 563                                         DBG("Relay data connection accepted, socket %d", newsock->fd);
 564                                 } else {
 565                                         assert(pollfd == control_sock->fd);
 566                                         newsock = control_sock->ops->accept(control_sock);
 567                                         if (!newsock) {
 568                                                 PERROR("accepting control sock");
 569                                                 free(relay_cmd);
 570                                                 goto error;
 571                                         }
 572                                         relay_cmd->type = RELAY_CONTROL;
 573                                         DBG("Relay control connection accepted, socket %d", newsock->fd);
 574                                 }
 575                                 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR,
 576                                                 &val, sizeof(int));
 577                                 if (ret < 0) {
 578                                         PERROR("setsockopt inet");
 579                                         lttcomm_destroy_sock(newsock);
 580                                         free(relay_cmd);
 581                                         goto error;
 582                                 }
 583                                 relay_cmd->sock = newsock;
 584                                 /*
 585                                  * Lock free enqueue the request.
 586                                  */
 587                                 cds_wfq_enqueue(&relay_cmd_queue.queue, &relay_cmd->node);
 588
 589                                 /*
 590                                  * Wake the dispatch queue futex. Implicit memory
 591                                  * barrier with the exchange in cds_wfq_enqueue.
 592                                  */
 593                                 futex_nto1_wake(&relay_cmd_queue.futex);
 594                         }
 595                 }
 596         }
 597
 598 exit:
 599 error:
 600 error_poll_add:
 601         lttng_poll_clean(&events);
 602 error_create_poll:
 603         if (data_sock->fd >= 0) {
 604                 ret = data_sock->ops->close(data_sock);
 605                 if (ret) {
 606                         PERROR("close");
 607                 }
 608         }
 609         lttcomm_destroy_sock(data_sock);
 610 error_sock_relay:
 611         if (control_sock->fd >= 0) {
 612                 ret = control_sock->ops->close(control_sock);
 613                 if (ret) {
 614                         PERROR("close");
 615                 }
 616         }
 617         lttcomm_destroy_sock(control_sock);
 618 error_sock_control:
 619         if (err) {
 620                 DBG("Thread exited with error");
 621         }
 622         DBG("Relay listener thread cleanup complete");
 623         stop_threads();
 624         return NULL;
 625 }
 626
 627 /*
 628  * This thread manages the dispatching of the requests to worker threads
 629  */
 630 static
 631 void *relay_thread_dispatcher(void *data)
 632 {
 633         int ret;
 634         struct cds_wfq_node *node;
 635         struct relay_command *relay_cmd = NULL;
 636
 637         DBG("[thread] Relay dispatcher started");
 638
 639         while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
 640                 /* Atomically prepare the queue futex */
 641                 futex_nto1_prepare(&relay_cmd_queue.futex);
 642
 643                 do {
 644                         /* Dequeue commands */
 645                         node = cds_wfq_dequeue_blocking(&relay_cmd_queue.queue);
 646                         if (node == NULL) {
 647                                 DBG("Woken up but nothing in the relay command queue");
 648                                 /* Continue thread execution */
 649                                 break;
 650                         }
 651
 652                         relay_cmd = caa_container_of(node, struct relay_command, node);
 653                         DBG("Dispatching request waiting on sock %d", relay_cmd->sock->fd);
 654
 655                         /*
 656                          * Inform worker thread of the new request. This
 657                          * call is blocking so we can be assured that the data will be read
 658                          * at some point in time or wait to the end of the world :)
 659                          */
 660                         do {
 661                                 ret = write(relay_cmd_pipe[1], relay_cmd,
 662                                                 sizeof(struct relay_command));
 663                         } while (ret < 0 && errno == EINTR);
 664                         free(relay_cmd);
 665                         if (ret < 0) {
 666                                 PERROR("write cmd pipe");
 667                                 goto error;
 668                         }
 669                 } while (node != NULL);
 670
 671                 /* Futex wait on queue. Blocking call on futex() */
 672                 futex_nto1_wait(&relay_cmd_queue.futex);
 673         }
 674
 675 error:
 676         DBG("Dispatch thread dying");
 677         stop_threads();
 678         return NULL;
 679 }
 680
 681 /*
 682  * Return the realpath(3) of the path even if the last directory token does not
 683  * exist. For example, with /tmp/test1/test2, if test2/ does not exist but the
 684  * /tmp/test1 does, the real path is returned. In normal time, realpath(3)
 685  * fails if the end point directory does not exist.
 686  */
 687 static
 688 char *expand_full_path(const char *path)
 689 {
 690         const char *end_path = path;
 691         char *next, *cut_path, *expanded_path, *respath;
 692
 693         /* Find last token delimited by '/' */
 694         while ((next = strpbrk(end_path + 1, "/"))) {
 695                 end_path = next;
 696         }
 697
 698         /* Cut last token from original path */
 699         cut_path = strndup(path, end_path - path);
 700
 701         expanded_path = malloc(PATH_MAX);
 702         if (expanded_path == NULL) {
 703                 respath = NULL;
 704                 goto end;
 705         }
 706
 707         respath = realpath(cut_path, expanded_path);
 708         if (respath == NULL) {
 709                 switch (errno) {
 710                 case ENOENT:
 711                         ERR("%s: No such file or directory", cut_path);
 712                         break;
 713                 default:
 714                         PERROR("realpath");
 715                         break;
 716                 }
 717                 free(expanded_path);
 718         } else {
 719                 /* Add end part to expanded path */
 720                 strcat(respath, end_path);
 721         }
 722 end:
 723         free(cut_path);
 724         return respath;
 725 }
 726
 727
 728 /*
 729  *  config_get_default_path
 730  *
 731  *  Returns the HOME directory path. Caller MUST NOT free(3) the return pointer.
 732  */
 733 static
 734 char *config_get_default_path(void)
 735 {
 736         return getenv("HOME");
 737 }
 738
 739 /*
 740  * Create recursively directory using the FULL path.
 741  */
 742 static
 743 int mkdir_recursive(char *path, mode_t mode)
 744 {
 745         char *p, tmp[PATH_MAX];
 746         struct stat statbuf;
 747         size_t len;
 748         int ret;
 749
 750         ret = snprintf(tmp, sizeof(tmp), "%s", path);
 751         if (ret < 0) {
 752                 PERROR("snprintf mkdir");
 753                 goto error;
 754         }
 755
 756         len = ret;
 757         if (tmp[len - 1] == '/') {
 758                 tmp[len - 1] = 0;
 759         }
 760
 761         for (p = tmp + 1; *p; p++) {
 762                 if (*p == '/') {
 763                         *p = 0;
 764                         if (tmp[strlen(tmp) - 1] == '.' &&
 765                                         tmp[strlen(tmp) - 2] == '.' &&
 766                                         tmp[strlen(tmp) - 3] == '/') {
 767                                 ERR("Using '/../' is not permitted in the trace path (%s)",
 768                                                 tmp);
 769                                 ret = -1;
 770                                 goto error;
 771                         }
 772                         ret = stat(tmp, &statbuf);
 773                         if (ret < 0) {
 774                                 ret = mkdir(tmp, mode);
 775                                 if (ret < 0) {
 776                                         if (errno != EEXIST) {
 777                                                 PERROR("mkdir recursive");
 778                                                 ret = -errno;
 779                                                 goto error;
 780                                         }
 781                                 }
 782                         }
 783                         *p = '/';
 784                 }
 785         }
 786
 787         ret = mkdir(tmp, mode);
 788         if (ret < 0) {
 789                 if (errno != EEXIST) {
 790                         PERROR("mkdir recursive last piece");
 791                         ret = -errno;
 792                 } else {
 793                         ret = 0;
 794                 }
 795         }
 796
 797 error:
 798         return ret;
 799 }
 800
 801 static
 802 char *create_output_path_auto(char *path_name)
 803 {
 804         int ret;
 805         char *traces_path = NULL;
 806         char *alloc_path = NULL;
 807         char *default_path;
 808
 809         default_path = config_get_default_path();
 810         if (default_path == NULL) {
 811                 ERR("Home path not found.\n \
 812                                 Please specify an output path using -o, --output PATH");
 813                 goto exit;
 814         }
 815         alloc_path = strdup(default_path);
 816         if (alloc_path == NULL) {
 817                 PERROR("Path allocation");
 818                 goto exit;
 819         }
 820         ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
 821                         "/%s", alloc_path, path_name);
 822         if (ret < 0) {
 823                 PERROR("asprintf trace dir name");
 824                 goto exit;
 825         }
 826 exit:
 827         free(alloc_path);
 828         return traces_path;
 829 }
 830
 831 static
 832 char *create_output_path_noauto(char *path_name)
 833 {
 834         int ret;
 835         char *traces_path = NULL;
 836         char *full_path;
 837
 838         full_path = expand_full_path(opt_output_path);
 839         ret = asprintf(&traces_path, "%s/%s", full_path, path_name);
 840         if (ret < 0) {
 841                 PERROR("asprintf trace dir name");
 842                 goto exit;
 843         }
 844 exit:
 845         free(full_path);
 846         return traces_path;
 847 }
 848
 849 /*
 850  * create_output_path: create the output trace directory
 851  */
 852 static
 853 char *create_output_path(char *path_name)
 854 {
 855         if (opt_output_path == NULL) {
 856                 return create_output_path_auto(path_name);
 857         } else {
 858                 return create_output_path_noauto(path_name);
 859         }
 860 }
 861
 862 static
 863 void deferred_free_stream(struct rcu_head *head)
 864 {
 865         struct relay_stream *stream =
 866                 caa_container_of(head, struct relay_stream, rcu_node);
 867         free(stream);
 868 }
 869
 870 /*
 871  * relay_delete_session: Free all memory associated with a session and
 872  * close all the FDs
 873  */
 874 static
 875 void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht)
 876 {
 877         struct lttng_ht_iter iter;
 878         struct lttng_ht_node_ulong *node;
 879         struct relay_stream *stream;
 880         int ret;
 881
 882         if (!cmd->session) {
 883                 return;
 884         }
 885
 886         DBG("Relay deleting session %" PRIu64, cmd->session->id);
 887
 888         lttcomm_destroy_sock(cmd->session->sock);
 889
 890         rcu_read_lock();
 891         cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) {
 892                 node = lttng_ht_iter_get_node_ulong(&iter);
 893                 if (node) {
 894                         stream = caa_container_of(node,
 895                                         struct relay_stream, stream_n);
 896                         if (stream->session == cmd->session) {
 897                                 ret = close(stream->fd);
 898                                 if (ret < 0) {
 899                                         PERROR("close stream fd on delete session");
 900                                 }
 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), 0);
 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), 0);
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                 delret = close(stream->fd);
1062                 if (delret < 0) {
1063                         PERROR("close stream");
1064                 }
1065                 delret = lttng_ht_del(streams_ht, &iter);
1066                 assert(!delret);
1067                 call_rcu(&stream->rcu_node,
1068                                 deferred_free_stream);
1069                 DBG("Closed tracefile %d from close stream", stream->fd);
1070         }
1071
1072 end_unlock:
1073         rcu_read_unlock();
1074
1075         if (ret < 0) {
1076                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1077         } else {
1078                 reply.ret_code = htobe32(LTTNG_OK);
1079         }
1080         send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1081                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1082         if (send_ret < 0) {
1083                 ERR("Relay sending stream id");
1084         }
1085
1086 end_no_session:
1087         return ret;
1088 }
1089
1090 /*
1091  * relay_unknown_command: send -1 if received unknown command
1092  */
1093 static
1094 void relay_unknown_command(struct relay_command *cmd)
1095 {
1096         struct lttcomm_relayd_generic_reply reply;
1097         int ret;
1098
1099         reply.ret_code = htobe32(LTTNG_ERR_UNK);
1100         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1101                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1102         if (ret < 0) {
1103                 ERR("Relay sending unknown command");
1104         }
1105 }
1106
1107 /*
1108  * relay_start: send an acknowledgment to the client to tell if we are
1109  * ready to receive data. We are ready if a session is established.
1110  */
1111 static
1112 int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1113                 struct relay_command *cmd)
1114 {
1115         int ret = htobe32(LTTNG_OK);
1116         struct lttcomm_relayd_generic_reply reply;
1117         struct relay_session *session = cmd->session;
1118
1119         if (!session) {
1120                 DBG("Trying to start the streaming without a session established");
1121                 ret = htobe32(LTTNG_ERR_UNK);
1122         }
1123
1124         reply.ret_code = ret;
1125         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1126                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1127         if (ret < 0) {
1128                 ERR("Relay sending start ack");
1129         }
1130
1131         return ret;
1132 }
1133
1134 /*
1135  * Get stream from stream id.
1136  * Need to be called with RCU read-side lock held.
1137  */
1138 static
1139 struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id,
1140                 struct lttng_ht *streams_ht)
1141 {
1142         struct lttng_ht_node_ulong *node;
1143         struct lttng_ht_iter iter;
1144         struct relay_stream *ret;
1145
1146         lttng_ht_lookup(streams_ht,
1147                         (void *)((unsigned long) stream_id),
1148                         &iter);
1149         node = lttng_ht_iter_get_node_ulong(&iter);
1150         if (node == NULL) {
1151                 DBG("Relay stream %" PRIu64 " not found", stream_id);
1152                 ret = NULL;
1153                 goto end;
1154         }
1155
1156         ret = caa_container_of(node, struct relay_stream, stream_n);
1157
1158 end:
1159         return ret;
1160 }
1161
1162 /*
1163  * Append padding to the file pointed by the file descriptor fd.
1164  */
1165 static int write_padding_to_file(int fd, uint32_t size)
1166 {
1167         int ret = 0;
1168         char *zeros;
1169
1170         if (size == 0) {
1171                 goto end;
1172         }
1173
1174         zeros = zmalloc(size);
1175         if (zeros == NULL) {
1176                 PERROR("zmalloc zeros for padding");
1177                 ret = -1;
1178                 goto end;
1179         }
1180
1181         do {
1182                 ret = write(fd, zeros, size);
1183         } while (ret < 0 && errno == EINTR);
1184         if (ret < 0) {
1185                 PERROR("write padding to file");
1186         }
1187
1188         free(zeros);
1189
1190 end:
1191         return ret;
1192 }
1193
1194 /*
1195  * relay_recv_metadata: receive the metada for the session.
1196  */
1197 static
1198 int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1199                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1200 {
1201         int ret = htobe32(LTTNG_OK);
1202         struct relay_session *session = cmd->session;
1203         struct lttcomm_relayd_metadata_payload *metadata_struct;
1204         struct relay_stream *metadata_stream;
1205         uint64_t data_size, payload_size;
1206
1207         if (!session) {
1208                 ERR("Metadata sent before version check");
1209                 ret = -1;
1210                 goto end;
1211         }
1212
1213         data_size = payload_size = be64toh(recv_hdr->data_size);
1214         if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1215                 ERR("Incorrect data size");
1216                 ret = -1;
1217                 goto end;
1218         }
1219         payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
1220
1221         if (data_buffer_size < data_size) {
1222                 /* In case the realloc fails, we can free the memory */
1223                 char *tmp_data_ptr = data_buffer;
1224                 data_buffer = realloc(data_buffer, data_size);
1225                 if (!data_buffer) {
1226                         ERR("Allocating data buffer");
1227                         free(tmp_data_ptr);
1228                         ret = -1;
1229                         goto end;
1230                 }
1231                 data_buffer_size = data_size;
1232         }
1233         memset(data_buffer, 0, data_size);
1234         DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
1235         ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
1236         if (ret < 0 || ret != data_size) {
1237                 ret = -1;
1238                 ERR("Relay didn't receive the whole metadata");
1239                 goto end;
1240         }
1241         metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
1242
1243         rcu_read_lock();
1244         metadata_stream = relay_stream_from_stream_id(
1245                         be64toh(metadata_struct->stream_id), streams_ht);
1246         if (!metadata_stream) {
1247                 ret = -1;
1248                 goto end_unlock;
1249         }
1250
1251         do {
1252                 ret = write(metadata_stream->fd, metadata_struct->payload,
1253                                 payload_size);
1254         } while (ret < 0 && errno == EINTR);
1255         if (ret < payload_size) {
1256                 ERR("Relay error writing metadata on file");
1257                 ret = -1;
1258                 goto end_unlock;
1259         }
1260
1261         ret = write_padding_to_file(metadata_stream->fd,
1262                         be32toh(metadata_struct->padding_size));
1263         if (ret < 0) {
1264                 goto end_unlock;
1265         }
1266
1267         DBG2("Relay metadata written");
1268
1269 end_unlock:
1270         rcu_read_unlock();
1271 end:
1272         return ret;
1273 }
1274
1275 /*
1276  * relay_send_version: send relayd version number
1277  */
1278 static
1279 int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1280                 struct relay_command *cmd)
1281 {
1282         int ret;
1283         struct lttcomm_relayd_version reply, msg;
1284         struct relay_session *session;
1285
1286         if (cmd->session == NULL) {
1287                 session = zmalloc(sizeof(struct relay_session));
1288                 if (session == NULL) {
1289                         PERROR("relay session zmalloc");
1290                         ret = -1;
1291                         goto end;
1292                 }
1293                 session->id = ++last_relay_session_id;
1294                 DBG("Created session %" PRIu64, session->id);
1295                 cmd->session = session;
1296         } else {
1297                 session = cmd->session;
1298         }
1299         session->version_check_done = 1;
1300
1301         /* Get version from the other side. */
1302         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1303         if (ret < 0 || ret != sizeof(msg)) {
1304                 ret = -1;
1305                 ERR("Relay failed to receive the version values.");
1306                 goto end;
1307         }
1308
1309         /*
1310          * For now, we just ignore the received version but after 2.1 stable
1311          * release, a check must be done to see if we either adapt to the other
1312          * side version (which MUST be lower than us) or keep the latest data
1313          * structure considering that the other side will adapt.
1314          */
1315
1316         ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
1317         if (ret < 2) {
1318                 ERR("Error in scanning version");
1319                 ret = -1;
1320                 goto end;
1321         }
1322         reply.major = htobe32(reply.major);
1323         reply.minor = htobe32(reply.minor);
1324         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1325                         sizeof(struct lttcomm_relayd_version), 0);
1326         if (ret < 0) {
1327                 ERR("Relay sending version");
1328         }
1329         DBG("Version check done (%u.%u)", be32toh(reply.major),
1330                         be32toh(reply.minor));
1331
1332 end:
1333         return ret;
1334 }
1335
1336 /*
1337  * Check for data pending for a given stream id from the session daemon.
1338  */
1339 static
1340 int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1341                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1342 {
1343         struct relay_session *session = cmd->session;
1344         struct lttcomm_relayd_data_pending msg;
1345         struct lttcomm_relayd_generic_reply reply;
1346         struct relay_stream *stream;
1347         int ret;
1348         struct lttng_ht_node_ulong *node;
1349         struct lttng_ht_iter iter;
1350         uint64_t last_net_seq_num, stream_id;
1351
1352         DBG("Data pending command received");
1353
1354         if (!session || session->version_check_done == 0) {
1355                 ERR("Trying to check for data before version check");
1356                 ret = -1;
1357                 goto end_no_session;
1358         }
1359
1360         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1361         if (ret < sizeof(msg)) {
1362                 ERR("Relay didn't receive valid data_pending struct size : %d", ret);
1363                 ret = -1;
1364                 goto end_no_session;
1365         }
1366
1367         stream_id = be64toh(msg.stream_id);
1368         last_net_seq_num = be64toh(msg.last_net_seq_num);
1369
1370         rcu_read_lock();
1371         lttng_ht_lookup(streams_ht, (void *)((unsigned long) stream_id), &iter);
1372         node = lttng_ht_iter_get_node_ulong(&iter);
1373         if (node == NULL) {
1374                 DBG("Relay stream %" PRIu64 " not found", stream_id);
1375                 ret = -1;
1376                 goto end_unlock;
1377         }
1378
1379         stream = caa_container_of(node, struct relay_stream, stream_n);
1380         assert(stream);
1381
1382         DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
1383                         " and last_seq %" PRIu64, stream_id, stream->prev_seq,
1384                         last_net_seq_num);
1385
1386         /* Avoid wrapping issue */
1387         if (((int64_t) (stream->prev_seq - last_net_seq_num)) <= 0) {
1388                 /* Data has in fact been written and is NOT pending */
1389                 ret = 0;
1390         } else {
1391                 /* Data still being streamed thus pending */
1392                 ret = 1;
1393         }
1394
1395 end_unlock:
1396         rcu_read_unlock();
1397
1398         reply.ret_code = htobe32(ret);
1399         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1400         if (ret < 0) {
1401                 ERR("Relay data pending ret code failed");
1402         }
1403
1404 end_no_session:
1405         return ret;
1406 }
1407
1408 /*
1409  * Wait for the control socket to reach a quiescent state.
1410  *
1411  * Note that for now, when receiving this command from the session daemon, this
1412  * means that every subsequent commands or data received on the control socket
1413  * has been handled. So, this is why we simply return OK here.
1414  */
1415 static
1416 int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
1417                 struct relay_command *cmd)
1418 {
1419         int ret;
1420         struct lttcomm_relayd_generic_reply reply;
1421
1422         DBG("Checking quiescent state on control socket");
1423
1424         reply.ret_code = htobe32(LTTNG_OK);
1425         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1426         if (ret < 0) {
1427                 ERR("Relay data quiescent control ret code failed");
1428         }
1429
1430         return ret;
1431 }
1432
1433 /*
1434  * relay_process_control: Process the commands received on the control socket
1435  */
1436 static
1437 int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1438                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1439 {
1440         int ret = 0;
1441
1442         switch (be32toh(recv_hdr->cmd)) {
1443                 /*
1444         case RELAYD_CREATE_SESSION:
1445                 ret = relay_create_session(recv_hdr, cmd);
1446                 break;
1447                 */
1448         case RELAYD_ADD_STREAM:
1449                 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1450                 break;
1451         case RELAYD_START_DATA:
1452                 ret = relay_start(recv_hdr, cmd);
1453                 break;
1454         case RELAYD_SEND_METADATA:
1455                 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1456                 break;
1457         case RELAYD_VERSION:
1458                 ret = relay_send_version(recv_hdr, cmd);
1459                 break;
1460         case RELAYD_CLOSE_STREAM:
1461                 ret = relay_close_stream(recv_hdr, cmd, streams_ht);
1462                 break;
1463         case RELAYD_DATA_PENDING:
1464                 ret = relay_data_pending(recv_hdr, cmd, streams_ht);
1465                 break;
1466         case RELAYD_QUIESCENT_CONTROL:
1467                 ret = relay_quiescent_control(recv_hdr, cmd);
1468                 break;
1469         case RELAYD_UPDATE_SYNC_INFO:
1470         default:
1471                 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1472                 relay_unknown_command(cmd);
1473                 ret = -1;
1474                 goto end;
1475         }
1476
1477 end:
1478         return ret;
1479 }
1480
1481 /*
1482  * relay_process_data: Process the data received on the data socket
1483  */
1484 static
1485 int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1486 {
1487         int ret = 0;
1488         struct relay_stream *stream;
1489         struct lttcomm_relayd_data_hdr data_hdr;
1490         uint64_t stream_id;
1491         uint64_t net_seq_num;
1492         uint32_t data_size;
1493
1494         ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1495                         sizeof(struct lttcomm_relayd_data_hdr), 0);
1496         if (ret <= 0) {
1497                 ERR("Connections seems to be closed");
1498                 ret = -1;
1499                 goto end;
1500         }
1501
1502         stream_id = be64toh(data_hdr.stream_id);
1503
1504         rcu_read_lock();
1505         stream = relay_stream_from_stream_id(stream_id, streams_ht);
1506         if (!stream) {
1507                 ret = -1;
1508                 goto end_unlock;
1509         }
1510
1511         data_size = be32toh(data_hdr.data_size);
1512         if (data_buffer_size < data_size) {
1513                 char *tmp_data_ptr = data_buffer;
1514                 data_buffer = realloc(data_buffer, data_size);
1515                 if (!data_buffer) {
1516                         ERR("Allocating data buffer");
1517                         free(tmp_data_ptr);
1518                         ret = -1;
1519                         goto end_unlock;
1520                 }
1521                 data_buffer_size = data_size;
1522         }
1523         memset(data_buffer, 0, data_size);
1524
1525         net_seq_num = be64toh(data_hdr.net_seq_num);
1526
1527         DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
1528                 data_size, stream_id, net_seq_num);
1529         ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
1530         if (ret <= 0) {
1531                 ret = -1;
1532                 goto end_unlock;
1533         }
1534
1535         do {
1536                 ret = write(stream->fd, data_buffer, data_size);
1537         } while (ret < 0 && errno == EINTR);
1538         if (ret < data_size) {
1539                 ERR("Relay error writing data to file");
1540                 ret = -1;
1541                 goto end_unlock;
1542         }
1543
1544         DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
1545                         ret, stream->stream_handle);
1546
1547         ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
1548         if (ret < 0) {
1549                 goto end_unlock;
1550         }
1551
1552         stream->prev_seq = net_seq_num;
1553
1554         /* Check if we need to close the FD */
1555         if (close_stream_check(stream)) {
1556                 int cret;
1557                 struct lttng_ht_iter iter;
1558
1559                 cret = close(stream->fd);
1560                 if (cret < 0) {
1561                         PERROR("close stream process data");
1562                 }
1563                 iter.iter.node = &stream->stream_n.node;
1564                 ret = lttng_ht_del(streams_ht, &iter);
1565                 assert(!ret);
1566                 call_rcu(&stream->rcu_node,
1567                         deferred_free_stream);
1568                 DBG("Closed tracefile %d after recv data", stream->fd);
1569         }
1570
1571 end_unlock:
1572         rcu_read_unlock();
1573 end:
1574         return ret;
1575 }
1576
1577 static
1578 void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1579 {
1580         int ret;
1581
1582         lttng_poll_del(events, pollfd);
1583
1584         ret = close(pollfd);
1585         if (ret < 0) {
1586                 ERR("Closing pollfd %d", pollfd);
1587         }
1588 }
1589
1590 static
1591 int relay_add_connection(int fd, struct lttng_poll_event *events,
1592                 struct lttng_ht *relay_connections_ht)
1593 {
1594         struct relay_command *relay_connection;
1595         int ret;
1596
1597         relay_connection = zmalloc(sizeof(struct relay_command));
1598         if (relay_connection == NULL) {
1599                 PERROR("Relay command zmalloc");
1600                 goto error;
1601         }
1602         ret = read(fd, relay_connection, sizeof(struct relay_command));
1603         if (ret < 0 || ret < sizeof(struct relay_command)) {
1604                 PERROR("read relay cmd pipe");
1605                 goto error_read;
1606         }
1607
1608         lttng_ht_node_init_ulong(&relay_connection->sock_n,
1609                         (unsigned long) relay_connection->sock->fd);
1610         rcu_read_lock();
1611         lttng_ht_add_unique_ulong(relay_connections_ht,
1612                         &relay_connection->sock_n);
1613         rcu_read_unlock();
1614         return lttng_poll_add(events,
1615                         relay_connection->sock->fd,
1616                         LPOLLIN | LPOLLRDHUP);
1617
1618 error_read:
1619         free(relay_connection);
1620 error:
1621         return -1;
1622 }
1623
1624 static
1625 void deferred_free_connection(struct rcu_head *head)
1626 {
1627         struct relay_command *relay_connection =
1628                 caa_container_of(head, struct relay_command, rcu_node);
1629
1630         lttcomm_destroy_sock(relay_connection->sock);
1631         free(relay_connection);
1632 }
1633
1634 static
1635 void relay_del_connection(struct lttng_ht *relay_connections_ht,
1636                 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1637                 struct relay_command *relay_connection)
1638 {
1639         int ret;
1640
1641         ret = lttng_ht_del(relay_connections_ht, iter);
1642         assert(!ret);
1643         if (relay_connection->type == RELAY_CONTROL) {
1644                 relay_delete_session(relay_connection, streams_ht);
1645         }
1646
1647         call_rcu(&relay_connection->rcu_node,
1648                 deferred_free_connection);
1649 }
1650
1651 /*
1652  * This thread does the actual work
1653  */
1654 static
1655 void *relay_thread_worker(void *data)
1656 {
1657         int i, ret, pollfd, err = -1;
1658         uint32_t revents, nb_fd;
1659         struct relay_command *relay_connection;
1660         struct lttng_poll_event events;
1661         struct lttng_ht *relay_connections_ht;
1662         struct lttng_ht_node_ulong *node;
1663         struct lttng_ht_iter iter;
1664         struct lttng_ht *streams_ht;
1665         struct lttcomm_relayd_hdr recv_hdr;
1666
1667         DBG("[thread] Relay worker started");
1668
1669         rcu_register_thread();
1670
1671         /* table of connections indexed on socket */
1672         relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1673         if (!relay_connections_ht) {
1674                 goto relay_connections_ht_error;
1675         }
1676
1677         /* tables of streams indexed by stream ID */
1678         streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1679         if (!streams_ht) {
1680                 goto streams_ht_error;
1681         }
1682
1683         ret = create_thread_poll_set(&events, 2);
1684         if (ret < 0) {
1685                 goto error_poll_create;
1686         }
1687
1688         ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1689         if (ret < 0) {
1690                 goto error;
1691         }
1692
1693         while (1) {
1694                 /* Zeroed the events structure */
1695                 lttng_poll_reset(&events);
1696
1697                 nb_fd = LTTNG_POLL_GETNB(&events);
1698
1699                 /* Infinite blocking call, waiting for transmission */
1700         restart:
1701                 DBG3("Relayd worker thread polling...");
1702                 ret = lttng_poll_wait(&events, -1);
1703                 if (ret < 0) {
1704                         /*
1705                          * Restart interrupted system call.
1706                          */
1707                         if (errno == EINTR) {
1708                                 goto restart;
1709                         }
1710                         goto error;
1711                 }
1712
1713                 for (i = 0; i < nb_fd; i++) {
1714                         /* Fetch once the poll data */
1715                         revents = LTTNG_POLL_GETEV(&events, i);
1716                         pollfd = LTTNG_POLL_GETFD(&events, i);
1717
1718                         /* Thread quit pipe has been closed. Killing thread. */
1719                         ret = check_thread_quit_pipe(pollfd, revents);
1720                         if (ret) {
1721                                 err = 0;
1722                                 goto exit;
1723                         }
1724
1725                         /* Inspect the relay cmd pipe for new connection */
1726                         if (pollfd == relay_cmd_pipe[0]) {
1727                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1728                                         ERR("Relay pipe error");
1729                                         goto error;
1730                                 } else if (revents & LPOLLIN) {
1731                                         DBG("Relay command received");
1732                                         ret = relay_add_connection(relay_cmd_pipe[0],
1733                                                         &events, relay_connections_ht);
1734                                         if (ret < 0) {
1735                                                 goto error;
1736                                         }
1737                                 }
1738                         } else if (revents > 0) {
1739                                 rcu_read_lock();
1740                                 lttng_ht_lookup(relay_connections_ht,
1741                                                 (void *)((unsigned long) pollfd),
1742                                                 &iter);
1743                                 node = lttng_ht_iter_get_node_ulong(&iter);
1744                                 if (node == NULL) {
1745                                         DBG2("Relay sock %d not found", pollfd);
1746                                         rcu_read_unlock();
1747                                         goto error;
1748                                 }
1749                                 relay_connection = caa_container_of(node,
1750                                                 struct relay_command, sock_n);
1751
1752                                 if (revents & (LPOLLERR)) {
1753                                         ERR("POLL ERROR");
1754                                         relay_cleanup_poll_connection(&events, pollfd);
1755                                         relay_del_connection(relay_connections_ht,
1756                                                         streams_ht, &iter,
1757                                                         relay_connection);
1758                                 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1759                                         DBG("Socket %d hung up", pollfd);
1760                                         relay_cleanup_poll_connection(&events, pollfd);
1761                                         relay_del_connection(relay_connections_ht,
1762                                                         streams_ht, &iter,
1763                                                         relay_connection);
1764                                 } else if (revents & LPOLLIN) {
1765                                         /* control socket */
1766                                         if (relay_connection->type == RELAY_CONTROL) {
1767                                                 ret = relay_connection->sock->ops->recvmsg(
1768                                                                 relay_connection->sock, &recv_hdr,
1769                                                                 sizeof(struct lttcomm_relayd_hdr), 0);
1770                                                 /* connection closed */
1771                                                 if (ret <= 0) {
1772                                                         relay_cleanup_poll_connection(&events, pollfd);
1773                                                         relay_del_connection(relay_connections_ht,
1774                                                                         streams_ht, &iter,
1775                                                                         relay_connection);
1776                                                         DBG("Control connection closed with %d", pollfd);
1777                                                 } else {
1778                                                         if (relay_connection->session) {
1779                                                                 DBG2("Relay worker receiving data for session : %" PRIu64,
1780                                                                                 relay_connection->session->id);
1781                                                         }
1782                                                         ret = relay_process_control(&recv_hdr,
1783                                                                         relay_connection,
1784                                                                         streams_ht);
1785                                                         /*
1786                                                          * there was an error in processing a control
1787                                                          * command: clear the session
1788                                                          * */
1789                                                         if (ret < 0) {
1790                                                                 relay_cleanup_poll_connection(&events, pollfd);
1791                                                                 relay_del_connection(relay_connections_ht,
1792                                                                                 streams_ht, &iter,
1793                                                                                 relay_connection);
1794                                                                 DBG("Connection closed with %d", pollfd);
1795                                                         }
1796                                                 }
1797                                                 /* data socket */
1798                                         } else if (relay_connection->type == RELAY_DATA) {
1799                                                 ret = relay_process_data(relay_connection, streams_ht);
1800                                                 /* connection closed */
1801                                                 if (ret < 0) {
1802                                                         relay_cleanup_poll_connection(&events, pollfd);
1803                                                         relay_del_connection(relay_connections_ht,
1804                                                                         streams_ht, &iter,
1805                                                                         relay_connection);
1806                                                         DBG("Data connection closed with %d", pollfd);
1807                                                 }
1808                                         }
1809                                 }
1810                                 rcu_read_unlock();
1811                         }
1812                 }
1813         }
1814
1815 exit:
1816 error:
1817         lttng_poll_clean(&events);
1818
1819         /* empty the hash table and free the memory */
1820         rcu_read_lock();
1821         cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1822                 node = lttng_ht_iter_get_node_ulong(&iter);
1823                 if (node) {
1824                         relay_connection = caa_container_of(node,
1825                                         struct relay_command, sock_n);
1826                         relay_del_connection(relay_connections_ht,
1827                                         streams_ht, &iter,
1828                                         relay_connection);
1829                 }
1830         }
1831         rcu_read_unlock();
1832 error_poll_create:
1833         lttng_ht_destroy(streams_ht);
1834 streams_ht_error:
1835         lttng_ht_destroy(relay_connections_ht);
1836 relay_connections_ht_error:
1837         /* Close relay cmd pipes */
1838         utils_close_pipe(relay_cmd_pipe);
1839         if (err) {
1840                 DBG("Thread exited with error");
1841         }
1842         DBG("Worker thread cleanup complete");
1843         free(data_buffer);
1844         stop_threads();
1845         rcu_unregister_thread();
1846         return NULL;
1847 }
1848
1849 /*
1850  * Create the relay command pipe to wake thread_manage_apps.
1851  * Closed in cleanup().
1852  */
1853 static int create_relay_cmd_pipe(void)
1854 {
1855         int ret;
1856
1857         ret = utils_create_pipe_cloexec(relay_cmd_pipe);
1858
1859         return ret;
1860 }
1861
1862 /*
1863  * main
1864  */
1865 int main(int argc, char **argv)
1866 {
1867         int ret = 0;
1868         void *status;
1869
1870         /* Create thread quit pipe */
1871         if ((ret = init_thread_quit_pipe()) < 0) {
1872                 goto error;
1873         }
1874
1875         /* Parse arguments */
1876         progname = argv[0];
1877         if ((ret = parse_args(argc, argv) < 0)) {
1878                 goto exit;
1879         }
1880
1881         if ((ret = set_signal_handler()) < 0) {
1882                 goto exit;
1883         }
1884
1885         /* Daemonize */
1886         if (opt_daemon) {
1887                 ret = daemon(0, 0);
1888                 if (ret < 0) {
1889                         PERROR("daemon");
1890                         goto exit;
1891                 }
1892         }
1893
1894         /* Check if daemon is UID = 0 */
1895         is_root = !getuid();
1896
1897         if (!is_root) {
1898                 if (control_uri->port < 1024 || data_uri->port < 1024) {
1899                         ERR("Need to be root to use ports < 1024");
1900                         ret = -1;
1901                         goto exit;
1902                 }
1903         }
1904
1905         /* Setup the thread apps communication pipe. */
1906         if ((ret = create_relay_cmd_pipe()) < 0) {
1907                 goto exit;
1908         }
1909
1910         /* Init relay command queue. */
1911         cds_wfq_init(&relay_cmd_queue.queue);
1912
1913         /* Set up max poll set size */
1914         lttng_poll_set_max_size();
1915
1916         /* Setup the dispatcher thread */
1917         ret = pthread_create(&dispatcher_thread, NULL,
1918                         relay_thread_dispatcher, (void *) NULL);
1919         if (ret != 0) {
1920                 PERROR("pthread_create dispatcher");
1921                 goto exit_dispatcher;
1922         }
1923
1924         /* Setup the worker thread */
1925         ret = pthread_create(&worker_thread, NULL,
1926                         relay_thread_worker, (void *) NULL);
1927         if (ret != 0) {
1928                 PERROR("pthread_create worker");
1929                 goto exit_worker;
1930         }
1931
1932         /* Setup the listener thread */
1933         ret = pthread_create(&listener_thread, NULL,
1934                         relay_thread_listener, (void *) NULL);
1935         if (ret != 0) {
1936                 PERROR("pthread_create listener");
1937                 goto exit_listener;
1938         }
1939
1940 exit_listener:
1941         ret = pthread_join(listener_thread, &status);
1942         if (ret != 0) {
1943                 PERROR("pthread_join");
1944                 goto error;     /* join error, exit without cleanup */
1945         }
1946
1947 exit_worker:
1948         ret = pthread_join(worker_thread, &status);
1949         if (ret != 0) {
1950                 PERROR("pthread_join");
1951                 goto error;     /* join error, exit without cleanup */
1952         }
1953
1954 exit_dispatcher:
1955         ret = pthread_join(dispatcher_thread, &status);
1956         if (ret != 0) {
1957                 PERROR("pthread_join");
1958                 goto error;     /* join error, exit without cleanup */
1959         }
1960
1961 exit:
1962         cleanup();
1963         if (!ret) {
1964                 exit(EXIT_SUCCESS);
1965         }
1966
1967 error:
1968         exit(EXIT_FAILURE);
1969 }