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