src/bin/lttng-relayd/main.c

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