src/bin/lttng-relayd/main.cpp

   1 /*
   2  * Copyright (C) 2012 Julien Desfossez <jdesfossez@efficios.com>
   3  * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
   4  * Copyright (C) 2013 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   5  * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   6  *
   7  * SPDX-License-Identifier: GPL-2.0-only
   8  *
   9  */
  10
  11 #define _LGPL_SOURCE
  12 #include <algorithm>
  13 #include <ctype.h>
  14 #include <fcntl.h>
  15 #include <getopt.h>
  16 #include <grp.h>
  17 #include <inttypes.h>
  18 #include <limits.h>
  19 #include <pthread.h>
  20 #include <signal.h>
  21 #include <stdio.h>
  22 #include <stdlib.h>
  23 #include <string.h>
  24 #include <strings.h>
  25 #include <sys/mman.h>
  26 #include <sys/mount.h>
  27 #include <sys/resource.h>
  28 #include <sys/socket.h>
  29 #include <sys/stat.h>
  30 #include <sys/types.h>
  31 #include <sys/wait.h>
  32 #include <unistd.h>
  33 #include <urcu/futex.h>
  34 #include <urcu/rculist.h>
  35 #include <urcu/uatomic.h>
  36
  37 #include <common/align.hpp>
  38 #include <common/buffer-view.hpp>
  39 #include <common/common.hpp>
  40 #include <common/compat/endian.hpp>
  41 #include <common/compat/getenv.hpp>
  42 #include <common/compat/poll.hpp>
  43 #include <common/compat/socket.hpp>
  44 #include <common/daemonize.hpp>
  45 #include <common/defaults.hpp>
  46 #include <common/dynamic-buffer.hpp>
  47 #include <common/fd-tracker/fd-tracker.hpp>
  48 #include <common/fd-tracker/utils.hpp>
  49 #include <common/futex.hpp>
  50 #include <common/ini-config/ini-config.hpp>
  51 #include <common/path.hpp>
  52 #include <common/sessiond-comm/inet.hpp>
  53 #include <common/sessiond-comm/relayd.hpp>
  54 #include <common/sessiond-comm/sessiond-comm.hpp>
  55 #include <common/string-utils/format.hpp>
  56 #include <common/uri.hpp>
  57 #include <common/utils.hpp>
  58 #include <lttng/lttng.h>
  59 #include <lttng/trace-format-descriptor-internal.hpp>
  60
  61 #include "backward-compatibility-group-by.hpp"
  62 #include "cmd.hpp"
  63 #include "connection.hpp"
  64 #include "ctf-trace.hpp"
  65 #include "health-relayd.hpp"
  66 #include "index.hpp"
  67 #include "live.hpp"
  68 #include "lttng-relayd.hpp"
  69 #include "session.hpp"
  70 #include "sessiond-trace-chunks.hpp"
  71 #include "stream.hpp"
  72 #include "tcp_keep_alive.hpp"
  73 #include "testpoint.hpp"
  74 #include "tracefile-array.hpp"
  75 #include "utils.hpp"
  76 #include "version.hpp"
  77 #include "viewer-stream.hpp"
  78
  79 static const char *help_msg =
  80 #ifdef LTTNG_EMBED_HELP
  81 #include <lttng-relayd.8.h>
  82 #else
  83 NULL
  84 #endif
  85 ;
  86
  87 enum relay_connection_status {
  88         RELAY_CONNECTION_STATUS_OK,
  89         /* An error occurred while processing an event on the connection. */
  90         RELAY_CONNECTION_STATUS_ERROR,
  91         /* Connection closed/shutdown cleanly. */
  92         RELAY_CONNECTION_STATUS_CLOSED,
  93 };
  94
  95 /* command line options */
  96 char *opt_output_path, *opt_working_directory;
  97 static int opt_daemon, opt_background, opt_print_version, opt_allow_clear = 1;
  98 static int opt_allow_ctf2 = 1;
  99 enum relay_group_output_by opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_UNKNOWN;
 100
 101 /* Argument variables */
 102 int lttng_opt_quiet;    /* not static in error.h */
 103 int lttng_opt_verbose;  /* not static in error.h */
 104 int lttng_opt_mi;       /* not static in error.h */
 105
 106 /*
 107  * We need to wait for listener and live listener threads, as well as
 108  * health check thread, before being ready to signal readiness.
 109  */
 110 #define NR_LTTNG_RELAY_READY    3
 111 static int lttng_relay_ready = NR_LTTNG_RELAY_READY;
 112
 113 /* Size of receive buffer. */
 114 #define RECV_DATA_BUFFER_SIZE           65536
 115
 116 static int recv_child_signal;   /* Set to 1 when a SIGUSR1 signal is received. */
 117 static pid_t child_ppid;        /* Internal parent PID use with daemonize. */
 118
 119 static struct lttng_uri *control_uri;
 120 static struct lttng_uri *data_uri;
 121 static struct lttng_uri *live_uri;
 122
 123 const char *progname;
 124
 125 const char *tracing_group_name = DEFAULT_TRACING_GROUP;
 126 static int tracing_group_name_override;
 127
 128 const char * const config_section_name = "relayd";
 129
 130 /*
 131  * Quit pipe for all threads. This permits a single cancellation point
 132  * for all threads when receiving an event on the pipe.
 133  */
 134 int thread_quit_pipe[2] = { -1, -1 };
 135
 136 /*
 137  * This pipe is used to inform the worker thread that a command is queued and
 138  * ready to be processed.
 139  */
 140 static int relay_conn_pipe[2] = { -1, -1 };
 141
 142 /* Shared between threads */
 143 static int dispatch_thread_exit;
 144
 145 static pthread_t listener_thread;
 146 static pthread_t dispatcher_thread;
 147 static pthread_t worker_thread;
 148 static pthread_t health_thread;
 149
 150 /*
 151  * last_relay_stream_id_lock protects last_relay_stream_id increment
 152  * atomicity on 32-bit architectures.
 153  */
 154 static pthread_mutex_t last_relay_stream_id_lock = PTHREAD_MUTEX_INITIALIZER;
 155 static uint64_t last_relay_stream_id;
 156
 157 /*
 158  * Relay command queue.
 159  *
 160  * The relay_thread_listener and relay_thread_dispatcher communicate with this
 161  * queue.
 162  */
 163 static struct relay_conn_queue relay_conn_queue;
 164
 165 /* Cap of file desriptors to be in simultaneous use by the relay daemon. */
 166 static unsigned int lttng_opt_fd_pool_size = -1;
 167
 168 /* Global relay stream hash table. */
 169 struct lttng_ht *relay_streams_ht;
 170
 171 /* Global relay viewer stream hash table. */
 172 struct lttng_ht *viewer_streams_ht;
 173
 174 /* Global relay sessions hash table. */
 175 struct lttng_ht *sessions_ht;
 176
 177 /* Relayd health monitoring */
 178 struct health_app *health_relayd;
 179
 180 struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry;
 181
 182 /* Global fd tracker. */
 183 struct fd_tracker *the_fd_tracker;
 184
 185 static struct option long_options[] = {
 186         { "control-port", 1, 0, 'C', },
 187         { "data-port", 1, 0, 'D', },
 188         { "live-port", 1, 0, 'L', },
 189         { "daemonize", 0, 0, 'd', },
 190         { "background", 0, 0, 'b', },
 191         { "group", 1, 0, 'g', },
 192         { "fd-pool-size", 1, 0, '\0', },
 193         { "help", 0, 0, 'h', },
 194         { "output", 1, 0, 'o', },
 195         { "verbose", 0, 0, 'v', },
 196         { "config", 1, 0, 'f' },
 197         { "version", 0, 0, 'V' },
 198         { "working-directory", 1, 0, 'w', },
 199         { "group-output-by-session", 0, 0, 's', },
 200         { "group-output-by-host", 0, 0, 'p', },
 201         { "disallow-clear", 0, 0, 'x' },
 202         { "disallow-ctf2", 0, 0, 'y' },
 203         { NULL, 0, 0, 0, },
 204 };
 205
 206 static const char *config_ignore_options[] = { "help", "config", "version" };
 207
 208 static void print_version(void) {
 209         fprintf(stdout, "%s\n", VERSION);
 210 }
 211
 212 static void relayd_config_log(void)
 213 {
 214         DBG("LTTng-relayd " VERSION " - " VERSION_NAME "%s%s",
 215                         GIT_VERSION[0] == '\0' ? "" : " - " GIT_VERSION,
 216                         EXTRA_VERSION_NAME[0] == '\0' ? "" : " - " EXTRA_VERSION_NAME);
 217         if (EXTRA_VERSION_DESCRIPTION[0] != '\0') {
 218                 DBG("LTTng-relayd extra version description:\n\t" EXTRA_VERSION_DESCRIPTION "\n");
 219         }
 220         if (EXTRA_VERSION_PATCHES[0] != '\0') {
 221                 DBG("LTTng-relayd extra patches:\n\t" EXTRA_VERSION_PATCHES "\n");
 222         }
 223
 224         DBG("Trace format supported: %s%s", "ctf1", opt_allow_ctf2 ? ", ctf2" : "");
 225 }
 226
 227 /*
 228  * Take an option from the getopt output and set it in the right variable to be
 229  * used later.
 230  *
 231  * Return 0 on success else a negative value.
 232  */
 233 static int set_option(int opt, const char *arg, const char *optname)
 234 {
 235         int ret;
 236
 237         switch (opt) {
 238         case 0:
 239                 if (!strcmp(optname, "fd-pool-size")) {
 240                         unsigned long v;
 241
 242                         errno = 0;
 243                         v = strtoul(arg, NULL, 0);
 244                         if (errno != 0 || !isdigit((unsigned char) arg[0])) {
 245                                 ERR("Wrong value in --fd-pool-size parameter: %s", arg);
 246                                 ret = -1;
 247                                 goto end;
 248                         }
 249                         if (v >= UINT_MAX) {
 250                                 ERR("File descriptor cap overflow in --fd-pool-size parameter: %s", arg);
 251                                 ret = -1;
 252                                 goto end;
 253                         }
 254                         lttng_opt_fd_pool_size = (unsigned int) v;
 255                 } else {
 256                         fprintf(stderr, "unknown option %s", optname);
 257                         if (arg) {
 258                                 fprintf(stderr, " with arg %s\n", arg);
 259                         }
 260                 }
 261                 break;
 262         case 'C':
 263                 if (lttng_is_setuid_setgid()) {
 264                         WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 265                                 "-C, --control-port");
 266                 } else {
 267                         ret = uri_parse(arg, &control_uri);
 268                         if (ret < 0) {
 269                                 ERR("Invalid control URI specified");
 270                                 goto end;
 271                         }
 272                         if (control_uri->port == 0) {
 273                                 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 274                         }
 275                 }
 276                 break;
 277         case 'D':
 278                 if (lttng_is_setuid_setgid()) {
 279                         WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 280                                 "-D, -data-port");
 281                 } else {
 282                         ret = uri_parse(arg, &data_uri);
 283                         if (ret < 0) {
 284                                 ERR("Invalid data URI specified");
 285                                 goto end;
 286                         }
 287                         if (data_uri->port == 0) {
 288                                 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
 289                         }
 290                 }
 291                 break;
 292         case 'L':
 293                 if (lttng_is_setuid_setgid()) {
 294                         WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 295                                 "-L, -live-port");
 296                 } else {
 297                         ret = uri_parse(arg, &live_uri);
 298                         if (ret < 0) {
 299                                 ERR("Invalid live URI specified");
 300                                 goto end;
 301                         }
 302                         if (live_uri->port == 0) {
 303                                 live_uri->port = DEFAULT_NETWORK_VIEWER_PORT;
 304                         }
 305                 }
 306                 break;
 307         case 'd':
 308                 opt_daemon = 1;
 309                 break;
 310         case 'b':
 311                 opt_background = 1;
 312                 break;
 313         case 'g':
 314                 if (lttng_is_setuid_setgid()) {
 315                         WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 316                                 "-g, --group");
 317                 } else {
 318                         tracing_group_name = strdup(arg);
 319                         if (tracing_group_name == NULL) {
 320                                 ret = -errno;
 321                                 PERROR("strdup");
 322                                 goto end;
 323                         }
 324                         tracing_group_name_override = 1;
 325                 }
 326                 break;
 327         case 'h':
 328                 ret = utils_show_help(8, "lttng-relayd", help_msg);
 329                 if (ret) {
 330                         ERR("Cannot show --help for `lttng-relayd`");
 331                         perror("exec");
 332                 }
 333                 exit(EXIT_FAILURE);
 334         case 'V':
 335                 opt_print_version = 1;
 336                 break;
 337         case 'o':
 338                 if (lttng_is_setuid_setgid()) {
 339                         WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 340                                 "-o, --output");
 341                 } else {
 342                         ret = asprintf(&opt_output_path, "%s", arg);
 343                         if (ret < 0) {
 344                                 ret = -errno;
 345                                 PERROR("asprintf opt_output_path");
 346                                 goto end;
 347                         }
 348                 }
 349                 break;
 350         case 'w':
 351                 if (lttng_is_setuid_setgid()) {
 352                         WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 353                                 "-w, --working-directory");
 354                 } else {
 355                         ret = asprintf(&opt_working_directory, "%s", arg);
 356                         if (ret < 0) {
 357                                 ret = -errno;
 358                                 PERROR("asprintf opt_working_directory");
 359                                 goto end;
 360                         }
 361                 }
 362                 break;
 363
 364         case 'v':
 365                 /* Verbose level can increase using multiple -v */
 366                 if (arg) {
 367                         lttng_opt_verbose = config_parse_value(arg);
 368                 } else {
 369                         /* Only 3 level of verbosity (-vvv). */
 370                         if (lttng_opt_verbose < 3) {
 371                                 lttng_opt_verbose += 1;
 372                         }
 373                 }
 374                 break;
 375         case 's':
 376                 if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
 377                         ERR("Cannot set --group-output-by-session, another --group-output-by argument is present");
 378                         exit(EXIT_FAILURE);
 379                 }
 380                 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_SESSION;
 381                 break;
 382         case 'p':
 383                 if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
 384                         ERR("Cannot set --group-output-by-host, another --group-output-by argument is present");
 385                         exit(EXIT_FAILURE);
 386                 }
 387                 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
 388                 break;
 389         case 'x':
 390                 /* Disallow clear */
 391                 opt_allow_clear = 0;
 392                 break;
 393         case 'y':
 394                 /* Disallow clear */
 395                 opt_allow_ctf2 = 0;
 396                 break;
 397         default:
 398                 /* Unknown option or other error.
 399                  * Error is printed by getopt, just return */
 400                 ret = -1;
 401                 goto end;
 402         }
 403
 404         /* All good. */
 405         ret = 0;
 406
 407 end:
 408         return ret;
 409 }
 410
 411 /*
 412  * config_entry_handler_cb used to handle options read from a config file.
 413  * See config_entry_handler_cb comment in common/config/session-config.h for the
 414  * return value conventions.
 415  */
 416 static int config_entry_handler(const struct config_entry *entry,
 417                 void *unused __attribute__((unused)))
 418 {
 419         int ret = 0, i;
 420
 421         if (!entry || !entry->name || !entry->value) {
 422                 ret = -EINVAL;
 423                 goto end;
 424         }
 425
 426         /* Check if the option is to be ignored */
 427         for (i = 0; i < sizeof(config_ignore_options) / sizeof(char *); i++) {
 428                 if (!strcmp(entry->name, config_ignore_options[i])) {
 429                         goto end;
 430                 }
 431         }
 432
 433         for (i = 0; i < (sizeof(long_options) / sizeof(struct option)) - 1; i++) {
 434                 /* Ignore if entry name is not fully matched. */
 435                 if (strcmp(entry->name, long_options[i].name)) {
 436                         continue;
 437                 }
 438
 439                 /*
 440                  * If the option takes no argument on the command line,
 441                  * we have to check if the value is "true". We support
 442                  * non-zero numeric values, true, on and yes.
 443                  */
 444                 if (!long_options[i].has_arg) {
 445                         ret = config_parse_value(entry->value);
 446                         if (ret <= 0) {
 447                                 if (ret) {
 448                                         WARN("Invalid configuration value \"%s\" for option %s",
 449                                                         entry->value, entry->name);
 450                                 }
 451                                 /* False, skip boolean config option. */
 452                                 goto end;
 453                         }
 454                 }
 455
 456                 ret = set_option(long_options[i].val, entry->value, entry->name);
 457                 goto end;
 458         }
 459
 460         WARN("Unrecognized option \"%s\" in daemon configuration file.",
 461                         entry->name);
 462
 463 end:
 464         return ret;
 465 }
 466
 467 static int parse_env_options(void)
 468 {
 469         int ret = 0;
 470         char *value = NULL;
 471
 472         value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_WORKING_DIRECTORY_ENV);
 473         if (value) {
 474                 opt_working_directory = strdup(value);
 475                 if (!opt_working_directory) {
 476                         ERR("Failed to allocate working directory string (\"%s\")",
 477                                         value);
 478                         ret = -1;
 479                 }
 480         }
 481         return ret;
 482 }
 483
 484 static int set_fd_pool_size(void)
 485 {
 486         int ret = 0;
 487         struct rlimit rlimit;
 488
 489         ret = getrlimit(RLIMIT_NOFILE, &rlimit);
 490         if (ret) {
 491                 PERROR("Failed to get file descriptor limit");
 492                 ret = -1;
 493                 goto end;
 494         }
 495
 496         DBG("File descriptor count limits are %" PRIu64 " (soft) and %" PRIu64 " (hard)",
 497                         (uint64_t) rlimit.rlim_cur,
 498                         (uint64_t) rlimit.rlim_max);
 499         if (lttng_opt_fd_pool_size == -1) {
 500                 /* Use default value (soft limit - reserve). */
 501                 if (rlimit.rlim_cur < DEFAULT_RELAYD_MIN_FD_POOL_SIZE) {
 502                         ERR("The process' file number limit is too low (%" PRIu64 "). The process' file number limit must be set to at least %i.",
 503                                         (uint64_t) rlimit.rlim_cur, DEFAULT_RELAYD_MIN_FD_POOL_SIZE);
 504                         ret = -1;
 505                         goto end;
 506                 }
 507                 lttng_opt_fd_pool_size = rlimit.rlim_cur -
 508                                 DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE;
 509                 goto end;
 510         }
 511
 512         if (lttng_opt_fd_pool_size < DEFAULT_RELAYD_MIN_FD_POOL_SIZE) {
 513                 ERR("File descriptor pool size must be set to at least %d",
 514                                 DEFAULT_RELAYD_MIN_FD_POOL_SIZE);
 515                 ret = -1;
 516                 goto end;
 517         }
 518
 519         if (lttng_opt_fd_pool_size > rlimit.rlim_cur) {
 520                 ERR("File descriptor pool size argument (%u) exceeds the process' soft limit (%" PRIu64 ").",
 521                                 lttng_opt_fd_pool_size, (uint64_t) rlimit.rlim_cur);
 522                 ret = -1;
 523                 goto end;
 524         }
 525
 526         DBG("File descriptor pool size argument (%u) adjusted to %u to accommodates transient fd uses",
 527                         lttng_opt_fd_pool_size,
 528                         lttng_opt_fd_pool_size - DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE);
 529         lttng_opt_fd_pool_size -= DEFAULT_RELAYD_FD_POOL_SIZE_RESERVE;
 530 end:
 531         return ret;
 532 }
 533
 534 static int set_options(int argc, char **argv)
 535 {
 536         int c, ret = 0, option_index = 0, retval = 0;
 537         int orig_optopt = optopt, orig_optind = optind;
 538         char *default_address, *optstring;
 539         char *config_path = NULL;
 540
 541         optstring = utils_generate_optstring(long_options,
 542                         sizeof(long_options) / sizeof(struct option));
 543         if (!optstring) {
 544                 retval = -ENOMEM;
 545                 goto exit;
 546         }
 547
 548         /* Check for the --config option */
 549
 550         while ((c = getopt_long(argc, argv, optstring, long_options,
 551                                         &option_index)) != -1) {
 552                 if (c == '?') {
 553                         retval = -EINVAL;
 554                         goto exit;
 555                 } else if (c != 'f') {
 556                         continue;
 557                 }
 558
 559                 if (lttng_is_setuid_setgid()) {
 560                         WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
 561                                 "-f, --config");
 562                 } else {
 563                         free(config_path);
 564                         config_path = utils_expand_path(optarg);
 565                         if (!config_path) {
 566                                 ERR("Failed to resolve path: %s", optarg);
 567                         }
 568                 }
 569         }
 570
 571         ret = config_get_section_entries(config_path, config_section_name,
 572                         config_entry_handler, NULL);
 573         if (ret) {
 574                 if (ret > 0) {
 575                         ERR("Invalid configuration option at line %i", ret);
 576                 }
 577                 retval = -1;
 578                 goto exit;
 579         }
 580
 581         /* Reset getopt's global state */
 582         optopt = orig_optopt;
 583         optind = orig_optind;
 584         while (1) {
 585                 c = getopt_long(argc, argv, optstring, long_options, &option_index);
 586                 if (c == -1) {
 587                         break;
 588                 }
 589
 590                 ret = set_option(c, optarg, long_options[option_index].name);
 591                 if (ret < 0) {
 592                         retval = -1;
 593                         goto exit;
 594                 }
 595         }
 596
 597         /* assign default values */
 598         if (control_uri == NULL) {
 599                 ret = asprintf(&default_address,
 600                         "tcp://" DEFAULT_NETWORK_CONTROL_BIND_ADDRESS ":%d",
 601                         DEFAULT_NETWORK_CONTROL_PORT);
 602                 if (ret < 0) {
 603                         PERROR("asprintf default data address");
 604                         retval = -1;
 605                         goto exit;
 606                 }
 607
 608                 ret = uri_parse(default_address, &control_uri);
 609                 free(default_address);
 610                 if (ret < 0) {
 611                         ERR("Invalid control URI specified");
 612                         retval = -1;
 613                         goto exit;
 614                 }
 615         }
 616         if (data_uri == NULL) {
 617                 ret = asprintf(&default_address,
 618                         "tcp://" DEFAULT_NETWORK_DATA_BIND_ADDRESS ":%d",
 619                         DEFAULT_NETWORK_DATA_PORT);
 620                 if (ret < 0) {
 621                         PERROR("asprintf default data address");
 622                         retval = -1;
 623                         goto exit;
 624                 }
 625
 626                 ret = uri_parse(default_address, &data_uri);
 627                 free(default_address);
 628                 if (ret < 0) {
 629                         ERR("Invalid data URI specified");
 630                         retval = -1;
 631                         goto exit;
 632                 }
 633         }
 634         if (live_uri == NULL) {
 635                 ret = asprintf(&default_address,
 636                         "tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS ":%d",
 637                         DEFAULT_NETWORK_VIEWER_PORT);
 638                 if (ret < 0) {
 639                         PERROR("asprintf default viewer control address");
 640                         retval = -1;
 641                         goto exit;
 642                 }
 643
 644                 ret = uri_parse(default_address, &live_uri);
 645                 free(default_address);
 646                 if (ret < 0) {
 647                         ERR("Invalid viewer control URI specified");
 648                         retval = -1;
 649                         goto exit;
 650                 }
 651         }
 652         ret = set_fd_pool_size();
 653         if (ret) {
 654                 retval = -1;
 655                 goto exit;
 656         }
 657
 658         if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
 659                 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
 660         }
 661         if (opt_allow_clear) {
 662                 /* Check if env variable exists. */
 663                 const char *value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_DISALLOW_CLEAR_ENV);
 664                 if (value) {
 665                         ret = config_parse_value(value);
 666                         if (ret < 0) {
 667                                 ERR("Invalid value for %s specified", DEFAULT_LTTNG_RELAYD_DISALLOW_CLEAR_ENV);
 668                                 retval = -1;
 669                                 goto exit;
 670                         }
 671                         opt_allow_clear = !ret;
 672                 }
 673         }
 674
 675         if (opt_allow_ctf2) {
 676                 /* Check if env variable exists. */
 677                 const char *value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_DISALLOW_CTF2_ENV);
 678                 if (value) {
 679                         ret = config_parse_value(value);
 680                         if (ret < 0) {
 681                                 ERR("Invalid value for %s specified",
 682                                                 DEFAULT_LTTNG_RELAYD_DISALLOW_CTF2_ENV);
 683                                 retval = -1;
 684                                 goto exit;
 685                         }
 686                         opt_allow_ctf2 = !ret;
 687                 }
 688         }
 689
 690 exit:
 691         free(config_path);
 692         free(optstring);
 693         return retval;
 694 }
 695
 696 static void print_global_objects(void)
 697 {
 698         print_viewer_streams();
 699         print_relay_streams();
 700         print_sessions();
 701 }
 702
 703 static int noop_close(void *data __attribute__((unused)),
 704                 int *fds __attribute__((unused)))
 705 {
 706         return 0;
 707 }
 708
 709 static void untrack_stdio(void)
 710 {
 711         int fds[] = { fileno(stdout), fileno(stderr) };
 712
 713         /*
 714          * noop_close is used since we don't really want to close
 715          * the stdio output fds; we merely want to stop tracking them.
 716          */
 717         (void) fd_tracker_close_unsuspendable_fd(the_fd_tracker,
 718                         fds, 2, noop_close, NULL);
 719 }
 720
 721 /*
 722  * Cleanup the daemon
 723  */
 724 static void relayd_cleanup(void)
 725 {
 726         print_global_objects();
 727
 728         DBG("Cleaning up");
 729
 730         if (viewer_streams_ht)
 731                 lttng_ht_destroy(viewer_streams_ht);
 732         if (relay_streams_ht)
 733                 lttng_ht_destroy(relay_streams_ht);
 734         if (sessions_ht)
 735                 lttng_ht_destroy(sessions_ht);
 736
 737         free(opt_output_path);
 738         free(opt_working_directory);
 739
 740         if (health_relayd) {
 741                 health_app_destroy(health_relayd);
 742         }
 743         /* Close thread quit pipes */
 744         if (health_quit_pipe[0] != -1) {
 745                 (void) fd_tracker_util_pipe_close(
 746                                 the_fd_tracker, health_quit_pipe);
 747         }
 748         if (thread_quit_pipe[0] != -1) {
 749                 (void) fd_tracker_util_pipe_close(
 750                                 the_fd_tracker, thread_quit_pipe);
 751         }
 752         if (sessiond_trace_chunk_registry) {
 753                 sessiond_trace_chunk_registry_destroy(
 754                                 sessiond_trace_chunk_registry);
 755         }
 756         if (the_fd_tracker) {
 757                 untrack_stdio();
 758                 /*
 759                  * fd_tracker_destroy() will log the contents of the fd-tracker
 760                  * if a leak is detected.
 761                  */
 762                 fd_tracker_destroy(the_fd_tracker);
 763         }
 764
 765         uri_free(control_uri);
 766         uri_free(data_uri);
 767         /* Live URI is freed in the live thread. */
 768
 769         if (tracing_group_name_override) {
 770                 free((void *) tracing_group_name);
 771         }
 772 }
 773
 774 /*
 775  * Write to writable pipe used to notify a thread.
 776  */
 777 static int notify_thread_pipe(int wpipe)
 778 {
 779         ssize_t ret;
 780
 781         ret = lttng_write(wpipe, "!", 1);
 782         if (ret < 1) {
 783                 PERROR("write poll pipe");
 784                 goto end;
 785         }
 786         ret = 0;
 787 end:
 788         return ret;
 789 }
 790
 791 static int notify_health_quit_pipe(int *pipe)
 792 {
 793         ssize_t ret;
 794
 795         ret = lttng_write(pipe[1], "4", 1);
 796         if (ret < 1) {
 797                 PERROR("write relay health quit");
 798                 goto end;
 799         }
 800         ret = 0;
 801 end:
 802         return ret;
 803 }
 804
 805 /*
 806  * Stop all relayd and relayd-live threads.
 807  */
 808 int lttng_relay_stop_threads(void)
 809 {
 810         int retval = 0;
 811
 812         /* Stopping all threads */
 813         DBG("Terminating all threads");
 814         if (notify_thread_pipe(thread_quit_pipe[1])) {
 815                 ERR("write error on thread quit pipe");
 816                 retval = -1;
 817         }
 818
 819         if (notify_health_quit_pipe(health_quit_pipe)) {
 820                 ERR("write error on health quit pipe");
 821         }
 822
 823         /* Dispatch thread */
 824         CMM_STORE_SHARED(dispatch_thread_exit, 1);
 825         futex_nto1_wake(&relay_conn_queue.futex);
 826
 827         if (relayd_live_stop()) {
 828                 ERR("Error stopping live threads");
 829                 retval = -1;
 830         }
 831         return retval;
 832 }
 833
 834 /*
 835  * Signal handler for the daemon
 836  *
 837  * Simply stop all worker threads, leaving main() return gracefully after
 838  * joining all threads and calling cleanup().
 839  */
 840 static void sighandler(int sig)
 841 {
 842         switch (sig) {
 843         case SIGINT:
 844                 DBG("SIGINT caught");
 845                 if (lttng_relay_stop_threads()) {
 846                         ERR("Error stopping threads");
 847                 }
 848                 break;
 849         case SIGTERM:
 850                 DBG("SIGTERM caught");
 851                 if (lttng_relay_stop_threads()) {
 852                         ERR("Error stopping threads");
 853                 }
 854                 break;
 855         case SIGUSR1:
 856                 CMM_STORE_SHARED(recv_child_signal, 1);
 857                 break;
 858         default:
 859                 break;
 860         }
 861 }
 862
 863 /*
 864  * Setup signal handler for :
 865  *              SIGINT, SIGTERM, SIGPIPE
 866  */
 867 static int set_signal_handler(void)
 868 {
 869         int ret = 0;
 870         struct sigaction sa;
 871         sigset_t sigset;
 872
 873         if ((ret = sigemptyset(&sigset)) < 0) {
 874                 PERROR("sigemptyset");
 875                 return ret;
 876         }
 877
 878         sa.sa_mask = sigset;
 879         sa.sa_flags = 0;
 880
 881         sa.sa_handler = sighandler;
 882         if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
 883                 PERROR("sigaction");
 884                 return ret;
 885         }
 886
 887         if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
 888                 PERROR("sigaction");
 889                 return ret;
 890         }
 891
 892         if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
 893                 PERROR("sigaction");
 894                 return ret;
 895         }
 896
 897         sa.sa_handler = SIG_IGN;
 898         if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
 899                 PERROR("sigaction");
 900                 return ret;
 901         }
 902
 903         DBG("Signal handler set for SIGTERM, SIGUSR1, SIGPIPE and SIGINT");
 904
 905         return ret;
 906 }
 907
 908 void lttng_relay_notify_ready(void)
 909 {
 910         /* Notify the parent of the fork() process that we are ready. */
 911         if (opt_daemon || opt_background) {
 912                 if (uatomic_sub_return(&lttng_relay_ready, 1) == 0) {
 913                         kill(child_ppid, SIGUSR1);
 914                 }
 915         }
 916 }
 917
 918 /*
 919  * Init thread quit pipe.
 920  *
 921  * Return -1 on error or 0 if all pipes are created.
 922  */
 923 static int init_thread_quit_pipe(void)
 924 {
 925         return fd_tracker_util_pipe_open_cloexec(
 926                         the_fd_tracker, "Quit pipe", thread_quit_pipe);
 927 }
 928
 929 /*
 930  * Init health quit pipe.
 931  *
 932  * Return -1 on error or 0 if all pipes are created.
 933  */
 934 static int init_health_quit_pipe(void)
 935 {
 936         return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
 937                         "Health quit pipe", health_quit_pipe);
 938 }
 939
 940 /*
 941  * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
 942  */
 943 static int create_named_thread_poll_set(struct lttng_poll_event *events,
 944                 int size, const char *name)
 945 {
 946         int ret;
 947
 948         if (events == NULL || size == 0) {
 949                 ret = -1;
 950                 goto error;
 951         }
 952
 953         ret = fd_tracker_util_poll_create(the_fd_tracker,
 954                         name, events, 1, LTTNG_CLOEXEC);
 955         if (ret) {
 956                 PERROR("Failed to create \"%s\" poll file descriptor", name);
 957                 goto error;
 958         }
 959
 960         /* Add quit pipe */
 961         ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
 962         if (ret < 0) {
 963                 goto error;
 964         }
 965
 966         return 0;
 967
 968 error:
 969         return ret;
 970 }
 971
 972 /*
 973  * Check if the thread quit pipe was triggered.
 974  *
 975  * Return 1 if it was triggered else 0;
 976  */
 977 static int check_thread_quit_pipe(int fd, uint32_t events)
 978 {
 979         if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
 980                 return 1;
 981         }
 982
 983         return 0;
 984 }
 985
 986 static int create_sock(void *data, int *out_fd)
 987 {
 988         int ret;
 989         struct lttcomm_sock *sock = (lttcomm_sock *) data;
 990
 991         ret = lttcomm_create_sock(sock);
 992         if (ret < 0) {
 993                 goto end;
 994         }
 995
 996         *out_fd = sock->fd;
 997 end:
 998         return ret;
 999 }
