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