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/sessiond-comm/relayd.h>
  51 #include <common/uri.h>
  52 #include <common/utils.h>
  53
  54 #include "cmd.h"
  55 #include "ctf-trace.h"
  56 #include "index.h"
  57 #include "utils.h"
  58 #include "lttng-relayd.h"
  59 #include "live.h"
  60
  61 /* command line options */
  62 char *opt_output_path;
  63 static int opt_daemon;
  64 static struct lttng_uri *control_uri;
  65 static struct lttng_uri *data_uri;
  66 static struct lttng_uri *live_uri;
  67
  68 const char *progname;
  69
  70 /*
  71  * Quit pipe for all threads. This permits a single cancellation point
  72  * for all threads when receiving an event on the pipe.
  73  */
  74 static int thread_quit_pipe[2] = { -1, -1 };
  75
  76 /*
  77  * This pipe is used to inform the worker thread that a command is queued and
  78  * ready to be processed.
  79  */
  80 static int relay_cmd_pipe[2] = { -1, -1 };
  81
  82 /* Shared between threads */
  83 static int dispatch_thread_exit;
  84
  85 static pthread_t listener_thread;
  86 static pthread_t dispatcher_thread;
  87 static pthread_t worker_thread;
  88
  89 static uint64_t last_relay_stream_id;
  90 static uint64_t last_relay_session_id;
  91
  92 /*
  93  * Relay command queue.
  94  *
  95  * The relay_thread_listener and relay_thread_dispatcher communicate with this
  96  * queue.
  97  */
  98 static struct relay_cmd_queue relay_cmd_queue;
  99
 100 /* buffer allocated at startup, used to store the trace data */
 101 static char *data_buffer;
 102 static unsigned int data_buffer_size;
 103
 104 /* Global hash table that stores relay index object. */
 105 static struct lttng_ht *indexes_ht;
 106
 107 /* We need those values for the file/dir creation. */
 108 static uid_t relayd_uid;
 109 static gid_t relayd_gid;
 110
 111 /* Global relay stream hash table. */
 112 struct lttng_ht *relay_streams_ht;
 113
 114 /*
 115  * usage function on stderr
 116  */
 117 static
 118 void usage(void)
 119 {
 120         fprintf(stderr, "Usage: %s OPTIONS\n\nOptions:\n", progname);
 121         fprintf(stderr, "  -h, --help                Display this usage.\n");
 122         fprintf(stderr, "  -d, --daemonize           Start as a daemon.\n");
 123         fprintf(stderr, "  -C, --control-port URL    Control port listening.\n");
 124         fprintf(stderr, "  -D, --data-port URL       Data port listening.\n");
 125         fprintf(stderr, "  -o, --output PATH         Output path for traces. Must use an absolute path.\n");
 126         fprintf(stderr, "  -v, --verbose             Verbose mode. Activate DBG() macro.\n");
 127 }
 128
 129 static
 130 int parse_args(int argc, char **argv)
 131 {
 132         int c;
 133         int ret = 0;
 134         char *default_address;
 135
 136         static struct option long_options[] = {
 137                 { "control-port", 1, 0, 'C', },
 138                 { "data-port", 1, 0, 'D', },
 139                 { "daemonize", 0, 0, 'd', },
 140                 { "help", 0, 0, 'h', },
 141                 { "output", 1, 0, 'o', },
 142                 { "verbose", 0, 0, 'v', },
 143                 { NULL, 0, 0, 0, },
 144         };
 145
 146         while (1) {
 147                 int option_index = 0;
 148                 c = getopt_long(argc, argv, "dhv" "C:D:o:",
 149                                 long_options, &option_index);
 150                 if (c == -1) {
 151                         break;
 152                 }
 153
 154                 switch (c) {
 155                 case 0:
 156                         fprintf(stderr, "option %s", long_options[option_index].name);
 157                         if (optarg) {
 158                                 fprintf(stderr, " with arg %s\n", optarg);
 159                         }
 160                         break;
 161                 case 'C':
 162                         ret = uri_parse(optarg, &control_uri);
 163                         if (ret < 0) {
 164                                 ERR("Invalid control URI specified");
 165                                 goto exit;
 166                         }
 167                         if (control_uri->port == 0) {
 168                                 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 169                         }
 170                         break;
 171                 case 'D':
 172                         ret = uri_parse(optarg, &data_uri);
 173                         if (ret < 0) {
 174                                 ERR("Invalid data URI specified");
 175                                 goto exit;
 176                         }
 177                         if (data_uri->port == 0) {
 178                                 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
 179                         }
 180                         break;
 181                 case 'd':
 182                         opt_daemon = 1;
 183                         break;
 184                 case 'h':
 185                         usage();
 186                         exit(EXIT_FAILURE);
 187                 case 'o':
 188                         ret = asprintf(&opt_output_path, "%s", optarg);
 189                         if (ret < 0) {
 190                                 PERROR("asprintf opt_output_path");
 191                                 goto exit;
 192                         }
 193                         break;
 194                 case 'v':
 195                         /* Verbose level can increase using multiple -v */
 196                         lttng_opt_verbose += 1;
 197                         break;
 198                 default:
 199                         /* Unknown option or other error.
 200                          * Error is printed by getopt, just return */
 201                         ret = -1;
 202                         goto exit;
 203                 }
 204         }
 205
 206         /* assign default values */
 207         if (control_uri == NULL) {
 208                 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
 209                                 DEFAULT_NETWORK_CONTROL_PORT);
 210                 if (ret < 0) {
 211                         PERROR("asprintf default data address");
 212                         goto exit;
 213                 }
 214
 215                 ret = uri_parse(default_address, &control_uri);
 216                 free(default_address);
 217                 if (ret < 0) {
 218                         ERR("Invalid control URI specified");
 219                         goto exit;
 220                 }
 221         }
 222         if (data_uri == NULL) {
 223                 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
 224                                 DEFAULT_NETWORK_DATA_PORT);
 225                 if (ret < 0) {
 226                         PERROR("asprintf default data address");
 227                         goto exit;
 228                 }
 229
 230                 ret = uri_parse(default_address, &data_uri);
 231                 free(default_address);
 232                 if (ret < 0) {
 233                         ERR("Invalid data URI specified");
 234                         goto exit;
 235                 }
 236         }
 237         if (live_uri == NULL) {
 238                 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
 239                                 DEFAULT_NETWORK_VIEWER_PORT);
 240                 if (ret < 0) {
 241                         PERROR("asprintf default viewer control address");
 242                         goto exit;
 243                 }
 244
 245                 ret = uri_parse(default_address, &live_uri);
 246                 free(default_address);
 247                 if (ret < 0) {
 248                         ERR("Invalid viewer control URI specified");
 249                         goto exit;
 250                 }
 251         }
 252
 253 exit:
 254         return ret;
 255 }
 256
 257 /*
 258  * Cleanup the daemon
 259  */
 260 static
 261 void cleanup(void)
 262 {
 263         DBG("Cleaning up");
 264
 265         /* free the dynamically allocated opt_output_path */
 266         free(opt_output_path);
 267
 268         /* Close thread quit pipes */
 269         utils_close_pipe(thread_quit_pipe);
 270
 271         uri_free(control_uri);
 272         uri_free(data_uri);
 273 }
 274
 275 /*
 276  * Write to writable pipe used to notify a thread.
 277  */
 278 static
 279 int notify_thread_pipe(int wpipe)
 280 {
 281         int ret;
 282
 283         do {
 284                 ret = write(wpipe, "!", 1);
 285         } while (ret < 0 && errno == EINTR);
 286         if (ret < 0 || ret != 1) {
 287                 PERROR("write poll pipe");
 288         }
 289
 290         return ret;
 291 }
 292
 293 /*
 294  * Stop all threads by closing the thread quit pipe.
 295  */
 296 static
 297 void stop_threads(void)
 298 {
 299         int ret;
 300
 301         /* Stopping all threads */
 302         DBG("Terminating all threads");
 303         ret = notify_thread_pipe(thread_quit_pipe[1]);
 304         if (ret < 0) {
 305                 ERR("write error on thread quit pipe");
 306         }
 307
 308         /* Dispatch thread */
 309         CMM_STORE_SHARED(dispatch_thread_exit, 1);
 310         futex_nto1_wake(&relay_cmd_queue.futex);
 311 }
 312
 313 /*
 314  * Signal handler for the daemon
 315  *
 316  * Simply stop all worker threads, leaving main() return gracefully after
 317  * joining all threads and calling cleanup().
 318  */
 319 static
 320 void sighandler(int sig)
 321 {
 322         switch (sig) {
 323         case SIGPIPE:
 324                 DBG("SIGPIPE caught");
 325                 return;
 326         case SIGINT:
 327                 DBG("SIGINT caught");
 328                 stop_threads();
 329                 break;
 330         case SIGTERM:
 331                 DBG("SIGTERM caught");
 332                 stop_threads();
 333                 break;
 334         default:
 335                 break;
 336         }
 337 }
 338
 339 /*
 340  * Setup signal handler for :
 341  *              SIGINT, SIGTERM, SIGPIPE
 342  */
 343 static
 344 int set_signal_handler(void)
 345 {
 346         int ret = 0;
 347         struct sigaction sa;
 348         sigset_t sigset;
 349
 350         if ((ret = sigemptyset(&sigset)) < 0) {
 351                 PERROR("sigemptyset");
 352                 return ret;
 353         }
 354
 355         sa.sa_handler = sighandler;
 356         sa.sa_mask = sigset;
 357         sa.sa_flags = 0;
 358         if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
 359                 PERROR("sigaction");
 360                 return ret;
 361         }
 362
 363         if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
 364                 PERROR("sigaction");
 365                 return ret;
 366         }
 367
 368         if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
 369                 PERROR("sigaction");
 370                 return ret;
 371         }
 372
 373         DBG("Signal handler set for SIGTERM, SIGPIPE and SIGINT");
 374
 375         return ret;
 376 }
 377
 378 /*
 379  * Init thread quit pipe.
 380  *
 381  * Return -1 on error or 0 if all pipes are created.
 382  */
 383 static
 384 int init_thread_quit_pipe(void)
 385 {
 386         int ret;
 387
 388         ret = utils_create_pipe_cloexec(thread_quit_pipe);
 389
 390         return ret;
 391 }
 392
 393 /*
 394  * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
 395  */
 396 static
 397 int create_thread_poll_set(struct lttng_poll_event *events, int size)
 398 {
 399         int ret;
 400
 401         if (events == NULL || size == 0) {
 402                 ret = -1;
 403                 goto error;
 404         }
 405
 406         ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
 407         if (ret < 0) {
 408                 goto error;
 409         }
 410
 411         /* Add quit pipe */
 412         ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN);
 413         if (ret < 0) {
 414                 goto error;
 415         }
 416
 417         return 0;
 418
 419 error:
 420         return ret;
 421 }
 422
 423 /*
 424  * Check if the thread quit pipe was triggered.
 425  *
 426  * Return 1 if it was triggered else 0;
 427  */
 428 static
 429 int check_thread_quit_pipe(int fd, uint32_t events)
 430 {
 431         if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
 432                 return 1;
 433         }
 434
 435         return 0;
 436 }
 437
 438 /*
 439  * Create and init socket from uri.
 440  */
 441 static
 442 struct lttcomm_sock *relay_init_sock(struct lttng_uri *uri)
 443 {
 444         int ret;
 445         struct lttcomm_sock *sock = NULL;
 446
 447         sock = lttcomm_alloc_sock_from_uri(uri);
 448         if (sock == NULL) {
 449                 ERR("Allocating socket");
 450                 goto error;
 451         }
 452
 453         ret = lttcomm_create_sock(sock);
 454         if (ret < 0) {
 455                 goto error;
 456         }
 457         DBG("Listening on sock %d", sock->fd);
 458
 459         ret = sock->ops->bind(sock);
 460         if (ret < 0) {
 461                 goto error;
 462         }
 463
 464         ret = sock->ops->listen(sock, -1);
 465         if (ret < 0) {
 466                 goto error;
 467
 468         }
 469
 470         return sock;
 471
 472 error:
 473         if (sock) {
 474                 lttcomm_destroy_sock(sock);
 475         }
 476         return NULL;
 477 }
 478
 479 /*
 480  * Return nonzero if stream needs to be closed.
 481  */
 482 static
 483 int close_stream_check(struct relay_stream *stream)
 484 {
 485
 486         if (stream->close_flag && stream->prev_seq == stream->last_net_seq_num) {
 487                 /*
 488                  * We are about to close the stream so set the data pending flag to 1
 489                  * which will make the end data pending command skip the stream which
 490                  * is now closed and ready. Note that after proceeding to a file close,
 491                  * the written file is ready for reading.
 492                  */
 493                 stream->data_pending_check_done = 1;
 494                 return 1;
 495         }
 496         return 0;
 497 }
 498
 499 /*
 500  * This thread manages the listening for new connections on the network
 501  */
 502 static
 503 void *relay_thread_listener(void *data)
 504 {
 505         int i, ret, pollfd, err = -1;
 506         int val = 1;
 507         uint32_t revents, nb_fd;
 508         struct lttng_poll_event events;
 509         struct lttcomm_sock *control_sock, *data_sock;
 510
 511         DBG("[thread] Relay listener started");
 512
 513         control_sock = relay_init_sock(control_uri);
 514         if (!control_sock) {
 515                 goto error_sock_control;
 516         }
 517
 518         data_sock = relay_init_sock(data_uri);
 519         if (!data_sock) {
 520                 goto error_sock_relay;
 521         }
 522
 523         /*
 524          * Pass 3 as size here for the thread quit pipe, control and data socket.
 525          */
 526         ret = create_thread_poll_set(&events, 3);
 527         if (ret < 0) {
 528                 goto error_create_poll;
 529         }
 530
 531         /* Add the control socket */
 532         ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
 533         if (ret < 0) {
 534                 goto error_poll_add;
 535         }
 536
 537         /* Add the data socket */
 538         ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
 539         if (ret < 0) {
 540                 goto error_poll_add;
 541         }
 542
 543         while (1) {
 544                 DBG("Listener accepting connections");
 545
 546 restart:
 547                 ret = lttng_poll_wait(&events, -1);
 548                 if (ret < 0) {
 549                         /*
 550                          * Restart interrupted system call.
 551                          */
 552                         if (errno == EINTR) {
 553                                 goto restart;
 554                         }
 555                         goto error;
 556                 }
 557
 558                 nb_fd = ret;
 559
 560                 DBG("Relay new connection received");
 561                 for (i = 0; i < nb_fd; i++) {
 562                         /* Fetch once the poll data */
 563                         revents = LTTNG_POLL_GETEV(&events, i);
 564                         pollfd = LTTNG_POLL_GETFD(&events, i);
 565
 566                         /* Thread quit pipe has been closed. Killing thread. */
 567                         ret = check_thread_quit_pipe(pollfd, revents);
 568                         if (ret) {
 569                                 err = 0;
 570                                 goto exit;
 571                         }
 572
 573                         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 574                                 ERR("socket poll error");
 575                                 goto error;
 576                         } else if (revents & LPOLLIN) {
 577                                 /*
 578                                  * Get allocated in this thread,
 579                                  * enqueued to a global queue, dequeued
 580                                  * and freed in the worker thread.
 581                                  */
 582                                 struct relay_command *relay_cmd;
 583                                 struct lttcomm_sock *newsock;
 584
 585                                 relay_cmd = zmalloc(sizeof(struct relay_command));
 586                                 if (relay_cmd == NULL) {
 587                                         PERROR("relay command zmalloc");
 588                                         goto error;
 589                                 }
 590
 591                                 if (pollfd == data_sock->fd) {
 592                                         newsock = data_sock->ops->accept(data_sock);
 593                                         if (!newsock) {
 594                                                 PERROR("accepting data sock");
 595                                                 free(relay_cmd);
 596                                                 goto error;
 597                                         }
 598                                         relay_cmd->type = RELAY_DATA;
 599                                         DBG("Relay data connection accepted, socket %d", newsock->fd);
 600                                 } else {
 601                                         assert(pollfd == control_sock->fd);
 602                                         newsock = control_sock->ops->accept(control_sock);
 603                                         if (!newsock) {
 604                                                 PERROR("accepting control sock");
 605                                                 free(relay_cmd);
 606                                                 goto error;
 607                                         }
 608                                         relay_cmd->type = RELAY_CONTROL;
 609                                         DBG("Relay control connection accepted, socket %d", newsock->fd);
 610                                 }
 611                                 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR,
 612                                                 &val, sizeof(int));
 613                                 if (ret < 0) {
 614                                         PERROR("setsockopt inet");
 615                                         lttcomm_destroy_sock(newsock);
 616                                         free(relay_cmd);
 617                                         goto error;
 618                                 }
 619                                 relay_cmd->sock = newsock;
 620                                 /*
 621                                  * Lock free enqueue the request.
 622                                  */
 623                                 cds_wfq_enqueue(&relay_cmd_queue.queue, &relay_cmd->node);
 624
 625                                 /*
 626                                  * Wake the dispatch queue futex. Implicit memory
 627                                  * barrier with the exchange in cds_wfq_enqueue.
 628                                  */
 629                                 futex_nto1_wake(&relay_cmd_queue.futex);
 630                         }
 631                 }
 632         }
 633
 634 exit:
 635 error:
 636 error_poll_add:
 637         lttng_poll_clean(&events);
 638 error_create_poll:
 639         if (data_sock->fd >= 0) {
 640                 ret = data_sock->ops->close(data_sock);
 641                 if (ret) {
 642                         PERROR("close");
 643                 }
 644         }
 645         lttcomm_destroy_sock(data_sock);
 646 error_sock_relay:
 647         if (control_sock->fd >= 0) {
 648                 ret = control_sock->ops->close(control_sock);
 649                 if (ret) {
 650                         PERROR("close");
 651                 }
 652         }
 653         lttcomm_destroy_sock(control_sock);
 654 error_sock_control:
 655         if (err) {
 656                 DBG("Thread exited with error");
 657         }
 658         DBG("Relay listener thread cleanup complete");
 659         stop_threads();
 660         return NULL;
 661 }
 662
 663 /*
 664  * This thread manages the dispatching of the requests to worker threads
 665  */
 666 static
 667 void *relay_thread_dispatcher(void *data)
 668 {
 669         int ret;
 670         struct cds_wfq_node *node;
 671         struct relay_command *relay_cmd = NULL;
 672
 673         DBG("[thread] Relay dispatcher started");
 674
 675         while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
 676                 /* Atomically prepare the queue futex */
 677                 futex_nto1_prepare(&relay_cmd_queue.futex);
 678
 679                 do {
 680                         /* Dequeue commands */
 681                         node = cds_wfq_dequeue_blocking(&relay_cmd_queue.queue);
 682                         if (node == NULL) {
 683                                 DBG("Woken up but nothing in the relay command queue");
 684                                 /* Continue thread execution */
 685                                 break;
 686                         }
 687
 688                         relay_cmd = caa_container_of(node, struct relay_command, node);
 689                         DBG("Dispatching request waiting on sock %d", relay_cmd->sock->fd);
 690
 691                         /*
 692                          * Inform worker thread of the new request. This
 693                          * call is blocking so we can be assured that the data will be read
 694                          * at some point in time or wait to the end of the world :)
 695                          */
 696                         do {
 697                                 ret = write(relay_cmd_pipe[1], relay_cmd,
 698                                                 sizeof(struct relay_command));
 699                         } while (ret < 0 && errno == EINTR);
 700                         free(relay_cmd);
 701                         if (ret < 0 || ret != sizeof(struct relay_command)) {
 702                                 PERROR("write cmd pipe");
 703                                 goto error;
 704                         }
 705                 } while (node != NULL);
 706
 707                 /* Futex wait on queue. Blocking call on futex() */
 708                 futex_nto1_wait(&relay_cmd_queue.futex);
 709         }
 710
 711 error:
 712         DBG("Dispatch thread dying");
 713         stop_threads();
 714         return NULL;
 715 }
 716
 717 /*
 718  * Get stream from stream id.
 719  * Need to be called with RCU read-side lock held.
 720  */
 721 struct relay_stream *relay_stream_find_by_id(uint64_t stream_id)
 722 {
 723         struct lttng_ht_node_ulong *node;
 724         struct lttng_ht_iter iter;
 725         struct relay_stream *ret;
 726
 727         lttng_ht_lookup(relay_streams_ht,
 728                         (void *)((unsigned long) stream_id),
 729                         &iter);
 730         node = lttng_ht_iter_get_node_ulong(&iter);
 731         if (node == NULL) {
 732                 DBG("Relay stream %" PRIu64 " not found", stream_id);
 733                 ret = NULL;
 734                 goto end;
 735         }
 736
 737         ret = caa_container_of(node, struct relay_stream, stream_n);
 738
 739 end:
 740         return ret;
 741 }
 742
 743 static
 744 void deferred_free_stream(struct rcu_head *head)
 745 {
 746         struct relay_stream *stream =
 747                 caa_container_of(head, struct relay_stream, rcu_node);
 748
 749         ctf_trace_try_destroy(stream->ctf_trace);
 750
 751         free(stream->path_name);
 752         free(stream->channel_name);
 753         free(stream);
 754 }
 755
 756 static
 757 void deferred_free_session(struct rcu_head *head)
 758 {
 759         struct relay_session *session =
 760                 caa_container_of(head, struct relay_session, rcu_node);
 761         free(session);
 762 }
 763
 764 static void close_stream(struct relay_stream *stream,
 765                 struct lttng_ht *viewer_streams_ht, struct lttng_ht *ctf_traces_ht)
 766 {
 767         int delret;
 768         struct relay_viewer_stream *vstream;
 769         struct lttng_ht_iter iter;
 770
 771         assert(stream);
 772         assert(viewer_streams_ht);
 773
 774         delret = close(stream->fd);
 775         if (delret < 0) {
 776                 PERROR("close stream");
 777         }
 778
 779         if (stream->index_fd >= 0) {
 780                 delret = close(stream->index_fd);
 781                 if (delret < 0) {
 782                         PERROR("close stream index_fd");
 783                 }
 784         }
 785
 786         vstream = live_find_viewer_stream_by_id(stream->stream_handle,
 787                         viewer_streams_ht);
 788         if (vstream) {
 789                 /*
 790                  * Set the last good value into the viewer stream. This is done
 791                  * right before the stream gets deleted from the hash table. The
 792                  * lookup failure on the live thread side of a stream indicates
 793                  * that the viewer stream index received value should be used.
 794                  */
 795                 vstream->total_index_received = stream->total_index_received;
 796         }
 797
 798         iter.iter.node = &stream->stream_n.node;
 799         delret = lttng_ht_del(relay_streams_ht, &iter);
 800         assert(!delret);
 801         iter.iter.node = &stream->ctf_trace_node.node;
 802         delret = lttng_ht_del(ctf_traces_ht, &iter);
 803         assert(!delret);
 804         call_rcu(&stream->rcu_node, deferred_free_stream);
 805         DBG("Closed tracefile %d from close stream", stream->fd);
 806 }
 807
 808 /*
 809  * relay_delete_session: Free all memory associated with a session and
 810  * close all the FDs
 811  */
 812 static
 813 void relay_delete_session(struct relay_command *cmd,
 814                 struct lttng_ht *sessions_ht)
 815 {
 816         struct lttng_ht_iter iter;
 817         struct lttng_ht_node_ulong *node;
 818         struct relay_stream *stream;
 819         int ret;
 820
 821         if (!cmd->session) {
 822                 return;
 823         }
 824
 825         DBG("Relay deleting session %" PRIu64, cmd->session->id);
 826
 827         rcu_read_lock();
 828         cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, node, node) {
 829                 node = lttng_ht_iter_get_node_ulong(&iter);
 830                 if (node) {
 831                         stream = caa_container_of(node,
 832                                         struct relay_stream, stream_n);
 833                         if (stream->session == cmd->session) {
 834                                 ret = close(stream->fd);
 835                                 if (ret < 0) {
 836                                         PERROR("close stream fd on delete session");
 837                                 }
 838                                 ret = lttng_ht_del(relay_streams_ht, &iter);
 839                                 assert(!ret);
 840                                 call_rcu(&stream->rcu_node,
 841                                         deferred_free_stream);
 842                         }
 843                         /* Cleanup index of that stream. */
 844                         relay_index_destroy_by_stream_id(stream->stream_handle,
 845                                         indexes_ht);
 846                 }
 847         }
 848         iter.iter.node = &cmd->session->session_n.node;
 849         ret = lttng_ht_del(sessions_ht, &iter);
 850         assert(!ret);
 851         call_rcu(&cmd->session->rcu_node,
 852                         deferred_free_session);
 853         rcu_read_unlock();
 854 }
 855
 856 /*
 857  * Copy index data from the control port to a given index object.
 858  */
 859 static void copy_index_control_data(struct relay_index *index,
 860                 struct lttcomm_relayd_index *data)
 861 {
 862         assert(index);
 863         assert(data);
 864
 865         /*
 866          * The index on disk is encoded in big endian, so we don't need to convert
 867          * the data received on the network. The data_offset value is NEVER
 868          * modified here and is updated by the data thread.
 869          */
 870         index->index_data.packet_size = data->packet_size;
 871         index->index_data.content_size = data->content_size;
 872         index->index_data.timestamp_begin = data->timestamp_begin;
 873         index->index_data.timestamp_end = data->timestamp_end;
 874         index->index_data.events_discarded = data->events_discarded;
 875         index->index_data.stream_id = data->stream_id;
 876 }
 877
 878 /*
 879  * Handle the RELAYD_CREATE_SESSION command.
 880  *
 881  * On success, send back the session id or else return a negative value.
 882  */
 883 static
 884 int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr,
 885                 struct relay_command *cmd,
 886                 struct lttng_ht *sessions_ht)
 887 {
 888         int ret = 0, send_ret;
 889         struct relay_session *session;
 890         struct lttcomm_relayd_status_session reply;
 891
 892         assert(recv_hdr);
 893         assert(cmd);
 894
 895         memset(&reply, 0, sizeof(reply));
 896
 897         session = zmalloc(sizeof(struct relay_session));
 898         if (session == NULL) {
 899                 PERROR("relay session zmalloc");
 900                 ret = -1;
 901                 goto error;
 902         }
 903
 904         session->id = ++last_relay_session_id;
 905         session->sock = cmd->sock;
 906         cmd->session = session;
 907
 908         reply.session_id = htobe64(session->id);
 909
 910         switch (cmd->minor) {
 911                 case 4: /* LTTng sessiond 2.4 */
 912                 default:
 913                         ret = cmd_create_session_2_4(cmd, session);
 914                         break;
 915         }
 916
 917         lttng_ht_node_init_ulong(&session->session_n,
 918                         (unsigned long) session->id);
 919         lttng_ht_add_unique_ulong(sessions_ht,
 920                         &session->session_n);
 921
 922         DBG("Created session %" PRIu64, session->id);
 923
 924 error:
 925         if (ret < 0) {
 926                 reply.ret_code = htobe32(LTTNG_ERR_FATAL);
 927         } else {
 928                 reply.ret_code = htobe32(LTTNG_OK);
 929         }
 930
 931         send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
 932         if (send_ret < 0) {
 933                 ERR("Relayd sending session id");
 934                 ret = send_ret;
 935         }
 936
 937         return ret;
 938 }
 939
 940 /*
 941  * relay_add_stream: allocate a new stream for a session
 942  */
 943 static
 944 int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
 945                 struct relay_command *cmd, struct lttng_ht *sessions_ht)
 946 {
 947         struct relay_session *session = cmd->session;
 948         struct relay_stream *stream = NULL;
 949         struct lttcomm_relayd_status_stream reply;
 950         int ret, send_ret;
 951
 952         if (!session || cmd->version_check_done == 0) {
 953                 ERR("Trying to add a stream before version check");
 954                 ret = -1;
 955                 goto end_no_session;
 956         }
 957
 958         stream = zmalloc(sizeof(struct relay_stream));
 959         if (stream == NULL) {
 960                 PERROR("relay stream zmalloc");
 961                 ret = -1;
 962                 goto end_no_session;
 963         }
 964
 965         switch (cmd->minor) {
 966         case 1: /* LTTng sessiond 2.1 */
 967                 ret = cmd_recv_stream_2_1(cmd, stream);
 968                 break;
 969         case 2: /* LTTng sessiond 2.2 */
 970         default:
 971                 ret = cmd_recv_stream_2_2(cmd, stream);
 972                 break;
 973         }
 974         if (ret < 0) {
 975                 goto err_free_stream;
 976         }
 977
 978         rcu_read_lock();
 979         stream->stream_handle = ++last_relay_stream_id;
 980         stream->prev_seq = -1ULL;
 981         stream->session = session;
 982         stream->index_fd = -1;
 983         stream->read_index_fd = -1;
 984         stream->ctf_trace = NULL;
 985         pthread_mutex_init(&stream->lock, NULL);
 986
 987         ret = utils_mkdir_recursive(stream->path_name, S_IRWXU | S_IRWXG);
 988         if (ret < 0) {
 989                 ERR("relay creating output directory");
 990                 goto end;
 991         }
 992
 993         /*
 994          * No need to use run_as API here because whatever we receives, the relayd
 995          * uses its own credentials for the stream files.
 996          */
 997         ret = utils_create_stream_file(stream->path_name, stream->channel_name,
 998                         stream->tracefile_size, 0, relayd_uid, relayd_gid, NULL);
 999         if (ret < 0) {
1000                 ERR("Create output file");
1001                 goto end;
1002         }
1003         stream->fd = ret;
1004         if (stream->tracefile_size) {
1005                 DBG("Tracefile %s/%s_0 created", stream->path_name, stream->channel_name);
1006         } else {
1007                 DBG("Tracefile %s/%s created", stream->path_name, stream->channel_name);
1008         }
1009
1010         if (!strncmp(stream->channel_name, DEFAULT_METADATA_NAME, NAME_MAX)) {
1011                 stream->metadata_flag = 1;
1012                 /*
1013                  * When we receive a new metadata stream, we create a new
1014                  * ctf_trace and we assign this ctf_trace to all streams with
1015                  * the same path.
1016                  *
1017                  * If later on we receive a new stream for the same ctf_trace,
1018                  * we copy the information from the first hit in the HT to the
1019                  * new stream.
1020                  */
1021                 stream->ctf_trace = ctf_trace_create();
1022                 if (!stream->ctf_trace) {
1023                         ret = -1;
1024                         goto end;
1025                 }
1026                 stream->ctf_trace->refcount++;
1027                 stream->ctf_trace->metadata_stream = stream;
1028         }
1029         ctf_trace_assign(cmd->ctf_traces_ht, stream);
1030
1031         lttng_ht_node_init_ulong(&stream->stream_n,
1032                         (unsigned long) stream->stream_handle);
1033         lttng_ht_add_unique_ulong(relay_streams_ht,
1034                         &stream->stream_n);
1035
1036         lttng_ht_node_init_str(&stream->ctf_trace_node, stream->path_name);
1037         lttng_ht_add_str(cmd->ctf_traces_ht, &stream->ctf_trace_node);
1038
1039         DBG("Relay new stream added %s with ID %" PRIu64, stream->channel_name,
1040                         stream->stream_handle);
1041
1042 end:
1043         reply.handle = htobe64(stream->stream_handle);
1044         /* send the session id to the client or a negative return code on error */
1045         if (ret < 0) {
1046                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1047                 /* stream was not properly added to the ht, so free it */
1048                 free(stream);
1049         } else {
1050                 reply.ret_code = htobe32(LTTNG_OK);
1051         }
1052
1053         send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1054                         sizeof(struct lttcomm_relayd_status_stream), 0);
1055         if (send_ret < 0) {
1056                 ERR("Relay sending stream id");
1057                 ret = send_ret;
1058         }
1059         rcu_read_unlock();
1060
1061 end_no_session:
1062         return ret;
1063
1064 err_free_stream:
1065         free(stream->path_name);
1066         free(stream->channel_name);
1067         free(stream);
1068         return ret;
1069 }
1070
1071 /*
1072  * relay_close_stream: close a specific stream
1073  */
1074 static
1075 int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
1076                 struct relay_command *cmd, struct lttng_ht *viewer_streams_ht)
1077 {
1078         int ret, send_ret;
1079         struct relay_session *session = cmd->session;
1080         struct lttcomm_relayd_close_stream stream_info;
1081         struct lttcomm_relayd_generic_reply reply;
1082         struct relay_stream *stream;
1083
1084         DBG("Close stream received");
1085
1086         if (!session || cmd->version_check_done == 0) {
1087                 ERR("Trying to close a stream before version check");
1088                 ret = -1;
1089                 goto end_no_session;
1090         }
1091
1092         ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
1093                         sizeof(struct lttcomm_relayd_close_stream), 0);
1094         if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
1095                 if (ret == 0) {
1096                         /* Orderly shutdown. Not necessary to print an error. */
1097                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1098                 } else {
1099                         ERR("Relay didn't receive valid add_stream struct size : %d", ret);
1100                 }
1101                 ret = -1;
1102                 goto end_no_session;
1103         }
1104
1105         rcu_read_lock();
1106         stream = relay_stream_find_by_id(be64toh(stream_info.stream_id));
1107         if (!stream) {
1108                 ret = -1;
1109                 goto end_unlock;
1110         }
1111
1112         stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
1113         stream->close_flag = 1;
1114
1115         if (close_stream_check(stream)) {
1116                 close_stream(stream, viewer_streams_ht, cmd->ctf_traces_ht);
1117         }
1118
1119 end_unlock:
1120         rcu_read_unlock();
1121
1122         if (ret < 0) {
1123                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1124         } else {
1125                 reply.ret_code = htobe32(LTTNG_OK);
1126         }
1127         send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1128                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1129         if (send_ret < 0) {
1130                 ERR("Relay sending stream id");
1131                 ret = send_ret;
1132         }
1133
1134 end_no_session:
1135         return ret;
1136 }
1137
1138 /*
1139  * relay_unknown_command: send -1 if received unknown command
1140  */
1141 static
1142 void relay_unknown_command(struct relay_command *cmd)
1143 {
1144         struct lttcomm_relayd_generic_reply reply;
1145         int ret;
1146
1147         reply.ret_code = htobe32(LTTNG_ERR_UNK);
1148         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1149                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1150         if (ret < 0) {
1151                 ERR("Relay sending unknown command");
1152         }
1153 }
1154
1155 /*
1156  * relay_start: send an acknowledgment to the client to tell if we are
1157  * ready to receive data. We are ready if a session is established.
1158  */
1159 static
1160 int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1161                 struct relay_command *cmd)
1162 {
1163         int ret = htobe32(LTTNG_OK);
1164         struct lttcomm_relayd_generic_reply reply;
1165         struct relay_session *session = cmd->session;
1166
1167         if (!session) {
1168                 DBG("Trying to start the streaming without a session established");
1169                 ret = htobe32(LTTNG_ERR_UNK);
1170         }
1171
1172         reply.ret_code = ret;
1173         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1174                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1175         if (ret < 0) {
1176                 ERR("Relay sending start ack");
1177         }
1178
1179         return ret;
1180 }
1181
1182 /*
1183  * Append padding to the file pointed by the file descriptor fd.
1184  */
1185 static int write_padding_to_file(int fd, uint32_t size)
1186 {
1187         int ret = 0;
1188         char *zeros;
1189
1190         if (size == 0) {
1191                 goto end;
1192         }
1193
1194         zeros = zmalloc(size);
1195         if (zeros == NULL) {
1196                 PERROR("zmalloc zeros for padding");
1197                 ret = -1;
1198                 goto end;
1199         }
1200
1201         do {
1202                 ret = write(fd, zeros, size);
1203         } while (ret < 0 && errno == EINTR);
1204         if (ret < 0 || ret != size) {
1205                 PERROR("write padding to file");
1206         }
1207
1208         free(zeros);
1209
1210 end:
1211         return ret;
1212 }
1213
1214 /*
1215  * relay_recv_metadata: receive the metada for the session.
1216  */
1217 static
1218 int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1219                 struct relay_command *cmd)
1220 {
1221         int ret = htobe32(LTTNG_OK);
1222         struct relay_session *session = cmd->session;
1223         struct lttcomm_relayd_metadata_payload *metadata_struct;
1224         struct relay_stream *metadata_stream;
1225         uint64_t data_size, payload_size;
1226
1227         if (!session) {
1228                 ERR("Metadata sent before version check");
1229                 ret = -1;
1230                 goto end;
1231         }
1232
1233         data_size = payload_size = be64toh(recv_hdr->data_size);
1234         if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1235                 ERR("Incorrect data size");
1236                 ret = -1;
1237                 goto end;
1238         }
1239         payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
1240
1241         if (data_buffer_size < data_size) {
1242                 /* In case the realloc fails, we can free the memory */
1243                 char *tmp_data_ptr;
1244
1245                 tmp_data_ptr = realloc(data_buffer, data_size);
1246                 if (!tmp_data_ptr) {
1247                         ERR("Allocating data buffer");
1248                         free(data_buffer);
1249                         ret = -1;
1250                         goto end;
1251                 }
1252                 data_buffer = tmp_data_ptr;
1253                 data_buffer_size = data_size;
1254         }
1255         memset(data_buffer, 0, data_size);
1256         DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
1257         ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
1258         if (ret < 0 || ret != data_size) {
1259                 if (ret == 0) {
1260                         /* Orderly shutdown. Not necessary to print an error. */
1261                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1262                 } else {
1263                         ERR("Relay didn't receive the whole metadata");
1264                 }
1265                 ret = -1;
1266                 goto end;
1267         }
1268         metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
1269
1270         rcu_read_lock();
1271         metadata_stream = relay_stream_find_by_id(
1272                         be64toh(metadata_struct->stream_id));
1273         if (!metadata_stream) {
1274                 ret = -1;
1275                 goto end_unlock;
1276         }
1277
1278         do {
1279                 ret = write(metadata_stream->fd, metadata_struct->payload,
1280                                 payload_size);
1281         } while (ret < 0 && errno == EINTR);
1282         if (ret < 0 || ret != payload_size) {
1283                 ERR("Relay error writing metadata on file");
1284                 ret = -1;
1285                 goto end_unlock;
1286         }
1287
1288         ret = write_padding_to_file(metadata_stream->fd,
1289                         be32toh(metadata_struct->padding_size));
1290         if (ret < 0) {
1291                 goto end_unlock;
1292         }
1293         metadata_stream->ctf_trace->metadata_received +=
1294                 payload_size + be32toh(metadata_struct->padding_size);
1295
1296         DBG2("Relay metadata written");
1297
1298 end_unlock:
1299         rcu_read_unlock();
1300 end:
1301         return ret;
1302 }
1303
1304 /*
1305  * relay_send_version: send relayd version number
1306  */
1307 static
1308 int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1309                 struct relay_command *cmd, struct lttng_ht *sessions_ht)
1310 {
1311         int ret;
1312         struct lttcomm_relayd_version reply, msg;
1313
1314         assert(cmd);
1315
1316         cmd->version_check_done = 1;
1317
1318         /* Get version from the other side. */
1319         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1320         if (ret < 0 || ret != sizeof(msg)) {
1321                 if (ret == 0) {
1322                         /* Orderly shutdown. Not necessary to print an error. */
1323                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1324                 } else {
1325                         ERR("Relay failed to receive the version values.");
1326                 }
1327                 ret = -1;
1328                 goto end;
1329         }
1330
1331         reply.major = RELAYD_VERSION_COMM_MAJOR;
1332         reply.minor = RELAYD_VERSION_COMM_MINOR;
1333
1334         /* Major versions must be the same */
1335         if (reply.major != be32toh(msg.major)) {
1336                 DBG("Incompatible major versions (%u vs %u), deleting session",
1337                                 reply.major, be32toh(msg.major));
1338                 relay_delete_session(cmd, sessions_ht);
1339                 ret = 0;
1340                 goto end;
1341         }
1342
1343         cmd->major = reply.major;
1344         /* We adapt to the lowest compatible version */
1345         if (reply.minor <= be32toh(msg.minor)) {
1346                 cmd->minor = reply.minor;
1347         } else {
1348                 cmd->minor = be32toh(msg.minor);
1349         }
1350
1351         reply.major = htobe32(reply.major);
1352         reply.minor = htobe32(reply.minor);
1353         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1354                         sizeof(struct lttcomm_relayd_version), 0);
1355         if (ret < 0) {
1356                 ERR("Relay sending version");
1357         }
1358
1359         DBG("Version check done using protocol %u.%u", cmd->major,
1360                         cmd->minor);
1361
1362 end:
1363         return ret;
1364 }
1365
1366 /*
1367  * Check for data pending for a given stream id from the session daemon.
1368  */
1369 static
1370 int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1371                 struct relay_command *cmd)
1372 {
1373         struct relay_session *session = cmd->session;
1374         struct lttcomm_relayd_data_pending msg;
1375         struct lttcomm_relayd_generic_reply reply;
1376         struct relay_stream *stream;
1377         int ret;
1378         uint64_t last_net_seq_num, stream_id;
1379
1380         DBG("Data pending command received");
1381
1382         if (!session || cmd->version_check_done == 0) {
1383                 ERR("Trying to check for data before version check");
1384                 ret = -1;
1385                 goto end_no_session;
1386         }
1387
1388         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1389         if (ret < sizeof(msg)) {
1390                 if (ret == 0) {
1391                         /* Orderly shutdown. Not necessary to print an error. */
1392                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1393                 } else {
1394                         ERR("Relay didn't receive valid data_pending struct size : %d",
1395                                         ret);
1396                 }
1397                 ret = -1;
1398                 goto end_no_session;
1399         }
1400
1401         stream_id = be64toh(msg.stream_id);
1402         last_net_seq_num = be64toh(msg.last_net_seq_num);
1403
1404         rcu_read_lock();
1405         stream = relay_stream_find_by_id(stream_id);
1406         if (stream == NULL) {
1407                 ret = -1;
1408                 goto end_unlock;
1409         }
1410
1411         DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
1412                         " and last_seq %" PRIu64, stream_id, stream->prev_seq,
1413                         last_net_seq_num);
1414
1415         /* Avoid wrapping issue */
1416         if (((int64_t) (stream->prev_seq - last_net_seq_num)) >= 0) {
1417                 /* Data has in fact been written and is NOT pending */
1418                 ret = 0;
1419         } else {
1420                 /* Data still being streamed thus pending */
1421                 ret = 1;
1422         }
1423
1424         /* Pending check is now done. */
1425         stream->data_pending_check_done = 1;
1426
1427 end_unlock:
1428         rcu_read_unlock();
1429
1430         reply.ret_code = htobe32(ret);
1431         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1432         if (ret < 0) {
1433                 ERR("Relay data pending ret code failed");
1434         }
1435
1436 end_no_session:
1437         return ret;
1438 }
1439
1440 /*
1441  * Wait for the control socket to reach a quiescent state.
1442  *
1443  * Note that for now, when receiving this command from the session daemon, this
1444  * means that every subsequent commands or data received on the control socket
1445  * has been handled. So, this is why we simply return OK here.
1446  */
1447 static
1448 int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
1449                 struct relay_command *cmd)
1450 {
1451         int ret;
1452         uint64_t stream_id;
1453         struct relay_stream *stream;
1454         struct lttng_ht_iter iter;
1455         struct lttcomm_relayd_quiescent_control msg;
1456         struct lttcomm_relayd_generic_reply reply;
1457
1458         DBG("Checking quiescent state on control socket");
1459
1460         if (!cmd->session || cmd->version_check_done == 0) {
1461                 ERR("Trying to check for data before version check");
1462                 ret = -1;
1463                 goto end_no_session;
1464         }
1465
1466         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1467         if (ret < sizeof(msg)) {
1468                 if (ret == 0) {
1469                         /* Orderly shutdown. Not necessary to print an error. */
1470                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1471                 } else {
1472                         ERR("Relay didn't receive valid begin data_pending struct size: %d",
1473                                         ret);
1474                 }
1475                 ret = -1;
1476                 goto end_no_session;
1477         }
1478
1479         stream_id = be64toh(msg.stream_id);
1480
1481         rcu_read_lock();
1482         cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
1483                         stream_n.node) {
1484                 if (stream->stream_handle == stream_id) {
1485                         stream->data_pending_check_done = 1;
1486                         DBG("Relay quiescent control pending flag set to %" PRIu64,
1487                                         stream_id);
1488                         break;
1489                 }
1490         }
1491         rcu_read_unlock();
1492
1493         reply.ret_code = htobe32(LTTNG_OK);
1494         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1495         if (ret < 0) {
1496                 ERR("Relay data quiescent control ret code failed");
1497         }
1498
1499 end_no_session:
1500         return ret;
1501 }
1502
1503 /*
1504  * Initialize a data pending command. This means that a client is about to ask
1505  * for data pending for each stream he/she holds. Simply iterate over all
1506  * streams of a session and set the data_pending_check_done flag.
1507  *
1508  * This command returns to the client a LTTNG_OK code.
1509  */
1510 static
1511 int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1512                 struct relay_command *cmd)
1513 {
1514         int ret;
1515         struct lttng_ht_iter iter;
1516         struct lttcomm_relayd_begin_data_pending msg;
1517         struct lttcomm_relayd_generic_reply reply;
1518         struct relay_stream *stream;
1519         uint64_t session_id;
1520
1521         assert(recv_hdr);
1522         assert(cmd);
1523
1524         DBG("Init streams for data pending");
1525
1526         if (!cmd->session || cmd->version_check_done == 0) {
1527                 ERR("Trying to check for data before version check");
1528                 ret = -1;
1529                 goto end_no_session;
1530         }
1531
1532         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1533         if (ret < sizeof(msg)) {
1534                 if (ret == 0) {
1535                         /* Orderly shutdown. Not necessary to print an error. */
1536                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1537                 } else {
1538                         ERR("Relay didn't receive valid begin data_pending struct size: %d",
1539                                         ret);
1540                 }
1541                 ret = -1;
1542                 goto end_no_session;
1543         }
1544
1545         session_id = be64toh(msg.session_id);
1546
1547         /*
1548          * Iterate over all streams to set the begin data pending flag. For now, the
1549          * streams are indexed by stream handle so we have to iterate over all
1550          * streams to find the one associated with the right session_id.
1551          */
1552         rcu_read_lock();
1553         cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
1554                         stream_n.node) {
1555                 if (stream->session->id == session_id) {
1556                         stream->data_pending_check_done = 0;
1557                         DBG("Set begin data pending flag to stream %" PRIu64,
1558                                         stream->stream_handle);
1559                 }
1560         }
1561         rcu_read_unlock();
1562
1563         /* All good, send back reply. */
1564         reply.ret_code = htobe32(LTTNG_OK);
1565
1566         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1567         if (ret < 0) {
1568                 ERR("Relay begin data pending send reply failed");
1569         }
1570
1571 end_no_session:
1572         return ret;
1573 }
1574
1575 /*
1576  * End data pending command. This will check, for a given session id, if each
1577  * stream associated with it has its data_pending_check_done flag set. If not,
1578  * this means that the client lost track of the stream but the data is still
1579  * being streamed on our side. In this case, we inform the client that data is
1580  * inflight.
1581  *
1582  * Return to the client if there is data in flight or not with a ret_code.
1583  */
1584 static
1585 int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1586                 struct relay_command *cmd)
1587 {
1588         int ret;
1589         struct lttng_ht_iter iter;
1590         struct lttcomm_relayd_end_data_pending msg;
1591         struct lttcomm_relayd_generic_reply reply;
1592         struct relay_stream *stream;
1593         uint64_t session_id;
1594         uint32_t is_data_inflight = 0;
1595
1596         assert(recv_hdr);
1597         assert(cmd);
1598
1599         DBG("End data pending command");
1600
1601         if (!cmd->session || cmd->version_check_done == 0) {
1602                 ERR("Trying to check for data before version check");
1603                 ret = -1;
1604                 goto end_no_session;
1605         }
1606
1607         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1608         if (ret < sizeof(msg)) {
1609                 if (ret == 0) {
1610                         /* Orderly shutdown. Not necessary to print an error. */
1611                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1612                 } else {
1613                         ERR("Relay didn't receive valid end data_pending struct size: %d",
1614                                         ret);
1615                 }
1616                 ret = -1;
1617                 goto end_no_session;
1618         }
1619
1620         session_id = be64toh(msg.session_id);
1621
1622         /* Iterate over all streams to see if the begin data pending flag is set. */
1623         rcu_read_lock();
1624         cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
1625                         stream_n.node) {
1626                 if (stream->session->id == session_id &&
1627                                 !stream->data_pending_check_done) {
1628                         is_data_inflight = 1;
1629                         DBG("Data is still in flight for stream %" PRIu64,
1630                                         stream->stream_handle);
1631                         break;
1632                 }
1633         }
1634         rcu_read_unlock();
1635
1636         /* All good, send back reply. */
1637         reply.ret_code = htobe32(is_data_inflight);
1638
1639         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1640         if (ret < 0) {
1641                 ERR("Relay end data pending send reply failed");
1642         }
1643
1644 end_no_session:
1645         return ret;
1646 }
1647
1648 /*
1649  * Receive an index for a specific stream.
1650  *
1651  * Return 0 on success else a negative value.
1652  */
1653 static
1654 int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
1655                 struct relay_command *cmd, struct lttng_ht *indexes_ht)
1656 {
1657         int ret, send_ret, index_created = 0;
1658         struct relay_session *session = cmd->session;
1659         struct lttcomm_relayd_index index_info;
1660         struct relay_index *index, *wr_index = NULL;
1661         struct lttcomm_relayd_generic_reply reply;
1662         struct relay_stream *stream;
1663         uint64_t net_seq_num;
1664
1665         assert(cmd);
1666         assert(indexes_ht);
1667
1668         DBG("Relay receiving index");
1669
1670         if (!session || cmd->version_check_done == 0) {
1671                 ERR("Trying to close a stream before version check");
1672                 ret = -1;
1673                 goto end_no_session;
1674         }
1675
1676         ret = cmd->sock->ops->recvmsg(cmd->sock, &index_info,
1677                         sizeof(index_info), 0);
1678         if (ret < sizeof(index_info)) {
1679                 if (ret == 0) {
1680                         /* Orderly shutdown. Not necessary to print an error. */
1681                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1682                 } else {
1683                         ERR("Relay didn't receive valid index struct size : %d", ret);
1684                 }
1685                 ret = -1;
1686                 goto end_no_session;
1687         }
1688
1689         net_seq_num = be64toh(index_info.net_seq_num);
1690
1691         rcu_read_lock();
1692         stream = relay_stream_find_by_id(be64toh(index_info.relay_stream_id));
1693         if (!stream) {
1694                 ret = -1;
1695                 goto end_rcu_unlock;
1696         }
1697
1698         /* Live beacon handling */
1699         if (index_info.packet_size == 0) {
1700                 DBG("Received live beacon for stream %" PRIu64, stream->stream_handle);
1701
1702                 /*
1703                  * Only flag a stream inactive when it has already received data.
1704                  */
1705                 if (stream->total_index_received > 0) {
1706                         stream->beacon_ts_end = be64toh(index_info.timestamp_end);
1707                 }
1708                 ret = 0;
1709                 goto end_rcu_unlock;
1710         } else {
1711                 stream->beacon_ts_end = -1ULL;
1712         }
1713
1714         index = relay_index_find(stream->stream_handle, net_seq_num, indexes_ht);
1715         if (!index) {
1716                 /* A successful creation will add the object to the HT. */
1717                 index = relay_index_create(stream->stream_handle, net_seq_num);
1718                 if (!index) {
1719                         goto end_rcu_unlock;
1720                 }
1721                 index_created = 1;
1722         }
1723
1724         copy_index_control_data(index, &index_info);
1725
1726         if (index_created) {
1727                 /*
1728                  * Try to add the relay index object to the hash table. If an object
1729                  * already exist, destroy back the index created, set the data in this
1730                  * object and write it on disk.
1731                  */
1732                 relay_index_add(index, indexes_ht, &wr_index);
1733                 if (wr_index) {
1734                         copy_index_control_data(wr_index, &index_info);
1735                         free(index);
1736                 }
1737         } else {
1738                 /* The index already exists so write it on disk. */
1739                 wr_index = index;
1740         }
1741
1742         /* Do we have a writable ready index to write on disk. */
1743         if (wr_index) {
1744                 /* Starting at 2.4, create the index file if none available. */
1745                 if (cmd->minor >= 4 && stream->index_fd < 0) {
1746                         ret = index_create_file(stream->path_name, stream->channel_name,
1747                                         relayd_uid, relayd_gid, stream->tracefile_size,
1748                                         stream->tracefile_count_current);
1749                         if (ret < 0) {
1750                                 goto end_rcu_unlock;
1751                         }
1752                         stream->index_fd = ret;
1753                 }
1754
1755                 ret = relay_index_write(wr_index->fd, wr_index, indexes_ht);
1756                 if (ret < 0) {
1757                         goto end_rcu_unlock;
1758                 }
1759                 stream->total_index_received++;
1760         }
1761
1762 end_rcu_unlock:
1763         rcu_read_unlock();
1764
1765         if (ret < 0) {
1766                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1767         } else {
1768                 reply.ret_code = htobe32(LTTNG_OK);
1769         }
1770         send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1771         if (send_ret < 0) {
1772                 ERR("Relay sending close index id reply");
1773                 ret = send_ret;
1774         }
1775
1776 end_no_session:
1777         return ret;
1778 }
1779
1780 /*
1781  * Process the commands received on the control socket
1782  */
1783 static
1784 int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1785                 struct relay_command *cmd, struct relay_local_data *ctx)
1786 {
1787         int ret = 0;
1788
1789         switch (be32toh(recv_hdr->cmd)) {
1790         case RELAYD_CREATE_SESSION:
1791                 ret = relay_create_session(recv_hdr, cmd, ctx->sessions_ht);
1792                 break;
1793         case RELAYD_ADD_STREAM:
1794                 ret = relay_add_stream(recv_hdr, cmd, ctx->sessions_ht);
1795                 break;
1796         case RELAYD_START_DATA:
1797                 ret = relay_start(recv_hdr, cmd);
1798                 break;
1799         case RELAYD_SEND_METADATA:
1800                 ret = relay_recv_metadata(recv_hdr, cmd);
1801                 break;
1802         case RELAYD_VERSION:
1803                 ret = relay_send_version(recv_hdr, cmd, ctx->sessions_ht);
1804                 break;
1805         case RELAYD_CLOSE_STREAM:
1806                 ret = relay_close_stream(recv_hdr, cmd, ctx->viewer_streams_ht);
1807                 break;
1808         case RELAYD_DATA_PENDING:
1809                 ret = relay_data_pending(recv_hdr, cmd);
1810                 break;
1811         case RELAYD_QUIESCENT_CONTROL:
1812                 ret = relay_quiescent_control(recv_hdr, cmd);
1813                 break;
1814         case RELAYD_BEGIN_DATA_PENDING:
1815                 ret = relay_begin_data_pending(recv_hdr, cmd);
1816                 break;
1817         case RELAYD_END_DATA_PENDING:
1818                 ret = relay_end_data_pending(recv_hdr, cmd);
1819                 break;
1820         case RELAYD_SEND_INDEX:
1821                 ret = relay_recv_index(recv_hdr, cmd, indexes_ht);
1822                 break;
1823         case RELAYD_UPDATE_SYNC_INFO:
1824         default:
1825                 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1826                 relay_unknown_command(cmd);
1827                 ret = -1;
1828                 goto end;
1829         }
1830
1831 end:
1832         return ret;
1833 }
1834
1835 /*
1836  * relay_process_data: Process the data received on the data socket
1837  */
1838 static
1839 int relay_process_data(struct relay_command *cmd,
1840                 struct lttng_ht *indexes_ht, struct lttng_ht *viewer_streams_ht)
1841 {
1842         int ret = 0, rotate_index = 0, index_created = 0;
1843         struct relay_stream *stream;
1844         struct relay_index *index, *wr_index = NULL;
1845         struct lttcomm_relayd_data_hdr data_hdr;
1846         uint64_t stream_id, data_offset;
1847         uint64_t net_seq_num;
1848         uint32_t data_size;
1849
1850         ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1851                         sizeof(struct lttcomm_relayd_data_hdr), 0);
1852         if (ret <= 0) {
1853                 if (ret == 0) {
1854                         /* Orderly shutdown. Not necessary to print an error. */
1855                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1856                 } else {
1857                         ERR("Unable to receive data header on sock %d", cmd->sock->fd);
1858                 }
1859                 ret = -1;
1860                 goto end;
1861         }
1862
1863         stream_id = be64toh(data_hdr.stream_id);
1864
1865         rcu_read_lock();
1866         stream = relay_stream_find_by_id(stream_id);
1867         if (!stream) {
1868                 ret = -1;
1869                 goto end_rcu_unlock;
1870         }
1871
1872         data_size = be32toh(data_hdr.data_size);
1873         if (data_buffer_size < data_size) {
1874                 char *tmp_data_ptr;
1875
1876                 tmp_data_ptr = realloc(data_buffer, data_size);
1877                 if (!tmp_data_ptr) {
1878                         ERR("Allocating data buffer");
1879                         free(data_buffer);
1880                         ret = -1;
1881                         goto end_rcu_unlock;
1882                 }
1883                 data_buffer = tmp_data_ptr;
1884                 data_buffer_size = data_size;
1885         }
1886         memset(data_buffer, 0, data_size);
1887
1888         net_seq_num = be64toh(data_hdr.net_seq_num);
1889
1890         DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
1891                 data_size, stream_id, net_seq_num);
1892         ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
1893         if (ret <= 0) {
1894                 if (ret == 0) {
1895                         /* Orderly shutdown. Not necessary to print an error. */
1896                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1897                 }
1898                 ret = -1;
1899                 goto end_rcu_unlock;
1900         }
1901
1902         /* Check if a rotation is needed. */
1903         if (stream->tracefile_size > 0 &&
1904                         (stream->tracefile_size_current + data_size) >
1905                         stream->tracefile_size) {
1906                 ret = utils_rotate_stream_file(stream->path_name, stream->channel_name,
1907                                 stream->tracefile_size, stream->tracefile_count,
1908                                 relayd_uid, relayd_gid, stream->fd,
1909                                 &(stream->tracefile_count_current), &stream->fd);
1910                 if (ret < 0) {
1911                         ERR("Rotating stream output file");
1912                         goto end_rcu_unlock;
1913                 }
1914                 /* Reset current size because we just perform a stream rotation. */
1915                 stream->tracefile_size_current = 0;
1916                 rotate_index = 1;
1917         }
1918
1919         /* Get data offset because we are about to update the index. */
1920         data_offset = htobe64(stream->tracefile_size_current);
1921
1922         /*
1923          * Lookup for an existing index for that stream id/sequence number. If on
1924          * exists, the control thread already received the data for it thus we need
1925          * to write it on disk.
1926          */
1927         index = relay_index_find(stream_id, net_seq_num, indexes_ht);
1928         if (!index) {
1929                 /* A successful creation will add the object to the HT. */
1930                 index = relay_index_create(stream->stream_handle, net_seq_num);
1931                 if (!index) {
1932                         goto end_rcu_unlock;
1933                 }
1934                 index_created = 1;
1935         }
1936
1937         if (rotate_index || stream->index_fd < 0) {
1938                 index->to_close_fd = stream->index_fd;
1939                 ret = index_create_file(stream->path_name, stream->channel_name,
1940                                 relayd_uid, relayd_gid, stream->tracefile_size,
1941                                 stream->tracefile_count_current);
1942                 if (ret < 0) {
1943                         /* This will close the stream's index fd if one. */
1944                         relay_index_free_safe(index);
1945                         goto end_rcu_unlock;
1946                 }
1947                 stream->index_fd = ret;
1948         }
1949         index->fd = stream->index_fd;
1950         index->index_data.offset = data_offset;
1951
1952         if (index_created) {
1953                 /*
1954                  * Try to add the relay index object to the hash table. If an object
1955                  * already exist, destroy back the index created and set the data.
1956                  */
1957                 relay_index_add(index, indexes_ht, &wr_index);
1958                 if (wr_index) {
1959                         /* Copy back data from the created index. */
1960                         wr_index->fd = index->fd;
1961                         wr_index->to_close_fd = index->to_close_fd;
1962                         wr_index->index_data.offset = data_offset;
1963                         free(index);
1964                 }
1965         } else {
1966                 /* The index already exists so write it on disk. */
1967                 wr_index = index;
1968         }
1969
1970         /* Do we have a writable ready index to write on disk. */
1971         if (wr_index) {
1972                 /* Starting at 2.4, create the index file if none available. */
1973                 if (cmd->minor >= 4 && stream->index_fd < 0) {
1974                         ret = index_create_file(stream->path_name, stream->channel_name,
1975                                         relayd_uid, relayd_gid, stream->tracefile_size,
1976                                         stream->tracefile_count_current);
1977                         if (ret < 0) {
1978                                 goto end_rcu_unlock;
1979                         }
1980                         stream->index_fd = ret;
1981                 }
1982
1983                 ret = relay_index_write(wr_index->fd, wr_index, indexes_ht);
1984                 if (ret < 0) {
1985                         goto end_rcu_unlock;
1986                 }
1987                 stream->total_index_received++;
1988         }
1989
1990         do {
1991                 ret = write(stream->fd, data_buffer, data_size);
1992         } while (ret < 0 && errno == EINTR);
1993         if (ret < 0 || ret != data_size) {
1994                 ERR("Relay error writing data to file");
1995                 ret = -1;
1996                 goto end_rcu_unlock;
1997         }
1998
1999         DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
2000                         ret, stream->stream_handle);
2001
2002         ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
2003         if (ret < 0) {
2004                 goto end_rcu_unlock;
2005         }
2006         stream->tracefile_size_current += data_size + be32toh(data_hdr.padding_size);
2007
2008         stream->prev_seq = net_seq_num;
2009
2010         /* Check if we need to close the FD */
2011         if (close_stream_check(stream)) {
2012                 close_stream(stream, viewer_streams_ht, cmd->ctf_traces_ht);
2013         }
2014
2015 end_rcu_unlock:
2016         rcu_read_unlock();
2017 end:
2018         return ret;
2019 }
2020
2021 static
2022 void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
2023 {
2024         int ret;
2025
2026         lttng_poll_del(events, pollfd);
2027
2028         ret = close(pollfd);
2029         if (ret < 0) {
2030                 ERR("Closing pollfd %d", pollfd);
2031         }
2032 }
2033
2034 static
2035 int relay_add_connection(int fd, struct lttng_poll_event *events,
2036                 struct lttng_ht *relay_connections_ht)
2037 {
2038         struct relay_command *relay_connection;
2039         int ret;
2040
2041         relay_connection = zmalloc(sizeof(struct relay_command));
2042         if (relay_connection == NULL) {
2043                 PERROR("Relay command zmalloc");
2044                 goto error;
2045         }
2046         do {
2047                 ret = read(fd, relay_connection, sizeof(struct relay_command));
2048         } while (ret < 0 && errno == EINTR);
2049         if (ret < 0 || ret < sizeof(struct relay_command)) {
2050                 PERROR("read relay cmd pipe");
2051                 goto error_read;
2052         }
2053
2054         relay_connection->ctf_traces_ht = lttng_ht_new(0, LTTNG_HT_TYPE_STRING);
2055         if (!relay_connection->ctf_traces_ht) {
2056                 goto error_read;
2057         }
2058
2059         lttng_ht_node_init_ulong(&relay_connection->sock_n,
2060                         (unsigned long) relay_connection->sock->fd);
2061         rcu_read_lock();
2062         lttng_ht_add_unique_ulong(relay_connections_ht,
2063                         &relay_connection->sock_n);
2064         rcu_read_unlock();
2065         return lttng_poll_add(events,
2066                         relay_connection->sock->fd,
2067                         LPOLLIN | LPOLLRDHUP);
2068
2069 error_read:
2070         free(relay_connection);
2071 error:
2072         return -1;
2073 }
2074
2075 static
2076 void deferred_free_connection(struct rcu_head *head)
2077 {
2078         struct relay_command *relay_connection =
2079                 caa_container_of(head, struct relay_command, rcu_node);
2080
2081         lttng_ht_destroy(relay_connection->ctf_traces_ht);
2082         lttcomm_destroy_sock(relay_connection->sock);
2083         free(relay_connection);
2084 }
2085
2086 static
2087 void relay_del_connection(struct lttng_ht *relay_connections_ht,
2088                 struct lttng_ht_iter *iter, struct relay_command *relay_connection,
2089                 struct lttng_ht *sessions_ht)
2090 {
2091         int ret;
2092
2093         ret = lttng_ht_del(relay_connections_ht, iter);
2094         assert(!ret);
2095         if (relay_connection->type == RELAY_CONTROL) {
2096                 relay_delete_session(relay_connection, sessions_ht);
2097         }
2098
2099         call_rcu(&relay_connection->rcu_node,
2100                 deferred_free_connection);
2101 }
2102
2103 /*
2104  * This thread does the actual work
2105  */
2106 static
2107 void *relay_thread_worker(void *data)
2108 {
2109         int ret, err = -1, last_seen_data_fd = -1;
2110         uint32_t nb_fd;
2111         struct relay_command *relay_connection;
2112         struct lttng_poll_event events;
2113         struct lttng_ht *relay_connections_ht;
2114         struct lttng_ht_node_ulong *node;
2115         struct lttng_ht_iter iter;
2116         struct lttcomm_relayd_hdr recv_hdr;
2117         struct relay_local_data *relay_ctx = (struct relay_local_data *) data;
2118         struct lttng_ht *sessions_ht = relay_ctx->sessions_ht;
2119
2120         DBG("[thread] Relay worker started");
2121
2122         rcu_register_thread();
2123
2124         /* table of connections indexed on socket */
2125         relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2126         if (!relay_connections_ht) {
2127                 goto relay_connections_ht_error;
2128         }
2129
2130         /* Tables of received indexes indexed by index handle and net_seq_num. */
2131         indexes_ht = lttng_ht_new(0, LTTNG_HT_TYPE_TWO_U64);
2132         if (!indexes_ht) {
2133                 goto indexes_ht_error;
2134         }
2135
2136         ret = create_thread_poll_set(&events, 2);
2137         if (ret < 0) {
2138                 goto error_poll_create;
2139         }
2140
2141         ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
2142         if (ret < 0) {
2143                 goto error;
2144         }
2145
2146 restart:
2147         while (1) {
2148                 int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
2149
2150                 /* Infinite blocking call, waiting for transmission */
2151                 DBG3("Relayd worker thread polling...");
2152                 ret = lttng_poll_wait(&events, -1);
2153                 if (ret < 0) {
2154                         /*
2155                          * Restart interrupted system call.
2156                          */
2157                         if (errno == EINTR) {
2158                                 goto restart;
2159                         }
2160                         goto error;
2161                 }
2162
2163                 nb_fd = ret;
2164
2165                 /*
2166                  * Process control. The control connection is prioritised so we don't
2167                  * starve it with high throughout put tracing data on the data
2168                  * connection.
2169                  */
2170                 for (i = 0; i < nb_fd; i++) {
2171                         /* Fetch once the poll data */
2172                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
2173                         int pollfd = LTTNG_POLL_GETFD(&events, i);
2174
2175                         /* Thread quit pipe has been closed. Killing thread. */
2176                         ret = check_thread_quit_pipe(pollfd, revents);
2177                         if (ret) {
2178                                 err = 0;
2179                                 goto exit;
2180                         }
2181
2182                         /* Inspect the relay cmd pipe for new connection */
2183                         if (pollfd == relay_cmd_pipe[0]) {
2184                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
2185                                         ERR("Relay pipe error");
2186                                         goto error;
2187                                 } else if (revents & LPOLLIN) {
2188                                         DBG("Relay command received");
2189                                         ret = relay_add_connection(relay_cmd_pipe[0],
2190                                                         &events, relay_connections_ht);
2191                                         if (ret < 0) {
2192                                                 goto error;
2193                                         }
2194                                 }
2195                         } else if (revents) {
2196                                 rcu_read_lock();
2197                                 lttng_ht_lookup(relay_connections_ht,
2198                                                 (void *)((unsigned long) pollfd),
2199                                                 &iter);
2200                                 node = lttng_ht_iter_get_node_ulong(&iter);
2201                                 if (node == NULL) {
2202                                         DBG2("Relay sock %d not found", pollfd);
2203                                         rcu_read_unlock();
2204                                         goto error;
2205                                 }
2206                                 relay_connection = caa_container_of(node,
2207                                                 struct relay_command, sock_n);
2208
2209                                 if (revents & (LPOLLERR)) {
2210                                         ERR("POLL ERROR");
2211                                         relay_cleanup_poll_connection(&events, pollfd);
2212                                         relay_del_connection(relay_connections_ht,
2213                                                         &iter, relay_connection, sessions_ht);
2214                                         if (last_seen_data_fd == pollfd) {
2215                                                 last_seen_data_fd = last_notdel_data_fd;
2216                                         }
2217                                 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
2218                                         DBG("Socket %d hung up", pollfd);
2219                                         relay_cleanup_poll_connection(&events, pollfd);
2220                                         relay_del_connection(relay_connections_ht,
2221                                                         &iter, relay_connection, sessions_ht);
2222                                         if (last_seen_data_fd == pollfd) {
2223                                                 last_seen_data_fd = last_notdel_data_fd;
2224                                         }
2225                                 } else if (revents & LPOLLIN) {
2226                                         /* control socket */
2227                                         if (relay_connection->type == RELAY_CONTROL) {
2228                                                 ret = relay_connection->sock->ops->recvmsg(
2229                                                                 relay_connection->sock, &recv_hdr,
2230                                                                 sizeof(struct lttcomm_relayd_hdr), 0);
2231                                                 /* connection closed */
2232                                                 if (ret <= 0) {
2233                                                         relay_cleanup_poll_connection(&events, pollfd);
2234                                                         relay_del_connection(relay_connections_ht,
2235                                                                         &iter, relay_connection, sessions_ht);
2236                                                         DBG("Control connection closed with %d", pollfd);
2237                                                 } else {
2238                                                         if (relay_connection->session) {
2239                                                                 DBG2("Relay worker receiving data for session : %" PRIu64,
2240                                                                                 relay_connection->session->id);
2241                                                         }
2242                                                         ret = relay_process_control(&recv_hdr,
2243                                                                         relay_connection, relay_ctx);
2244                                                         if (ret < 0) {
2245                                                                 /* Clear the session on error. */
2246                                                                 relay_cleanup_poll_connection(&events, pollfd);
2247                                                                 relay_del_connection(relay_connections_ht,
2248                                                                                 &iter, relay_connection, sessions_ht);
2249                                                                 DBG("Connection closed with %d", pollfd);
2250                                                         }
2251                                                         seen_control = 1;
2252                                                 }
2253                                         } else {
2254                                                 /*
2255                                                  * Flag the last seen data fd not deleted. It will be
2256                                                  * used as the last seen fd if any fd gets deleted in
2257                                                  * this first loop.
2258                                                  */
2259                                                 last_notdel_data_fd = pollfd;
2260                                         }
2261                                 }
2262                                 rcu_read_unlock();
2263                         }
2264                 }
2265
2266                 /*
2267                  * The last loop handled a control request, go back to poll to make
2268                  * sure we prioritise the control socket.
2269                  */
2270                 if (seen_control) {
2271                         continue;
2272                 }
2273
2274                 if (last_seen_data_fd >= 0) {
2275                         for (i = 0; i < nb_fd; i++) {
2276                                 int pollfd = LTTNG_POLL_GETFD(&events, i);
2277                                 if (last_seen_data_fd == pollfd) {
2278                                         idx = i;
2279                                         break;
2280                                 }
2281                         }
2282                 }
2283
2284                 /* Process data connection. */
2285                 for (i = idx + 1; i < nb_fd; i++) {
2286                         /* Fetch the poll data. */
2287                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
2288                         int pollfd = LTTNG_POLL_GETFD(&events, i);
2289
2290                         /* Skip the command pipe. It's handled in the first loop. */
2291                         if (pollfd == relay_cmd_pipe[0]) {
2292                                 continue;
2293                         }
2294
2295                         if (revents) {
2296                                 rcu_read_lock();
2297                                 lttng_ht_lookup(relay_connections_ht,
2298                                                 (void *)((unsigned long) pollfd),
2299                                                 &iter);
2300                                 node = lttng_ht_iter_get_node_ulong(&iter);
2301                                 if (node == NULL) {
2302                                         /* Skip it. Might be removed before. */
2303                                         rcu_read_unlock();
2304                                         continue;
2305                                 }
2306                                 relay_connection = caa_container_of(node,
2307                                                 struct relay_command, sock_n);
2308
2309                                 if (revents & LPOLLIN) {
2310                                         if (relay_connection->type != RELAY_DATA) {
2311                                                 continue;
2312                                         }
2313
2314                                         ret = relay_process_data(relay_connection, indexes_ht,
2315                                                         relay_ctx->viewer_streams_ht);
2316                                         /* connection closed */
2317                                         if (ret < 0) {
2318                                                 relay_cleanup_poll_connection(&events, pollfd);
2319                                                 relay_del_connection(relay_connections_ht,
2320                                                                 &iter, relay_connection, sessions_ht);
2321                                                 DBG("Data connection closed with %d", pollfd);
2322                                                 /*
2323                                                  * Every goto restart call sets the last seen fd where
2324                                                  * here we don't really care since we gracefully
2325                                                  * continue the loop after the connection is deleted.
2326                                                  */
2327                                         } else {
2328                                                 /* Keep last seen port. */
2329                                                 last_seen_data_fd = pollfd;
2330                                                 rcu_read_unlock();
2331                                                 goto restart;
2332                                         }
2333                                 }
2334                                 rcu_read_unlock();
2335                         }
2336                 }
2337                 last_seen_data_fd = -1;
2338         }
2339
2340 exit:
2341 error:
2342         lttng_poll_clean(&events);
2343
2344         /* empty the hash table and free the memory */
2345         rcu_read_lock();
2346         cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
2347                 node = lttng_ht_iter_get_node_ulong(&iter);
2348                 if (node) {
2349                         relay_connection = caa_container_of(node,
2350                                         struct relay_command, sock_n);
2351                         relay_del_connection(relay_connections_ht,
2352                                         &iter, relay_connection, sessions_ht);
2353                 }
2354         }
2355 error_poll_create:
2356         {
2357                 struct relay_index *index;
2358                 cds_lfht_for_each_entry(indexes_ht->ht, &iter.iter, index, index_n.node) {
2359                         relay_index_delete(index, indexes_ht);
2360                 }
2361                 lttng_ht_destroy(indexes_ht);
2362         }
2363         rcu_read_unlock();
2364 indexes_ht_error:
2365         lttng_ht_destroy(relay_connections_ht);
2366 relay_connections_ht_error:
2367         /* Close relay cmd pipes */
2368         utils_close_pipe(relay_cmd_pipe);
2369         if (err) {
2370                 DBG("Thread exited with error");
2371         }
2372         DBG("Worker thread cleanup complete");
2373         free(data_buffer);
2374         stop_threads();
2375         rcu_unregister_thread();
2376         return NULL;
2377 }
2378
2379 /*
2380  * Create the relay command pipe to wake thread_manage_apps.
2381  * Closed in cleanup().
2382  */
2383 static int create_relay_cmd_pipe(void)
2384 {
2385         int ret;
2386
2387         ret = utils_create_pipe_cloexec(relay_cmd_pipe);
2388
2389         return ret;
2390 }
2391
2392 /*
2393  * main
2394  */
2395 int main(int argc, char **argv)
2396 {
2397         int ret = 0;
2398         void *status;
2399         struct relay_local_data *relay_ctx;
2400
2401         /* Create thread quit pipe */
2402         if ((ret = init_thread_quit_pipe()) < 0) {
2403                 goto error;
2404         }
2405
2406         /* Parse arguments */
2407         progname = argv[0];
2408         if ((ret = parse_args(argc, argv)) < 0) {
2409                 goto exit;
2410         }
2411
2412         if ((ret = set_signal_handler()) < 0) {
2413                 goto exit;
2414         }
2415
2416         /* Try to create directory if -o, --output is specified. */
2417         if (opt_output_path) {
2418                 if (*opt_output_path != '/') {
2419                         ERR("Please specify an absolute path for -o, --output PATH");
2420                         goto exit;
2421                 }
2422
2423                 ret = utils_mkdir_recursive(opt_output_path, S_IRWXU | S_IRWXG);
2424                 if (ret < 0) {
2425                         ERR("Unable to create %s", opt_output_path);
2426                         goto exit;
2427                 }
2428         }
2429
2430         /* Daemonize */
2431         if (opt_daemon) {
2432                 ret = daemon(0, 0);
2433                 if (ret < 0) {
2434                         PERROR("daemon");
2435                         goto exit;
2436                 }
2437         }
2438
2439         /* We need those values for the file/dir creation. */
2440         relayd_uid = getuid();
2441         relayd_gid = getgid();
2442
2443         /* Check if daemon is UID = 0 */
2444         if (relayd_uid == 0) {
2445                 if (control_uri->port < 1024 || data_uri->port < 1024) {
2446                         ERR("Need to be root to use ports < 1024");
2447                         ret = -1;
2448                         goto exit;
2449                 }
2450         }
2451
2452         /* Setup the thread apps communication pipe. */
2453         if ((ret = create_relay_cmd_pipe()) < 0) {
2454                 goto exit;
2455         }
2456
2457         /* Init relay command queue. */
2458         cds_wfq_init(&relay_cmd_queue.queue);
2459
2460         /* Set up max poll set size */
2461         lttng_poll_set_max_size();
2462
2463         /* Initialize communication library */
2464         lttcomm_init();
2465
2466         relay_ctx = zmalloc(sizeof(struct relay_local_data));
2467         if (!relay_ctx) {
2468                 PERROR("relay_ctx");
2469                 goto exit;
2470         }
2471
2472         /* tables of sessions indexed by session ID */
2473         relay_ctx->sessions_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2474         if (!relay_ctx->sessions_ht) {
2475                 goto exit_relay_ctx_sessions;
2476         }
2477
2478         /* tables of streams indexed by stream ID */
2479         relay_streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2480         if (!relay_streams_ht) {
2481                 goto exit_relay_ctx_streams;
2482         }
2483
2484         /* tables of streams indexed by stream ID */
2485         relay_ctx->viewer_streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2486         if (!relay_ctx->viewer_streams_ht) {
2487                 goto exit_relay_ctx_viewer_streams;
2488         }
2489
2490         /* Setup the dispatcher thread */
2491         ret = pthread_create(&dispatcher_thread, NULL,
2492                         relay_thread_dispatcher, (void *) NULL);
2493         if (ret != 0) {
2494                 PERROR("pthread_create dispatcher");
2495                 goto exit_dispatcher;
2496         }
2497
2498         /* Setup the worker thread */
2499         ret = pthread_create(&worker_thread, NULL,
2500                         relay_thread_worker, (void *) relay_ctx);
2501         if (ret != 0) {
2502                 PERROR("pthread_create worker");
2503                 goto exit_worker;
2504         }
2505
2506         /* Setup the listener thread */
2507         ret = pthread_create(&listener_thread, NULL,
2508                         relay_thread_listener, (void *) NULL);
2509         if (ret != 0) {
2510                 PERROR("pthread_create listener");
2511                 goto exit_listener;
2512         }
2513
2514         ret = live_start_threads(live_uri, relay_ctx);
2515         if (ret != 0) {
2516                 ERR("Starting live viewer threads");
2517         }
2518
2519 exit_listener:
2520         ret = pthread_join(listener_thread, &status);
2521         if (ret != 0) {
2522                 PERROR("pthread_join");
2523                 goto error;     /* join error, exit without cleanup */
2524         }
2525
2526 exit_worker:
2527         ret = pthread_join(worker_thread, &status);
2528         if (ret != 0) {
2529                 PERROR("pthread_join");
2530                 goto error;     /* join error, exit without cleanup */
2531         }
2532
2533 exit_dispatcher:
2534         ret = pthread_join(dispatcher_thread, &status);
2535         if (ret != 0) {
2536                 PERROR("pthread_join");
2537                 goto error;     /* join error, exit without cleanup */
2538         }
2539         lttng_ht_destroy(relay_ctx->viewer_streams_ht);
2540
2541 exit_relay_ctx_viewer_streams:
2542         lttng_ht_destroy(relay_streams_ht);
2543
2544 exit_relay_ctx_streams:
2545         lttng_ht_destroy(relay_ctx->sessions_ht);
2546
2547 exit_relay_ctx_sessions:
2548         free(relay_ctx);
2549
2550 exit:
2551         live_stop_threads();
2552         cleanup();
2553         if (!ret) {
2554                 exit(EXIT_SUCCESS);
2555         }
2556
2557 error:
2558         exit(EXIT_FAILURE);
2559 }