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         ret = utils_create_stream_file(stream->path_name, stream->channel_name,
 867                         stream->tracefile_size, 0, getuid(), getgid());
 868         if (ret < 0) {
 869                 ERR("Create output file");
 870                 goto end;
 871         }
 872         stream->fd = ret;
 873         if (stream->tracefile_size) {
 874                 DBG("Tracefile %s/%s_0 created", stream->path_name, stream->channel_name);
 875         } else {
 876                 DBG("Tracefile %s/%s created", stream->path_name, stream->channel_name);
 877         }
 878
 879         lttng_ht_node_init_ulong(&stream->stream_n,
 880                         (unsigned long) stream->stream_handle);
 881         lttng_ht_add_unique_ulong(streams_ht,
 882                         &stream->stream_n);
 883
 884         DBG("Relay new stream added %s", stream->channel_name);
 885
 886 end:
 887         reply.handle = htobe64(stream->stream_handle);
 888         /* send the session id to the client or a negative return code on error */
 889         if (ret < 0) {
 890                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
 891                 /* stream was not properly added to the ht, so free it */
 892                 free(stream);
 893         } else {
 894                 reply.ret_code = htobe32(LTTNG_OK);
 895         }
 896
 897         send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
 898                         sizeof(struct lttcomm_relayd_status_stream), 0);
 899         if (send_ret < 0) {
 900                 ERR("Relay sending stream id");
 901                 ret = send_ret;
 902         }
 903         rcu_read_unlock();
 904
 905 end_no_session:
 906         return ret;
 907
 908 err_free_stream:
 909         free(stream->path_name);
 910         free(stream->channel_name);
 911         free(stream);
 912         return ret;
 913 }
 914
 915 /*
 916  * relay_close_stream: close a specific stream
 917  */
 918 static
 919 int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
 920                 struct relay_command *cmd, struct lttng_ht *streams_ht)
 921 {
 922         struct relay_session *session = cmd->session;
 923         struct lttcomm_relayd_close_stream stream_info;
 924         struct lttcomm_relayd_generic_reply reply;
 925         struct relay_stream *stream;
 926         int ret, send_ret;
 927         struct lttng_ht_iter iter;
 928
 929         DBG("Close stream received");
 930
 931         if (!session || cmd->version_check_done == 0) {
 932                 ERR("Trying to close a stream before version check");
 933                 ret = -1;
 934                 goto end_no_session;
 935         }
 936
 937         ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
 938                         sizeof(struct lttcomm_relayd_close_stream), 0);
 939         if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
 940                 if (ret == 0) {
 941                         /* Orderly shutdown. Not necessary to print an error. */
 942                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
 943                 } else {
 944                         ERR("Relay didn't receive valid add_stream struct size : %d", ret);
 945                 }
 946                 ret = -1;
 947                 goto end_no_session;
 948         }
 949
 950         rcu_read_lock();
 951         stream = relay_stream_from_stream_id(be64toh(stream_info.stream_id),
 952                         streams_ht);
 953         if (!stream) {
 954                 ret = -1;
 955                 goto end_unlock;
 956         }
 957
 958         stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
 959         stream->close_flag = 1;
 960
 961         if (close_stream_check(stream)) {
 962                 int delret;
 963
 964                 delret = close(stream->fd);
 965                 if (delret < 0) {
 966                         PERROR("close stream");
 967                 }
 968                 iter.iter.node = &stream->stream_n.node;
 969                 delret = lttng_ht_del(streams_ht, &iter);
 970                 assert(!delret);
 971                 call_rcu(&stream->rcu_node,
 972                                 deferred_free_stream);
 973                 DBG("Closed tracefile %d from close stream", stream->fd);
 974         }
 975
 976 end_unlock:
 977         rcu_read_unlock();
 978
 979         if (ret < 0) {
 980                 reply.ret_code = htobe32(LTTNG_ERR_UNK);
 981         } else {
 982                 reply.ret_code = htobe32(LTTNG_OK);
 983         }
 984         send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
 985                         sizeof(struct lttcomm_relayd_generic_reply), 0);
 986         if (send_ret < 0) {
 987                 ERR("Relay sending stream id");
 988                 ret = send_ret;
 989         }
 990
 991 end_no_session:
 992         return ret;
 993 }
 994
 995 /*
 996  * relay_unknown_command: send -1 if received unknown command
 997  */
 998 static
 999 void relay_unknown_command(struct relay_command *cmd)