1000
1001 static int close_sock(void *data, int *in_fd __attribute__((unused)))
1002 {
1003         struct lttcomm_sock *sock = (lttcomm_sock *) data;
1004
1005         return sock->ops->close(sock);
1006 }
1007
1008 static int accept_sock(void *data, int *out_fd)
1009 {
1010         int ret = 0;
1011         /* Socks is an array of in_sock, out_sock. */
1012         struct lttcomm_sock **socks = (lttcomm_sock **) data;
1013         struct lttcomm_sock *in_sock = socks[0];
1014
1015         socks[1] = in_sock->ops->accept(in_sock);
1016         if (!socks[1]) {
1017                 ret = -1;
1018                 goto end;
1019         }
1020         *out_fd = socks[1]->fd;
1021 end:
1022         return ret;
1023 }
1024
1025 /*
1026  * Create and init socket from uri.
1027  */
1028 static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri,
1029                 const char *name)
1030 {
1031         int ret, sock_fd;
1032         struct lttcomm_sock *sock = NULL;
1033         char uri_str[PATH_MAX];
1034         char *formated_name = NULL;
1035
1036         sock = lttcomm_alloc_sock_from_uri(uri);
1037         if (sock == NULL) {
1038                 ERR("Allocating socket");
1039                 goto error;
1040         }
1041
1042         /*
1043          * Don't fail to create the socket if the name can't be built as it is
1044          * only used for debugging purposes.
1045          */
1046         ret = uri_to_str_url(uri, uri_str, sizeof(uri_str));
1047         uri_str[sizeof(uri_str) - 1] = '\0';
1048         if (ret >= 0) {
1049                 ret = asprintf(&formated_name, "%s socket @ %s", name,
1050                                 uri_str);
1051                 if (ret < 0) {
1052                         formated_name = NULL;
1053                 }
1054         }
1055
1056         ret = fd_tracker_open_unsuspendable_fd(the_fd_tracker, &sock_fd,
1057                         (const char **) (formated_name ? &formated_name : NULL),
1058                         1, create_sock, sock);
1059         if (ret) {
1060                 PERROR("Failed to open \"%s\" relay socket",
1061                                 formated_name ?: "Unknown");
1062                 goto error;
1063         }
1064         DBG("Listening on %s socket %d", name, sock->fd);
1065
1066         ret = sock->ops->bind(sock);
1067         if (ret < 0) {
1068                 PERROR("Failed to bind socket");
1069                 goto error;
1070         }
1071
1072         ret = sock->ops->listen(sock, -1);
1073         if (ret < 0) {
1074                 goto error;
1075
1076         }
1077
1078         free(formated_name);
1079         return sock;
1080
1081 error:
1082         if (sock) {
1083                 lttcomm_destroy_sock(sock);
1084         }
1085         free(formated_name);
1086         return NULL;
1087 }
1088
1089 static
1090 struct lttcomm_sock *accept_relayd_sock(struct lttcomm_sock *listening_sock,
1091                 const char *name)
1092 {
1093         int out_fd, ret;
1094         struct lttcomm_sock *socks[2] = { listening_sock, NULL };
1095         struct lttcomm_sock *new_sock = NULL;
1096
1097         ret = fd_tracker_open_unsuspendable_fd(
1098                         the_fd_tracker, &out_fd,
1099                         (const char **) &name,
1100                         1, accept_sock, &socks);
1101         if (ret) {
1102                 goto end;
1103         }
1104         new_sock = socks[1];
1105         DBG("%s accepted, socket %d", name, new_sock->fd);
1106 end:
1107         return new_sock;
1108 }
1109
1110 /*
1111  * This thread manages the listening for new connections on the network
1112  */
1113 static void *relay_thread_listener(void *data __attribute__((unused)))
1114 {
1115         int i, ret, pollfd, err = -1;
1116         uint32_t revents, nb_fd;
1117         struct lttng_poll_event events;
1118         struct lttcomm_sock *control_sock, *data_sock;
1119
1120         DBG("[thread] Relay listener started");
1121
1122         rcu_register_thread();
1123         health_register(health_relayd, HEALTH_RELAYD_TYPE_LISTENER);
1124
1125         health_code_update();
1126
1127         control_sock = relay_socket_create(control_uri, "Control listener");
1128         if (!control_sock) {
1129                 goto error_sock_control;
1130         }
1131
1132         data_sock = relay_socket_create(data_uri, "Data listener");
1133         if (!data_sock) {
1134                 goto error_sock_relay;
1135         }
1136
1137         /*
1138          * Pass 3 as size here for the thread quit pipe, control and
1139          * data socket.
1140          */
1141         ret = create_named_thread_poll_set(&events, 3, "Listener thread epoll");
1142         if (ret < 0) {
1143                 goto error_create_poll;
1144         }
1145
1146         /* Add the control socket */
1147         ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
1148         if (ret < 0) {
1149                 goto error_poll_add;
1150         }
1151
1152         /* Add the data socket */
1153         ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
1154         if (ret < 0) {
1155                 goto error_poll_add;
1156         }
1157
1158         lttng_relay_notify_ready();
1159
1160         if (testpoint(relayd_thread_listener)) {
1161                 goto error_testpoint;
1162         }
1163
1164         while (1) {
1165                 health_code_update();
1166
1167                 DBG("Listener accepting connections");
1168
1169 restart:
1170                 health_poll_entry();
1171                 ret = lttng_poll_wait(&events, -1);
1172                 health_poll_exit();
1173                 if (ret < 0) {
1174                         /*
1175                          * Restart interrupted system call.
1176                          */
1177                         if (errno == EINTR) {
1178                                 goto restart;
1179                         }
1180                         goto error;
1181                 }
1182
1183                 nb_fd = ret;
1184
1185                 DBG("Relay new connection received");
1186                 for (i = 0; i < nb_fd; i++) {
1187                         health_code_update();
1188
1189                         /* Fetch once the poll data */
1190                         revents = LTTNG_POLL_GETEV(&events, i);
1191                         pollfd = LTTNG_POLL_GETFD(&events, i);
1192
1193                         /* Thread quit pipe has been closed. Killing thread. */
1194                         ret = check_thread_quit_pipe(pollfd, revents);
1195                         if (ret) {
1196                                 err = 0;
1197                                 goto exit;
1198                         }
1199
1200                         if (revents & LPOLLIN) {
1201                                 /*
1202                                  * A new connection is requested, therefore a
1203                                  * sessiond/consumerd connection is allocated in
1204                                  * this thread, enqueued to a global queue and
1205                                  * dequeued (and freed) in the worker thread.
1206                                  */
1207                                 int val = 1;
1208                                 struct relay_connection *new_conn;
1209                                 struct lttcomm_sock *newsock = NULL;
1210                                 enum connection_type type;
1211
1212                                 if (pollfd == data_sock->fd) {
1213                                         type = RELAY_DATA;
1214                                         newsock = accept_relayd_sock(data_sock,
1215                                                         "Data socket to relayd");
1216                                 } else {
1217                                         LTTNG_ASSERT(pollfd == control_sock->fd);
1218                                         type = RELAY_CONTROL;
1219                                         newsock = accept_relayd_sock(control_sock,
1220                                                         "Control socket to relayd");
1221                                 }
1222                                 if (!newsock) {
1223                                         PERROR("accepting sock");
1224                                         goto error;
1225                                 }
1226
1227                                 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
1228                                                 sizeof(val));
1229                                 if (ret < 0) {
1230                                         PERROR("setsockopt inet");
1231                                         lttcomm_destroy_sock(newsock);
1232                                         goto error;
1233                                 }
1234
1235                                 ret = socket_apply_keep_alive_config(newsock->fd);
1236                                 if (ret < 0) {
1237                                         ERR("Failed to apply TCP keep-alive configuration on socket (%i)",
1238                                                         newsock->fd);
1239                                         lttcomm_destroy_sock(newsock);
1240                                         goto error;
1241                                 }
1242
1243                                 new_conn = connection_create(newsock, type);
1244                                 if (!new_conn) {
1245                                         lttcomm_destroy_sock(newsock);
1246                                         goto error;
1247                                 }
1248
1249                                 /* Enqueue request for the dispatcher thread. */
1250                                 cds_wfcq_head_ptr_t head;
1251                                 head.h = &relay_conn_queue.head;
1252                                 cds_wfcq_enqueue(head, &relay_conn_queue.tail,
1253                                                  &new_conn->qnode);
1254
1255                                 /*
1256                                  * Wake the dispatch queue futex.
1257                                  * Implicit memory barrier with the
1258                                  * exchange in cds_wfcq_enqueue.
1259                                  */
1260                                 futex_nto1_wake(&relay_conn_queue.futex);
1261                         } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1262                                 ERR("socket poll error");
1263                                 goto error;
1264                         } else {
1265                                 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
1266                                 goto error;
1267                         }
1268                 }
1269         }
1270
1271 exit:
1272 error:
1273 error_poll_add:
1274 error_testpoint:
1275         (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
1276 error_create_poll:
1277         if (data_sock->fd >= 0) {
1278                 int data_sock_fd = data_sock->fd;
1279
1280                 ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
1281                                 &data_sock_fd, 1, close_sock,
1282                                 data_sock);
1283                 if (ret) {
1284                         PERROR("Failed to close the data listener socket file descriptor");
1285                 }
1286                 data_sock->fd = -1;
1287         }
1288         lttcomm_destroy_sock(data_sock);
1289 error_sock_relay:
1290         if (control_sock->fd >= 0) {
1291                 int control_sock_fd = control_sock->fd;
1292
1293                 ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker,
1294                                 &control_sock_fd, 1, close_sock,
1295                                 control_sock);
1296                 if (ret) {
1297                         PERROR("Failed to close the control listener socket file descriptor");
1298                 }
1299                 control_sock->fd = -1;
1300         }
1301         lttcomm_destroy_sock(control_sock);
1302 error_sock_control:
1303         if (err) {
1304                 health_error();
1305                 ERR("Health error occurred in %s", __func__);
1306         }
1307         health_unregister(health_relayd);
1308         rcu_unregister_thread();
1309         DBG("Relay listener thread cleanup complete");
1310         lttng_relay_stop_threads();
1311         return NULL;
1312 }
1313
1314 /*
1315  * This thread manages the dispatching of the requests to worker threads
1316  */
1317 static void *relay_thread_dispatcher(void *data __attribute__((unused)))
1318 {
1319         int err = -1;
1320         ssize_t ret;
1321         struct cds_wfcq_node *node;
1322         struct relay_connection *new_conn = NULL;
1323
1324         DBG("[thread] Relay dispatcher started");
1325
1326         health_register(health_relayd, HEALTH_RELAYD_TYPE_DISPATCHER);
1327
1328         if (testpoint(relayd_thread_dispatcher)) {
1329                 goto error_testpoint;
1330         }
1331
1332         health_code_update();
1333
1334         for (;;) {
1335                 health_code_update();
1336
1337                 /* Atomically prepare the queue futex */
1338                 futex_nto1_prepare(&relay_conn_queue.futex);
1339
1340                 if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
1341                         break;
1342                 }
1343
1344                 do {
1345                         health_code_update();
1346
1347                         /* Dequeue commands */
1348                         node = cds_wfcq_dequeue_blocking(&relay_conn_queue.head,
1349                                                          &relay_conn_queue.tail);
1350                         if (node == NULL) {
1351                                 DBG("Woken up but nothing in the relay command queue");
1352                                 /* Continue thread execution */
1353                                 break;
1354                         }
1355                         new_conn = lttng::utils::container_of(node, &relay_connection::qnode);
1356
1357                         DBG("Dispatching request waiting on sock %d", new_conn->sock->fd);
1358
1359                         /*
1360                          * Inform worker thread of the new request. This
1361                          * call is blocking so we can be assured that
1362                          * the data will be read at some point in time
1363                          * or wait to the end of the world :)
1364                          */
1365                         ret = lttng_write(relay_conn_pipe[1], &new_conn, sizeof(new_conn));
1366                         if (ret < 0) {
1367                                 PERROR("write connection pipe");
1368                                 connection_put(new_conn);
1369                                 goto error;
1370                         }
1371                 } while (node != NULL);
1372
1373                 /* Futex wait on queue. Blocking call on futex() */
1374                 health_poll_entry();
1375                 futex_nto1_wait(&relay_conn_queue.futex);
1376                 health_poll_exit();
1377         }
1378
1379         /* Normal exit, no error */
1380         err = 0;
1381
1382 error:
1383 error_testpoint:
1384         if (err) {
1385                 health_error();
1386                 ERR("Health error occurred in %s", __func__);
1387         }
1388         health_unregister(health_relayd);
1389         DBG("Dispatch thread dying");
1390         lttng_relay_stop_threads();
1391         return NULL;
1392 }
1393
1394 static bool session_streams_have_index(const struct relay_session *session)
1395 {
1396         return session->minor >= 4 && !session->snapshot;
1397 }
1398
1399 /*
1400  * Handle the RELAYD_CREATE_SESSION command.
1401  *
1402  * On success, send back the session id or else return a negative value.
1403  */
1404 static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1405                 struct relay_connection *conn,
1406                 const struct lttng_buffer_view *payload)
1407 {
1408         int ret = 0;
1409         enum lttng_error_code ret_code;
1410         ssize_t send_ret;
1411         struct relay_session *session = NULL;
1412         lttcomm_relayd_create_session_reply_2_15 reply = {};
1413         char session_name[LTTNG_NAME_MAX] = {};
1414         char hostname[LTTNG_HOST_NAME_MAX] = {};
1415         uint32_t live_timer = 0;
1416         bool snapshot = false;
1417         bool session_name_contains_creation_timestamp = false;
1418         /* Left nil for peers < 2.11. */
1419         char base_path[LTTNG_PATH_MAX] = {};
1420         lttng_uuid sessiond_uuid = {};
1421         LTTNG_OPTIONAL(uint64_t) id_sessiond = {};
1422         LTTNG_OPTIONAL(uint64_t) current_chunk_id = {};
1423         LTTNG_OPTIONAL(time_t) creation_time = {};
1424         /* The default trace format is CTF1 */
1425         enum relayd_trace_format trace_format = RELAYD_TRACE_FORMAT_CTF_1;
1426
1427         struct lttng_dynamic_buffer reply_payload;
1428
1429         lttng_dynamic_buffer_init(&reply_payload);
1430
1431         if (conn->minor < 4) {
1432                 /* From 2.1 to 2.3 */
1433                 ret = 0;
1434         } else if (conn->minor >= 4 && conn->minor < 11) {
1435                 /* From 2.4 to 2.10 */
1436                 ret = cmd_create_session_2_4(
1437                                 payload, session_name, hostname, &live_timer, &snapshot);
1438         } else if (conn->minor >= 11 && conn->minor < 15) {
1439                 /* From 2.11 to 2.15 */
1440                 bool has_current_chunk;
1441                 uint64_t current_chunk_id_value;
1442                 time_t creation_time_value;
1443                 uint64_t id_sessiond_value;
1444
1445                 /* From 2.11 to ... */
1446                 ret = cmd_create_session_2_11(payload, session_name, hostname,
1447                                 base_path, &live_timer, &snapshot, &id_sessiond_value,
1448                                 sessiond_uuid, &has_current_chunk,
1449                                 &current_chunk_id_value, &creation_time_value,
1450                                 &session_name_contains_creation_timestamp);
1451                 if (lttng_uuid_is_nil(sessiond_uuid)) {
1452                         /* The nil UUID is reserved for pre-2.11 clients. */
1453                         ERR("Illegal nil UUID announced by peer in create session command");
1454                         ret_code = LTTNG_ERR_FATAL;
1455                         goto send_reply;
1456                 }
1457                 LTTNG_OPTIONAL_SET(&id_sessiond, id_sessiond_value);
1458                 LTTNG_OPTIONAL_SET(&creation_time, creation_time_value);
1459                 if (has_current_chunk) {
1460                         LTTNG_OPTIONAL_SET(&current_chunk_id, current_chunk_id_value);
1461                 }
1462         } else {
1463                 /* From 2.15 to ... */
1464                 bool has_current_chunk;
1465                 uint64_t current_chunk_id_value;
1466                 time_t creation_time_value;
1467                 uint64_t id_sessiond_value;
1468
1469                 /* From 2.11 to ... */
1470                 ret = cmd_create_session_2_15(payload, session_name, hostname, base_path,
1471                                 &live_timer, &snapshot, &id_sessiond_value, sessiond_uuid,
1472                                 &has_current_chunk, &current_chunk_id_value, &creation_time_value,
1473                                 &session_name_contains_creation_timestamp, trace_format);
1474                 if (lttng_uuid_is_nil(sessiond_uuid)) {
1475                         /* The nil UUID is reserved for pre-2.11 clients. */
1476                         ERR("Illegal nil UUID announced by peer in create session command");
1477                         ret_code = LTTNG_ERR_FATAL;
1478                         goto send_reply;
1479                 }
1480
1481                 LTTNG_OPTIONAL_SET(&id_sessiond, id_sessiond_value);
1482                 LTTNG_OPTIONAL_SET(&creation_time, creation_time_value);
1483                 if (has_current_chunk) {
1484                         LTTNG_OPTIONAL_SET(&current_chunk_id,
1485                                         current_chunk_id_value);
1486                 }
1487         }
1488
1489         if (ret < 0) {
1490                 ret_code = LTTNG_ERR_FATAL;
1491                 goto send_reply;
1492         }
1493
1494         /*
1495          * Trace format check. TODO: move to a separate function or inside
1496          * session create?
1497          */
1498         if (trace_format == RELAYD_TRACE_FORMAT_CTF_2 && !opt_allow_ctf2) {
1499                 ret_code = LTTNG_ERR_TRACE_FORMAT_UNSUPPORTED_RELAY_DAEMON;
1500                 goto send_reply;
1501         }
1502
1503         session = session_create(session_name, hostname, base_path, live_timer, snapshot,
1504                         sessiond_uuid, id_sessiond.is_set ? &id_sessiond.value : NULL,
1505                         current_chunk_id.is_set ? &current_chunk_id.value : NULL,
1506                         creation_time.is_set ? &creation_time.value : NULL, conn->major,
1507                         conn->minor, session_name_contains_creation_timestamp, trace_format);
1508         if (!session) {
1509                 ret_code = LTTNG_ERR_FATAL;
1510                 goto send_reply;
1511         }
1512
1513         LTTNG_ASSERT(!conn->session);
1514         conn->session = session;
1515         DBG("Created session %" PRIu64, session->id);
1516
1517         reply.generic.session_id = htobe64(session->id);
1518         ret_code = LTTNG_OK;
1519
1520 send_reply:
1521         reply.generic.ret_code = htobe32(ret_code);
1522
1523         if (conn->minor < 11) {
1524                 /* From 2.1 to 2.10 */
1525                 ret = lttng_dynamic_buffer_append(&reply_payload,
1526                                 &reply.generic, sizeof(reply.generic));
1527                 if (ret) {
1528                         ERR("Failed to append \"create session\" command reply header to payload buffer");
1529                         ret = -1;
1530                         goto end;
1531                 }
1532         } else {
1533                 /* From 2.11 to ... */
1534                 const uint32_t output_path_length = session ? strlen(session->output_path) + 1 : 0;
1535
1536                 reply.output_path_length = htobe32(output_path_length);
1537                 ret = lttng_dynamic_buffer_append(
1538                                 &reply_payload, &reply, sizeof(reply));
1539                 if (ret) {
1540                         ERR("Failed to append \"create session\" command reply header to payload buffer");
1541                         goto end;
1542                 }
1543
1544                 if (output_path_length) {
1545                         ret = lttng_dynamic_buffer_append(&reply_payload,
1546                                         session->output_path,
1547                                         output_path_length);
1548                         if (ret) {
1549                                 ERR("Failed to append \"create session\" command reply path to payload buffer");
1550                                 goto end;
1551                         }
1552                 }
1553         }
1554
1555         send_ret = conn->sock->ops->sendmsg(conn->sock, reply_payload.data,
1556                         reply_payload.size, 0);
1557         if (send_ret < (ssize_t) reply_payload.size) {
1558                 ERR("Failed to send \"create session\" command reply of %zu bytes (ret = %zd)",
1559                                 reply_payload.size, send_ret);
1560                 ret = -1;
1561         }
1562 end:
1563         if (ret < 0 && session) {
1564                 session_put(session);
1565         }
1566         lttng_dynamic_buffer_reset(&reply_payload);
1567         return ret;
1568 }
1569
1570 /*
1571  * When we have received all the streams and the metadata for a channel,
1572  * we make them visible to the viewer threads.
1573  */
1574 static void publish_connection_local_streams(struct relay_connection *conn)
1575 {
1576         struct relay_stream *stream;
1577         struct relay_session *session = conn->session;
1578
1579         /*
1580          * We publish all streams belonging to a session atomically wrt
1581          * session lock.
1582          */
1583         pthread_mutex_lock(&session->lock);
1584         rcu_read_lock();
1585         cds_list_for_each_entry_rcu(stream, &session->recv_list,
1586                         recv_node) {
1587                 stream_publish(stream);
1588         }
1589         rcu_read_unlock();
1590
1591         /*
1592          * Inform the viewer that there are new streams in the session.
1593          */
1594         if (session->viewer_attached) {
1595                 uatomic_set(&session->new_streams, 1);
1596         }
1597         pthread_mutex_unlock(&session->lock);
1598 }
1599
1600 static int conform_channel_path(char *channel_path)
1601 {
1602         int ret = 0;
1603
1604         if (strstr("../", channel_path)) {
1605                 ERR("Refusing channel path as it walks up the path hierarchy: \"%s\"",
1606                                 channel_path);
1607                 ret = -1;
1608                 goto end;
1609         }
1610
1611         if (*channel_path == '/') {
1612                 const size_t len = strlen(channel_path);
1613
1614                 /*
1615                  * Channel paths from peers prior to 2.11 are expressed as an
1616                  * absolute path that is, in reality, relative to the relay
1617                  * daemon's output directory. Remove the leading slash so it
1618                  * is correctly interpreted as a relative path later on.
1619                  *
1620                  * len (and not len - 1) is used to copy the trailing NULL.
1621                  */
1622                 bcopy(channel_path + 1, channel_path, len);
1623         }
1624 end:
1625         return ret;
1626 }
1627
1628 /*
1629  * relay_add_stream: allocate a new stream for a session
1630  */
1631 static int relay_add_stream(
1632                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1633                 struct relay_connection *conn,
1634                 const struct lttng_buffer_view *payload)
1635 {
1636         int ret;
1637         ssize_t send_ret;
1638         struct relay_session *session = conn->session;
1639         struct relay_stream *stream = NULL;
1640         struct lttcomm_relayd_status_stream reply;
1641         struct ctf_trace *trace = NULL;
1642         uint64_t stream_handle = -1ULL;
1643         char *path_name = NULL, *channel_name = NULL;
1644         uint64_t tracefile_size = 0, tracefile_count = 0;
1645         LTTNG_OPTIONAL(uint64_t) stream_chunk_id = {};
1646
1647         if (!session || !conn->version_check_done) {
1648                 ERR("Trying to add a stream before version check");
1649                 ret = -1;
1650                 goto end_no_session;
1651         }
1652
1653         if (session->minor == 1) {
1654                 /* For 2.1 */
1655                 ret = cmd_recv_stream_2_1(payload, &path_name,
1656                         &channel_name);
1657         } else if (session->minor > 1 && session->minor < 11) {
1658                 /* From 2.2 to 2.10 */
1659                 ret = cmd_recv_stream_2_2(payload, &path_name,
1660                         &channel_name, &tracefile_size, &tracefile_count);
1661         } else {
1662                 /* From 2.11 to ... */
1663                 ret = cmd_recv_stream_2_11(payload, &path_name,
1664                         &channel_name, &tracefile_size, &tracefile_count,
1665                         &stream_chunk_id.value);
1666                 stream_chunk_id.is_set = true;
1667         }
1668
1669         if (ret < 0) {
1670                 goto send_reply;
1671         }
1672
1673         if (conform_channel_path(path_name)) {
1674                 goto send_reply;
1675         }
1676
1677         /*
1678          * Backward compatibility for --group-output-by-session.
1679          * Prior to lttng 2.11, the complete path is passed by the stream.
1680          * Starting at 2.11, lttng-relayd uses chunk. When dealing with producer
1681          * >=2.11 the chunk is responsible for the output path. When dealing
1682          * with producer < 2.11 the chunk output_path is the root output path
1683          * and the stream carries the complete path (path_name).
1684          * To support --group-output-by-session with older producer (<2.11), we
1685          * need to craft the path based on the stream path.
1686          */
1687         if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_SESSION) {
1688                 if (conn->minor < 4) {
1689                         /*
1690                          * From 2.1 to 2.3, the session_name is not passed on
1691                          * the RELAYD_CREATE_SESSION command. The session name
1692                          * is necessary to detect the presence of a base_path
1693                          * inside the stream path. Without it we cannot perform
1694                          * a valid group-output-by-session transformation.
1695                          */
1696                         WARN("Unable to perform a --group-by-session transformation for session %" PRIu64
1697                              " for stream with path \"%s\" as it is produced by a peer using a protocol older than v2.4",
1698                                         session->id, path_name);
1699                 } else if (conn->minor >= 4 && conn->minor < 11) {
1700                         char *group_by_session_path_name;
1701
1702                         LTTNG_ASSERT(session->session_name[0] != '\0');
1703
1704                         group_by_session_path_name =
1705                                         backward_compat_group_by_session(
1706                                                         path_name,
1707                                                         session->session_name,
1708                                                         session->creation_time.value);
1709                         if (!group_by_session_path_name) {
1710                                 ERR("Failed to apply group by session to stream of session %" PRIu64,
1711                                                 session->id);
1712                                 goto send_reply;
1713                         }
1714
1715                         DBG("Transformed session path from \"%s\" to \"%s\" to honor per-session name grouping",
1716                                         path_name, group_by_session_path_name);
1717
1718                         free(path_name);
1719                         path_name = group_by_session_path_name;
1720                 }
1721         }
1722
1723         trace = ctf_trace_get_by_path_or_create(session, path_name);
1724         if (!trace) {
1725                 goto send_reply;
1726         }
1727
1728         /* This stream here has one reference on the trace. */
1729         pthread_mutex_lock(&last_relay_stream_id_lock);
1730         stream_handle = ++last_relay_stream_id;
1731         pthread_mutex_unlock(&last_relay_stream_id_lock);
1732
1733         /* We pass ownership of path_name and channel_name. */
1734         stream = stream_create(trace, stream_handle, path_name,
1735                 channel_name, tracefile_size, tracefile_count);
1736         path_name = NULL;
1737         channel_name = NULL;
1738
1739         /*
1740          * Streams are the owners of their trace. Reference to trace is
1741          * kept within stream_create().
1742          */
1743         ctf_trace_put(trace);
1744
1745 send_reply:
1746         memset(&reply, 0, sizeof(reply));
1747         reply.handle = htobe64(stream_handle);
1748         if (!stream) {
1749                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1750         } else {
1751                 reply.ret_code = htobe32(LTTNG_OK);
1752         }
1753
1754         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1755                         sizeof(struct lttcomm_relayd_status_stream), 0);
1756         if (send_ret < (ssize_t) sizeof(reply)) {
1757                 ERR("Failed to send \"add stream\" command reply (ret = %zd)",
1758                                 send_ret);
1759                 ret = -1;
1760         }
1761
1762 end_no_session:
1763         free(path_name);
1764         free(channel_name);
1765         return ret;
1766 }
1767
1768 /*
1769  * relay_close_stream: close a specific stream
1770  */
1771 static int relay_close_stream(
1772                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1773                 struct relay_connection *conn,
1774                 const struct lttng_buffer_view *payload)
1775 {
1776         int ret;
1777         ssize_t send_ret;
1778         struct relay_session *session = conn->session;
1779         struct lttcomm_relayd_close_stream stream_info;
1780         struct lttcomm_relayd_generic_reply reply;
1781         struct relay_stream *stream;
1782
1783         DBG("Close stream received");
1784
1785         if (!session || !conn->version_check_done) {
1786                 ERR("Trying to close a stream before version check");
1787                 ret = -1;
1788                 goto end_no_session;
1789         }
1790
1791         if (payload->size < sizeof(stream_info)) {
1792                 ERR("Unexpected payload size in \"relay_close_stream\": expected >= %zu bytes, got %zu bytes",
1793                                 sizeof(stream_info), payload->size);
1794                 ret = -1;
1795                 goto end_no_session;
1796         }
1797         memcpy(&stream_info, payload->data, sizeof(stream_info));
1798         stream_info.stream_id = be64toh(stream_info.stream_id);
1799         stream_info.last_net_seq_num = be64toh(stream_info.last_net_seq_num);
1800
1801         stream = stream_get_by_id(stream_info.stream_id);
1802         if (!stream) {
1803                 ret = -1;
1804                 goto end;
1805         }
1806
1807         /*
1808          * Set last_net_seq_num before the close flag. Required by data
1809          * pending check.
1810          */
1811         pthread_mutex_lock(&stream->lock);
1812         stream->last_net_seq_num = stream_info.last_net_seq_num;
1813         pthread_mutex_unlock(&stream->lock);
1814
1815         /*
1816          * This is one of the conditions which may trigger a stream close
1817          * with the others being:
1818          *     1) A close command is received for a stream
1819          *     2) The control connection owning the stream is closed
1820          *     3) We have received all of the stream's data _after_ a close
1821          *        request.
1822          */
1823         try_stream_close(stream);
1824         stream_put(stream);
1825         ret = 0;
1826
1827 end:
1828         memset(&reply, 0, sizeof(reply));
1829         if (ret < 0) {
1830                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1831         } else {
1832                 reply.ret_code = htobe32(LTTNG_OK);
1833         }
1834         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1835                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1836         if (send_ret < (ssize_t) sizeof(reply)) {
1837                 ERR("Failed to send \"close stream\" command reply (ret = %zd)",
1838                                 send_ret);
1839                 ret = -1;
1840         }
1841
1842 end_no_session:
1843         return ret;
1844 }
1845
1846 /*
1847  * relay_reset_metadata: reset a metadata stream
1848  */
1849 static
1850 int relay_reset_metadata(
1851                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1852                 struct relay_connection *conn,
1853                 const struct lttng_buffer_view *payload)
1854 {
1855         int ret;
1856         ssize_t send_ret;
1857         struct relay_session *session = conn->session;
1858         struct lttcomm_relayd_reset_metadata stream_info;
1859         struct lttcomm_relayd_generic_reply reply;
1860         struct relay_stream *stream;
1861
1862         DBG("Reset metadata received");
1863
1864         if (!session || !conn->version_check_done) {
1865                 ERR("Trying to reset a metadata stream before version check");
1866                 ret = -1;
1867                 goto end_no_session;
1868         }
1869
1870         if (payload->size < sizeof(stream_info)) {
1871                 ERR("Unexpected payload size in \"relay_reset_metadata\": expected >= %zu bytes, got %zu bytes",
1872                                 sizeof(stream_info), payload->size);
1873                 ret = -1;
1874                 goto end_no_session;
1875         }
1876         memcpy(&stream_info, payload->data, sizeof(stream_info));
1877         stream_info.stream_id = be64toh(stream_info.stream_id);
1878         stream_info.version = be64toh(stream_info.version);
1879
1880         DBG("Update metadata to version %" PRIu64, stream_info.version);
1881
1882         /* Unsupported for live sessions for now. */
1883         if (session->live_timer != 0) {
1884                 ret = -1;
1885                 goto end;
1886         }
1887
1888         stream = stream_get_by_id(stream_info.stream_id);
1889         if (!stream) {
1890                 ret = -1;
1891                 goto end;
1892         }
1893         pthread_mutex_lock(&stream->lock);
1894         if (!stream->is_metadata) {
1895                 ret = -1;
1896                 goto end_unlock;
1897         }
1898
1899         ret = stream_reset_file(stream);
1900         if (ret < 0) {
1901                 ERR("Failed to reset metadata stream %" PRIu64
1902                                 ": stream_path = %s, channel = %s",
1903                                 stream->stream_handle, stream->path_name,
1904                                 stream->channel_name);
1905                 goto end_unlock;
1906         }
1907 end_unlock:
1908         pthread_mutex_unlock(&stream->lock);
1909         stream_put(stream);
1910
1911 end:
1912         memset(&reply, 0, sizeof(reply));
1913         if (ret < 0) {
1914                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1915         } else {
1916                 reply.ret_code = htobe32(LTTNG_OK);
1917         }
1918         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1919                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1920         if (send_ret < (ssize_t) sizeof(reply)) {
1921                 ERR("Failed to send \"reset metadata\" command reply (ret = %zd)",
1922                                 send_ret);
1923                 ret = -1;
1924         }
1925
1926 end_no_session:
1927         return ret;
1928 }
1929
1930 /*
1931  * relay_unknown_command: send -1 if received unknown command
1932  */
1933 static void relay_unknown_command(struct relay_connection *conn)
1934 {
1935         struct lttcomm_relayd_generic_reply reply;
1936         ssize_t send_ret;
1937
1938         memset(&reply, 0, sizeof(reply));
1939         reply.ret_code = htobe32(LTTNG_ERR_UNK);
1940         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1941         if (send_ret < sizeof(reply)) {
1942                 ERR("Failed to send \"unknown command\" command reply (ret = %zd)", send_ret);
1943         }
1944 }
1945
1946 /*
1947  * relay_start: send an acknowledgment to the client to tell if we are
1948  * ready to receive data. We are ready if a session is established.
1949  */
1950 static int relay_start(
1951                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
1952                 struct relay_connection *conn,
1953                 const struct lttng_buffer_view *payload __attribute__((unused)))
1954 {
1955         int ret = 0;
1956         ssize_t send_ret;
1957         struct lttcomm_relayd_generic_reply reply;
1958         struct relay_session *session = conn->session;
1959
1960         if (!session) {
1961                 DBG("Trying to start the streaming without a session established");
1962                 ret = htobe32(LTTNG_ERR_UNK);
1963         }
1964
1965         memset(&reply, 0, sizeof(reply));
1966         reply.ret_code = htobe32(LTTNG_OK);
1967         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1968                         sizeof(reply), 0);
1969         if (send_ret < (ssize_t) sizeof(reply)) {
1970                 ERR("Failed to send \"relay_start\" command reply (ret = %zd)",
1971                                 send_ret);
1972                 ret = -1;
1973         }
1974
1975         return ret;
1976 }
1977
1978 /*
1979  * relay_recv_metadata: receive the metadata for the session.
1980  */
1981 static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
1982                 struct relay_connection *conn,
1983                 const struct lttng_buffer_view *payload)
1984 {
1985         int ret = 0;
1986         struct relay_session *session = conn->session;
1987         struct lttcomm_relayd_metadata_payload metadata_payload_header;
1988         struct relay_stream *metadata_stream;
1989         uint64_t metadata_payload_size;
1990         struct lttng_buffer_view packet_view;
1991
1992         if (!session) {
1993                 ERR("Metadata sent before version check");
1994                 ret = -1;
1995                 goto end;
1996         }
1997
1998         if (recv_hdr->data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1999                 ERR("Incorrect data size");
2000                 ret = -1;
2001                 goto end;
2002         }
2003         metadata_payload_size = recv_hdr->data_size -
2004                         sizeof(struct lttcomm_relayd_metadata_payload);
2005
2006         memcpy(&metadata_payload_header, payload->data,
2007                         sizeof(metadata_payload_header));
2008         metadata_payload_header.stream_id = be64toh(
2009                         metadata_payload_header.stream_id);
2010         metadata_payload_header.padding_size = be32toh(
2011                         metadata_payload_header.padding_size);
2012
2013         metadata_stream = stream_get_by_id(metadata_payload_header.stream_id);
2014         if (!metadata_stream) {
2015                 ret = -1;
2016                 goto end;
2017         }
2018
2019         packet_view = lttng_buffer_view_from_view(payload,
2020                         sizeof(metadata_payload_header), metadata_payload_size);
2021         if (!lttng_buffer_view_is_valid(&packet_view)) {
2022                 ERR("Invalid metadata packet length announced by header");
2023                 ret = -1;
2024                 goto end_put;
2025         }
2026
2027         pthread_mutex_lock(&metadata_stream->lock);
2028         ret = stream_write(metadata_stream, &packet_view,
2029                         metadata_payload_header.padding_size);
2030         pthread_mutex_unlock(&metadata_stream->lock);
2031         if (ret){
2032                 ret = -1;
2033                 goto end_put;
2034         }
2035 end_put:
2036         stream_put(metadata_stream);
2037 end:
2038         return ret;
2039 }
2040
2041 /*
2042  * relay_send_version: send relayd version number
2043  */
2044 static int relay_send_version(
2045                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2046                 struct relay_connection *conn,
2047                 const struct lttng_buffer_view *payload)
2048 {
2049         int ret;
2050         ssize_t send_ret;
2051         struct lttcomm_relayd_version reply, msg;
2052         bool compatible = true;
2053
2054         conn->version_check_done = true;
2055
2056         /* Get version from the other side. */
2057         if (payload->size < sizeof(msg)) {
2058                 ERR("Unexpected payload size in \"relay_send_version\": expected >= %zu bytes, got %zu bytes",
2059                                 sizeof(msg), payload->size);
2060                 ret = -1;
2061                 goto end;
2062         }
2063
2064         memcpy(&msg, payload->data, sizeof(msg));
2065         msg.major = be32toh(msg.major);
2066         msg.minor = be32toh(msg.minor);
2067
2068         memset(&reply, 0, sizeof(reply));
2069         reply.major = RELAYD_VERSION_COMM_MAJOR;
2070         reply.minor = RELAYD_VERSION_COMM_MINOR;
2071
2072         /* Major versions must be the same */
2073         if (reply.major != msg.major) {
2074                 DBG("Incompatible major versions (%u vs %u), deleting session",
2075                                 reply.major, msg.major);
2076                 compatible = false;
2077         }
2078
2079         conn->major = reply.major;
2080         /* We adapt to the lowest compatible version */
2081         if (reply.minor <= msg.minor) {
2082                 conn->minor = reply.minor;
2083         } else {
2084                 conn->minor = msg.minor;
2085         }
2086
2087         reply.major = htobe32(reply.major);
2088         reply.minor = htobe32(reply.minor);
2089         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
2090                         sizeof(reply), 0);
2091         if (send_ret < (ssize_t) sizeof(reply)) {
2092                 ERR("Failed to send \"send version\" command reply (ret = %zd)",
2093                                 send_ret);
2094                 ret = -1;
2095                 goto end;
2096         } else {
2097                 ret = 0;
2098         }
2099
2100         if (!compatible) {
2101                 ret = -1;
2102                 goto end;
2103         }
2104
2105         DBG("Version check done using protocol %u.%u", conn->major,
2106                         conn->minor);
2107
2108 end:
2109         return ret;
2110 }
2111
2112 /*
2113  * Check for data pending for a given stream id from the session daemon.
2114  */
2115 static int relay_data_pending(
2116                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2117                 struct relay_connection *conn,
2118                 const struct lttng_buffer_view *payload)
2119 {
2120         struct relay_session *session = conn->session;
2121         struct lttcomm_relayd_data_pending msg;
2122         struct lttcomm_relayd_generic_reply reply;
2123         struct relay_stream *stream;
2124         ssize_t send_ret;
2125         int ret;
2126         uint64_t stream_seq;
2127
2128         DBG("Data pending command received");
2129
2130         if (!session || !conn->version_check_done) {
2131                 ERR("Trying to check for data before version check");
2132                 ret = -1;
2133                 goto end_no_session;
2134         }
2135
2136         if (payload->size < sizeof(msg)) {
2137                 ERR("Unexpected payload size in \"relay_data_pending\": expected >= %zu bytes, got %zu bytes",
2138                                 sizeof(msg), payload->size);
2139                 ret = -1;
2140                 goto end_no_session;
2141         }
2142         memcpy(&msg, payload->data, sizeof(msg));
2143         msg.stream_id = be64toh(msg.stream_id);
2144         msg.last_net_seq_num = be64toh(msg.last_net_seq_num);
2145
2146         stream = stream_get_by_id(msg.stream_id);
2147         if (stream == NULL) {
2148                 ret = -1;
2149                 goto end;
2150         }
2151
2152         pthread_mutex_lock(&stream->lock);
2153
2154         if (session_streams_have_index(session)) {
2155                 /*
2156                  * Ensure that both the index and stream data have been
2157                  * flushed up to the requested point.
2158                  */
2159                 stream_seq = std::min(stream->prev_data_seq, stream->prev_index_seq);
2160         } else {
2161                 stream_seq = stream->prev_data_seq;
2162         }
2163         DBG("Data pending for stream id %" PRIu64 ": prev_data_seq %" PRIu64
2164                         ", prev_index_seq %" PRIu64
2165                         ", and last_seq %" PRIu64, msg.stream_id,
2166                         stream->prev_data_seq, stream->prev_index_seq,
2167                         msg.last_net_seq_num);
2168
2169         /* Avoid wrapping issue */
2170         if (((int64_t) (stream_seq - msg.last_net_seq_num)) >= 0) {
2171                 /* Data has in fact been written and is NOT pending */
2172                 ret = 0;
2173         } else {
2174                 /* Data still being streamed thus pending */
2175                 ret = 1;
2176         }
2177
2178         stream->data_pending_check_done = true;
2179         pthread_mutex_unlock(&stream->lock);
2180
2181         stream_put(stream);
2182 end:
2183
2184         memset(&reply, 0, sizeof(reply));
2185         reply.ret_code = htobe32(ret);
2186         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2187         if (send_ret < (ssize_t) sizeof(reply)) {
2188                 ERR("Failed to send \"data pending\" command reply (ret = %zd)",
2189                                 send_ret);
2190                 ret = -1;
2191         }
2192
2193 end_no_session:
2194         return ret;
2195 }
2196
2197 /*
2198  * Wait for the control socket to reach a quiescent state.
2199  *
2200  * Note that for now, when receiving this command from the session
2201  * daemon, this means that every subsequent commands or data received on
2202  * the control socket has been handled. So, this is why we simply return
2203  * OK here.
2204  */
2205 static int relay_quiescent_control(
2206                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2207                 struct relay_connection *conn,
2208                 const struct lttng_buffer_view *payload)
2209 {
2210         int ret;
2211         ssize_t send_ret;
2212         struct relay_stream *stream;
2213         struct lttcomm_relayd_quiescent_control msg;
2214         struct lttcomm_relayd_generic_reply reply;
2215
2216         DBG("Checking quiescent state on control socket");
2217
2218         if (!conn->session || !conn->version_check_done) {
2219                 ERR("Trying to check for data before version check");
2220                 ret = -1;
2221                 goto end_no_session;
2222         }
2223
2224         if (payload->size < sizeof(msg)) {
2225                 ERR("Unexpected payload size in \"relay_quiescent_control\": expected >= %zu bytes, got %zu bytes",
2226                                 sizeof(msg), payload->size);
2227                 ret = -1;
2228                 goto end_no_session;
2229         }
2230         memcpy(&msg, payload->data, sizeof(msg));
2231         msg.stream_id = be64toh(msg.stream_id);
2232
2233         stream = stream_get_by_id(msg.stream_id);
2234         if (!stream) {
2235                 goto reply;
2236         }
2237         pthread_mutex_lock(&stream->lock);
2238         stream->data_pending_check_done = true;
2239         pthread_mutex_unlock(&stream->lock);
2240
2241         DBG("Relay quiescent control pending flag set to %" PRIu64, msg.stream_id);
2242         stream_put(stream);
2243 reply:
2244         memset(&reply, 0, sizeof(reply));
2245         reply.ret_code = htobe32(LTTNG_OK);
2246         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2247         if (send_ret < (ssize_t) sizeof(reply)) {
2248                 ERR("Failed to send \"quiescent control\" command reply (ret = %zd)",
2249                                 send_ret);
2250                 ret = -1;
2251         } else {
2252                 ret = 0;
2253         }
2254
2255 end_no_session:
2256         return ret;
2257 }
2258
2259 /*
2260  * Initialize a data pending command. This means that a consumer is about
2261  * to ask for data pending for each stream it holds. Simply iterate over
2262  * all streams of a session and set the data_pending_check_done flag.
2263  *
2264  * This command returns to the client a LTTNG_OK code.
2265  */
2266 static int relay_begin_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
2267                 struct relay_connection *conn,
2268                 const struct lttng_buffer_view *payload)
2269 {
2270         int ret;
2271         ssize_t send_ret;
2272         struct lttng_ht_iter iter;
2273         struct lttcomm_relayd_begin_data_pending msg;
2274         struct lttcomm_relayd_generic_reply reply;
2275         struct relay_stream *stream;
2276
2277         LTTNG_ASSERT(recv_hdr);
2278         LTTNG_ASSERT(conn);
2279
2280         DBG("Init streams for data pending");
2281
2282         if (!conn->session || !conn->version_check_done) {
2283                 ERR("Trying to check for data before version check");
2284                 ret = -1;
2285                 goto end_no_session;
2286         }
2287
2288         if (payload->size < sizeof(msg)) {
2289                 ERR("Unexpected payload size in \"relay_begin_data_pending\": expected >= %zu bytes, got %zu bytes",
2290                                 sizeof(msg), payload->size);
2291                 ret = -1;
2292                 goto end_no_session;
2293         }
2294         memcpy(&msg, payload->data, sizeof(msg));
2295         msg.session_id = be64toh(msg.session_id);
2296
2297         /*
2298          * Iterate over all streams to set the begin data pending flag.
2299          * For now, the streams are indexed by stream handle so we have
2300          * to iterate over all streams to find the one associated with
2301          * the right session_id.
2302          */
2303         rcu_read_lock();
2304         cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2305                         node.node) {
2306                 if (!stream_get(stream)) {
2307                         continue;
2308                 }
2309                 if (stream->trace->session->id == msg.session_id) {
2310                         pthread_mutex_lock(&stream->lock);
2311                         stream->data_pending_check_done = false;
2312                         pthread_mutex_unlock(&stream->lock);
2313                         DBG("Set begin data pending flag to stream %" PRIu64,
2314                                         stream->stream_handle);
2315                 }
2316                 stream_put(stream);
2317         }
2318         rcu_read_unlock();
2319
2320         memset(&reply, 0, sizeof(reply));
2321         /* All good, send back reply. */
2322         reply.ret_code = htobe32(LTTNG_OK);
2323
2324         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2325         if (send_ret < (ssize_t) sizeof(reply)) {
2326                 ERR("Failed to send \"begin data pending\" command reply (ret = %zd)",
2327                         send_ret);
2328                 ret = -1;
2329         } else {
2330                 ret = 0;
2331         }
2332
2333 end_no_session:
2334         return ret;
2335 }
2336
2337 /*
2338  * End data pending command. This will check, for a given session id, if
2339  * each stream associated with it has its data_pending_check_done flag
2340  * set. If not, this means that the client lost track of the stream but
2341  * the data is still being streamed on our side. In this case, we inform
2342  * the client that data is in flight.
2343  *
2344  * Return to the client if there is data in flight or not with a ret_code.
2345  */
2346 static int relay_end_data_pending(
2347                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2348                 struct relay_connection *conn,
2349                 const struct lttng_buffer_view *payload)
2350 {
2351         int ret;
2352         ssize_t send_ret;
2353         struct lttng_ht_iter iter;
2354         struct lttcomm_relayd_end_data_pending msg;
2355         struct lttcomm_relayd_generic_reply reply;
2356         struct relay_stream *stream;
2357         uint32_t is_data_inflight = 0;
2358
2359         DBG("End data pending command");
2360
2361         if (!conn->session || !conn->version_check_done) {
2362                 ERR("Trying to check for data before version check");
2363                 ret = -1;
2364                 goto end_no_session;
2365         }
2366
2367         if (payload->size < sizeof(msg)) {
2368                 ERR("Unexpected payload size in \"relay_end_data_pending\": expected >= %zu bytes, got %zu bytes",
2369                                 sizeof(msg), payload->size);
2370                 ret = -1;
2371                 goto end_no_session;
2372         }
2373         memcpy(&msg, payload->data, sizeof(msg));
2374         msg.session_id = be64toh(msg.session_id);
2375
2376         /*
2377          * Iterate over all streams to see if the begin data pending
2378          * flag is set.
2379          */
2380         rcu_read_lock();
2381         cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2382                         node.node) {
2383                 if (!stream_get(stream)) {
2384                         continue;
2385                 }
2386                 if (stream->trace->session->id != msg.session_id) {
2387                         stream_put(stream);
2388                         continue;
2389                 }
2390                 pthread_mutex_lock(&stream->lock);
2391                 if (!stream->data_pending_check_done) {
2392                         uint64_t stream_seq;
2393
2394                         if (session_streams_have_index(conn->session)) {
2395                                 /*
2396                                  * Ensure that both the index and stream data have been
2397                                  * flushed up to the requested point.
2398                                  */
2399                                 stream_seq = std::min(stream->prev_data_seq, stream->prev_index_seq);
2400                         } else {
2401                                 stream_seq = stream->prev_data_seq;
2402                         }
2403                         if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
2404                                 is_data_inflight = 1;
2405                                 DBG("Data is still in flight for stream %" PRIu64,
2406                                                 stream->stream_handle);
2407                                 pthread_mutex_unlock(&stream->lock);
2408                                 stream_put(stream);
2409                                 break;
2410                         }
2411                 }
2412                 pthread_mutex_unlock(&stream->lock);
2413                 stream_put(stream);
2414         }
2415         rcu_read_unlock();
2416
2417         memset(&reply, 0, sizeof(reply));
2418         /* All good, send back reply. */
2419         reply.ret_code = htobe32(is_data_inflight);
2420
2421         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2422         if (send_ret < (ssize_t) sizeof(reply)) {
2423                 ERR("Failed to send \"end data pending\" command reply (ret = %zd)",
2424                         send_ret);
2425                 ret = -1;
2426         } else {
2427                 ret = 0;
2428         }
2429
2430 end_no_session:
2431         return ret;
2432 }
2433
2434 /*
2435  * Receive an index for a specific stream.
2436  *
2437  * Return 0 on success else a negative value.
2438  */
2439 static int relay_recv_index(
2440                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2441                 struct relay_connection *conn,
2442                 const struct lttng_buffer_view *payload)
2443 {
2444         int ret;
2445         ssize_t send_ret;
2446         struct relay_session *session = conn->session;
2447         struct lttcomm_relayd_index index_info;
2448         struct lttcomm_relayd_generic_reply reply;
2449         struct relay_stream *stream;
2450         size_t msg_len;
2451
2452         LTTNG_ASSERT(conn);
2453
2454         DBG("Relay receiving index");
2455
2456         if (!session || !conn->version_check_done) {
2457                 ERR("Trying to close a stream before version check");
2458                 ret = -1;
2459                 goto end_no_session;
2460         }
2461
2462         msg_len = lttcomm_relayd_index_len(
2463                         lttng_to_index_major(conn->major, conn->minor),
2464                         lttng_to_index_minor(conn->major, conn->minor));
2465         if (payload->size < msg_len) {
2466                 ERR("Unexpected payload size in \"relay_recv_index\": expected >= %zu bytes, got %zu bytes",
2467                                 msg_len, payload->size);
2468                 ret = -1;
2469                 goto end_no_session;
2470         }
2471         memcpy(&index_info, payload->data, msg_len);
2472         index_info.relay_stream_id = be64toh(index_info.relay_stream_id);
2473         index_info.net_seq_num = be64toh(index_info.net_seq_num);
2474         index_info.packet_size = be64toh(index_info.packet_size);
2475         index_info.content_size = be64toh(index_info.content_size);
2476         index_info.timestamp_begin = be64toh(index_info.timestamp_begin);
2477         index_info.timestamp_end = be64toh(index_info.timestamp_end);
2478         index_info.events_discarded = be64toh(index_info.events_discarded);
2479         index_info.stream_id = be64toh(index_info.stream_id);
2480
2481         if (conn->minor >= 8) {
2482                 index_info.stream_instance_id =
2483                                 be64toh(index_info.stream_instance_id);
2484                 index_info.packet_seq_num = be64toh(index_info.packet_seq_num);
2485         } else {
2486                 index_info.stream_instance_id = -1ULL;
2487                 index_info.packet_seq_num = -1ULL;
2488         }
2489
2490         stream = stream_get_by_id(index_info.relay_stream_id);
2491         if (!stream) {
2492                 ERR("stream_get_by_id not found");
2493                 ret = -1;
2494                 goto end;
2495         }
2496
2497         pthread_mutex_lock(&stream->lock);
2498         ret = stream_add_index(stream, &index_info);
2499         pthread_mutex_unlock(&stream->lock);
2500         if (ret) {
2501                 goto end_stream_put;
2502         }
2503
2504 end_stream_put:
2505         stream_put(stream);
2506 end:
2507         memset(&reply, 0, sizeof(reply));
2508         if (ret < 0) {
2509                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
2510         } else {
2511                 reply.ret_code = htobe32(LTTNG_OK);
2512         }
2513         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2514         if (send_ret < (ssize_t) sizeof(reply)) {
2515                 ERR("Failed to send \"recv index\" command reply (ret = %zd)", send_ret);
2516                 ret = -1;
2517         }
2518
2519 end_no_session:
2520         return ret;
2521 }
2522
2523 /*
2524  * Receive the streams_sent message.
2525  *
2526  * Return 0 on success else a negative value.
2527  */
2528 static int relay_streams_sent(
2529                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2530                 struct relay_connection *conn,
2531                 const struct lttng_buffer_view *payload __attribute__((unused)))
2532 {
2533         int ret;
2534         ssize_t send_ret;
2535         struct lttcomm_relayd_generic_reply reply;
2536
2537         LTTNG_ASSERT(conn);
2538
2539         DBG("Relay receiving streams_sent");
2540
2541         if (!conn->session || !conn->version_check_done) {
2542                 ERR("Trying to close a stream before version check");
2543                 ret = -1;
2544                 goto end_no_session;
2545         }
2546
2547         /*
2548          * Publish every pending stream in the connection recv list which are
2549          * now ready to be used by the viewer.
2550          */
2551         publish_connection_local_streams(conn);
2552
2553         memset(&reply, 0, sizeof(reply));
2554         reply.ret_code = htobe32(LTTNG_OK);
2555         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2556         if (send_ret < (ssize_t) sizeof(reply)) {
2557                 ERR("Failed to send \"streams sent\" command reply (ret = %zd)",
2558                         send_ret);
2559                 ret = -1;
2560         } else {
2561                 /* Success. */
2562                 ret = 0;
2563         }
2564
2565 end_no_session:
2566         return ret;
2567 }
2568
2569 static ssize_t relay_unpack_rotate_streams_header(
2570                 const struct lttng_buffer_view *payload,
2571                 struct lttcomm_relayd_rotate_streams *_rotate_streams)
2572 {
2573         struct lttcomm_relayd_rotate_streams rotate_streams;
2574         /*
2575          * Set to the smallest version (packed) of `lttcomm_relayd_rotate_streams`.
2576          * This is the smallest version of this structure, but it can be larger;
2577          * this variable is updated once the proper size of the structure is known.
2578          *
2579          * See comment at the declaration of this structure for more information.
2580          */
2581         ssize_t header_len = sizeof(struct lttcomm_relayd_rotate_streams_packed);
2582         size_t expected_payload_size_no_padding,
2583                 expected_payload_size_3_bytes_padding,
2584                 expected_payload_size_7_bytes_padding;
2585
2586         if (payload->size < header_len) {
2587                 ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected >= %zu bytes, got %zu bytes",
2588                                 header_len, payload->size);
2589                 goto error;
2590         }
2591
2592         /*
2593          * Some versions incorrectly omitted the LTTNG_PACKED annotation on the
2594          * `new_chunk_id` optional field of struct lttcomm_relayd_rotate_streams.
2595          *
2596          * We start by "unpacking" `stream_count` to figure out the padding length
2597          * emited by our peer.
2598          */
2599         {
2600                 decltype(rotate_streams.stream_count) stream_count;
2601
2602                 memcpy(&stream_count, payload->data, sizeof(stream_count));
2603                 rotate_streams.stream_count = be32toh(stream_count);
2604         }
2605
2606         rotate_streams.new_chunk_id = LTTNG_OPTIONAL_INIT_UNSET;
2607
2608         /*
2609          * Payload size expected given the possible padding lengths in
2610          * `struct lttcomm_relayd_rotate_streams`.
2611          */
2612         expected_payload_size_no_padding = (rotate_streams.stream_count *
2613                 sizeof(*rotate_streams.rotation_positions)) +
2614                 sizeof(lttcomm_relayd_rotate_streams_packed);
2615         expected_payload_size_3_bytes_padding = (rotate_streams.stream_count *
2616                 sizeof(*rotate_streams.rotation_positions)) +
2617                 sizeof(lttcomm_relayd_rotate_streams_3_bytes_padding);
2618         expected_payload_size_7_bytes_padding = (rotate_streams.stream_count *
2619                 sizeof(*rotate_streams.rotation_positions)) +
2620                 sizeof(lttcomm_relayd_rotate_streams_7_bytes_padding);
2621
2622         if (payload->size == expected_payload_size_no_padding) {
2623                 struct lttcomm_relayd_rotate_streams_packed packed_rotate_streams;
2624
2625                 /*
2626                  * This handles cases where someone might build with
2627                  * -fpack-struct or any other toolchain that wouldn't produce
2628                  * padding to align `value`.
2629                  */
2630                 DBG("Received `struct lttcomm_relayd_rotate_streams` with no padding");
2631
2632                 header_len = sizeof(packed_rotate_streams);
2633                 memcpy(&packed_rotate_streams, payload->data, header_len);
2634
2635                 /* Unpack the packed structure to the natively-packed version. */
2636                 _rotate_streams->new_chunk_id = (typeof(_rotate_streams->new_chunk_id)){
2637                         .is_set = !!packed_rotate_streams.new_chunk_id.is_set,
2638                         .value = be64toh(packed_rotate_streams.new_chunk_id.value),
2639                 };
2640                 _rotate_streams->stream_count = be32toh(packed_rotate_streams.stream_count);
2641         } else if (payload->size == expected_payload_size_3_bytes_padding) {
2642                 struct lttcomm_relayd_rotate_streams_3_bytes_padding padded_rotate_streams;
2643
2644                 DBG("Received `struct lttcomm_relayd_rotate_streams` with 3 bytes of padding (4-byte aligned peer)");
2645
2646                 header_len = sizeof(padded_rotate_streams);
2647                 memcpy(&padded_rotate_streams, payload->data, header_len);
2648
2649                 /* Unpack the 3-byte padded structure to the natively-packed version. */
2650                 _rotate_streams->new_chunk_id = (typeof(_rotate_streams->new_chunk_id)){
2651                         .is_set = !!padded_rotate_streams.new_chunk_id.is_set,
2652                         .value = be64toh(padded_rotate_streams.new_chunk_id.value),
2653                 };
2654                 _rotate_streams->stream_count = be32toh(padded_rotate_streams.stream_count);
2655         } else if (payload->size == expected_payload_size_7_bytes_padding) {
2656                 struct lttcomm_relayd_rotate_streams_7_bytes_padding padded_rotate_streams;
2657
2658                 DBG("Received `struct lttcomm_relayd_rotate_streams` with 7 bytes of padding (8-byte aligned peer)");
2659
2660                 header_len = sizeof(padded_rotate_streams);
2661                 memcpy(&padded_rotate_streams, payload->data, header_len);
2662
2663                 /* Unpack the 7-byte padded structure to the natively-packed version. */
2664                 _rotate_streams->new_chunk_id = (typeof(_rotate_streams->new_chunk_id)){
2665                         .is_set = !!padded_rotate_streams.new_chunk_id.is_set,
2666                         .value = be64toh(padded_rotate_streams.new_chunk_id.value),
2667                 };
2668                 _rotate_streams->stream_count = be32toh(padded_rotate_streams.stream_count);
2669
2670                 header_len = sizeof(padded_rotate_streams);
2671         } else {
2672                 ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected %zu, %zu or %zu bytes, got %zu bytes",
2673                                 expected_payload_size_no_padding,
2674                                 expected_payload_size_3_bytes_padding,
2675                                 expected_payload_size_7_bytes_padding,
2676                                 payload->size);
2677                 goto error;
2678         }
2679
2680         return header_len;
2681 error:
2682         return -1;
2683 }
2684
2685 /*
2686  * relay_rotate_session_stream: rotate a stream to a new tracefile for the
2687  * session rotation feature (not the tracefile rotation feature).
2688  */
2689 static int relay_rotate_session_streams(
2690                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2691                 struct relay_connection *conn,
2692                 const struct lttng_buffer_view *payload)
2693 {
2694         int ret = 0;
2695         uint32_t i;
2696         ssize_t send_ret;
2697         enum lttng_error_code reply_code = LTTNG_ERR_UNK;
2698         struct relay_session *session = conn->session;
2699         struct lttcomm_relayd_rotate_streams rotate_streams;
2700         struct lttcomm_relayd_generic_reply reply = {};
2701         struct relay_stream *stream = NULL;
2702         struct lttng_trace_chunk *next_trace_chunk = NULL;
2703         struct lttng_buffer_view stream_positions;
2704         char chunk_id_buf[MAX_INT_DEC_LEN(uint64_t)];
2705         const char *chunk_id_str = "none";
2706         ssize_t header_len;
2707
2708         if (!session || !conn->version_check_done) {
2709                 ERR("Trying to rotate a stream before version check");
2710                 ret = -1;
2711                 goto end_no_reply;
2712         }
2713
2714         if (session->major == 2 && session->minor < 11) {
2715                 ERR("Unsupported feature before 2.11");
2716                 ret = -1;
2717                 goto end_no_reply;
2718         }
2719
2720         header_len = relay_unpack_rotate_streams_header(payload, &rotate_streams);
2721         if (header_len < 0) {
2722                 ret = -1;
2723                 goto end_no_reply;
2724         }
2725
2726         if (rotate_streams.new_chunk_id.is_set) {
2727                 /*
2728                  * Retrieve the trace chunk the stream must transition to. As
2729                  * per the protocol, this chunk should have been created
2730                  * before this command is received.
2731                  */
2732                 next_trace_chunk = sessiond_trace_chunk_registry_get_chunk(
2733                                 sessiond_trace_chunk_registry,
2734                                 session->sessiond_uuid,
2735                                 conn->session->id_sessiond.is_set ?
2736                                         conn->session->id_sessiond.value :
2737                                         conn->session->id,
2738                                 rotate_streams.new_chunk_id.value);
2739                 if (!next_trace_chunk) {
2740                         char uuid_str[LTTNG_UUID_STR_LEN];
2741
2742                         lttng_uuid_to_str(session->sessiond_uuid, uuid_str);
2743                         ERR("Unknown next trace chunk in ROTATE_STREAMS command: sessiond_uuid = {%s}, session_id = %" PRIu64
2744                                         ", trace_chunk_id = %" PRIu64,
2745                                         uuid_str, session->id,
2746                                         rotate_streams.new_chunk_id.value);
2747                         reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2748                         ret = -1;
2749                         goto end;
2750                 }
2751
2752                 ret = snprintf(chunk_id_buf, sizeof(chunk_id_buf), "%" PRIu64,
2753                                 rotate_streams.new_chunk_id.value);
2754                 if (ret < 0 || ret >= sizeof(chunk_id_buf)) {
2755                         chunk_id_str = "formatting error";
2756                 } else {
2757                         chunk_id_str = chunk_id_buf;
2758                 }
2759         }
2760
2761         DBG("Rotate %" PRIu32 " streams of session \"%s\" to chunk \"%s\"",
2762                         rotate_streams.stream_count, session->session_name,
2763                         chunk_id_str);
2764
2765         stream_positions = lttng_buffer_view_from_view(payload,
2766                         header_len, -1);
2767         if (!stream_positions.data ||
2768                         stream_positions.size <
2769                                         (rotate_streams.stream_count *
2770                                                         sizeof(struct lttcomm_relayd_stream_rotation_position))) {
2771                 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2772                 ret = -1;
2773                 goto end;
2774         }
2775
2776         for (i = 0; i < rotate_streams.stream_count; i++) {
2777                 struct lttcomm_relayd_stream_rotation_position *position_comm =
2778                                 &((typeof(position_comm)) stream_positions.data)[i];
2779                 const struct lttcomm_relayd_stream_rotation_position pos = {
2780                         .stream_id = be64toh(position_comm->stream_id),
2781                         .rotate_at_seq_num = be64toh(
2782                                         position_comm->rotate_at_seq_num),
2783                 };
2784
2785                 stream = stream_get_by_id(pos.stream_id);
2786                 if (!stream) {
2787                         reply_code = LTTNG_ERR_INVALID;
2788                         ret = -1;
2789                         goto end;
2790                 }
2791
2792                 pthread_mutex_lock(&stream->lock);
2793                 ret = stream_set_pending_rotation(stream, next_trace_chunk,
2794                                 pos.rotate_at_seq_num);
2795                 pthread_mutex_unlock(&stream->lock);
2796                 if (ret) {
2797                         reply_code = LTTNG_ERR_FILE_CREATION_ERROR;
2798                         goto end;
2799                 }
2800
2801                 stream_put(stream);
2802                 stream = NULL;
2803         }
2804
2805         reply_code = LTTNG_OK;
2806         ret = 0;
2807 end:
2808         if (stream) {
2809                 stream_put(stream);
2810         }
2811
2812         reply.ret_code = htobe32((uint32_t) reply_code);
2813         send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
2814                         sizeof(struct lttcomm_relayd_generic_reply), 0);
2815         if (send_ret < (ssize_t) sizeof(reply)) {
2816                 ERR("Failed to send \"rotate session stream\" command reply (ret = %zd)",
2817                                 send_ret);
2818                 ret = -1;
2819         }
2820 end_no_reply:
2821         lttng_trace_chunk_put(next_trace_chunk);
2822         return ret;
2823 }
2824
2825 /*
2826  * relay_create_trace_chunk: create a new trace chunk
2827  */
2828 static int relay_create_trace_chunk(
2829                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
2830                 struct relay_connection *conn,
2831                 const struct lttng_buffer_view *payload)
2832 {
2833         int ret = 0;
2834         ssize_t send_ret;
2835         struct relay_session *session = conn->session;
2836         struct lttcomm_relayd_create_trace_chunk *msg;
2837         struct lttcomm_relayd_generic_reply reply = {};
2838         struct lttng_buffer_view header_view;
2839         struct lttng_trace_chunk *chunk = NULL, *published_chunk = NULL;
2840         enum lttng_error_code reply_code = LTTNG_OK;
2841         enum lttng_trace_chunk_status chunk_status;
2842         const char *new_path;
2843
2844         if (!session || !conn->version_check_done) {
2845                 ERR("Trying to create a trace chunk before version check");
2846                 ret = -1;
2847                 goto end_no_reply;
2848         }
2849
2850         if (session->major == 2 && session->minor < 11) {
2851                 ERR("Chunk creation command is unsupported before 2.11");
2852                 ret = -1;
2853                 goto end_no_reply;
2854         }
2855
2856         header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2857         if (!lttng_buffer_view_is_valid(&header_view)) {
2858                 ERR("Failed to receive payload of chunk creation command");
2859                 ret = -1;
2860                 goto end_no_reply;
2861         }
2862
2863         /* Convert to host endianness. */
2864         msg = (typeof(msg)) header_view.data;
2865         msg->chunk_id = be64toh(msg->chunk_id);
2866         msg->creation_timestamp = be64toh(msg->creation_timestamp);
2867         msg->override_name_length = be32toh(msg->override_name_length);
2868
2869         pthread_mutex_lock(&conn->session->lock);
2870         session->ongoing_rotation = true;
2871         if (session->current_trace_chunk &&
2872                         !lttng_trace_chunk_get_name_overridden(session->current_trace_chunk)) {
2873                 chunk_status = lttng_trace_chunk_rename_path(session->current_trace_chunk,
2874                                         DEFAULT_CHUNK_TMP_OLD_DIRECTORY);
2875                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2876                         ERR("Failed to rename old chunk");
2877                         ret = -1;
2878                         reply_code = LTTNG_ERR_UNK;
2879                         goto end;
2880                 }
2881         }
2882         if (!session->current_trace_chunk) {
2883                 if (!session->has_rotated) {
2884                         new_path = "";
2885                 } else {
2886                         new_path = NULL;
2887                 }
2888         } else {
2889                 new_path = DEFAULT_CHUNK_TMP_NEW_DIRECTORY;
2890         }
2891         chunk = lttng_trace_chunk_create(
2892                         msg->chunk_id, msg->creation_timestamp, new_path);
2893         if (!chunk) {
2894                 ERR("Failed to create trace chunk in trace chunk creation command");
2895                 ret = -1;
2896                 reply_code = LTTNG_ERR_NOMEM;
2897                 goto end;
2898         }
2899         lttng_trace_chunk_set_fd_tracker(chunk, the_fd_tracker);
2900
2901         if (msg->override_name_length) {
2902                 const char *name;
2903                 const struct lttng_buffer_view chunk_name_view =
2904                                 lttng_buffer_view_from_view(payload,
2905                                                 sizeof(*msg),
2906                                                 msg->override_name_length);
2907
2908                 if (!lttng_buffer_view_is_valid(&chunk_name_view)) {
2909                         ERR("Invalid payload of chunk creation command (protocol error): buffer too short for expected name length");
2910                         ret = -1;
2911                         reply_code = LTTNG_ERR_INVALID;
2912                         goto end;
2913                 }
2914
2915                 name = chunk_name_view.data;
2916                 if (name[msg->override_name_length - 1]) {
2917                         ERR("Invalid payload of chunk creation command (protocol error): name is not null-terminated");
2918                         ret = -1;
2919                         reply_code = LTTNG_ERR_INVALID;
2920                         goto end;
2921                 }
2922
2923                 chunk_status = lttng_trace_chunk_override_name(
2924                                 chunk, chunk_name_view.data);
2925                 switch (chunk_status) {
2926                 case LTTNG_TRACE_CHUNK_STATUS_OK:
2927                         break;
2928                 case LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT:
2929                         ERR("Failed to set the name of new trace chunk in trace chunk creation command (invalid name)");
2930                         reply_code = LTTNG_ERR_INVALID;
2931                         ret = -1;
2932                         goto end;
2933                 default:
2934                         ERR("Failed to set the name of new trace chunk in trace chunk creation command (unknown error)");
2935                         reply_code = LTTNG_ERR_UNK;
2936                         ret = -1;
2937                         goto end;
2938                 }
2939         }
2940
2941         chunk_status = lttng_trace_chunk_set_credentials_current_user(chunk);
2942         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2943                 reply_code = LTTNG_ERR_UNK;
2944                 ret = -1;
2945                 goto end;
2946         }
2947
2948         LTTNG_ASSERT(conn->session->output_directory);
2949         chunk_status = lttng_trace_chunk_set_as_owner(chunk,
2950                         conn->session->output_directory);
2951         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2952                 reply_code = LTTNG_ERR_UNK;
2953                 ret = -1;
2954                 goto end;
2955         }
2956
2957         published_chunk = sessiond_trace_chunk_registry_publish_chunk(
2958                         sessiond_trace_chunk_registry,
2959                         conn->session->sessiond_uuid,
2960                         conn->session->id_sessiond.is_set ?
2961                                 conn->session->id_sessiond.value :
2962                                 conn->session->id,
2963                         chunk);
2964         if (!published_chunk) {
2965                 char uuid_str[LTTNG_UUID_STR_LEN];
2966
2967                 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
2968                 ERR("Failed to publish chunk: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2969                                 uuid_str,
2970                                 conn->session->id,
2971                                 msg->chunk_id);
2972                 ret = -1;
2973                 reply_code = LTTNG_ERR_NOMEM;
2974                 goto end;
2975         }
2976
2977         if (conn->session->pending_closure_trace_chunk) {
2978                 /*
2979                  * Invalid; this means a second create_trace_chunk command was
2980                  * received before a close_trace_chunk.
2981                  */
2982                 ERR("Invalid trace chunk close command received; a trace chunk is already waiting for a trace chunk close command");
2983                 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2984                 ret = -1;
2985                 goto end;
2986         }
2987         conn->session->pending_closure_trace_chunk =
2988                         conn->session->current_trace_chunk;
2989         conn->session->current_trace_chunk = published_chunk;
2990         published_chunk = NULL;
2991         if (!conn->session->pending_closure_trace_chunk) {
2992                 session->ongoing_rotation = false;
2993         }
2994 end:
2995         pthread_mutex_unlock(&conn->session->lock);
2996         reply.ret_code = htobe32((uint32_t) reply_code);
2997         send_ret = conn->sock->ops->sendmsg(conn->sock,
2998                         &reply,
2999                         sizeof(struct lttcomm_relayd_generic_reply),
3000                         0);
3001         if (send_ret < (ssize_t) sizeof(reply)) {
3002                 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
3003                                 send_ret);
3004                 ret = -1;
3005         }
3006 end_no_reply:
3007         lttng_trace_chunk_put(chunk);
3008         lttng_trace_chunk_put(published_chunk);
3009         return ret;
3010 }
3011
3012 /*
3013  * relay_close_trace_chunk: close a trace chunk
3014  */
3015 static int relay_close_trace_chunk(
3016                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
3017                 struct relay_connection *conn,
3018                 const struct lttng_buffer_view *payload)
3019 {
3020         int ret = 0, buf_ret;
3021         ssize_t send_ret;
3022         struct relay_session *session = conn->session;
3023         struct lttcomm_relayd_close_trace_chunk *msg;
3024         struct lttcomm_relayd_close_trace_chunk_reply reply = {};
3025         struct lttng_buffer_view header_view;
3026         struct lttng_trace_chunk *chunk = NULL;
3027         enum lttng_error_code reply_code = LTTNG_OK;
3028         enum lttng_trace_chunk_status chunk_status;
3029         uint64_t chunk_id;
3030         LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {};
3031         time_t close_timestamp;
3032         char closed_trace_chunk_path[LTTNG_PATH_MAX];
3033         size_t path_length = 0;
3034         const char *chunk_name = NULL;
3035         struct lttng_dynamic_buffer reply_payload;
3036         const char *new_path;
3037
3038         lttng_dynamic_buffer_init(&reply_payload);
3039
3040         if (!session || !conn->version_check_done) {
3041                 ERR("Trying to close a trace chunk before version check");
3042                 ret = -1;
3043                 goto end_no_reply;
3044         }
3045
3046         if (session->major == 2 && session->minor < 11) {
3047                 ERR("Chunk close command is unsupported before 2.11");
3048                 ret = -1;
3049                 goto end_no_reply;
3050         }
3051
3052         header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
3053         if (!lttng_buffer_view_is_valid(&header_view)) {
3054                 ERR("Failed to receive payload of chunk close command");
3055                 ret = -1;
3056                 goto end_no_reply;
3057         }
3058
3059         /* Convert to host endianness. */
3060         msg = (typeof(msg)) header_view.data;
3061         chunk_id = be64toh(msg->chunk_id);
3062         close_timestamp = (time_t) be64toh(msg->close_timestamp);
3063         close_command.value = (lttng_trace_chunk_command_type) be32toh(msg->close_command.value);
3064         close_command.is_set = msg->close_command.is_set;
3065
3066         chunk = sessiond_trace_chunk_registry_get_chunk(
3067                         sessiond_trace_chunk_registry,
3068                         conn->session->sessiond_uuid,
3069                         conn->session->id_sessiond.is_set ?
3070                                 conn->session->id_sessiond.value :
3071                                 conn->session->id,
3072                         chunk_id);
3073         if (!chunk) {
3074                 char uuid_str[LTTNG_UUID_STR_LEN];
3075
3076                 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
3077                 ERR("Failed to find chunk to close: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
3078                                 uuid_str,
3079                                 conn->session->id,
3080                                 msg->chunk_id);
3081                 ret = -1;
3082                 reply_code = LTTNG_ERR_NOMEM;
3083                 goto end;
3084         }
3085
3086         pthread_mutex_lock(&session->lock);
3087         if (close_command.is_set &&
3088                         close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_DELETE) {
3089                 /*
3090                  * Clear command. It is a protocol error to ask for a
3091                  * clear on a relay which does not allow it. Querying
3092                  * the configuration allows figuring out whether
3093                  * clearing is allowed before doing the clear.
3094                  */
3095                 if (!opt_allow_clear) {
3096                         ret = -1;
3097                         reply_code = LTTNG_ERR_INVALID_PROTOCOL;
3098                         goto end_unlock_session;
3099                 }
3100         }
3101         if (session->pending_closure_trace_chunk &&
3102                         session->pending_closure_trace_chunk != chunk) {
3103                 ERR("Trace chunk close command for session \"%s\" does not target the trace chunk pending closure",
3104                                 session->session_name);
3105                 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
3106                 ret = -1;
3107                 goto end_unlock_session;
3108         }
3109
3110         if (session->current_trace_chunk && session->current_trace_chunk != chunk &&
3111                         !lttng_trace_chunk_get_name_overridden(session->current_trace_chunk)) {
3112                 if (close_command.is_set &&
3113                                 close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_DELETE &&
3114                                 !session->has_rotated) {
3115                         /* New chunk stays in session output directory. */
3116                         new_path = "";
3117                 } else {
3118                         /* Use chunk name for new chunk. */
3119                         new_path = NULL;
3120                 }
3121                 /* Rename new chunk path. */
3122                 chunk_status = lttng_trace_chunk_rename_path(session->current_trace_chunk,
3123                                 new_path);
3124                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3125                         ret = -1;
3126                         goto end_unlock_session;
3127                 }
3128                 session->ongoing_rotation = false;
3129         }
3130         if ((!close_command.is_set ||
3131                         close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_NO_OPERATION) &&
3132                         !lttng_trace_chunk_get_name_overridden(chunk)) {
3133                 const char *old_path;
3134
3135                 if (!session->has_rotated) {
3136                         old_path = "";
3137                 } else {
3138                         old_path = NULL;
3139                 }
3140                 /* We need to move back the .tmp_old_chunk to its rightful place. */
3141                 chunk_status = lttng_trace_chunk_rename_path(chunk, old_path);
3142                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3143                         ret = -1;
3144                         goto end_unlock_session;
3145                 }
3146         }
3147         chunk_status = lttng_trace_chunk_set_close_timestamp(
3148                         chunk, close_timestamp);
3149         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3150                 ERR("Failed to set trace chunk close timestamp");
3151                 ret = -1;
3152                 reply_code = LTTNG_ERR_UNK;
3153                 goto end_unlock_session;
3154         }
3155
3156         if (close_command.is_set) {
3157                 chunk_status = lttng_trace_chunk_set_close_command(
3158                                 chunk, close_command.value);
3159                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3160                         ret = -1;
3161                         reply_code = LTTNG_ERR_INVALID;
3162                         goto end_unlock_session;
3163                 }
3164         }
3165         chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name, NULL);
3166         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
3167                 ERR("Failed to get chunk name");
3168                 ret = -1;
3169                 reply_code = LTTNG_ERR_UNK;
3170                 goto end_unlock_session;
3171         }
3172         if (!session->has_rotated && !session->snapshot) {
3173                 ret = lttng_strncpy(closed_trace_chunk_path,
3174                                 session->output_path,
3175                                 sizeof(closed_trace_chunk_path));
3176                 if (ret) {
3177                         ERR("Failed to send trace chunk path: path length of %zu bytes exceeds the maximal allowed length of %zu bytes",
3178                                         strlen(session->output_path),
3179                                         sizeof(closed_trace_chunk_path));
3180                         reply_code = LTTNG_ERR_NOMEM;
3181                         ret = -1;
3182                         goto end_unlock_session;
3183                 }
3184         } else {
3185                 if (session->snapshot) {
3186                         ret = snprintf(closed_trace_chunk_path,
3187                                         sizeof(closed_trace_chunk_path),
3188                                         "%s/%s", session->output_path,
3189                                         chunk_name);
3190                 } else {
3191                         ret = snprintf(closed_trace_chunk_path,
3192                                         sizeof(closed_trace_chunk_path),
3193                                         "%s/" DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY
3194                                         "/%s",
3195                                         session->output_path, chunk_name);
3196                 }
3197                 if (ret < 0 || ret == sizeof(closed_trace_chunk_path)) {
3198                         ERR("Failed to format closed trace chunk resulting path");
3199                         reply_code = ret < 0 ? LTTNG_ERR_UNK : LTTNG_ERR_NOMEM;
3200                         ret = -1;
3201                         goto end_unlock_session;
3202                 }
3203         }
3204         if (close_command.is_set &&
3205                         close_command.value == LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED) {
3206                 session->has_rotated = true;
3207         }
3208         DBG("Reply chunk path on close: %s", closed_trace_chunk_path);
3209         path_length = strlen(closed_trace_chunk_path) + 1;
3210         if (path_length > UINT32_MAX) {
3211                 ERR("Closed trace chunk path exceeds the maximal length allowed by the protocol");
3212                 ret = -1;
3213                 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
3214                 goto end_unlock_session;
3215         }
3216
3217         if (session->current_trace_chunk == chunk) {
3218                 /*
3219                  * After a trace chunk close command, no new streams
3220                  * referencing the chunk may be created. Hence, on the
3221                  * event that no new trace chunk have been created for
3222                  * the session, the reference to the current trace chunk
3223                  * is released in order to allow it to be reclaimed when
3224                  * the last stream releases its reference to it.
3225                  */
3226                 lttng_trace_chunk_put(session->current_trace_chunk);
3227                 session->current_trace_chunk = NULL;
3228         }
3229         lttng_trace_chunk_put(session->pending_closure_trace_chunk);
3230         session->pending_closure_trace_chunk = NULL;
3231 end_unlock_session:
3232         pthread_mutex_unlock(&session->lock);
3233
3234 end:
3235         reply.generic.ret_code = htobe32((uint32_t) reply_code);
3236         reply.path_length = htobe32((uint32_t) path_length);
3237         buf_ret = lttng_dynamic_buffer_append(
3238                         &reply_payload, &reply, sizeof(reply));
3239         if (buf_ret) {
3240                 ERR("Failed to append \"close trace chunk\" command reply header to payload buffer");
3241                 goto end_no_reply;
3242         }
3243
3244         if (reply_code == LTTNG_OK) {
3245                 buf_ret = lttng_dynamic_buffer_append(&reply_payload,
3246                                 closed_trace_chunk_path, path_length);
3247                 if (buf_ret) {
3248                         ERR("Failed to append \"close trace chunk\" command reply path to payload buffer");
3249                         goto end_no_reply;
3250                 }
3251         }
3252
3253         send_ret = conn->sock->ops->sendmsg(conn->sock,
3254                         reply_payload.data,
3255                         reply_payload.size,
3256                         0);
3257         if (send_ret < reply_payload.size) {
3258                 ERR("Failed to send \"close trace chunk\" command reply of %zu bytes (ret = %zd)",
3259                                 reply_payload.size, send_ret);
3260                 ret = -1;
3261                 goto end_no_reply;
3262         }
3263 end_no_reply:
3264         lttng_trace_chunk_put(chunk);
3265         lttng_dynamic_buffer_reset(&reply_payload);
3266         return ret;
3267 }
3268
3269 /*
3270  * relay_trace_chunk_exists: check if a trace chunk exists
3271  */
3272 static int relay_trace_chunk_exists(
3273                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
3274                 struct relay_connection *conn,
3275                 const struct lttng_buffer_view *payload)
3276 {
3277         int ret = 0;
3278         ssize_t send_ret;
3279         struct relay_session *session = conn->session;
3280         struct lttcomm_relayd_trace_chunk_exists *msg;
3281         struct lttcomm_relayd_trace_chunk_exists_reply reply = {};
3282         struct lttng_buffer_view header_view;
3283         uint64_t chunk_id;
3284         bool chunk_exists;
3285
3286         if (!session || !conn->version_check_done) {
3287                 ERR("Trying to check for the presence of a trace chunk before version check");
3288                 ret = -1;
3289                 goto end_no_reply;
3290         }
3291
3292         if (session->major == 2 && session->minor < 11) {
3293                 ERR("Chunk exists command is unsupported before 2.11");
3294                 ret = -1;
3295                 goto end_no_reply;
3296         }
3297
3298         header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
3299         if (!lttng_buffer_view_is_valid(&header_view)) {
3300                 ERR("Failed to receive payload of chunk exists command");
3301                 ret = -1;
3302                 goto end_no_reply;
3303         }
3304
3305         /* Convert to host endianness. */
3306         msg = (typeof(msg)) header_view.data;
3307         chunk_id = be64toh(msg->chunk_id);
3308
3309         ret = sessiond_trace_chunk_registry_chunk_exists(
3310                         sessiond_trace_chunk_registry,
3311                         conn->session->sessiond_uuid,
3312                         conn->session->id,
3313                         chunk_id, &chunk_exists);
3314         /*
3315          * If ret is not 0, send the reply and report the error to the caller.
3316          * It is a protocol (or internal) error and the session/connection
3317          * should be torn down.
3318          */
3319         reply.generic.ret_code = htobe32((uint32_t) (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL));
3320         reply.trace_chunk_exists = ret == 0 ? chunk_exists : 0;
3321
3322         send_ret = conn->sock->ops->sendmsg(
3323                         conn->sock, &reply, sizeof(reply), 0);
3324         if (send_ret < (ssize_t) sizeof(reply)) {
3325                 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
3326                                 send_ret);
3327                 ret = -1;
3328         }
3329 end_no_reply:
3330         return ret;
3331 }
3332
3333 /*
3334  * relay_get_configuration: query whether feature is available
3335  */
3336 static int relay_get_configuration(
3337                 const struct lttcomm_relayd_hdr *recv_hdr __attribute__((unused)),
3338                 struct relay_connection *conn,
3339                 const struct lttng_buffer_view *payload)
3340 {
3341         int ret = 0;
3342         ssize_t send_ret;
3343         struct lttcomm_relayd_get_configuration *msg;
3344         struct lttcomm_relayd_get_configuration_reply reply = {};
3345         struct lttng_buffer_view header_view;
3346         struct lttng_dynamic_buffer buffer;
3347         struct lttng_dynamic_buffer extra_payload;
3348         uint64_t query_flags = 0;
3349         uint64_t result_flags = 0;
3350
3351         lttng_dynamic_buffer_init(&buffer);
3352         lttng_dynamic_buffer_init(&extra_payload);
3353
3354         header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
3355         if (!lttng_buffer_view_is_valid(&header_view)) {
3356                 ERR("Failed to receive payload of chunk close command");
3357                 ret = -1;
3358                 goto end_no_reply;
3359         }
3360
3361         /* Convert to host endianness. */
3362         msg = (typeof(msg)) header_view.data;
3363         query_flags = be64toh(msg->query_flags);
3364
3365         /* FIXME: Do we want to return invalid protocol or do we want to return
3366          * an array of lttcomm_relayd_get_configuration_specialized_query_reply
3367          * for each unknown query bit ?
3368          */
3369         if (query_flags & ~LTTCOMM_RELAYD_CONFIGURATION_QUERY_FLAG_MASK) {
3370                 ret = LTTNG_ERR_INVALID_PROTOCOL;
3371                 goto reply;
3372         }
3373
3374         if (opt_allow_clear) {
3375                 result_flags |= LTTCOMM_RELAYD_CONFIGURATION_FLAG_CLEAR_ALLOWED;
3376         }
3377
3378         if (query_flags & LTTCOMM_RELAYD_CONFIGURATION_QUERY_FLAG_SUPPORTED_TRACE_FORMAT) {
3379                 uint64_t supported_trace_format = 0;
3380
3381                 supported_trace_format |= LTTCOMM_RELAYD_CONFIGURATION_TRACE_FORMAT_SUPPORTED_CTF1;
3382
3383                 if (opt_allow_ctf2) {
3384                         supported_trace_format |=
3385                                         LTTCOMM_RELAYD_CONFIGURATION_TRACE_FORMAT_SUPPORTED_CTF2;
3386                 }
3387
3388                 supported_trace_format = htobe64(supported_trace_format);
3389
3390                 lttcomm_relayd_get_configuration_specialized_query_reply s_reply = {};
3391                 s_reply.query_flag = htobe64(
3392                                 LTTCOMM_RELAYD_CONFIGURATION_QUERY_FLAG_SUPPORTED_TRACE_FORMAT);
3393                 s_reply.payload_len = htobe64((uint64_t) sizeof(supported_trace_format));
3394                 lttng_dynamic_buffer_append(&extra_payload, &s_reply, sizeof(s_reply));
3395                 lttng_dynamic_buffer_append(&extra_payload, &supported_trace_format,
3396                                 sizeof(supported_trace_format));
3397         }
3398
3399 reply:
3400         reply.generic.ret_code =
3401                         htobe32((uint32_t) (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL));
3402         reply.relayd_configuration_flags = htobe64(result_flags);
3403
3404         lttng_dynamic_buffer_append(&buffer, &reply, sizeof(reply));
3405         lttng_dynamic_buffer_append_buffer(&buffer, &extra_payload);
3406
3407         send_ret = conn->sock->ops->sendmsg(conn->sock, buffer.data, buffer.size, 0);
3408         if (send_ret < (ssize_t) buffer.size) {
3409                 ERR("Failed to send \"get configuration\" command reply (ret = %zd)", send_ret);
3410                 ret = -1;
3411         }
3412 end_no_reply:
3413         lttng_dynamic_buffer_reset(&extra_payload);
3414         lttng_dynamic_buffer_reset(&buffer);
3415         return ret;
3416 }
3417
3418 static int relay_process_control_command(struct relay_connection *conn,
3419                 const struct lttcomm_relayd_hdr *header,
3420                 const struct lttng_buffer_view *payload)
3421 {
3422         int ret = 0;
3423
3424         DBG3("Processing \"%s\" command for socket %i",
3425                         lttcomm_relayd_command_str((lttcomm_relayd_command) header->cmd),
3426                         conn->sock->fd);
3427         switch (header->cmd) {
3428         case RELAYD_CREATE_SESSION:
3429                 ret = relay_create_session(header, conn, payload);
3430                 break;
3431         case RELAYD_ADD_STREAM:
3432                 ret = relay_add_stream(header, conn, payload);
3433                 break;
3434         case RELAYD_START_DATA:
3435                 ret = relay_start(header, conn, payload);
3436                 break;
3437         case RELAYD_SEND_METADATA:
3438                 ret = relay_recv_metadata(header, conn, payload);
3439                 break;
3440         case RELAYD_VERSION:
3441                 ret = relay_send_version(header, conn, payload);
3442                 break;
3443         case RELAYD_CLOSE_STREAM:
3444                 ret = relay_close_stream(header, conn, payload);
3445                 break;
3446         case RELAYD_DATA_PENDING:
3447                 ret = relay_data_pending(header, conn, payload);
3448                 break;
3449         case RELAYD_QUIESCENT_CONTROL:
3450                 ret = relay_quiescent_control(header, conn, payload);
3451                 break;
3452         case RELAYD_BEGIN_DATA_PENDING:
3453                 ret = relay_begin_data_pending(header, conn, payload);
3454                 break;
3455         case RELAYD_END_DATA_PENDING:
3456                 ret = relay_end_data_pending(header, conn, payload);
3457                 break;
3458         case RELAYD_SEND_INDEX:
3459                 ret = relay_recv_index(header, conn, payload);
3460                 break;
3461         case RELAYD_STREAMS_SENT:
3462                 ret = relay_streams_sent(header, conn, payload);
3463                 break;
3464         case RELAYD_RESET_METADATA:
3465                 ret = relay_reset_metadata(header, conn, payload);
3466                 break;
3467         case RELAYD_ROTATE_STREAMS:
3468                 ret = relay_rotate_session_streams(header, conn, payload);
3469                 break;
3470         case RELAYD_CREATE_TRACE_CHUNK:
3471                 ret = relay_create_trace_chunk(header, conn, payload);
3472                 break;
3473         case RELAYD_CLOSE_TRACE_CHUNK:
3474                 ret = relay_close_trace_chunk(header, conn, payload);
3475                 break;
3476         case RELAYD_TRACE_CHUNK_EXISTS:
3477                 ret = relay_trace_chunk_exists(header, conn, payload);
3478                 break;
3479         case RELAYD_GET_CONFIGURATION:
3480                 ret = relay_get_configuration(header, conn, payload);
3481                 break;
3482         case RELAYD_UPDATE_SYNC_INFO:
3483         default:
3484                 ERR("Received unknown command (%u)", header->cmd);
3485                 relay_unknown_command(conn);
3486                 ret = -1;
3487                 goto end;
3488         }
3489
3490 end:
3491         return ret;
3492 }
3493
3494 static enum relay_connection_status relay_process_control_receive_payload(
3495                 struct relay_connection *conn)
3496 {
3497         int ret = 0;
3498         enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3499         struct lttng_dynamic_buffer *reception_buffer =
3500                         &conn->protocol.ctrl.reception_buffer;
3501         struct ctrl_connection_state_receive_payload *state =
3502                         &conn->protocol.ctrl.state.receive_payload;
3503         struct lttng_buffer_view payload_view;
3504
3505         if (state->left_to_receive == 0) {
3506                 /* Short-circuit for payload-less commands. */
3507                 goto reception_complete;
3508         }
3509
3510         ret = conn->sock->ops->recvmsg(conn->sock,
3511                         reception_buffer->data + state->received,
3512                         state->left_to_receive, MSG_DONTWAIT);
3513         if (ret < 0) {
3514                 DIAGNOSTIC_PUSH
3515                 DIAGNOSTIC_IGNORE_LOGICAL_OP
3516                 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3517                 DIAGNOSTIC_POP
3518                         PERROR("Unable to receive command payload on sock %d",
3519                                         conn->sock->fd);
3520                         status = RELAY_CONNECTION_STATUS_ERROR;
3521                 }
3522                 goto end;
3523         } else if (ret == 0) {
3524                 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3525                 status = RELAY_CONNECTION_STATUS_CLOSED;
3526                 goto end;
3527         }
3528
3529         LTTNG_ASSERT(ret > 0);
3530         LTTNG_ASSERT(ret <= state->left_to_receive);
3531
3532         state->left_to_receive -= ret;
3533         state->received += ret;
3534
3535         if (state->left_to_receive > 0) {
3536                 /*
3537                  * Can't transition to the protocol's next state, wait to
3538                  * receive the rest of the header.
3539                  */
3540                 DBG3("Partial reception of control connection protocol payload (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
3541                                 state->received, state->left_to_receive,
3542                                 conn->sock->fd);
3543                 goto end;
3544         }
3545
3546 reception_complete:
3547         DBG("Done receiving control command payload: fd = %i, payload size = %" PRIu64 " bytes",
3548                         conn->sock->fd, state->received);
3549         /*
3550          * The payload required to process the command has been received.
3551          * A view to the reception buffer is forwarded to the various
3552          * commands and the state of the control is reset on success.
3553          *
3554          * Commands are responsible for sending their reply to the peer.
3555          */
3556         payload_view = lttng_buffer_view_from_dynamic_buffer(reception_buffer,
3557                         0, -1);
3558         ret = relay_process_control_command(conn,
3559                         &state->header, &payload_view);
3560         if (ret < 0) {
3561                 status = RELAY_CONNECTION_STATUS_ERROR;
3562                 goto end;
3563         }
3564
3565         ret = connection_reset_protocol_state(conn);
3566         if (ret) {
3567                 status = RELAY_CONNECTION_STATUS_ERROR;
3568         }
3569 end:
3570         return status;
3571 }
3572
3573 static enum relay_connection_status relay_process_control_receive_header(
3574                 struct relay_connection *conn)
3575 {
3576         int ret = 0;
3577         enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3578         struct lttcomm_relayd_hdr header;
3579         struct lttng_dynamic_buffer *reception_buffer =
3580                         &conn->protocol.ctrl.reception_buffer;
3581         struct ctrl_connection_state_receive_header *state =
3582                         &conn->protocol.ctrl.state.receive_header;
3583
3584         LTTNG_ASSERT(state->left_to_receive != 0);
3585
3586         ret = conn->sock->ops->recvmsg(conn->sock,
3587                         reception_buffer->data + state->received,
3588                         state->left_to_receive, MSG_DONTWAIT);
3589         if (ret < 0) {
3590                 DIAGNOSTIC_PUSH
3591                 DIAGNOSTIC_IGNORE_LOGICAL_OP
3592                 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3593                 DIAGNOSTIC_POP
3594                         PERROR("Unable to receive control command header on sock %d",
3595                                         conn->sock->fd);
3596                         status = RELAY_CONNECTION_STATUS_ERROR;
3597                 }
3598                 goto end;
3599         } else if (ret == 0) {
3600                 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3601                 status = RELAY_CONNECTION_STATUS_CLOSED;
3602                 goto end;
3603         }
3604
3605         LTTNG_ASSERT(ret > 0);
3606         LTTNG_ASSERT(ret <= state->left_to_receive);
3607
3608         state->left_to_receive -= ret;
3609         state->received += ret;
3610
3611         if (state->left_to_receive > 0) {
3612                 /*
3613                  * Can't transition to the protocol's next state, wait to
3614                  * receive the rest of the header.
3615                  */
3616                 DBG3("Partial reception of control connection protocol header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
3617                                 state->received, state->left_to_receive,
3618                                 conn->sock->fd);
3619                 goto end;
3620         }
3621
3622         /* Transition to next state: receiving the command's payload. */
3623         conn->protocol.ctrl.state_id =
3624                         CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD;
3625         memcpy(&header, reception_buffer->data, sizeof(header));
3626         header.circuit_id = be64toh(header.circuit_id);
3627         header.data_size = be64toh(header.data_size);
3628         header.cmd = be32toh(header.cmd);
3629         header.cmd_version = be32toh(header.cmd_version);
3630         memcpy(&conn->protocol.ctrl.state.receive_payload.header,
3631                         &header, sizeof(header));
3632
3633         DBG("Done receiving control command header: fd = %i, cmd = %s, cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes",
3634                         conn->sock->fd, lttcomm_relayd_command_str((enum lttcomm_relayd_command) header.cmd),
3635                         header.cmd_version, header.data_size);
3636
3637         if (header.data_size > DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE) {
3638                 ERR("Command header indicates a payload (%" PRIu64 " bytes) that exceeds the maximal payload size allowed on a control connection.",
3639                                 header.data_size);
3640                 status = RELAY_CONNECTION_STATUS_ERROR;
3641                 goto end;
3642         }
3643
3644         conn->protocol.ctrl.state.receive_payload.left_to_receive =
3645                         header.data_size;
3646         conn->protocol.ctrl.state.receive_payload.received = 0;
3647         ret = lttng_dynamic_buffer_set_size(reception_buffer,
3648                         header.data_size);
3649         if (ret) {
3650                 status = RELAY_CONNECTION_STATUS_ERROR;
3651                 goto end;
3652         }
3653
3654         if (header.data_size == 0) {
3655                 /*
3656                  * Manually invoke the next state as the poll loop
3657                  * will not wake-up to allow us to proceed further.
3658                  */
3659                 status = relay_process_control_receive_payload(conn);
3660         }
3661 end:
3662         return status;
3663 }
3664
3665 /*
3666  * Process the commands received on the control socket
3667  */
3668 static enum relay_connection_status relay_process_control(
3669                 struct relay_connection *conn)
3670 {
3671         enum relay_connection_status status;
3672
3673         switch (conn->protocol.ctrl.state_id) {
3674         case CTRL_CONNECTION_STATE_RECEIVE_HEADER:
3675                 status = relay_process_control_receive_header(conn);
3676                 break;
3677         case CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD:
3678                 status = relay_process_control_receive_payload(conn);
3679                 break;
3680         default:
3681                 ERR("Unknown control connection protocol state encountered.");
3682                 abort();
3683         }
3684
3685         return status;
3686 }
3687
3688 static enum relay_connection_status relay_process_data_receive_header(
3689                 struct relay_connection *conn)
3690 {
3691         int ret;
3692         enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3693         struct data_connection_state_receive_header *state =
3694                         &conn->protocol.data.state.receive_header;
3695         struct lttcomm_relayd_data_hdr header;
3696         struct relay_stream *stream;
3697
3698         LTTNG_ASSERT(state->left_to_receive != 0);
3699
3700         ret = conn->sock->ops->recvmsg(conn->sock,
3701                         state->header_reception_buffer + state->received,
3702                         state->left_to_receive, MSG_DONTWAIT);
3703         if (ret < 0) {
3704                 DIAGNOSTIC_PUSH
3705                 DIAGNOSTIC_IGNORE_LOGICAL_OP
3706                 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3707                 DIAGNOSTIC_POP
3708                         PERROR("Unable to receive data header on sock %d", conn->sock->fd);
3709                         status = RELAY_CONNECTION_STATUS_ERROR;
3710                 }
3711                 goto end;
3712         } else if (ret == 0) {
3713                 /* Orderly shutdown. Not necessary to print an error. */
3714                 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3715                 status = RELAY_CONNECTION_STATUS_CLOSED;
3716                 goto end;
3717         }
3718
3719         LTTNG_ASSERT(ret > 0);
3720         LTTNG_ASSERT(ret <= state->left_to_receive);
3721
3722         state->left_to_receive -= ret;
3723         state->received += ret;
3724
3725         if (state->left_to_receive > 0) {
3726                 /*
3727                  * Can't transition to the protocol's next state, wait to
3728                  * receive the rest of the header.
3729                  */
3730                 DBG3("Partial reception of data connection header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
3731                                 state->received, state->left_to_receive,
3732                                 conn->sock->fd);
3733                 goto end;
3734         }
3735
3736         /* Transition to next state: receiving the payload. */
3737         conn->protocol.data.state_id = DATA_CONNECTION_STATE_RECEIVE_PAYLOAD;
3738
3739         memcpy(&header, state->header_reception_buffer, sizeof(header));
3740         header.circuit_id = be64toh(header.circuit_id);
3741         header.stream_id = be64toh(header.stream_id);
3742         header.data_size = be32toh(header.data_size);
3743         header.net_seq_num = be64toh(header.net_seq_num);
3744         header.padding_size = be32toh(header.padding_size);
3745         memcpy(&conn->protocol.data.state.receive_payload.header, &header, sizeof(header));
3746
3747         conn->protocol.data.state.receive_payload.left_to_receive =
3748                         header.data_size;
3749         conn->protocol.data.state.receive_payload.received = 0;
3750         conn->protocol.data.state.receive_payload.rotate_index = false;
3751
3752         DBG("Received data connection header on fd %i: circuit_id = %" PRIu64 ", stream_id = %" PRIu64 ", data_size = %" PRIu32 ", net_seq_num = %" PRIu64 ", padding_size = %" PRIu32,
3753                         conn->sock->fd, header.circuit_id,
3754                         header.stream_id, header.data_size,
3755                         header.net_seq_num, header.padding_size);
3756
3757         stream = stream_get_by_id(header.stream_id);
3758         if (!stream) {
3759                 DBG("relay_process_data_receive_payload: Cannot find stream %" PRIu64,
3760                                 header.stream_id);
3761                 /* Protocol error. */
3762                 status = RELAY_CONNECTION_STATUS_ERROR;
3763                 goto end;
3764         }
3765
3766         pthread_mutex_lock(&stream->lock);
3767         /* Prepare stream for the reception of a new packet. */
3768         ret = stream_init_packet(stream, header.data_size,
3769                         &conn->protocol.data.state.receive_payload.rotate_index);
3770         pthread_mutex_unlock(&stream->lock);
3771         if (ret) {
3772                 ERR("Failed to rotate stream output file");
3773                 status = RELAY_CONNECTION_STATUS_ERROR;
3774                 goto end_stream_unlock;
3775         }
3776
3777 end_stream_unlock:
3778         stream_put(stream);
3779 end:
3780         return status;
3781 }
3782
3783 static enum relay_connection_status relay_process_data_receive_payload(
3784                 struct relay_connection *conn)
3785 {
3786         int ret;
3787         enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3788         struct relay_stream *stream;
3789         struct data_connection_state_receive_payload *state =
3790                         &conn->protocol.data.state.receive_payload;
3791         const size_t chunk_size = RECV_DATA_BUFFER_SIZE;
3792         char data_buffer[chunk_size];
3793         bool partial_recv = false;
3794         bool new_stream = false, close_requested = false, index_flushed = false;
3795         uint64_t left_to_receive = state->left_to_receive;
3796         struct relay_session *session;
3797
3798         DBG3("Receiving data for stream id %" PRIu64 " seqnum %" PRIu64 ", %" PRIu64" bytes received, %" PRIu64 " bytes left to receive",
3799                         state->header.stream_id, state->header.net_seq_num,
3800                         state->received, left_to_receive);
3801
3802         stream = stream_get_by_id(state->header.stream_id);
3803         if (!stream) {
3804                 /* Protocol error. */
3805                 ERR("relay_process_data_receive_payload: cannot find stream %" PRIu64,
3806                                 state->header.stream_id);
3807                 status = RELAY_CONNECTION_STATUS_ERROR;
3808                 goto end;
3809         }
3810
3811         pthread_mutex_lock(&stream->lock);
3812         session = stream->trace->session;
3813         if (!conn->session) {
3814                 ret = connection_set_session(conn, session);
3815                 if (ret) {
3816                         status = RELAY_CONNECTION_STATUS_ERROR;
3817                         goto end_stream_unlock;
3818                 }
3819         }
3820
3821         /*
3822          * The size of the "chunk" received on any iteration is bounded by:
3823          *   - the data left to receive,
3824          *   - the data immediately available on the socket,
3825          *   - the on-stack data buffer
3826          */
3827         while (left_to_receive > 0 && !partial_recv) {
3828                 size_t recv_size = std::min<uint64_t>(left_to_receive, chunk_size);
3829                 struct lttng_buffer_view packet_chunk;
3830
3831                 ret = conn->sock->ops->recvmsg(conn->sock, data_buffer,
3832                                 recv_size, MSG_DONTWAIT);
3833                 if (ret < 0) {
3834                         DIAGNOSTIC_PUSH
3835                         DIAGNOSTIC_IGNORE_LOGICAL_OP
3836                         if (errno != EAGAIN && errno != EWOULDBLOCK) {
3837                         DIAGNOSTIC_POP
3838                                 PERROR("Socket %d error", conn->sock->fd);
3839                                 status = RELAY_CONNECTION_STATUS_ERROR;
3840                         }
3841                         goto end_stream_unlock;
3842                 } else if (ret == 0) {
3843                         /* No more data ready to be consumed on socket. */
3844                         DBG3("No more data ready for consumption on data socket of stream id %" PRIu64,
3845                                         state->header.stream_id);
3846                         status = RELAY_CONNECTION_STATUS_CLOSED;
3847                         break;
3848                 } else if (ret < (int) recv_size) {
3849                         /*
3850                          * All the data available on the socket has been
3851                          * consumed.
3852                          */
3853                         partial_recv = true;
3854                         recv_size = ret;
3855                 }
3856
3857                 packet_chunk = lttng_buffer_view_init(data_buffer,
3858                                 0, recv_size);
3859                 LTTNG_ASSERT(packet_chunk.data);
3860
3861                 ret = stream_write(stream, &packet_chunk, 0);
3862                 if (ret) {
3863                         ERR("Relay error writing data to file");
3864                         status = RELAY_CONNECTION_STATUS_ERROR;
3865                         goto end_stream_unlock;
3866                 }
3867
3868                 left_to_receive -= recv_size;
3869                 state->received += recv_size;
3870                 state->left_to_receive = left_to_receive;
3871         }
3872
3873         if (state->left_to_receive > 0) {
3874                 /*
3875                  * Did not receive all the data expected, wait for more data to
3876                  * become available on the socket.
3877                  */
3878                 DBG3("Partial receive on data connection of stream id %" PRIu64 ", %" PRIu64 " bytes received, %" PRIu64 " bytes left to receive",
3879                                 state->header.stream_id, state->received,
3880                                 state->left_to_receive);
3881                 goto end_stream_unlock;
3882         }
3883
3884         ret = stream_write(stream, NULL, state->header.padding_size);
3885         if (ret) {
3886                 status = RELAY_CONNECTION_STATUS_ERROR;
3887                 goto end_stream_unlock;
3888         }
3889
3890         if (session_streams_have_index(session)) {
3891                 ret = stream_update_index(stream, state->header.net_seq_num,
3892                                 state->rotate_index, &index_flushed,
3893                                 state->header.data_size + state->header.padding_size);
3894                 if (ret < 0) {
3895                         ERR("Failed to update index: stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d",
3896                                         stream->stream_handle,
3897                                         state->header.net_seq_num, ret);
3898                         status = RELAY_CONNECTION_STATUS_ERROR;
3899                         goto end_stream_unlock;
3900                 }
3901         }
3902
3903         if (stream->prev_data_seq == -1ULL) {
3904                 new_stream = true;
3905         }
3906
3907         ret = stream_complete_packet(stream, state->header.data_size +
3908                         state->header.padding_size, state->header.net_seq_num,
3909                         index_flushed);
3910         if (ret) {
3911                 status = RELAY_CONNECTION_STATUS_ERROR;
3912                 goto end_stream_unlock;
3913         }
3914
3915         /*
3916          * Resetting the protocol state (to RECEIVE_HEADER) will trash the
3917          * contents of *state which are aliased (union) to the same location as
3918          * the new state. Don't use it beyond this point.
3919          */
3920         connection_reset_protocol_state(conn);
3921         state = NULL;
3922
3923 end_stream_unlock:
3924         close_requested = stream->close_requested;
3925         pthread_mutex_unlock(&stream->lock);
3926         if (close_requested && left_to_receive == 0) {
3927                 try_stream_close(stream);
3928         }
3929
3930         if (new_stream) {
3931                 pthread_mutex_lock(&session->lock);
3932                 uatomic_set(&session->new_streams, 1);
3933                 pthread_mutex_unlock(&session->lock);
3934         }
3935
3936         stream_put(stream);
3937 end:
3938         return status;
3939 }
3940
3941 /*
3942  * relay_process_data: Process the data received on the data socket
3943  */
3944 static enum relay_connection_status relay_process_data(
3945                 struct relay_connection *conn)
3946 {
3947         enum relay_connection_status status;
3948
3949         switch (conn->protocol.data.state_id) {
3950         case DATA_CONNECTION_STATE_RECEIVE_HEADER:
3951                 status = relay_process_data_receive_header(conn);
3952                 break;
3953         case DATA_CONNECTION_STATE_RECEIVE_PAYLOAD:
3954                 status = relay_process_data_receive_payload(conn);
3955                 break;
3956         default:
3957                 ERR("Unexpected data connection communication state.");
3958                 abort();
3959         }
3960
3961         return status;
3962 }
3963
3964 static void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd)
3965 {
3966         int ret;
3967
3968         (void) lttng_poll_del(events, pollfd);
3969
3970         ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, &pollfd, 1,
3971                         fd_tracker_util_close_fd, NULL);
3972         if (ret < 0) {
3973                 ERR("Closing pollfd %d", pollfd);
3974         }
3975 }
3976
3977 static void relay_thread_close_connection(struct lttng_poll_event *events,
3978                 int pollfd, struct relay_connection *conn)
3979 {
3980         const char *type_str;
3981
3982         switch (conn->type) {
3983         case RELAY_DATA:
3984                 type_str = "Data";
3985                 break;
3986         case RELAY_CONTROL:
3987                 type_str = "Control";
3988                 break;
3989         case RELAY_VIEWER_COMMAND:
3990                 type_str = "Viewer Command";
3991                 break;
3992         case RELAY_VIEWER_NOTIFICATION:
3993                 type_str = "Viewer Notification";
3994                 break;
3995         default:
3996                 type_str = "Unknown";
3997         }
3998         cleanup_connection_pollfd(events, pollfd);
3999         connection_put(conn);
4000         DBG("%s connection closed with %d", type_str, pollfd);
4001 }
4002
4003 /*
4004  * This thread does the actual work
4005  */
4006 static void *relay_thread_worker(void *data __attribute__((unused)))
4007 {
4008         int ret, err = -1, last_seen_data_fd = -1;
4009         uint32_t nb_fd;
4010         struct lttng_poll_event events;
4011         struct lttng_ht *relay_connections_ht;
4012         struct lttng_ht_iter iter;
4013         struct relay_connection *destroy_conn = NULL;
4014
4015         DBG("[thread] Relay worker started");
4016
4017         rcu_register_thread();
4018
4019         health_register(health_relayd, HEALTH_RELAYD_TYPE_WORKER);
4020
4021         if (testpoint(relayd_thread_worker)) {
4022                 goto error_testpoint;
4023         }
4024
4025         health_code_update();
4026
4027         /* table of connections indexed on socket */
4028         relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
4029         if (!relay_connections_ht) {
4030                 goto relay_connections_ht_error;
4031         }
4032
4033         ret = create_named_thread_poll_set(&events, 2, "Worker thread epoll");
4034         if (ret < 0) {
4035                 goto error_poll_create;
4036         }
4037
4038         ret = lttng_poll_add(&events, relay_conn_pipe[0], LPOLLIN | LPOLLRDHUP);
4039         if (ret < 0) {
4040                 goto error;
4041         }
4042
4043 restart:
4044         while (1) {
4045                 int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
4046
4047                 health_code_update();
4048
4049                 /* Infinite blocking call, waiting for transmission */
4050                 DBG3("Relayd worker thread polling...");
4051                 health_poll_entry();
4052                 ret = lttng_poll_wait(&events, -1);
4053                 health_poll_exit();
4054                 if (ret < 0) {
4055                         /*
4056                          * Restart interrupted system call.
4057                          */
4058                         if (errno == EINTR) {
4059                                 goto restart;
4060                         }
4061                         goto error;
4062                 }
4063
4064                 nb_fd = ret;
4065
4066                 /*
4067                  * Process control. The control connection is
4068                  * prioritized so we don't starve it with high
4069                  * throughput tracing data on the data connection.
4070                  */
4071                 for (i = 0; i < nb_fd; i++) {
4072                         /* Fetch once the poll data */
4073                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
4074                         int pollfd = LTTNG_POLL_GETFD(&events, i);
4075
4076                         health_code_update();
4077
4078                         /* Thread quit pipe has been closed. Killing thread. */
4079                         ret = check_thread_quit_pipe(pollfd, revents);
4080                         if (ret) {
4081                                 err = 0;
4082                                 goto exit;
4083                         }
4084
4085                         /* Inspect the relay conn pipe for new connection */
4086                         if (pollfd == relay_conn_pipe[0]) {
4087                                 if (revents & LPOLLIN) {
4088                                         struct relay_connection *conn;
4089
4090                                         ret = lttng_read(relay_conn_pipe[0], &conn, sizeof(conn));
4091                                         if (ret < 0) {
4092                                                 goto error;
4093                                         }
4094                                         ret = lttng_poll_add(&events,
4095                                                         conn->sock->fd,
4096                                                         LPOLLIN | LPOLLRDHUP);
4097                                         if (ret) {
4098                                                 ERR("Failed to add new connection file descriptor to poll set");
4099                                                 goto error;
4100                                         }
4101                                         connection_ht_add(relay_connections_ht, conn);
4102                                         DBG("Connection socket %d added", conn->sock->fd);
4103                                 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
4104                                         ERR("Relay connection pipe error");
4105                                         goto error;
4106                                 } else {
4107                                         ERR("Unexpected poll events %u for sock %d", revents, pollfd);
4108                                         goto error;
4109                                 }
4110                         } else {
4111                                 struct relay_connection *ctrl_conn;
4112
4113                                 ctrl_conn = connection_get_by_sock(relay_connections_ht, pollfd);
4114                                 /* If not found, there is a synchronization issue. */
4115                                 LTTNG_ASSERT(ctrl_conn);
4116
4117                                 if (ctrl_conn->type == RELAY_DATA) {
4118                                         if (revents & LPOLLIN) {
4119                                                 /*
4120                                                  * Flag the last seen data fd not deleted. It will be
4121                                                  * used as the last seen fd if any fd gets deleted in
4122                                                  * this first loop.
4123                                                  */
4124                                                 last_notdel_data_fd = pollfd;
4125                                         }
4126                                         goto put_ctrl_connection;
4127                                 }
4128                                 LTTNG_ASSERT(ctrl_conn->type == RELAY_CONTROL);
4129
4130                                 if (revents & LPOLLIN) {
4131                                         enum relay_connection_status status;
4132
4133                                         status = relay_process_control(ctrl_conn);
4134                                         if (status != RELAY_CONNECTION_STATUS_OK) {
4135                                                 /*
4136                                                  * On socket error flag the session as aborted to force
4137                                                  * the cleanup of its stream otherwise it can leak
4138                                                  * during the lifetime of the relayd.
4139                                                  *
4140                                                  * This prevents situations in which streams can be
4141                                                  * left opened because an index was received, the
4142                                                  * control connection is closed, and the data
4143                                                  * connection is closed (uncleanly) before the packet's
4144                                                  * data provided.
4145                                                  *
4146                                                  * Since the control connection encountered an error,
4147                                                  * it is okay to be conservative and close the
4148                                                  * session right now as we can't rely on the protocol
4149                                                  * being respected anymore.
4150                                                  */
4151                                                 if (status == RELAY_CONNECTION_STATUS_ERROR) {
4152                                                         session_abort(ctrl_conn->session);
4153                                                 }
4154
4155                                                 /* Clear the connection on error or close. */
4156                                                 relay_thread_close_connection(&events,
4157                                                                 pollfd,
4158                                                                 ctrl_conn);
4159                                         }
4160                                         seen_control = 1;
4161                                 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
4162                                         relay_thread_close_connection(&events,
4163                                                         pollfd, ctrl_conn);
4164                                         if (last_seen_data_fd == pollfd) {
4165                                                 last_seen_data_fd = last_notdel_data_fd;
4166                                         }
4167                                 } else {
4168                                         ERR("Unexpected poll events %u for control sock %d",
4169                                                         revents, pollfd);
4170                                         connection_put(ctrl_conn);
4171                                         goto error;
4172                                 }
4173                         put_ctrl_connection:
4174                                 connection_put(ctrl_conn);
4175                         }
4176                 }
4177
4178                 /*
4179                  * The last loop handled a control request, go back to poll to make
4180                  * sure we prioritise the control socket.
4181                  */
4182                 if (seen_control) {
4183                         continue;
4184                 }
4185
4186                 if (last_seen_data_fd >= 0) {
4187                         for (i = 0; i < nb_fd; i++) {
4188                                 int pollfd = LTTNG_POLL_GETFD(&events, i);
4189
4190                                 health_code_update();
4191
4192                                 if (last_seen_data_fd == pollfd) {
4193                                         idx = i;
4194                                         break;
4195                                 }
4196                         }
4197                 }
4198
4199                 /* Process data connection. */
4200                 for (i = idx + 1; i < nb_fd; i++) {
4201                         /* Fetch the poll data. */
4202                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
4203                         int pollfd = LTTNG_POLL_GETFD(&events, i);
4204                         struct relay_connection *data_conn;
4205
4206                         health_code_update();
4207
4208                         if (!revents) {
4209                                 /* No activity for this FD (poll implementation). */
4210                                 continue;
4211                         }
4212
4213                         /* Skip the command pipe. It's handled in the first loop. */
4214                         if (pollfd == relay_conn_pipe[0]) {
4215                                 continue;
4216                         }
4217
4218                         data_conn = connection_get_by_sock(relay_connections_ht, pollfd);
4219                         if (!data_conn) {
4220                                 /* Skip it. Might be removed before. */
4221                                 continue;
4222                         }
4223                         if (data_conn->type == RELAY_CONTROL) {
4224                                 goto put_data_connection;
4225                         }
4226                         LTTNG_ASSERT(data_conn->type == RELAY_DATA);
4227
4228                         if (revents & LPOLLIN) {
4229                                 enum relay_connection_status status;
4230
4231                                 status = relay_process_data(data_conn);
4232                                 /* Connection closed or error. */
4233                                 if (status != RELAY_CONNECTION_STATUS_OK) {
4234                                         /*
4235                                          * On socket error flag the session as aborted to force
4236                                          * the cleanup of its stream otherwise it can leak
4237                                          * during the lifetime of the relayd.
4238                                          *
4239                                          * This prevents situations in which streams can be
4240                                          * left opened because an index was received, the
4241                                          * control connection is closed, and the data
4242                                          * connection is closed (uncleanly) before the packet's
4243                                          * data provided.
4244                                          *
4245                                          * Since the data connection encountered an error,
4246                                          * it is okay to be conservative and close the
4247                                          * session right now as we can't rely on the protocol
4248                                          * being respected anymore.
4249                                          */
4250                                         if (status == RELAY_CONNECTION_STATUS_ERROR) {
4251                                                 session_abort(data_conn->session);
4252                                         }
4253                                         relay_thread_close_connection(&events, pollfd,
4254                                                         data_conn);
4255                                         /*
4256                                          * Every goto restart call sets the last seen fd where
4257                                          * here we don't really care since we gracefully
4258                                          * continue the loop after the connection is deleted.
4259                                          */
4260                                 } else {
4261                                         /* Keep last seen port. */
4262                                         last_seen_data_fd = pollfd;
4263                                         connection_put(data_conn);
4264                                         goto restart;
4265                                 }
4266                         } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
4267                                 relay_thread_close_connection(&events, pollfd,
4268                                                 data_conn);
4269                         } else {
4270                                 ERR("Unknown poll events %u for data sock %d",
4271                                                 revents, pollfd);
4272                         }
4273                 put_data_connection:
4274                         connection_put(data_conn);
4275                 }
4276                 last_seen_data_fd = -1;
4277         }
4278
4279         /* Normal exit, no error */
4280         ret = 0;
4281
4282 exit:
4283 error:
4284         /* Cleanup remaining connection object. */
4285         rcu_read_lock();
4286         cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter,
4287                         destroy_conn,
4288                         sock_n.node) {
4289                 health_code_update();
4290
4291                 session_abort(destroy_conn->session);
4292
4293                 /*
4294                  * No need to grab another ref, because we own
4295                  * destroy_conn.
4296                  */
4297                 relay_thread_close_connection(&events, destroy_conn->sock->fd,
4298                                 destroy_conn);
4299         }
4300         rcu_read_unlock();
4301
4302         (void) fd_tracker_util_poll_clean(the_fd_tracker, &events);
4303 error_poll_create:
4304         lttng_ht_destroy(relay_connections_ht);
4305 relay_connections_ht_error:
4306         /* Close relay conn pipes */
4307         (void) fd_tracker_util_pipe_close(the_fd_tracker,
4308                         relay_conn_pipe);
4309         if (err) {
4310                 DBG("Thread exited with error");
4311         }
4312         DBG("Worker thread cleanup complete");
4313 error_testpoint:
4314         if (err) {
4315                 health_error();
4316                 ERR("Health error occurred in %s", __func__);
4317         }
4318         health_unregister(health_relayd);
4319         rcu_unregister_thread();
4320         lttng_relay_stop_threads();
4321         return NULL;
4322 }
4323
4324 /*
4325  * Create the relay command pipe to wake thread_manage_apps.
4326  * Closed in cleanup().
4327  */
4328 static int create_relay_conn_pipe(void)
4329 {
4330         return fd_tracker_util_pipe_open_cloexec(the_fd_tracker,
4331                         "Relayd connection pipe", relay_conn_pipe);
4332 }
4333
4334 static int stdio_open(void *data __attribute__((unused)), int *fds)
4335 {
4336         fds[0] = fileno(stdout);
4337         fds[1] = fileno(stderr);
4338         return 0;
4339 }
4340
4341 static int track_stdio(void)
4342 {
4343         int fds[2];
4344         const char *names[] = { "stdout", "stderr" };
4345
4346         return fd_tracker_open_unsuspendable_fd(the_fd_tracker, fds,
4347                         names, 2, stdio_open, NULL);
4348 }
4349
4350 /*
4351  * main
4352  */
4353 int main(int argc, char **argv)
4354 {
4355         bool thread_is_rcu_registered = false;
4356         int ret = 0, retval = 0;
4357         void *status;
4358         char *unlinked_file_directory_path = NULL, *output_path = NULL;
4359
4360         /* Parse environment variables */
4361         ret = parse_env_options();
4362         if (ret) {
4363                 retval = -1;
4364                 goto exit_options;
4365         }
4366
4367         /*
4368          * Parse arguments.
4369          * Command line arguments overwrite environment.
4370          */
4371         progname = argv[0];
4372         if (set_options(argc, argv)) {
4373                 retval = -1;
4374                 goto exit_options;
4375         }
4376
4377         if (set_signal_handler()) {
4378                 retval = -1;
4379                 goto exit_options;
4380         }
4381
4382         relayd_config_log();
4383
4384         if (opt_print_version) {
4385                 print_version();
4386                 retval = 0;
4387                 goto exit_options;
4388         }
4389
4390         ret = fclose(stdin);
4391         if (ret) {
4392                 PERROR("Failed to close stdin");
4393                 goto exit_options;
4394         }
4395
4396         DBG("Clear command %s", opt_allow_clear ? "allowed" : "disallowed");
4397
4398         /* Try to create directory if -o, --output is specified. */
4399         if (opt_output_path) {
4400                 if (*opt_output_path != '/') {
4401                         ERR("Please specify an absolute path for -o, --output PATH");
4402                         retval = -1;
4403                         goto exit_options;
4404                 }
4405
4406                 ret = utils_mkdir_recursive(opt_output_path, S_IRWXU | S_IRWXG,
4407                                 -1, -1);
4408                 if (ret < 0) {
4409                         ERR("Unable to create %s", opt_output_path);
4410                         retval = -1;
4411                         goto exit_options;
4412                 }
4413         }
4414
4415         /* Daemonize */
4416         if (opt_daemon || opt_background) {
4417                 ret = lttng_daemonize(&child_ppid, &recv_child_signal,
4418                         !opt_background);
4419                 if (ret < 0) {
4420                         retval = -1;
4421                         goto exit_options;
4422                 }
4423         }
4424
4425         if (opt_working_directory) {
4426                 ret = utils_change_working_directory(opt_working_directory);
4427                 if (ret) {
4428                         /* All errors are already logged. */
4429                         goto exit_options;
4430                 }
4431         }
4432
4433         sessiond_trace_chunk_registry = sessiond_trace_chunk_registry_create();
4434         if (!sessiond_trace_chunk_registry) {
4435                 ERR("Failed to initialize session daemon trace chunk registry");
4436                 retval = -1;
4437                 goto exit_options;
4438         }
4439
4440         /*
4441          * The RCU thread registration (and use, through the fd-tracker's
4442          * creation) is done after the daemonization to allow us to not
4443          * deal with liburcu's fork() management as the call RCU needs to
4444          * be restored.
4445          */
4446         rcu_register_thread();
4447         thread_is_rcu_registered = true;
4448
4449         output_path = create_output_path("");
4450         if (!output_path) {
4451                 ERR("Failed to get output path");
4452                 retval = -1;
4453                 goto exit_options;
4454         }
4455         ret = asprintf(&unlinked_file_directory_path, "%s/%s", output_path,
4456                         DEFAULT_UNLINKED_FILES_DIRECTORY);
4457         free(output_path);
4458         if (ret < 0) {
4459                 ERR("Failed to format unlinked file directory path");
4460                 retval = -1;
4461                 goto exit_options;
4462         }
4463         the_fd_tracker = fd_tracker_create(
4464                         unlinked_file_directory_path, lttng_opt_fd_pool_size);
4465         free(unlinked_file_directory_path);
4466         if (!the_fd_tracker) {
4467                 retval = -1;
4468                 goto exit_options;
4469         }
4470
4471         ret = track_stdio();
4472         if (ret) {
4473                 retval = -1;
4474                 goto exit_options;
4475         }
4476
4477         /* Initialize thread health monitoring */
4478         health_relayd = health_app_create(NR_HEALTH_RELAYD_TYPES);
4479         if (!health_relayd) {
4480                 PERROR("health_app_create error");
4481                 retval = -1;
4482                 goto exit_options;
4483         }
4484
4485         /* Create thread quit pipe */
4486         if (init_thread_quit_pipe()) {
4487                 retval = -1;
4488                 goto exit_options;
4489         }
4490
4491         /* Setup the thread apps communication pipe. */
4492         if (create_relay_conn_pipe()) {
4493                 retval = -1;
4494                 goto exit_options;
4495         }
4496
4497         /* Init relay command queue. */
4498         cds_wfcq_init(&relay_conn_queue.head, &relay_conn_queue.tail);
4499
4500         /* Initialize communication library */
4501         lttcomm_init();
4502         lttcomm_inet_init();
4503
4504         /* tables of sessions indexed by session ID */
4505         sessions_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
4506         if (!sessions_ht) {
4507                 retval = -1;
4508                 goto exit_options;
4509         }
4510
4511         /* tables of streams indexed by stream ID */
4512         relay_streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
4513         if (!relay_streams_ht) {
4514                 retval = -1;
4515                 goto exit_options;
4516         }
4517
4518         /* tables of streams indexed by stream ID */
4519         viewer_streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
4520         if (!viewer_streams_ht) {
4521                 retval = -1;
4522                 goto exit_options;
4523         }
4524
4525         ret = init_health_quit_pipe();
4526         if (ret) {
4527                 retval = -1;
4528                 goto exit_options;
4529         }
4530
4531         /* Create thread to manage the client socket */
4532         ret = pthread_create(&health_thread, default_pthread_attr(),
4533                         thread_manage_health_relayd, (void *) NULL);
4534         if (ret) {
4535                 errno = ret;
4536                 PERROR("pthread_create health");
4537                 retval = -1;
4538                 goto exit_options;
4539         }
4540
4541         /* Setup the dispatcher thread */
4542         ret = pthread_create(&dispatcher_thread, default_pthread_attr(),
4543                         relay_thread_dispatcher, (void *) NULL);
4544         if (ret) {
4545                 errno = ret;
4546                 PERROR("pthread_create dispatcher");
4547                 retval = -1;
4548                 goto exit_dispatcher_thread;
4549         }
4550
4551         /* Setup the worker thread */
4552         ret = pthread_create(&worker_thread, default_pthread_attr(),
4553                         relay_thread_worker, NULL);
4554         if (ret) {
4555                 errno = ret;
4556                 PERROR("pthread_create worker");
4557                 retval = -1;
4558                 goto exit_worker_thread;
4559         }
4560
4561         /* Setup the listener thread */
4562         ret = pthread_create(&listener_thread, default_pthread_attr(),
4563                         relay_thread_listener, (void *) NULL);
4564         if (ret) {
4565                 errno = ret;
4566                 PERROR("pthread_create listener");
4567                 retval = -1;
4568                 goto exit_listener_thread;
4569         }
4570
4571         ret = relayd_live_create(live_uri);
4572         if (ret) {
4573                 ERR("Starting live viewer threads");
4574                 retval = -1;
4575                 goto exit_live;
4576         }
4577
4578         /*
4579          * This is where we start awaiting program completion (e.g. through
4580          * signal that asks threads to teardown).
4581          */
4582
4583         ret = relayd_live_join();
4584         if (ret) {
4585                 retval = -1;
4586         }
4587 exit_live:
4588
4589         ret = pthread_join(listener_thread, &status);
4590         if (ret) {
4591                 errno = ret;
4592                 PERROR("pthread_join listener_thread");
4593                 retval = -1;
4594         }
4595
4596 exit_listener_thread:
4597         ret = pthread_join(worker_thread, &status);
4598         if (ret) {
4599                 errno = ret;
4600                 PERROR("pthread_join worker_thread");
4601                 retval = -1;
4602         }
4603
4604 exit_worker_thread:
4605         ret = pthread_join(dispatcher_thread, &status);
4606         if (ret) {
4607                 errno = ret;
4608                 PERROR("pthread_join dispatcher_thread");
4609                 retval = -1;
4610         }
4611 exit_dispatcher_thread:
4612
4613         ret = pthread_join(health_thread, &status);
4614         if (ret) {
4615                 errno = ret;
4616                 PERROR("pthread_join health_thread");
4617                 retval = -1;
4618         }
4619 exit_options:
4620         /*
4621          * Wait for all pending call_rcu work to complete before tearing
4622          * down data structures. call_rcu worker may be trying to
4623          * perform lookups in those structures.
4624          */
4625         rcu_barrier();
4626         relayd_cleanup();
4627
4628         /* Ensure all prior call_rcu are done. */
4629         rcu_barrier();
4630
4631         if (thread_is_rcu_registered) {
4632                 rcu_unregister_thread();
4633         }
4634
4635         if (!retval) {
4636                 exit(EXIT_SUCCESS);
4637         } else {
4638                 exit(EXIT_FAILURE);
4639         }
4640 }