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