1000 {
1001         struct lttcomm_relayd_generic_reply reply;
1002         int ret;
1003
1004         reply.ret_code = htobe32(LTTNG_ERR_UNK);
1005         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1006                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1007         if (ret < 0) {
1008                 ERR("Relay sending unknown command");
1009         }
1010 }
1011
1012 /*
1013  * relay_start: send an acknowledgment to the client to tell if we are
1014  * ready to receive data. We are ready if a session is established.
1015  */
1016 static
1017 int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1018                 struct relay_command *cmd)
1019 {
1020         int ret = htobe32(LTTNG_OK);
1021         struct lttcomm_relayd_generic_reply reply;
1022         struct relay_session *session = cmd->session;
1023
1024         if (!session) {
1025                 DBG("Trying to start the streaming without a session established");
1026                 ret = htobe32(LTTNG_ERR_UNK);
1027         }
1028
1029         reply.ret_code = ret;
1030         ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1031                         sizeof(struct lttcomm_relayd_generic_reply), 0);
1032         if (ret < 0) {
1033                 ERR("Relay sending start ack");
1034         }
1035
1036         return ret;
1037 }
1038
1039 /*
1040  * Append padding to the file pointed by the file descriptor fd.
1041  */
1042 static int write_padding_to_file(int fd, uint32_t size)
1043 {
1044         int ret = 0;
1045         char *zeros;
1046
1047         if (size == 0) {
1048                 goto end;
1049         }
1050
1051         zeros = zmalloc(size);
1052         if (zeros == NULL) {
1053                 PERROR("zmalloc zeros for padding");
1054                 ret = -1;
1055                 goto end;
1056         }
1057
1058         do {
1059                 ret = write(fd, zeros, size);
1060         } while (ret < 0 && errno == EINTR);
1061         if (ret < 0 || ret != size) {
1062                 PERROR("write padding to file");
1063         }
1064
1065         free(zeros);
1066
1067 end:
1068         return ret;
1069 }
1070
1071 /*
1072  * relay_recv_metadata: receive the metada for the session.
1073  */
1074 static
1075 int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1076                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1077 {
1078         int ret = htobe32(LTTNG_OK);
1079         struct relay_session *session = cmd->session;
1080         struct lttcomm_relayd_metadata_payload *metadata_struct;
1081         struct relay_stream *metadata_stream;
1082         uint64_t data_size, payload_size;
1083
1084         if (!session) {
1085                 ERR("Metadata sent before version check");
1086                 ret = -1;
1087                 goto end;
1088         }
1089
1090         data_size = payload_size = be64toh(recv_hdr->data_size);
1091         if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1092                 ERR("Incorrect data size");
1093                 ret = -1;
1094                 goto end;
1095         }
1096         payload_size -= sizeof(struct lttcomm_relayd_metadata_payload);
1097
1098         if (data_buffer_size < data_size) {
1099                 /* In case the realloc fails, we can free the memory */
1100                 char *tmp_data_ptr;
1101
1102                 tmp_data_ptr = realloc(data_buffer, data_size);
1103                 if (!tmp_data_ptr) {
1104                         ERR("Allocating data buffer");
1105                         free(data_buffer);
1106                         ret = -1;
1107                         goto end;
1108                 }
1109                 data_buffer = tmp_data_ptr;
1110                 data_buffer_size = data_size;
1111         }
1112         memset(data_buffer, 0, data_size);
1113         DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
1114         ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
1115         if (ret < 0 || ret != data_size) {
1116                 if (ret == 0) {
1117                         /* Orderly shutdown. Not necessary to print an error. */
1118                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1119                 } else {
1120                         ERR("Relay didn't receive the whole metadata");
1121                 }
1122                 ret = -1;
1123                 goto end;
1124         }
1125         metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
1126
1127         rcu_read_lock();
1128         metadata_stream = relay_stream_from_stream_id(
1129                         be64toh(metadata_struct->stream_id), streams_ht);
1130         if (!metadata_stream) {
1131                 ret = -1;
1132                 goto end_unlock;
1133         }
1134
1135         do {
1136                 ret = write(metadata_stream->fd, metadata_struct->payload,
1137                                 payload_size);
1138         } while (ret < 0 && errno == EINTR);
1139         if (ret < 0 || ret != payload_size) {
1140                 ERR("Relay error writing metadata on file");
1141                 ret = -1;
1142                 goto end_unlock;
1143         }
1144
1145         ret = write_padding_to_file(metadata_stream->fd,
1146                         be32toh(metadata_struct->padding_size));
1147         if (ret < 0) {
1148                 goto end_unlock;
1149         }
1150
1151         DBG2("Relay metadata written");
1152
1153 end_unlock:
1154         rcu_read_unlock();
1155 end:
1156         return ret;
1157 }
1158
1159 /*
1160  * relay_send_version: send relayd version number
1161  */
1162 static
1163 int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1164                 struct relay_command *cmd, struct lttng_ht *streams_ht)
1165 {
1166         int ret;
1167         struct lttcomm_relayd_version reply, msg;
1168
1169         assert(cmd);
1170
1171         cmd->version_check_done = 1;
1172
1173         /* Get version from the other side. */
1174         ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
1175         if (ret < 0 || ret != sizeof(msg)) {
1176                 if (ret == 0) {
1177                         /* Orderly shutdown. Not necessary to print an error. */
1178                         DBG("Socket %d did an orderly shutdown", cmd->sock->fd);
1179                 } else {
1180                         ERR("Relay failed to receive the version values.");
1181                 }
1182                 ret = -1;
1183                 goto end;
1184         }
1185
1186         ret = sscanf(VERSION, "%10u.%10u", &reply.major, &reply.minor);
1187         if (ret < 2) {
1188                 ERR("Error in scanning version");
1189                 ret = -1;
1190                 goto end;
1191         }
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, getuid(), getgid(),
1629                                 stream->fd, &(stream->tracefile_count_current));
1630                 if (ret < 0) {
1631                         ERR("Rotating output file");
1632                         goto end;
1633                 }
1634                 stream->fd = ret;
1635         }
1636         stream->tracefile_size_current += data_size;
1637         do {
1638                 ret = write(stream->fd, data_buffer, data_size);
1639         } while (ret < 0 && errno == EINTR);
1640         if (ret < 0 || ret != data_size) {
1641                 ERR("Relay error writing data to file");
1642                 ret = -1;
1643                 goto end_unlock;
1644         }
1645
1646         DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
1647                         ret, stream->stream_handle);
1648
1649         ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
1650         if (ret < 0) {
1651                 goto end_unlock;
1652         }
1653
1654         stream->prev_seq = net_seq_num;
1655
1656         /* Check if we need to close the FD */
1657         if (close_stream_check(stream)) {
1658                 int cret;
1659                 struct lttng_ht_iter iter;
1660
1661                 cret = close(stream->fd);
1662                 if (cret < 0) {
1663                         PERROR("close stream process data");
1664                 }
1665                 iter.iter.node = &stream->stream_n.node;
1666                 ret = lttng_ht_del(streams_ht, &iter);
1667                 assert(!ret);
1668                 call_rcu(&stream->rcu_node,
1669                         deferred_free_stream);
1670                 DBG("Closed tracefile %d after recv data", stream->fd);
1671         }
1672
1673 end_unlock:
1674         rcu_read_unlock();
1675 end:
1676         return ret;
1677 }
1678
1679 static
1680 void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1681 {
1682         int ret;
1683
1684         lttng_poll_del(events, pollfd);
1685
1686         ret = close(pollfd);
1687         if (ret < 0) {
1688                 ERR("Closing pollfd %d", pollfd);
1689         }
1690 }
1691
1692 static
1693 int relay_add_connection(int fd, struct lttng_poll_event *events,
1694                 struct lttng_ht *relay_connections_ht)
1695 {
1696         struct relay_command *relay_connection;
1697         int ret;
1698
1699         relay_connection = zmalloc(sizeof(struct relay_command));
1700         if (relay_connection == NULL) {
1701                 PERROR("Relay command zmalloc");
1702                 goto error;
1703         }
1704         do {
1705                 ret = read(fd, relay_connection, sizeof(struct relay_command));
1706         } while (ret < 0 && errno == EINTR);
1707         if (ret < 0 || ret < sizeof(struct relay_command)) {
1708                 PERROR("read relay cmd pipe");
1709                 goto error_read;
1710         }
1711
1712         lttng_ht_node_init_ulong(&relay_connection->sock_n,
1713                         (unsigned long) relay_connection->sock->fd);
1714         rcu_read_lock();
1715         lttng_ht_add_unique_ulong(relay_connections_ht,
1716                         &relay_connection->sock_n);
1717         rcu_read_unlock();
1718         return lttng_poll_add(events,
1719                         relay_connection->sock->fd,
1720                         LPOLLIN | LPOLLRDHUP);
1721
1722 error_read:
1723         free(relay_connection);
1724 error:
1725         return -1;
1726 }
1727
1728 static
1729 void deferred_free_connection(struct rcu_head *head)
1730 {
1731         struct relay_command *relay_connection =
1732                 caa_container_of(head, struct relay_command, rcu_node);
1733
1734         lttcomm_destroy_sock(relay_connection->sock);
1735         free(relay_connection);
1736 }
1737
1738 static
1739 void relay_del_connection(struct lttng_ht *relay_connections_ht,
1740                 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1741                 struct relay_command *relay_connection)
1742 {
1743         int ret;
1744
1745         ret = lttng_ht_del(relay_connections_ht, iter);
1746         assert(!ret);
1747         if (relay_connection->type == RELAY_CONTROL) {
1748                 relay_delete_session(relay_connection, streams_ht);
1749         }
1750
1751         call_rcu(&relay_connection->rcu_node,
1752                 deferred_free_connection);
1753 }
1754
1755 /*
1756  * This thread does the actual work
1757  */
1758 static
1759 void *relay_thread_worker(void *data)
1760 {
1761         int ret, err = -1, last_seen_data_fd = -1;
1762         uint32_t nb_fd;
1763         struct relay_command *relay_connection;
1764         struct lttng_poll_event events;
1765         struct lttng_ht *relay_connections_ht;
1766         struct lttng_ht_node_ulong *node;
1767         struct lttng_ht_iter iter;
1768         struct lttng_ht *streams_ht;
1769         struct lttcomm_relayd_hdr recv_hdr;
1770
1771         DBG("[thread] Relay worker started");
1772
1773         rcu_register_thread();
1774
1775         /* table of connections indexed on socket */
1776         relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1777         if (!relay_connections_ht) {
1778                 goto relay_connections_ht_error;
1779         }
1780
1781         /* tables of streams indexed by stream ID */
1782         streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1783         if (!streams_ht) {
1784                 goto streams_ht_error;
1785         }
1786
1787         ret = create_thread_poll_set(&events, 2);
1788         if (ret < 0) {
1789                 goto error_poll_create;
1790         }
1791
1792         ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1793         if (ret < 0) {
1794                 goto error;
1795         }
1796
1797 restart:
1798         while (1) {
1799                 int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
1800
1801                 /* Infinite blocking call, waiting for transmission */
1802                 DBG3("Relayd worker thread polling...");
1803                 ret = lttng_poll_wait(&events, -1);
1804                 if (ret < 0) {
1805                         /*
1806                          * Restart interrupted system call.
1807                          */
1808                         if (errno == EINTR) {
1809                                 goto restart;
1810                         }
1811                         goto error;
1812                 }
1813
1814                 nb_fd = ret;
1815
1816                 /*
1817                  * Process control. The control connection is prioritised so we don't
1818                  * starve it with high throughout put tracing data on the data
1819                  * connection.
1820                  */
1821                 for (i = 0; i < nb_fd; i++) {
1822                         /* Fetch once the poll data */
1823                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1824                         int pollfd = LTTNG_POLL_GETFD(&events, i);
1825
1826                         /* Thread quit pipe has been closed. Killing thread. */
1827                         ret = check_thread_quit_pipe(pollfd, revents);
1828                         if (ret) {
1829                                 err = 0;
1830                                 goto exit;
1831                         }
1832
1833                         /* Inspect the relay cmd pipe for new connection */
1834                         if (pollfd == relay_cmd_pipe[0]) {
1835                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1836                                         ERR("Relay pipe error");
1837                                         goto error;
1838                                 } else if (revents & LPOLLIN) {
1839                                         DBG("Relay command received");
1840                                         ret = relay_add_connection(relay_cmd_pipe[0],
1841                                                         &events, relay_connections_ht);
1842                                         if (ret < 0) {
1843                                                 goto error;
1844                                         }
1845                                 }
1846                         } else if (revents) {
1847                                 rcu_read_lock();
1848                                 lttng_ht_lookup(relay_connections_ht,
1849                                                 (void *)((unsigned long) pollfd),
1850                                                 &iter);
1851                                 node = lttng_ht_iter_get_node_ulong(&iter);
1852                                 if (node == NULL) {
1853                                         DBG2("Relay sock %d not found", pollfd);
1854                                         rcu_read_unlock();
1855                                         goto error;
1856                                 }
1857                                 relay_connection = caa_container_of(node,
1858                                                 struct relay_command, sock_n);
1859
1860                                 if (revents & (LPOLLERR)) {
1861                                         ERR("POLL ERROR");
1862                                         relay_cleanup_poll_connection(&events, pollfd);
1863                                         relay_del_connection(relay_connections_ht,
1864                                                         streams_ht, &iter,
1865                                                         relay_connection);
1866                                         if (last_seen_data_fd == pollfd) {
1867                                                 last_seen_data_fd = last_notdel_data_fd;
1868                                         }
1869                                 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1870                                         DBG("Socket %d hung up", pollfd);
1871                                         relay_cleanup_poll_connection(&events, pollfd);
1872                                         relay_del_connection(relay_connections_ht,
1873                                                         streams_ht, &iter,
1874                                                         relay_connection);
1875                                         if (last_seen_data_fd == pollfd) {
1876                                                 last_seen_data_fd = last_notdel_data_fd;
1877                                         }
1878                                 } else if (revents & LPOLLIN) {
1879                                         /* control socket */
1880                                         if (relay_connection->type == RELAY_CONTROL) {
1881                                                 ret = relay_connection->sock->ops->recvmsg(
1882                                                                 relay_connection->sock, &recv_hdr,
1883                                                                 sizeof(struct lttcomm_relayd_hdr), 0);
1884                                                 /* connection closed */
1885                                                 if (ret <= 0) {
1886                                                         relay_cleanup_poll_connection(&events, pollfd);
1887                                                         relay_del_connection(relay_connections_ht,
1888                                                                         streams_ht, &iter,
1889                                                                         relay_connection);
1890                                                         DBG("Control connection closed with %d", pollfd);
1891                                                 } else {
1892                                                         if (relay_connection->session) {
1893                                                                 DBG2("Relay worker receiving data for session : %" PRIu64,
1894                                                                                 relay_connection->session->id);
1895                                                         }
1896                                                         ret = relay_process_control(&recv_hdr,
1897                                                                         relay_connection,
1898                                                                         streams_ht);
1899                                                         if (ret < 0) {
1900                                                                 /* Clear the session on error. */
1901                                                                 relay_cleanup_poll_connection(&events, pollfd);
1902                                                                 relay_del_connection(relay_connections_ht,
1903                                                                                 streams_ht, &iter,
1904                                                                                 relay_connection);
1905                                                                 DBG("Connection closed with %d", pollfd);
1906                                                         }
1907                                                         seen_control = 1;
1908                                                 }
1909                                         } else {
1910                                                 /*
1911                                                  * Flag the last seen data fd not deleted. It will be
1912                                                  * used as the last seen fd if any fd gets deleted in
1913                                                  * this first loop.
1914                                                  */
1915                                                 last_notdel_data_fd = pollfd;
1916                                         }
1917                                 }
1918                                 rcu_read_unlock();
1919                         }
1920                 }
1921
1922                 /*
1923                  * The last loop handled a control request, go back to poll to make
1924                  * sure we prioritise the control socket.
1925                  */
1926                 if (seen_control) {
1927                         continue;
1928                 }
1929
1930                 if (last_seen_data_fd >= 0) {
1931                         for (i = 0; i < nb_fd; i++) {
1932                                 int pollfd = LTTNG_POLL_GETFD(&events, i);
1933                                 if (last_seen_data_fd == pollfd) {
1934                                         idx = i;
1935                                         break;
1936                                 }
1937                         }
1938                 }
1939
1940                 /* Process data connection. */
1941                 for (i = idx + 1; i < nb_fd; i++) {
1942                         /* Fetch the poll data. */
1943                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1944                         int pollfd = LTTNG_POLL_GETFD(&events, i);
1945
1946                         /* Skip the command pipe. It's handled in the first loop. */
1947                         if (pollfd == relay_cmd_pipe[0]) {
1948                                 continue;
1949                         }
1950
1951                         if (revents) {
1952                                 rcu_read_lock();
1953                                 lttng_ht_lookup(relay_connections_ht,
1954                                                 (void *)((unsigned long) pollfd),
1955                                                 &iter);
1956                                 node = lttng_ht_iter_get_node_ulong(&iter);
1957                                 if (node == NULL) {
1958                                         /* Skip it. Might be removed before. */
1959                                         rcu_read_unlock();
1960                                         continue;
1961                                 }
1962                                 relay_connection = caa_container_of(node,
1963                                                 struct relay_command, sock_n);
1964
1965                                 if (revents & LPOLLIN) {
1966                                         if (relay_connection->type != RELAY_DATA) {
1967                                                 continue;
1968                                         }
1969
1970                                         ret = relay_process_data(relay_connection, streams_ht);
1971                                         /* connection closed */
1972                                         if (ret < 0) {
1973                                                 relay_cleanup_poll_connection(&events, pollfd);
1974                                                 relay_del_connection(relay_connections_ht,
1975                                                                 streams_ht, &iter,
1976                                                                 relay_connection);
1977                                                 DBG("Data connection closed with %d", pollfd);
1978                                                 /*
1979                                                  * Every goto restart call sets the last seen fd where
1980                                                  * here we don't really care since we gracefully
1981                                                  * continue the loop after the connection is deleted.
1982                                                  */
1983                                         } else {
1984                                                 /* Keep last seen port. */
1985                                                 last_seen_data_fd = pollfd;
1986                                                 rcu_read_unlock();
1987                                                 goto restart;
1988                                         }
1989                                 }
1990                                 rcu_read_unlock();
1991                         }
1992                 }
1993                 last_seen_data_fd = -1;
1994         }
1995
1996 exit:
1997 error:
1998         lttng_poll_clean(&events);
1999
2000         /* empty the hash table and free the memory */
2001         rcu_read_lock();
2002         cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
2003                 node = lttng_ht_iter_get_node_ulong(&iter);
2004                 if (node) {
2005                         relay_connection = caa_container_of(node,
2006                                         struct relay_command, sock_n);
2007                         relay_del_connection(relay_connections_ht,
2008                                         streams_ht, &iter,
2009                                         relay_connection);
2010                 }
2011         }
2012         rcu_read_unlock();
2013 error_poll_create:
2014         lttng_ht_destroy(streams_ht);
2015 streams_ht_error:
2016         lttng_ht_destroy(relay_connections_ht);
2017 relay_connections_ht_error:
2018         /* Close relay cmd pipes */
2019         utils_close_pipe(relay_cmd_pipe);
2020         if (err) {
2021                 DBG("Thread exited with error");
2022         }
2023         DBG("Worker thread cleanup complete");
2024         free(data_buffer);
2025         stop_threads();
2026         rcu_unregister_thread();
2027         return NULL;
2028 }
2029
2030 /*
2031  * Create the relay command pipe to wake thread_manage_apps.
2032  * Closed in cleanup().
2033  */
2034 static int create_relay_cmd_pipe(void)
2035 {
2036         int ret;
2037
2038         ret = utils_create_pipe_cloexec(relay_cmd_pipe);
2039
2040         return ret;
2041 }
2042
2043 /*
2044  * main
2045  */
2046 int main(int argc, char **argv)
2047 {
2048         int ret = 0;
2049         void *status;
2050
2051         /* Create thread quit pipe */
2052         if ((ret = init_thread_quit_pipe()) < 0) {
2053                 goto error;
2054         }
2055
2056         /* Parse arguments */
2057         progname = argv[0];
2058         if ((ret = parse_args(argc, argv)) < 0) {
2059                 goto exit;
2060         }
2061
2062         if ((ret = set_signal_handler()) < 0) {
2063                 goto exit;
2064         }
2065
2066         /* Try to create directory if -o, --output is specified. */
2067         if (opt_output_path) {
2068                 ret = utils_mkdir_recursive(opt_output_path, S_IRWXU | S_IRWXG);
2069                 if (ret < 0) {
2070                         ERR("Unable to create %s", opt_output_path);
2071                         goto exit;
2072                 }
2073         }
2074
2075         /* Daemonize */
2076         if (opt_daemon) {
2077                 ret = daemon(0, 0);
2078                 if (ret < 0) {
2079                         PERROR("daemon");
2080                         goto exit;
2081                 }
2082         }
2083
2084         /* Check if daemon is UID = 0 */
2085         is_root = !getuid();
2086
2087         if (!is_root) {
2088                 if (control_uri->port < 1024 || data_uri->port < 1024) {
2089                         ERR("Need to be root to use ports < 1024");
2090                         ret = -1;
2091                         goto exit;
2092                 }
2093         }
2094
2095         /* Setup the thread apps communication pipe. */
2096         if ((ret = create_relay_cmd_pipe()) < 0) {
2097                 goto exit;
2098         }
2099
2100         /* Init relay command queue. */
2101         cds_wfq_init(&relay_cmd_queue.queue);
2102
2103         /* Set up max poll set size */
2104         lttng_poll_set_max_size();
2105
2106         /* Setup the dispatcher thread */
2107         ret = pthread_create(&dispatcher_thread, NULL,
2108                         relay_thread_dispatcher, (void *) NULL);
2109         if (ret != 0) {
2110                 PERROR("pthread_create dispatcher");
2111                 goto exit_dispatcher;
2112         }
2113
2114         /* Setup the worker thread */
2115         ret = pthread_create(&worker_thread, NULL,
2116                         relay_thread_worker, (void *) NULL);
2117         if (ret != 0) {
2118                 PERROR("pthread_create worker");
2119                 goto exit_worker;
2120         }
2121
2122         /* Setup the listener thread */
2123         ret = pthread_create(&listener_thread, NULL,
2124                         relay_thread_listener, (void *) NULL);
2125         if (ret != 0) {
2126                 PERROR("pthread_create listener");
2127                 goto exit_listener;
2128         }
2129
2130 exit_listener:
2131         ret = pthread_join(listener_thread, &status);
2132         if (ret != 0) {
2133                 PERROR("pthread_join");
2134                 goto error;     /* join error, exit without cleanup */
2135         }
2136
2137 exit_worker:
2138         ret = pthread_join(worker_thread, &status);
2139         if (ret != 0) {
2140                 PERROR("pthread_join");
2141                 goto error;     /* join error, exit without cleanup */
2142         }
2143
2144 exit_dispatcher:
2145         ret = pthread_join(dispatcher_thread, &status);
2146         if (ret != 0) {
2147                 PERROR("pthread_join");
2148                 goto error;     /* join error, exit without cleanup */
2149         }
2150
2151 exit:
2152         cleanup();
2153         if (!ret) {
2154                 exit(EXIT_SUCCESS);
2155         }
2156
2157 error:
2158         exit(EXIT_FAILURE);
2159 }