src/bin/lttng-relayd/main.c

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