src/bin/lttng-relayd/main.c

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