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                 Control port listening (URI)\n");
 111         fprintf(stderr, "  -D, --data-port                    Data port listening (URI)\n");
 112         fprintf(stderr, "  -o, --output                       Output path for traces (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);
 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         ret = sscanf(VERSION, "%10u.%10u", &reply.major, &reply.minor);
1191         if (ret < 2) {
1192                 ERR("Error in scanning version");
1193                 ret = -1;
1194                 goto end;
1195         }
1196
1197         /* Major versions must be the same */
1198         if (reply.major != be32toh(msg.major)) {
1199                 DBG("Incompatible major versions (%u vs %u), deleting session",
1200                                 reply.major, be32toh(msg.major));
1201                 relay_delete_session(cmd, streams_ht);
1202                 ret = 0;
1203                 goto end;
1204         }
1205
1206         cmd->major = reply.major;
1207         /* We adapt to the lowest compatible version */
1208         if (reply.minor <= be32toh(msg.minor)) {
1209                 cmd->minor = reply.minor;
1210         } else {
1211                 cmd->minor = be32toh(msg.minor);
1212         }
1213
1214         reply.major = htobe32(reply.major);
1215         reply.minor = htobe32(reply.minor);
1216         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1217                         sizeof(struct lttcomm_relayd_version), 0);
1218         if (ret < 0) {
1219                 ERR("Relay sending version");
1220         }
1221
1222         DBG("Version check done using protocol %u.%u", cmd->major,
1223                         cmd->minor);
1224
1225 end:
1226         return ret;
1227 }
1228
1229 /*
1230  * Check for data pending for a given stream id from the session daemon.
1231  */
1232 static
1233 int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1234                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1235 {
1236         struct relay_session *session = cmd->session;
1237         struct lttcomm_relayd_data_pending msg;
1238         struct lttcomm_relayd_generic_reply reply;
1239         struct relay_stream *stream;
1240         int ret;
1241         uint64_t last_net_seq_num, stream_id;
1242
1243         DBG("Data pending command received");
1244
1245         if (!session || cmd->version_check_done == 0) {
1246                 ERR("Trying to check for data before version check");
1247                 ret = -1;
1248                 goto end_no_session;
1249         }
1250
1251         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1252         if (ret < sizeof(msg)) {
1253                 if (ret == 0) {
1254                         /* Orderly shutdown. Not necessary to print an error. */
1255                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1256                 } else {
1257                         ERR("Relay didn't receive valid data_pending struct size : %d",
1258                                         ret);
1259                 }
1260                 ret = -1;
1261                 goto end_no_session;
1262         }
1263
1264         stream_id = be64toh(msg.stream_id);
1265         last_net_seq_num = be64toh(msg.last_net_seq_num);
1266
1267         rcu_read_lock();
1268         stream = relay_stream_from_stream_id(stream_id, streams_ht);
1269         if (stream == NULL) {
1270                 ret = -1;
1271                 goto end_unlock;
1272         }
1273
1274         DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
1275                         " and last_seq %" PRIu64, stream_id, stream->prev_seq,
1276                         last_net_seq_num);
1277
1278         /* Avoid wrapping issue */
1279         if (((int64_t) (stream->prev_seq - last_net_seq_num)) >= 0) {
1280                 /* Data has in fact been written and is NOT pending */
1281                 ret = 0;
1282         } else {
1283                 /* Data still being streamed thus pending */
1284                 ret = 1;
1285         }
1286
1287         /* Pending check is now done. */
1288         stream->data_pending_check_done = 1;
1289
1290 end_unlock:
1291         rcu_read_unlock();
1292
1293         reply.ret_code = htobe32(ret);
1294         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1295         if (ret < 0) {
1296                 ERR("Relay data pending ret code failed");
1297         }
1298
1299 end_no_session:
1300         return ret;
1301 }
1302
1303 /*
1304  * Wait for the control socket to reach a quiescent state.
1305  *
1306  * Note that for now, when receiving this command from the session daemon, this
1307  * means that every subsequent commands or data received on the control socket
1308  * has been handled. So, this is why we simply return OK here.
1309  */
1310 static
1311 int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
1312                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1313 {
1314         int ret;
1315         uint64_t stream_id;
1316         struct relay_stream *stream;
1317         struct lttng_ht_iter iter;
1318         struct lttcomm_relayd_quiescent_control msg;
1319         struct lttcomm_relayd_generic_reply reply;
1320
1321         DBG("Checking quiescent state on control socket");
1322
1323         if (!cmd->session || cmd->version_check_done == 0) {
1324                 ERR("Trying to check for data before version check");
1325                 ret = -1;
1326                 goto end_no_session;
1327         }
1328
1329         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1330         if (ret < sizeof(msg)) {
1331                 if (ret == 0) {
1332                         /* Orderly shutdown. Not necessary to print an error. */
1333                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1334                 } else {
1335                         ERR("Relay didn't receive valid begin data_pending struct size: %d",
1336                                         ret);
1337                 }
1338                 ret = -1;
1339                 goto end_no_session;
1340         }
1341
1342         stream_id = be64toh(msg.stream_id);
1343
1344         rcu_read_lock();
1345         cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, stream, stream_n.node) {
1346                 if (stream->stream_handle == stream_id) {
1347                         stream->data_pending_check_done = 1;
1348                         DBG("Relay quiescent control pending flag set to %" PRIu64,
1349                                         stream_id);
1350                         break;
1351                 }
1352         }
1353         rcu_read_unlock();
1354
1355         reply.ret_code = htobe32(LTTNG_OK);
1356         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1357         if (ret < 0) {
1358                 ERR("Relay data quiescent control ret code failed");
1359         }
1360
1361 end_no_session:
1362         return ret;
1363 }
1364
1365 /*
1366  * Initialize a data pending command. This means that a client is about to ask
1367  * for data pending for each stream he/she holds. Simply iterate over all
1368  * streams of a session and set the data_pending_check_done flag.
1369  *
1370  * This command returns to the client a LTTNG_OK code.
1371  */
1372 static
1373 int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1374                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1375 {
1376         int ret;
1377         struct lttng_ht_iter iter;
1378         struct lttcomm_relayd_begin_data_pending msg;
1379         struct lttcomm_relayd_generic_reply reply;
1380         struct relay_stream *stream;
1381         uint64_t session_id;
1382
1383         assert(recv_hdr);
1384         assert(cmd);
1385         assert(streams_ht);
1386
1387         DBG("Init streams for data pending");
1388
1389         if (!cmd->session || cmd->version_check_done == 0) {
1390                 ERR("Trying to check for data before version check");
1391                 ret = -1;
1392                 goto end_no_session;
1393         }
1394
1395         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1396         if (ret < sizeof(msg)) {
1397                 if (ret == 0) {
1398                         /* Orderly shutdown. Not necessary to print an error. */
1399                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1400                 } else {
1401                         ERR("Relay didn't receive valid begin data_pending struct size: %d",
1402                                         ret);
1403                 }
1404                 ret = -1;
1405                 goto end_no_session;
1406         }
1407
1408         session_id = be64toh(msg.session_id);
1409
1410         /*
1411          * Iterate over all streams to set the begin data pending flag. For now, the
1412          * streams are indexed by stream handle so we have to iterate over all
1413          * streams to find the one associated with the right session_id.
1414          */
1415         rcu_read_lock();
1416         cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, stream, stream_n.node) {
1417                 if (stream->session->id == session_id) {
1418                         stream->data_pending_check_done = 0;
1419                         DBG("Set begin data pending flag to stream %" PRIu64,
1420                                         stream->stream_handle);
1421                 }
1422         }
1423         rcu_read_unlock();
1424
1425         /* All good, send back reply. */
1426         reply.ret_code = htobe32(LTTNG_OK);
1427
1428         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1429         if (ret < 0) {
1430                 ERR("Relay begin data pending send reply failed");
1431         }
1432
1433 end_no_session:
1434         return ret;
1435 }
1436
1437 /*
1438  * End data pending command. This will check, for a given session id, if each
1439  * stream associated with it has its data_pending_check_done flag set. If not,
1440  * this means that the client lost track of the stream but the data is still
1441  * being streamed on our side. In this case, we inform the client that data is
1442  * inflight.
1443  *
1444  * Return to the client if there is data in flight or not with a ret_code.
1445  */
1446 static
1447 int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr,
1448                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1449 {
1450         int ret;
1451         struct lttng_ht_iter iter;
1452         struct lttcomm_relayd_end_data_pending msg;
1453         struct lttcomm_relayd_generic_reply reply;
1454         struct relay_stream *stream;
1455         uint64_t session_id;
1456         uint32_t is_data_inflight = 0;
1457
1458         assert(recv_hdr);
1459         assert(cmd);
1460         assert(streams_ht);
1461
1462         DBG("End data pending command");
1463
1464         if (!cmd->session || cmd->version_check_done == 0) {
1465                 ERR("Trying to check for data before version check");
1466                 ret = -1;
1467                 goto end_no_session;
1468         }
1469
1470         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1471         if (ret < sizeof(msg)) {
1472                 if (ret == 0) {
1473                         /* Orderly shutdown. Not necessary to print an error. */
1474                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1475                 } else {
1476                         ERR("Relay didn't receive valid end data_pending struct size: %d",
1477                                         ret);
1478                 }
1479                 ret = -1;
1480                 goto end_no_session;
1481         }
1482
1483         session_id = be64toh(msg.session_id);
1484
1485         /* Iterate over all streams to see if the begin data pending flag is set. */
1486         rcu_read_lock();
1487         cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, stream, stream_n.node) {
1488                 if (stream->session->id == session_id &&
1489                                 !stream->data_pending_check_done) {
1490                         is_data_inflight = 1;
1491                         DBG("Data is still in flight for stream %" PRIu64,
1492                                         stream->stream_handle);
1493                         break;
1494                 }
1495         }
1496         rcu_read_unlock();
1497
1498         /* All good, send back reply. */
1499         reply.ret_code = htobe32(is_data_inflight);
1500
1501         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
1502         if (ret < 0) {
1503                 ERR("Relay end data pending send reply failed");
1504         }
1505
1506 end_no_session:
1507         return ret;
1508 }
1509
1510 /*
1511  * relay_process_control: Process the commands received on the control socket
1512  */
1513 static
1514 int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1515                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1516 {
1517         int ret = 0;
1518
1519         switch (be32toh(recv_hdr->cmd)) {
1520         case RELAYD_CREATE_SESSION:
1521                 ret = relay_create_session(recv_hdr, cmd);
1522                 break;
1523         case RELAYD_ADD_STREAM:
1524                 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1525                 break;
1526         case RELAYD_START_DATA:
1527                 ret = relay_start(recv_hdr, cmd);
1528                 break;
1529         case RELAYD_SEND_METADATA:
1530                 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1531                 break;
1532         case RELAYD_VERSION:
1533                 ret = relay_send_version(recv_hdr, cmd, streams_ht);
1534                 break;
1535         case RELAYD_CLOSE_STREAM:
1536                 ret = relay_close_stream(recv_hdr, cmd, streams_ht);
1537                 break;
1538         case RELAYD_DATA_PENDING:
1539                 ret = relay_data_pending(recv_hdr, cmd, streams_ht);
1540                 break;
1541         case RELAYD_QUIESCENT_CONTROL:
1542                 ret = relay_quiescent_control(recv_hdr, cmd, streams_ht);
1543                 break;
1544         case RELAYD_BEGIN_DATA_PENDING:
1545                 ret = relay_begin_data_pending(recv_hdr, cmd, streams_ht);
1546                 break;
1547         case RELAYD_END_DATA_PENDING:
1548                 ret = relay_end_data_pending(recv_hdr, cmd, streams_ht);
1549                 break;
1550         case RELAYD_UPDATE_SYNC_INFO:
1551         default:
1552                 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1553                 relay_unknown_command(cmd);
1554                 ret = -1;
1555                 goto end;
1556         }
1557
1558 end:
1559         return ret;
1560 }
1561
1562 /*
1563  * relay_process_data: Process the data received on the data socket
1564  */
1565 static
1566 int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1567 {
1568         int ret = 0;
1569         struct relay_stream *stream;
1570         struct lttcomm_relayd_data_hdr data_hdr;
1571         uint64_t stream_id;
1572         uint64_t net_seq_num;
1573         uint32_t data_size;
1574
1575         ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1576                         sizeof(struct lttcomm_relayd_data_hdr), 0);
1577         if (ret <= 0) {
1578                 if (ret == 0) {
1579                         /* Orderly shutdown. Not necessary to print an error. */
1580                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1581                 } else {
1582                         ERR("Unable to receive data header on sock %d", cmd->sock->fd);
1583                 }
1584                 ret = -1;
1585                 goto end;
1586         }
1587
1588         stream_id = be64toh(data_hdr.stream_id);
1589
1590         rcu_read_lock();
1591         stream = relay_stream_from_stream_id(stream_id, streams_ht);
1592         if (!stream) {
1593                 ret = -1;
1594                 goto end_unlock;
1595         }
1596
1597         data_size = be32toh(data_hdr.data_size);
1598         if (data_buffer_size < data_size) {
1599                 char *tmp_data_ptr;
1600
1601                 tmp_data_ptr = realloc(data_buffer, data_size);
1602                 if (!tmp_data_ptr) {
1603                         ERR("Allocating data buffer");
1604                         free(data_buffer);
1605                         ret = -1;
1606                         goto end_unlock;
1607                 }
1608                 data_buffer = tmp_data_ptr;
1609                 data_buffer_size = data_size;
1610         }
1611         memset(data_buffer, 0, data_size);
1612
1613         net_seq_num = be64toh(data_hdr.net_seq_num);
1614
1615         DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
1616                 data_size, stream_id, net_seq_num);
1617         ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
1618         if (ret <= 0) {
1619                 if (ret == 0) {
1620                         /* Orderly shutdown. Not necessary to print an error. */
1621                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1622                 }
1623                 ret = -1;
1624                 goto end_unlock;
1625         }
1626
1627         if (stream->tracefile_size > 0 &&
1628                         (stream->tracefile_size_current + data_size) >
1629                         stream->tracefile_size) {
1630                 ret = utils_rotate_stream_file(stream->path_name,
1631                                 stream->channel_name, stream->tracefile_size,
1632                                 stream->tracefile_count, -1, -1,
1633                                 stream->fd, &(stream->tracefile_count_current));
1634                 if (ret < 0) {
1635                         ERR("Rotating output file");
1636                         goto end;
1637                 }
1638                 stream->fd = ret;
1639                 /* Reset current size because we just perform a stream rotation. */
1640                 stream->tracefile_size_current = 0;
1641         }
1642         stream->tracefile_size_current += data_size;
1643         do {
1644                 ret = write(stream->fd, data_buffer, data_size);
1645         } while (ret < 0 && errno == EINTR);
1646         if (ret < 0 || ret != data_size) {
1647                 ERR("Relay error writing data to file");
1648                 ret = -1;
1649                 goto end_unlock;
1650         }
1651
1652         DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
1653                         ret, stream->stream_handle);
1654
1655         ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
1656         if (ret < 0) {
1657                 goto end_unlock;
1658         }
1659
1660         stream->prev_seq = net_seq_num;
1661
1662         /* Check if we need to close the FD */
1663         if (close_stream_check(stream)) {
1664                 int cret;
1665                 struct lttng_ht_iter iter;
1666
1667                 cret = close(stream->fd);
1668                 if (cret < 0) {
1669                         PERROR("close stream process data");
1670                 }
1671                 iter.iter.node = &stream->stream_n.node;
1672                 ret = lttng_ht_del(streams_ht, &iter);
1673                 assert(!ret);
1674                 call_rcu(&stream->rcu_node,
1675                         deferred_free_stream);
1676                 DBG("Closed tracefile %d after recv data", stream->fd);
1677         }
1678
1679 end_unlock:
1680         rcu_read_unlock();
1681 end:
1682         return ret;
1683 }
1684
1685 static
1686 void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1687 {
1688         int ret;
1689
1690         lttng_poll_del(events, pollfd);
1691
1692         ret = close(pollfd);
1693         if (ret < 0) {
1694                 ERR("Closing pollfd %d", pollfd);
1695         }
1696 }
1697
1698 static
1699 int relay_add_connection(int fd, struct lttng_poll_event *events,
1700                 struct lttng_ht *relay_connections_ht)
1701 {
1702         struct relay_command *relay_connection;
1703         int ret;
1704
1705         relay_connection = zmalloc(sizeof(struct relay_command));
1706         if (relay_connection == NULL) {
1707                 PERROR("Relay command zmalloc");
1708                 goto error;
1709         }
1710         do {
1711                 ret = read(fd, relay_connection, sizeof(struct relay_command));
1712         } while (ret < 0 && errno == EINTR);
1713         if (ret < 0 || ret < sizeof(struct relay_command)) {
1714                 PERROR("read relay cmd pipe");
1715                 goto error_read;
1716         }
1717
1718         lttng_ht_node_init_ulong(&relay_connection->sock_n,
1719                         (unsigned long) relay_connection->sock->fd);
1720         rcu_read_lock();
1721         lttng_ht_add_unique_ulong(relay_connections_ht,
1722                         &relay_connection->sock_n);
1723         rcu_read_unlock();
1724         return lttng_poll_add(events,
1725                         relay_connection->sock->fd,
1726                         LPOLLIN | LPOLLRDHUP);
1727
1728 error_read:
1729         free(relay_connection);
1730 error:
1731         return -1;
1732 }
1733
1734 static
1735 void deferred_free_connection(struct rcu_head *head)
1736 {
1737         struct relay_command *relay_connection =
1738                 caa_container_of(head, struct relay_command, rcu_node);
1739
1740         lttcomm_destroy_sock(relay_connection->sock);
1741         free(relay_connection);
1742 }
1743
1744 static
1745 void relay_del_connection(struct lttng_ht *relay_connections_ht,
1746                 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1747                 struct relay_command *relay_connection)
1748 {
1749         int ret;
1750
1751         ret = lttng_ht_del(relay_connections_ht, iter);
1752         assert(!ret);
1753         if (relay_connection->type == RELAY_CONTROL) {
1754                 relay_delete_session(relay_connection, streams_ht);
1755         }
1756
1757         call_rcu(&relay_connection->rcu_node,
1758                 deferred_free_connection);
1759 }
1760
1761 /*
1762  * This thread does the actual work
1763  */
1764 static
1765 void *relay_thread_worker(void *data)
1766 {
1767         int ret, err = -1, last_seen_data_fd = -1;
1768         uint32_t nb_fd;
1769         struct relay_command *relay_connection;
1770         struct lttng_poll_event events;
1771         struct lttng_ht *relay_connections_ht;
1772         struct lttng_ht_node_ulong *node;
1773         struct lttng_ht_iter iter;
1774         struct lttng_ht *streams_ht;
1775         struct lttcomm_relayd_hdr recv_hdr;
1776
1777         DBG("[thread] Relay worker started");
1778
1779         rcu_register_thread();
1780
1781         /* table of connections indexed on socket */
1782         relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1783         if (!relay_connections_ht) {
1784                 goto relay_connections_ht_error;
1785         }
1786
1787         /* tables of streams indexed by stream ID */
1788         streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1789         if (!streams_ht) {
1790                 goto streams_ht_error;
1791         }
1792
1793         ret = create_thread_poll_set(&events, 2);
1794         if (ret < 0) {
1795                 goto error_poll_create;
1796         }
1797
1798         ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1799         if (ret < 0) {
1800                 goto error;
1801         }
1802
1803 restart:
1804         while (1) {
1805                 int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
1806
1807                 /* Infinite blocking call, waiting for transmission */
1808                 DBG3("Relayd worker thread polling...");
1809                 ret = lttng_poll_wait(&events, -1);
1810                 if (ret < 0) {
1811                         /*
1812                          * Restart interrupted system call.
1813                          */
1814                         if (errno == EINTR) {
1815                                 goto restart;
1816                         }
1817                         goto error;
1818                 }
1819
1820                 nb_fd = ret;
1821
1822                 /*
1823                  * Process control. The control connection is prioritised so we don't
1824                  * starve it with high throughout put tracing data on the data
1825                  * connection.
1826                  */
1827                 for (i = 0; i < nb_fd; i++) {
1828                         /* Fetch once the poll data */
1829                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1830                         int pollfd = LTTNG_POLL_GETFD(&events, i);
1831
1832                         /* Thread quit pipe has been closed. Killing thread. */
1833                         ret = check_thread_quit_pipe(pollfd, revents);
1834                         if (ret) {
1835                                 err = 0;
1836                                 goto exit;
1837                         }
1838
1839                         /* Inspect the relay cmd pipe for new connection */
1840                         if (pollfd == relay_cmd_pipe[0]) {
1841                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1842                                         ERR("Relay pipe error");
1843                                         goto error;
1844                                 } else if (revents & LPOLLIN) {
1845                                         DBG("Relay command received");
1846                                         ret = relay_add_connection(relay_cmd_pipe[0],
1847                                                         &events, relay_connections_ht);
1848                                         if (ret < 0) {
1849                                                 goto error;
1850                                         }
1851                                 }
1852                         } else if (revents) {
1853                                 rcu_read_lock();
1854                                 lttng_ht_lookup(relay_connections_ht,
1855                                                 (void *)((unsigned long) pollfd),
1856                                                 &iter);
1857                                 node = lttng_ht_iter_get_node_ulong(&iter);
1858                                 if (node == NULL) {
1859                                         DBG2("Relay sock %d not found", pollfd);
1860                                         rcu_read_unlock();
1861                                         goto error;
1862                                 }
1863                                 relay_connection = caa_container_of(node,
1864                                                 struct relay_command, sock_n);
1865
1866                                 if (revents & (LPOLLERR)) {
1867                                         ERR("POLL ERROR");
1868                                         relay_cleanup_poll_connection(&events, pollfd);
1869                                         relay_del_connection(relay_connections_ht,
1870                                                         streams_ht, &iter,
1871                                                         relay_connection);
1872                                         if (last_seen_data_fd == pollfd) {
1873                                                 last_seen_data_fd = last_notdel_data_fd;
1874                                         }
1875                                 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1876                                         DBG("Socket %d hung up", pollfd);
1877                                         relay_cleanup_poll_connection(&events, pollfd);
1878                                         relay_del_connection(relay_connections_ht,
1879                                                         streams_ht, &iter,
1880                                                         relay_connection);
1881                                         if (last_seen_data_fd == pollfd) {
1882                                                 last_seen_data_fd = last_notdel_data_fd;
1883                                         }
1884                                 } else if (revents & LPOLLIN) {
1885                                         /* control socket */
1886                                         if (relay_connection->type == RELAY_CONTROL) {
1887                                                 ret = relay_connection->sock->ops->recvmsg(
1888                                                                 relay_connection->sock, &recv_hdr,
1889                                                                 sizeof(struct lttcomm_relayd_hdr), 0);
1890                                                 /* connection closed */
1891                                                 if (ret <= 0) {
1892                                                         relay_cleanup_poll_connection(&events, pollfd);
1893                                                         relay_del_connection(relay_connections_ht,
1894                                                                         streams_ht, &iter,
1895                                                                         relay_connection);
1896                                                         DBG("Control connection closed with %d", pollfd);
1897                                                 } else {
1898                                                         if (relay_connection->session) {
1899                                                                 DBG2("Relay worker receiving data for session : %" PRIu64,
1900                                                                                 relay_connection->session->id);
1901                                                         }
1902                                                         ret = relay_process_control(&recv_hdr,
1903                                                                         relay_connection,
1904                                                                         streams_ht);
1905                                                         if (ret < 0) {
1906                                                                 /* Clear the session on error. */
1907                                                                 relay_cleanup_poll_connection(&events, pollfd);
1908                                                                 relay_del_connection(relay_connections_ht,
1909                                                                                 streams_ht, &iter,
1910                                                                                 relay_connection);
1911                                                                 DBG("Connection closed with %d", pollfd);
1912                                                         }
1913                                                         seen_control = 1;
1914                                                 }
1915                                         } else {
1916                                                 /*
1917                                                  * Flag the last seen data fd not deleted. It will be
1918                                                  * used as the last seen fd if any fd gets deleted in
1919                                                  * this first loop.
1920                                                  */
1921                                                 last_notdel_data_fd = pollfd;
1922                                         }
1923                                 }
1924                                 rcu_read_unlock();
1925                         }
1926                 }
1927
1928                 /*
1929                  * The last loop handled a control request, go back to poll to make
1930                  * sure we prioritise the control socket.
1931                  */
1932                 if (seen_control) {
1933                         continue;
1934                 }
1935
1936                 if (last_seen_data_fd >= 0) {
1937                         for (i = 0; i < nb_fd; i++) {
1938                                 int pollfd = LTTNG_POLL_GETFD(&events, i);
1939                                 if (last_seen_data_fd == pollfd) {
1940                                         idx = i;
1941                                         break;
1942                                 }
1943                         }
1944                 }
1945
1946                 /* Process data connection. */
1947                 for (i = idx + 1; i < nb_fd; i++) {
1948                         /* Fetch the poll data. */
1949                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1950                         int pollfd = LTTNG_POLL_GETFD(&events, i);
1951
1952                         /* Skip the command pipe. It's handled in the first loop. */
1953                         if (pollfd == relay_cmd_pipe[0]) {
1954                                 continue;
1955                         }
1956
1957                         if (revents) {
1958                                 rcu_read_lock();
1959                                 lttng_ht_lookup(relay_connections_ht,
1960                                                 (void *)((unsigned long) pollfd),
1961                                                 &iter);
1962                                 node = lttng_ht_iter_get_node_ulong(&iter);
1963                                 if (node == NULL) {
1964                                         /* Skip it. Might be removed before. */
1965                                         rcu_read_unlock();
1966                                         continue;
1967                                 }
1968                                 relay_connection = caa_container_of(node,
1969                                                 struct relay_command, sock_n);
1970
1971                                 if (revents & LPOLLIN) {
1972                                         if (relay_connection->type != RELAY_DATA) {
1973                                                 continue;
1974                                         }
1975
1976                                         ret = relay_process_data(relay_connection, streams_ht);
1977                                         /* connection closed */
1978                                         if (ret < 0) {
1979                                                 relay_cleanup_poll_connection(&events, pollfd);
1980                                                 relay_del_connection(relay_connections_ht,
1981                                                                 streams_ht, &iter,
1982                                                                 relay_connection);
1983                                                 DBG("Data connection closed with %d", pollfd);
1984                                                 /*
1985                                                  * Every goto restart call sets the last seen fd where
1986                                                  * here we don't really care since we gracefully
1987                                                  * continue the loop after the connection is deleted.
1988                                                  */
1989                                         } else {
1990                                                 /* Keep last seen port. */
1991                                                 last_seen_data_fd = pollfd;
1992                                                 rcu_read_unlock();
1993                                                 goto restart;
1994                                         }
1995                                 }
1996                                 rcu_read_unlock();
1997                         }
1998                 }
1999                 last_seen_data_fd = -1;
2000         }
2001
2002 exit:
2003 error:
2004         lttng_poll_clean(&events);
2005
2006         /* empty the hash table and free the memory */
2007         rcu_read_lock();
2008         cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
2009                 node = lttng_ht_iter_get_node_ulong(&iter);
2010                 if (node) {
2011                         relay_connection = caa_container_of(node,
2012                                         struct relay_command, sock_n);
2013                         relay_del_connection(relay_connections_ht,
2014                                         streams_ht, &iter,
2015                                         relay_connection);
2016                 }
2017         }
2018         rcu_read_unlock();
2019 error_poll_create:
2020         lttng_ht_destroy(streams_ht);
2021 streams_ht_error:
2022         lttng_ht_destroy(relay_connections_ht);
2023 relay_connections_ht_error:
2024         /* Close relay cmd pipes */
2025         utils_close_pipe(relay_cmd_pipe);
2026         if (err) {
2027                 DBG("Thread exited with error");
2028         }
2029         DBG("Worker thread cleanup complete");
2030         free(data_buffer);
2031         stop_threads();
2032         rcu_unregister_thread();
2033         return NULL;
2034 }
2035
2036 /*
2037  * Create the relay command pipe to wake thread_manage_apps.
2038  * Closed in cleanup().
2039  */
2040 static int create_relay_cmd_pipe(void)
2041 {
2042         int ret;
2043
2044         ret = utils_create_pipe_cloexec(relay_cmd_pipe);
2045
2046         return ret;
2047 }
2048
2049 /*
2050  * main
2051  */
2052 int main(int argc, char **argv)
2053 {
2054         int ret = 0;
2055         void *status;
2056
2057         /* Create thread quit pipe */
2058         if ((ret = init_thread_quit_pipe()) < 0) {
2059                 goto error;
2060         }
2061
2062         /* Parse arguments */
2063         progname = argv[0];
2064         if ((ret = parse_args(argc, argv)) < 0) {
2065                 goto exit;
2066         }
2067
2068         if ((ret = set_signal_handler()) < 0) {
2069                 goto exit;
2070         }
2071
2072         /* Try to create directory if -o, --output is specified. */
2073         if (opt_output_path) {
2074                 ret = utils_mkdir_recursive(opt_output_path, S_IRWXU | S_IRWXG);
2075                 if (ret < 0) {
2076                         ERR("Unable to create %s", opt_output_path);
2077                         goto exit;
2078                 }
2079         }
2080
2081         /* Daemonize */
2082         if (opt_daemon) {
2083                 ret = daemon(0, 0);
2084                 if (ret < 0) {
2085                         PERROR("daemon");
2086                         goto exit;
2087                 }
2088         }
2089
2090         /* Check if daemon is UID = 0 */
2091         is_root = !getuid();
2092
2093         if (!is_root) {
2094                 if (control_uri->port < 1024 || data_uri->port < 1024) {
2095                         ERR("Need to be root to use ports < 1024");
2096                         ret = -1;
2097                         goto exit;
2098                 }
2099         }
2100
2101         /* Setup the thread apps communication pipe. */
2102         if ((ret = create_relay_cmd_pipe()) < 0) {
2103                 goto exit;
2104         }
2105
2106         /* Init relay command queue. */
2107         cds_wfq_init(&relay_cmd_queue.queue);
2108
2109         /* Set up max poll set size */
2110         lttng_poll_set_max_size();
2111
2112         /* Setup the dispatcher thread */
2113         ret = pthread_create(&dispatcher_thread, NULL,
2114                         relay_thread_dispatcher, (void *) NULL);
2115         if (ret != 0) {
2116                 PERROR("pthread_create dispatcher");
2117                 goto exit_dispatcher;
2118         }
2119
2120         /* Setup the worker thread */
2121         ret = pthread_create(&worker_thread, NULL,
2122                         relay_thread_worker, (void *) NULL);
2123         if (ret != 0) {
2124                 PERROR("pthread_create worker");
2125                 goto exit_worker;
2126         }
2127
2128         /* Setup the listener thread */
2129         ret = pthread_create(&listener_thread, NULL,
2130                         relay_thread_listener, (void *) NULL);
2131         if (ret != 0) {
2132                 PERROR("pthread_create listener");
2133                 goto exit_listener;
2134         }
2135
2136 exit_listener:
2137         ret = pthread_join(listener_thread, &status);
2138         if (ret != 0) {
2139                 PERROR("pthread_join");
2140                 goto error;     /* join error, exit without cleanup */
2141         }
2142
2143 exit_worker:
2144         ret = pthread_join(worker_thread, &status);
2145         if (ret != 0) {
2146                 PERROR("pthread_join");
2147                 goto error;     /* join error, exit without cleanup */
2148         }
2149
2150 exit_dispatcher:
2151         ret = pthread_join(dispatcher_thread, &status);
2152         if (ret != 0) {
2153                 PERROR("pthread_join");
2154                 goto error;     /* join error, exit without cleanup */
2155         }
2156
2157 exit:
2158         cleanup();
2159         if (!ret) {
2160                 exit(EXIT_SUCCESS);
2161         }
2162
2163 error:
2164         exit(EXIT_FAILURE);
2165 }