src/bin/lttng-relayd/live.c

   1 /*
   2  * Copyright (C) 2013 - 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/index/index.h>
  49 #include <common/sessiond-comm/sessiond-comm.h>
  50 #include <common/sessiond-comm/inet.h>
  51 #include <common/sessiond-comm/relayd.h>
  52 #include <common/uri.h>
  53 #include <common/utils.h>
  54
  55 #include "cmd.h"
  56 #include "live.h"
  57 #include "lttng-relayd.h"
  58 #include "utils.h"
  59 #include "health-relayd.h"
  60 #include "testpoint.h"
  61 #include "viewer-stream.h"
  62
  63 static struct lttng_uri *live_uri;
  64
  65 /*
  66  * This pipe is used to inform the worker thread that a command is queued and
  67  * ready to be processed.
  68  */
  69 static int live_relay_cmd_pipe[2] = { -1, -1 };
  70
  71 /* Shared between threads */
  72 static int live_dispatch_thread_exit;
  73
  74 static pthread_t live_listener_thread;
  75 static pthread_t live_dispatcher_thread;
  76 static pthread_t live_worker_thread;
  77
  78 /*
  79  * Relay command queue.
  80  *
  81  * The live_thread_listener and live_thread_dispatcher communicate with this
  82  * queue.
  83  */
  84 static struct relay_cmd_queue viewer_cmd_queue;
  85
  86 static uint64_t last_relay_viewer_session_id;
  87
  88 /*
  89  * Cleanup the daemon
  90  */
  91 static
  92 void cleanup(void)
  93 {
  94         DBG("Cleaning up");
  95
  96         free(live_uri);
  97 }
  98
  99 /*
 100  * Receive a request buffer using a given socket, destination allocated buffer
 101  * of length size.
 102  *
 103  * Return the size of the received message or else a negative value on error
 104  * with errno being set by recvmsg() syscall.
 105  */
 106 static
 107 ssize_t recv_request(struct lttcomm_sock *sock, void *buf, size_t size)
 108 {
 109         ssize_t ret;
 110
 111         assert(sock);
 112         assert(buf);
 113
 114         ret = sock->ops->recvmsg(sock, buf, size, 0);
 115         if (ret < 0 || ret != size) {
 116                 if (ret == 0) {
 117                         /* Orderly shutdown. Not necessary to print an error. */
 118                         DBG("Socket %d did an orderly shutdown", sock->fd);
 119                 } else {
 120                         ERR("Relay failed to receive request.");
 121                 }
 122                 ret = -1;
 123         }
 124
 125         return ret;
 126 }
 127
 128 /*
 129  * Send a response buffer using a given socket, source allocated buffer of
 130  * length size.
 131  *
 132  * Return the size of the sent message or else a negative value on error with
 133  * errno being set by sendmsg() syscall.
 134  */
 135 static
 136 ssize_t send_response(struct lttcomm_sock *sock, void *buf, size_t size)
 137 {
 138         ssize_t ret;
 139
 140         assert(sock);
 141         assert(buf);
 142
 143         ret = sock->ops->sendmsg(sock, buf, size, 0);
 144         if (ret < 0) {
 145                 ERR("Relayd failed to send response.");
 146         }
 147
 148         return ret;
 149 }
 150
 151 /*
 152  * Atomically check if new streams got added in the session since the last
 153  * check and reset the flag to 0.
 154  *
 155  * Returns 1 if new streams got added, 0 if nothing changed, a negative value
 156  * on error.
 157  */
 158 static
 159 int check_new_streams(uint64_t session_id, struct lttng_ht *sessions_ht)
 160 {
 161         int ret;
 162         unsigned long current_val;
 163         struct relay_session *session;
 164
 165         assert(sessions_ht);
 166
 167         session = session_find_by_id(sessions_ht, session_id);
 168         if (!session) {
 169                 DBG("Relay session %" PRIu64 " not found", session_id);
 170                 ret = -1;
 171                 goto error;
 172         }
 173
 174         current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
 175         ret = current_val;
 176
 177 error:
 178         return ret;
 179 }
 180
 181 /*
 182  * Send viewer streams to the given socket. The ignore_sent_flag indicates if
 183  * this function should ignore the sent flag or not.
 184  *
 185  * Return 0 on success or else a negative value.
 186  */
 187 static
 188 ssize_t send_viewer_streams(struct lttcomm_sock *sock,
 189                 struct relay_session *session, unsigned int ignore_sent_flag)
 190 {
 191         ssize_t ret;
 192         struct lttng_viewer_stream send_stream;
 193         struct lttng_ht_iter iter;
 194         struct relay_viewer_stream *vstream;
 195
 196         assert(session);
 197
 198         rcu_read_lock();
 199
 200         cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, vstream,
 201                         stream_n.node) {
 202                 health_code_update();
 203
 204                 /* Ignore if not the same session. */
 205                 if (vstream->session_id != session->id ||
 206                                 (!ignore_sent_flag && vstream->sent_flag)) {
 207                         continue;
 208                 }
 209
 210                 send_stream.id = htobe64(vstream->stream_handle);
 211                 send_stream.ctf_trace_id = htobe64(vstream->ctf_trace->id);
 212                 send_stream.metadata_flag = htobe32(vstream->metadata_flag);
 213                 strncpy(send_stream.path_name, vstream->path_name,
 214                                 sizeof(send_stream.path_name));
 215                 strncpy(send_stream.channel_name, vstream->channel_name,
 216                                 sizeof(send_stream.channel_name));
 217
 218                 DBG("Sending stream %" PRIu64 " to viewer", vstream->stream_handle);
 219                 ret = send_response(sock, &send_stream, sizeof(send_stream));
 220                 if (ret < 0) {
 221                         goto end_unlock;
 222                 }
 223                 vstream->sent_flag = 1;
 224         }
 225
 226         ret = 0;
 227
 228 end_unlock:
 229         rcu_read_unlock();
 230         return ret;
 231 }
 232
 233 /*
 234  * Create every viewer stream possible for the given session with the seek
 235  * type. Three counters *can* be return which are in order the total amount of
 236  * viewer stream of the session, the number of unsent stream and the number of
 237  * stream created. Those counters can be NULL and thus will be ignored.
 238  *
 239  * Return 0 on success or else a negative value.
 240  */
 241 static
 242 int make_viewer_streams(struct relay_session *session,
 243                 enum lttng_viewer_seek seek_t, uint32_t *nb_total, uint32_t *nb_unsent,
 244                 uint32_t *nb_created)
 245 {
 246         int ret;
 247         struct relay_stream *stream;
 248         struct lttng_ht_iter iter;
 249
 250         assert(session);
 251
 252         /*
 253          * This is to make sure we create viewer streams for a full received
 254          * channel. For instance, if we have 8 streams for a channel that are
 255          * concurrently being flagged ready, we can end up creating just a subset
 256          * of the 8 streams (the ones that are flagged). This lock avoids this
 257          * limbo state.
 258          */
 259         pthread_mutex_lock(&session->viewer_ready_lock);
 260
 261         /*
 262          * Create viewer streams for relay streams that are ready to be used for a
 263          * the given session id only.
 264          */
 265         cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
 266                         stream_n.node) {
 267                 struct relay_viewer_stream *vstream;
 268
 269                 health_code_update();
 270
 271                 if (stream->session->id != session->id ||
 272                                 !stream->ctf_trace || !stream->viewer_ready) {
 273                         /*
 274                          * Ignore stream from a different session. Don't create streams
 275                          * with no ctf_trace or not ready for the viewer.
 276                          */
 277                         continue;
 278                 }
 279
 280                 vstream = viewer_stream_find_by_id(stream->stream_handle);
 281                 if (!vstream) {
 282                         vstream = viewer_stream_create(stream, seek_t);
 283                         if (!vstream) {
 284                                 ret = -1;
 285                                 goto error_unlock;
 286                         }
 287                         if (nb_created) {
 288                                 /* Update number of created stream counter. */
 289                                 (*nb_created)++;
 290                         }
 291                 } else if (!vstream->sent_flag && nb_unsent) {
 292                         /* Update number of unsent stream counter. */
 293                         (*nb_unsent)++;
 294                 }
 295                 /* Update number of total stream counter. */
 296                 if (nb_total) {
 297                         (*nb_total)++;
 298                 }
 299         }
 300
 301         ret = 0;
 302
 303 error_unlock:
 304         pthread_mutex_unlock(&session->viewer_ready_lock);
 305         return ret;
 306 }
 307
 308 /*
 309  * Write to writable pipe used to notify a thread.
 310  */
 311 static
 312 int notify_thread_pipe(int wpipe)
 313 {
 314         ssize_t ret;
 315
 316         ret = lttng_write(wpipe, "!", 1);
 317         if (ret < 1) {
 318                 PERROR("write poll pipe");
 319         }
 320
 321         return (int) ret;
 322 }
 323
 324 /*
 325  * Stop all threads by closing the thread quit pipe.
 326  */
 327 static
 328 void stop_threads(void)
 329 {
 330         int ret;
 331
 332         /* Stopping all threads */
 333         DBG("Terminating all live threads");
 334         ret = notify_thread_pipe(thread_quit_pipe[1]);
 335         if (ret < 0) {
 336                 ERR("write error on thread quit pipe");
 337         }
 338
 339         /* Dispatch thread */
 340         CMM_STORE_SHARED(live_dispatch_thread_exit, 1);
 341         futex_nto1_wake(&viewer_cmd_queue.futex);
 342 }
 343
 344 /*
 345  * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
 346  */
 347 static
 348 int create_thread_poll_set(struct lttng_poll_event *events, int size)
 349 {
 350         int ret;
 351
 352         if (events == NULL || size == 0) {
 353                 ret = -1;
 354                 goto error;
 355         }
 356
 357         ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
 358         if (ret < 0) {
 359                 goto error;
 360         }
 361
 362         /* Add quit pipe */
 363         ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
 364         if (ret < 0) {
 365                 goto error;
 366         }
 367
 368         return 0;
 369
 370 error:
 371         return ret;
 372 }
 373
 374 /*
 375  * Check if the thread quit pipe was triggered.
 376  *
 377  * Return 1 if it was triggered else 0;
 378  */
 379 static
 380 int check_thread_quit_pipe(int fd, uint32_t events)
 381 {
 382         if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
 383                 return 1;
 384         }
 385
 386         return 0;
 387 }
 388
 389 /*
 390  * Create and init socket from uri.
 391  */
 392 static
 393 struct lttcomm_sock *init_socket(struct lttng_uri *uri)
 394 {
 395         int ret;
 396         struct lttcomm_sock *sock = NULL;
 397
 398         sock = lttcomm_alloc_sock_from_uri(uri);
 399         if (sock == NULL) {
 400                 ERR("Allocating socket");
 401                 goto error;
 402         }
 403
 404         ret = lttcomm_create_sock(sock);
 405         if (ret < 0) {
 406                 goto error;
 407         }
 408         DBG("Listening on sock %d for live", sock->fd);
 409
 410         ret = sock->ops->bind(sock);
 411         if (ret < 0) {
 412                 goto error;
 413         }
 414
 415         ret = sock->ops->listen(sock, -1);
 416         if (ret < 0) {
 417                 goto error;
 418
 419         }
 420
 421         return sock;
 422
 423 error:
 424         if (sock) {
 425                 lttcomm_destroy_sock(sock);
 426         }
 427         return NULL;
 428 }
 429
 430 /*
 431  * This thread manages the listening for new connections on the network
 432  */
 433 static
 434 void *thread_listener(void *data)
 435 {
 436         int i, ret, pollfd, err = -1;
 437         int val = 1;
 438         uint32_t revents, nb_fd;
 439         struct lttng_poll_event events;
 440         struct lttcomm_sock *live_control_sock;
 441
 442         DBG("[thread] Relay live listener started");
 443
 444         health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_LISTENER);
 445
 446         health_code_update();
 447
 448         live_control_sock = init_socket(live_uri);
 449         if (!live_control_sock) {
 450                 goto error_sock_control;
 451         }
 452
 453         /* Pass 2 as size here for the thread quit pipe and control sockets. */
 454         ret = create_thread_poll_set(&events, 2);
 455         if (ret < 0) {
 456                 goto error_create_poll;
 457         }
 458
 459         /* Add the control socket */
 460         ret = lttng_poll_add(&events, live_control_sock->fd, LPOLLIN | LPOLLRDHUP);
 461         if (ret < 0) {
 462                 goto error_poll_add;
 463         }
 464
 465         lttng_relay_notify_ready();
 466
 467         if (testpoint(relayd_thread_live_listener)) {
 468                 goto error_testpoint;
 469         }
 470
 471         while (1) {
 472                 health_code_update();
 473
 474                 DBG("Listener accepting live viewers connections");
 475
 476 restart:
 477                 health_poll_entry();
 478                 ret = lttng_poll_wait(&events, -1);
 479                 health_poll_exit();
 480                 if (ret < 0) {
 481                         /*
 482                          * Restart interrupted system call.
 483                          */
 484                         if (errno == EINTR) {
 485                                 goto restart;
 486                         }
 487                         goto error;
 488                 }
 489                 nb_fd = ret;
 490
 491                 DBG("Relay new viewer connection received");
 492                 for (i = 0; i < nb_fd; i++) {
 493                         health_code_update();
 494
 495                         /* Fetch once the poll data */
 496                         revents = LTTNG_POLL_GETEV(&events, i);
 497                         pollfd = LTTNG_POLL_GETFD(&events, i);
 498
 499                         /* Thread quit pipe has been closed. Killing thread. */
 500                         ret = check_thread_quit_pipe(pollfd, revents);
 501                         if (ret) {
 502                                 err = 0;
 503                                 goto exit;
 504                         }
 505
 506                         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 507                                 ERR("socket poll error");
 508                                 goto error;
 509                         } else if (revents & LPOLLIN) {
 510                                 /*
 511                                  * Get allocated in this thread, enqueued to a global queue,
 512                                  * dequeued and freed in the worker thread.
 513                                  */
 514                                 struct relay_command *relay_cmd;
 515                                 struct lttcomm_sock *newsock;
 516
 517                                 relay_cmd = zmalloc(sizeof(*relay_cmd));
 518                                 if (!relay_cmd) {
 519                                         PERROR("relay command zmalloc");
 520                                         goto error;
 521                                 }
 522
 523                                 assert(pollfd == live_control_sock->fd);
 524                                 newsock = live_control_sock->ops->accept(live_control_sock);
 525                                 if (!newsock) {
 526                                         PERROR("accepting control sock");
 527                                         free(relay_cmd);
 528                                         goto error;
 529                                 }
 530                                 DBG("Relay viewer connection accepted socket %d", newsock->fd);
 531                                 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
 532                                                 sizeof(int));
 533                                 if (ret < 0) {
 534                                         PERROR("setsockopt inet");
 535                                         lttcomm_destroy_sock(newsock);
 536                                         free(relay_cmd);
 537                                         goto error;
 538                                 }
 539                                 relay_cmd->sock = newsock;
 540
 541                                 /*
 542                                  * Lock free enqueue the request.
 543                                  */
 544                                 cds_wfq_enqueue(&viewer_cmd_queue.queue, &relay_cmd->node);
 545
 546                                 /*
 547                                  * Wake the dispatch queue futex. Implicit memory
 548                                  * barrier with the exchange in cds_wfq_enqueue.
 549                                  */
 550                                 futex_nto1_wake(&viewer_cmd_queue.futex);
 551                         }
 552                 }
 553         }
 554
 555 exit:
 556 error:
 557 error_poll_add:
 558 error_testpoint:
 559         lttng_poll_clean(&events);
 560 error_create_poll:
 561         if (live_control_sock->fd >= 0) {
 562                 ret = live_control_sock->ops->close(live_control_sock);
 563                 if (ret) {
 564                         PERROR("close");
 565                 }
 566         }
 567         lttcomm_destroy_sock(live_control_sock);
 568 error_sock_control:
 569         if (err) {
 570                 health_error();
 571                 DBG("Live viewer listener thread exited with error");
 572         }
 573         health_unregister(health_relayd);
 574         DBG("Live viewer listener thread cleanup complete");
 575         stop_threads();
 576         return NULL;
 577 }
 578
 579 /*
 580  * This thread manages the dispatching of the requests to worker threads
 581  */
 582 static
 583 void *thread_dispatcher(void *data)
 584 {
 585         int err = -1;
 586         ssize_t ret;
 587         struct cds_wfq_node *node;
 588         struct relay_command *relay_cmd = NULL;
 589
 590         DBG("[thread] Live viewer relay dispatcher started");
 591
 592         health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_DISPATCHER);
 593
 594         if (testpoint(relayd_thread_live_dispatcher)) {
 595                 goto error_testpoint;
 596         }
 597
 598         health_code_update();
 599
 600         while (!CMM_LOAD_SHARED(live_dispatch_thread_exit)) {
 601                 health_code_update();
 602
 603                 /* Atomically prepare the queue futex */
 604                 futex_nto1_prepare(&viewer_cmd_queue.futex);
 605
 606                 do {
 607                         health_code_update();
 608
 609                         /* Dequeue commands */
 610                         node = cds_wfq_dequeue_blocking(&viewer_cmd_queue.queue);
 611                         if (node == NULL) {
 612                                 DBG("Woken up but nothing in the live-viewer "
 613                                                 "relay command queue");
 614                                 /* Continue thread execution */
 615                                 break;
 616                         }
 617
 618                         relay_cmd = caa_container_of(node, struct relay_command, node);
 619                         DBG("Dispatching viewer request waiting on sock %d",
 620                                         relay_cmd->sock->fd);
 621
 622                         /*
 623                          * Inform worker thread of the new request. This call is blocking
 624                          * so we can be assured that the data will be read at some point in
 625                          * time or wait to the end of the world :)
 626                          */
 627                         ret = lttng_write(live_relay_cmd_pipe[1], relay_cmd,
 628                                         sizeof(*relay_cmd));
 629                         free(relay_cmd);
 630                         if (ret < sizeof(struct relay_command)) {
 631                                 PERROR("write cmd pipe");
 632                                 goto error;
 633                         }
 634                 } while (node != NULL);
 635
 636                 /* Futex wait on queue. Blocking call on futex() */
 637                 health_poll_entry();
 638                 futex_nto1_wait(&viewer_cmd_queue.futex);
 639                 health_poll_exit();
 640         }
 641
 642         /* Normal exit, no error */
 643         err = 0;
 644
 645 error:
 646 error_testpoint:
 647         if (err) {
 648                 health_error();
 649                 ERR("Health error occurred in %s", __func__);
 650         }
 651         health_unregister(health_relayd);
 652         DBG("Live viewer dispatch thread dying");
 653         stop_threads();
 654         return NULL;
 655 }
 656
 657 /*
 658  * Establish connection with the viewer and check the versions.
 659  *
 660  * Return 0 on success or else negative value.
 661  */
 662 static
 663 int viewer_connect(struct relay_command *cmd)
 664 {
 665         int ret;
 666         struct lttng_viewer_connect reply, msg;
 667
 668         assert(cmd);
 669
 670         cmd->version_check_done = 1;
 671
 672         health_code_update();
 673
 674         DBG("Viewer is establishing a connection to the relayd.");
 675
 676         ret = recv_request(cmd->sock, &msg, sizeof(msg));
 677         if (ret < 0) {
 678                 goto end;
 679         }
 680
 681         health_code_update();
 682
 683         reply.major = RELAYD_VERSION_COMM_MAJOR;
 684         reply.minor = RELAYD_VERSION_COMM_MINOR;
 685
 686         /* Major versions must be the same */
 687         if (reply.major != be32toh(msg.major)) {
 688                 DBG("Incompatible major versions ([relayd] %u vs [client] %u)",
 689                                 reply.major, be32toh(msg.major));
 690                 ret = -1;
 691                 goto end;
 692         }
 693
 694         cmd->major = reply.major;
 695         /* We adapt to the lowest compatible version */
 696         if (reply.minor <= be32toh(msg.minor)) {
 697                 cmd->minor = reply.minor;
 698         } else {
 699                 cmd->minor = be32toh(msg.minor);
 700         }
 701
 702         if (be32toh(msg.type) == VIEWER_CLIENT_COMMAND) {
 703                 cmd->type = RELAY_VIEWER_COMMAND;
 704         } else if (be32toh(msg.type) == VIEWER_CLIENT_NOTIFICATION) {
 705                 cmd->type = RELAY_VIEWER_NOTIFICATION;
 706         } else {
 707                 ERR("Unknown connection type : %u", be32toh(msg.type));
 708                 ret = -1;
 709                 goto end;
 710         }
 711
 712         reply.major = htobe32(reply.major);
 713         reply.minor = htobe32(reply.minor);
 714         if (cmd->type == RELAY_VIEWER_COMMAND) {
 715                 reply.viewer_session_id = htobe64(++last_relay_viewer_session_id);
 716         }
 717
 718         health_code_update();
 719
 720
 721         ret = send_response(cmd->sock, &reply, sizeof(reply));
 722         if (ret < 0) {
 723                 goto end;
 724         }
 725
 726         health_code_update();
 727
 728         DBG("Version check done using protocol %u.%u", cmd->major, cmd->minor);
 729         ret = 0;
 730
 731 end:
 732         return ret;
 733 }
 734
 735 /*
 736  * Send the viewer the list of current sessions.
 737  *
 738  * Return 0 on success or else a negative value.
 739  */
 740 static
 741 int viewer_list_sessions(struct relay_command *cmd,
 742                 struct lttng_ht *sessions_ht)
 743 {
 744         int ret;
 745         struct lttng_viewer_list_sessions session_list;
 746         unsigned long count;
 747         long approx_before, approx_after;
 748         struct lttng_ht_iter iter;
 749         struct lttng_viewer_session send_session;
 750         struct relay_session *session;
 751
 752         DBG("List sessions received");
 753
 754         rcu_read_lock();
 755         cds_lfht_count_nodes(sessions_ht->ht, &approx_before, &count, &approx_after);
 756         session_list.sessions_count = htobe32(count);
 757
 758         health_code_update();
 759
 760         ret = send_response(cmd->sock, &session_list, sizeof(session_list));
 761         if (ret < 0) {
 762                 goto end_unlock;
 763         }
 764
 765         health_code_update();
 766
 767         cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, session,
 768                         session_n.node) {
 769                 health_code_update();
 770
 771                 strncpy(send_session.session_name, session->session_name,
 772                                 sizeof(send_session.session_name));
 773                 strncpy(send_session.hostname, session->hostname,
 774                                 sizeof(send_session.hostname));
 775                 send_session.id = htobe64(session->id);
 776                 send_session.live_timer = htobe32(session->live_timer);
 777                 send_session.clients = htobe32(session->viewer_attached);
 778                 send_session.streams = htobe32(session->stream_count);
 779
 780                 health_code_update();
 781
 782                 ret = send_response(cmd->sock, &send_session, sizeof(send_session));
 783                 if (ret < 0) {
 784                         goto end_unlock;
 785                 }
 786         }
 787         health_code_update();
 788
 789         rcu_read_unlock();
 790         ret = 0;
 791         goto end;
 792
 793 end_unlock:
 794         rcu_read_unlock();
 795
 796 end:
 797         return ret;
 798 }
 799
 800 /*
 801  * Send the viewer the list of current sessions.
 802  */
 803 static
 804 int viewer_get_new_streams(struct relay_command *cmd,
 805                 struct lttng_ht *sessions_ht)
 806 {
 807         int ret, send_streams = 0;
 808         uint32_t nb_created = 0, nb_unsent = 0, nb_streams = 0;
 809         struct lttng_viewer_new_streams_request request;
 810         struct lttng_viewer_new_streams_response response;
 811         struct relay_session *session;
 812
 813         assert(cmd);
 814         assert(sessions_ht);
 815
 816         DBG("Get new streams received");
 817
 818         health_code_update();
 819
 820         /* Receive the request from the connected client. */
 821         ret = recv_request(cmd->sock, &request, sizeof(request));
 822         if (ret < 0) {
 823                 goto error;
 824         }
 825
 826         health_code_update();
 827
 828         rcu_read_lock();
 829         session = session_find_by_id(sessions_ht, be64toh(request.session_id));
 830         if (!session) {
 831                 DBG("Relay session %" PRIu64 " not found",
 832                                 be64toh(request.session_id));
 833                 response.status = htobe32(VIEWER_NEW_STREAMS_ERR);
 834                 goto send_reply;
 835         }
 836
 837         if (cmd->session_id == session->id) {
 838                 /* We confirmed the viewer is asking for the same session. */
 839                 send_streams = 1;
 840                 response.status = htobe32(VIEWER_NEW_STREAMS_OK);
 841         } else {
 842                 send_streams = 0;
 843                 response.status = htobe32(VIEWER_NEW_STREAMS_ERR);
 844                 goto send_reply;
 845         }
 846
 847         if (!send_streams) {
 848                 goto send_reply;
 849         }
 850
 851         ret = make_viewer_streams(session, VIEWER_SEEK_LAST, NULL, &nb_unsent,
 852                         &nb_created);
 853         if (ret < 0) {
 854                 goto end_unlock;
 855         }
 856         /* Only send back the newly created streams with the unsent ones. */
 857         nb_streams = nb_created + nb_unsent;
 858         response.streams_count = htobe32(nb_streams);
 859
 860 send_reply:
 861         health_code_update();
 862         ret = send_response(cmd->sock, &response, sizeof(response));
 863         if (ret < 0) {
 864                 goto end_unlock;
 865         }
 866         health_code_update();
 867
 868         /*
 869          * Unknown or empty session, just return gracefully, the viewer knows what
 870          * is happening.
 871          */
 872         if (!send_streams || !nb_streams) {
 873                 ret = 0;
 874                 goto end_unlock;
 875         }
 876
 877         /*
 878          * Send stream and *DON'T* ignore the sent flag so every viewer streams
 879          * that were not sent from that point will be sent to the viewer.
 880          */
 881         ret = send_viewer_streams(cmd->sock, session, 0);
 882         if (ret < 0) {
 883                 goto end_unlock;
 884         }
 885
 886 end_unlock:
 887         rcu_read_unlock();
 888 error:
 889         return ret;
 890 }
 891
 892 /*
 893  * Send the viewer the list of current sessions.
 894  */
 895 static
 896 int viewer_attach_session(struct relay_command *cmd,
 897                 struct lttng_ht *sessions_ht)
 898 {
 899         int send_streams = 0;
 900         ssize_t ret;
 901         uint32_t nb_streams = 0;
 902         enum lttng_viewer_seek seek_type;
 903         struct lttng_viewer_attach_session_request request;
 904         struct lttng_viewer_attach_session_response response;
 905         struct relay_session *session;
 906
 907         assert(cmd);
 908         assert(sessions_ht);
 909
 910         DBG("Attach session received");
 911
 912         health_code_update();
 913
 914         /* Receive the request from the connected client. */
 915         ret = recv_request(cmd->sock, &request, sizeof(request));
 916         if (ret < 0) {
 917                 goto error;
 918         }
 919
 920         health_code_update();
 921
 922         rcu_read_lock();
 923         session = session_find_by_id(sessions_ht, be64toh(request.session_id));
 924         if (!session) {
 925                 DBG("Relay session %" PRIu64 " not found",
 926                                 be64toh(request.session_id));
 927                 response.status = htobe32(VIEWER_ATTACH_UNK);
 928                 goto send_reply;
 929         }
 930
 931         if (cmd->session_id == session->id) {
 932                 /* Same viewer already attached, just send the stream list. */
 933                 send_streams = 1;
 934                 response.status = htobe32(VIEWER_ATTACH_OK);
 935         } else if (session->viewer_attached != 0) {
 936                 DBG("Already a viewer attached");
 937                 response.status = htobe32(VIEWER_ATTACH_ALREADY);
 938                 goto send_reply;
 939         } else if (session->live_timer == 0) {
 940                 DBG("Not live session");
 941                 response.status = htobe32(VIEWER_ATTACH_NOT_LIVE);
 942                 goto send_reply;
 943         } else {
 944                 session->viewer_attached++;
 945                 send_streams = 1;
 946                 response.status = htobe32(VIEWER_ATTACH_OK);
 947                 cmd->session_id = session->id;
 948                 cmd->session = session;
 949         }
 950
 951         switch (be32toh(request.seek)) {
 952         case VIEWER_SEEK_BEGINNING:
 953         case VIEWER_SEEK_LAST:
 954                 seek_type = be32toh(request.seek);
 955                 break;
 956         default:
 957                 ERR("Wrong seek parameter");
 958                 response.status = htobe32(VIEWER_ATTACH_SEEK_ERR);
 959                 send_streams = 0;
 960                 goto send_reply;
 961         }
 962
 963         if (!send_streams) {
 964                 goto send_reply;
 965         }
 966
 967         ret = make_viewer_streams(session, seek_type, &nb_streams, NULL, NULL);
 968         if (ret < 0) {
 969                 goto end_unlock;
 970         }
 971         response.streams_count = htobe32(nb_streams);
 972
 973 send_reply:
 974         health_code_update();
 975         ret = send_response(cmd->sock, &response, sizeof(response));
 976         if (ret < 0) {
 977                 goto end_unlock;
 978         }
 979         health_code_update();
 980
 981         /*
 982          * Unknown or empty session, just return gracefully, the viewer knows what
 983          * is happening.
 984          */
 985         if (!send_streams || !nb_streams) {
 986                 ret = 0;
 987                 goto end_unlock;
 988         }
 989
 990         /* Send stream and ignore the sent flag. */
 991         ret = send_viewer_streams(cmd->sock, session, 1);
 992         if (ret < 0) {
 993                 goto end_unlock;
 994         }
 995
 996 end_unlock:
 997         rcu_read_unlock();
 998 error:
 999         return ret;
1000 }
1001
1002 /*
1003  * Send the next index for a stream.
1004  *
1005  * Return 0 on success or else a negative value.
1006  */
1007 static
1008 int viewer_get_next_index(struct relay_command *cmd,
1009                 struct lttng_ht *sessions_ht)
1010 {
1011         int ret;
1012         struct lttng_viewer_get_next_index request_index;
1013         struct lttng_viewer_index viewer_index;
1014         struct ctf_packet_index packet_index;
1015         struct relay_viewer_stream *vstream;
1016         struct relay_stream *rstream;
1017
1018         assert(cmd);
1019         assert(sessions_ht);
1020
1021         DBG("Viewer get next index");
1022
1023         health_code_update();
1024
1025         ret = recv_request(cmd->sock, &request_index, sizeof(request_index));
1026         if (ret < 0) {
1027                 goto end;
1028         }
1029         health_code_update();
1030
1031         rcu_read_lock();
1032         vstream = viewer_stream_find_by_id(be64toh(request_index.stream_id));
1033         if (!vstream) {
1034                 ret = -1;
1035                 goto end_unlock;
1036         }
1037
1038         memset(&viewer_index, 0, sizeof(viewer_index));
1039
1040         /*
1041          * The viewer should not ask for index on metadata stream.
1042          */
1043         if (vstream->metadata_flag) {
1044                 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1045                 goto send_reply;
1046         }
1047
1048         /* First time, we open the index file */
1049         if (vstream->index_read_fd < 0) {
1050                 ret = index_open(vstream->path_name, vstream->channel_name,
1051                                 vstream->tracefile_count, vstream->tracefile_count_current);
1052                 if (ret == -ENOENT) {
1053                         /*
1054                          * The index is created only when the first data packet arrives, it
1055                          * might not be ready at the beginning of the session
1056                          */
1057                         viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1058                         goto send_reply;
1059                 } else if (ret < 0) {
1060                         viewer_index.status = htobe32(VIEWER_INDEX_ERR);
1061                         goto send_reply;
1062                 }
1063                 vstream->index_read_fd = ret;
1064         }
1065
1066         rstream = relay_stream_find_by_id(vstream->stream_handle);
1067         if (rstream) {
1068                 if (vstream->abort_flag) {
1069                         /* Rotate on abort (overwrite). */
1070                         DBG("Viewer rotate because of overwrite");
1071                         ret = viewer_stream_rotate(vstream, rstream);
1072                         if (ret < 0) {
1073                                 goto end_unlock;
1074                         } else if (ret == 1) {
1075                                 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1076                                 viewer_stream_delete(vstream);
1077                                 viewer_stream_destroy(vstream);
1078                                 goto send_reply;
1079                         }
1080                 }
1081                 pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
1082                 if (rstream->tracefile_count_current == vstream->tracefile_count_current) {
1083                         if (rstream->beacon_ts_end != -1ULL &&
1084                                 vstream->last_sent_index == rstream->total_index_received) {
1085                                 viewer_index.status = htobe32(VIEWER_INDEX_INACTIVE);
1086                                 viewer_index.timestamp_end = htobe64(rstream->beacon_ts_end);
1087                                 pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
1088                                 goto send_reply;
1089                         /*
1090                          * Reader and writer are working in the same tracefile, so we care
1091                          * about the number of index received and sent. Otherwise, we read
1092                          * up to EOF.
1093                          */
1094                         } else if (rstream->total_index_received <= vstream->last_sent_index
1095                                         && !vstream->close_write_flag) {
1096                                 pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
1097                                 /* No new index to send, retry later. */
1098                                 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1099                                 goto send_reply;
1100                         }
1101                 }
1102                 pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
1103         } else if (!rstream && vstream->close_write_flag &&
1104                         vstream->total_index_received == vstream->last_sent_index) {
1105                 /* Last index sent and current tracefile closed in write */
1106                 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1107                 viewer_stream_delete(vstream);
1108                 viewer_stream_destroy(vstream);
1109                 goto send_reply;
1110         } else {
1111                 vstream->close_write_flag = 1;
1112         }
1113
1114         if (!vstream->ctf_trace->metadata_received ||
1115                         vstream->ctf_trace->metadata_received >
1116                         vstream->ctf_trace->metadata_sent) {
1117                 viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
1118         }
1119
1120         ret = check_new_streams(vstream->session_id, sessions_ht);
1121         if (ret < 0) {
1122                 goto end_unlock;
1123         } else if (ret == 1) {
1124                 viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM;
1125         }
1126
1127         pthread_mutex_lock(&vstream->overwrite_lock);
1128         if (vstream->abort_flag) {
1129                 /*
1130                  * The file is being overwritten by the writer, we cannot
1131                  * use it.
1132                  */
1133                 viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1134                 pthread_mutex_unlock(&vstream->overwrite_lock);
1135                 ret = viewer_stream_rotate(vstream, rstream);
1136                 if (ret < 0) {
1137                         goto end_unlock;
1138                 } else if (ret == 1) {
1139                         viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1140                         viewer_stream_delete(vstream);
1141                         viewer_stream_destroy(vstream);
1142                         goto send_reply;
1143                 }
1144                 goto send_reply;
1145         }
1146
1147         ret = lttng_read(vstream->index_read_fd, &packet_index,
1148                         sizeof(packet_index));
1149         pthread_mutex_unlock(&vstream->overwrite_lock);
1150         if (ret < sizeof(packet_index)) {
1151                 /*
1152                  * The tracefile is closed in write, so we read up to EOF.
1153                  */
1154                 if (vstream->close_write_flag == 1) {
1155                         viewer_index.status = htobe32(VIEWER_INDEX_RETRY);
1156                         /* Rotate on normal EOF */
1157                         ret = viewer_stream_rotate(vstream, rstream);
1158                         if (ret < 0) {
1159                                 goto end_unlock;
1160                         } else if (ret == 1) {
1161                                 viewer_index.status = htobe32(VIEWER_INDEX_HUP);
1162                                 viewer_stream_delete(vstream);
1163                                 viewer_stream_destroy(vstream);
1164                                 goto send_reply;
1165                         }
1166                 } else {
1167                         PERROR("Relay reading index file %d", vstream->index_read_fd);
1168                         viewer_index.status = htobe32(VIEWER_INDEX_ERR);
1169                 }
1170                 goto send_reply;
1171         } else {
1172                 viewer_index.status = htobe32(VIEWER_INDEX_OK);
1173                 vstream->last_sent_index++;
1174         }
1175
1176         /*
1177          * Indexes are stored in big endian, no need to switch before sending.
1178          */
1179         viewer_index.offset = packet_index.offset;
1180         viewer_index.packet_size = packet_index.packet_size;
1181         viewer_index.content_size = packet_index.content_size;
1182         viewer_index.timestamp_begin = packet_index.timestamp_begin;
1183         viewer_index.timestamp_end = packet_index.timestamp_end;
1184         viewer_index.events_discarded = packet_index.events_discarded;
1185         viewer_index.stream_id = packet_index.stream_id;
1186
1187 send_reply:
1188         viewer_index.flags = htobe32(viewer_index.flags);
1189         health_code_update();
1190
1191         ret = send_response(cmd->sock, &viewer_index, sizeof(viewer_index));
1192         if (ret < 0) {
1193                 goto end_unlock;
1194         }
1195         health_code_update();
1196
1197         DBG("Index %" PRIu64 " for stream %" PRIu64 " sent",
1198                         vstream->last_sent_index, vstream->stream_handle);
1199
1200 end_unlock:
1201         rcu_read_unlock();
1202
1203 end:
1204         return ret;
1205 }
1206
1207 /*
1208  * Send the next index for a stream
1209  *
1210  * Return 0 on success or else a negative value.
1211  */
1212 static
1213 int viewer_get_packet(struct relay_command *cmd,
1214                 struct lttng_ht *sessions_ht)
1215 {
1216         int ret, send_data = 0;
1217         char *data = NULL;
1218         uint32_t len = 0;
1219         ssize_t read_len;
1220         struct lttng_viewer_get_packet get_packet_info;
1221         struct lttng_viewer_trace_packet reply;
1222         struct relay_viewer_stream *stream;
1223
1224         assert(cmd);
1225
1226         DBG2("Relay get data packet");
1227
1228         health_code_update();
1229
1230         ret = recv_request(cmd->sock, &get_packet_info, sizeof(get_packet_info));
1231         if (ret < 0) {
1232                 goto end;
1233         }
1234         health_code_update();
1235
1236         /* From this point on, the error label can be reached. */
1237         memset(&reply, 0, sizeof(reply));
1238
1239         rcu_read_lock();
1240         stream = viewer_stream_find_by_id(be64toh(get_packet_info.stream_id));
1241         if (!stream) {
1242                 goto error;
1243         }
1244         assert(stream->ctf_trace);
1245
1246         /*
1247          * First time we read this stream, we need open the tracefile, we should
1248          * only arrive here if an index has already been sent to the viewer, so the
1249          * tracefile must exist, if it does not it is a fatal error.
1250          */
1251         if (stream->read_fd < 0) {
1252                 char fullpath[PATH_MAX];
1253
1254                 if (stream->tracefile_count > 0) {
1255                         ret = snprintf(fullpath, PATH_MAX, "%s/%s_%" PRIu64, stream->path_name,
1256                                         stream->channel_name,
1257                                         stream->tracefile_count_current);
1258                 } else {
1259                         ret = snprintf(fullpath, PATH_MAX, "%s/%s", stream->path_name,
1260                                         stream->channel_name);
1261                 }
1262                 if (ret < 0) {
1263                         goto error;
1264                 }
1265                 ret = open(fullpath, O_RDONLY);
1266                 if (ret < 0) {
1267                         PERROR("Relay opening trace file");
1268                         goto error;
1269                 }
1270                 stream->read_fd = ret;
1271         }
1272
1273         if (!stream->ctf_trace->metadata_received ||
1274                         stream->ctf_trace->metadata_received >
1275                         stream->ctf_trace->metadata_sent) {
1276                 reply.status = htobe32(VIEWER_GET_PACKET_ERR);
1277                 reply.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
1278                 goto send_reply;
1279         }
1280
1281         ret = check_new_streams(stream->session_id, sessions_ht);
1282         if (ret < 0) {
1283                 goto end_unlock;
1284         } else if (ret == 1) {
1285                 reply.status = htobe32(VIEWER_GET_PACKET_ERR);
1286                 reply.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM;
1287                 goto send_reply;
1288         }
1289
1290         len = be32toh(get_packet_info.len);
1291         data = zmalloc(len);
1292         if (!data) {
1293                 PERROR("relay data zmalloc");
1294                 goto error;
1295         }
1296
1297         ret = lseek(stream->read_fd, be64toh(get_packet_info.offset), SEEK_SET);
1298         if (ret < 0) {
1299                 /*
1300                  * If the read fd was closed by the streaming side, the
1301                  * abort_flag will be set to 1, otherwise it is an error.
1302                  */
1303                 if (stream->abort_flag == 0) {
1304                         PERROR("lseek");
1305                         goto error;
1306                 }
1307                 reply.status = htobe32(VIEWER_GET_PACKET_EOF);
1308                 goto send_reply;
1309         }
1310         read_len = lttng_read(stream->read_fd, data, len);
1311         if (read_len < len) {
1312                 /*
1313                  * If the read fd was closed by the streaming side, the
1314                  * abort_flag will be set to 1, otherwise it is an error.
1315                  */
1316                 if (stream->abort_flag == 0) {
1317                         PERROR("Relay reading trace file, fd: %d, offset: %" PRIu64,
1318                                         stream->read_fd,
1319                                         be64toh(get_packet_info.offset));
1320                         goto error;
1321                 } else {
1322                         reply.status = htobe32(VIEWER_GET_PACKET_EOF);
1323                         goto send_reply;
1324                 }
1325         }
1326         reply.status = htobe32(VIEWER_GET_PACKET_OK);
1327         reply.len = htobe32(len);
1328         send_data = 1;
1329         goto send_reply;
1330
1331 error:
1332         reply.status = htobe32(VIEWER_GET_PACKET_ERR);
1333
1334 send_reply:
1335         reply.flags = htobe32(reply.flags);
1336
1337         health_code_update();
1338
1339         ret = send_response(cmd->sock, &reply, sizeof(reply));
1340         if (ret < 0) {
1341                 goto end_unlock;
1342         }
1343         health_code_update();
1344
1345         if (send_data) {
1346                 health_code_update();
1347                 ret = send_response(cmd->sock, data, len);
1348                 if (ret < 0) {
1349                         goto end_unlock;
1350                 }
1351                 health_code_update();
1352         }
1353
1354         DBG("Sent %u bytes for stream %" PRIu64, len,
1355                         be64toh(get_packet_info.stream_id));
1356
1357 end_unlock:
1358         free(data);
1359         rcu_read_unlock();
1360
1361 end:
1362         return ret;
1363 }
1364
1365 /*
1366  * Send the session's metadata
1367  *
1368  * Return 0 on success else a negative value.
1369  */
1370 static
1371 int viewer_get_metadata(struct relay_command *cmd)
1372 {
1373         int ret = 0;
1374         ssize_t read_len;
1375         uint64_t len = 0;
1376         char *data = NULL;
1377         struct lttng_viewer_get_metadata request;
1378         struct lttng_viewer_metadata_packet reply;
1379         struct relay_viewer_stream *stream;
1380
1381         assert(cmd);
1382
1383         DBG("Relay get metadata");
1384
1385         health_code_update();
1386
1387         ret = recv_request(cmd->sock, &request, sizeof(request));
1388         if (ret < 0) {
1389                 goto end;
1390         }
1391         health_code_update();
1392
1393         rcu_read_lock();
1394         stream = viewer_stream_find_by_id(be64toh(request.stream_id));
1395         if (!stream || !stream->metadata_flag) {
1396                 ERR("Invalid metadata stream");
1397                 goto error;
1398         }
1399         assert(stream->ctf_trace);
1400         assert(stream->ctf_trace->metadata_sent <=
1401                         stream->ctf_trace->metadata_received);
1402
1403         len = stream->ctf_trace->metadata_received -
1404                 stream->ctf_trace->metadata_sent;
1405         if (len == 0) {
1406                 reply.status = htobe32(VIEWER_NO_NEW_METADATA);
1407                 goto send_reply;
1408         }
1409
1410         /* first time, we open the metadata file */
1411         if (stream->read_fd < 0) {
1412                 char fullpath[PATH_MAX];
1413
1414                 ret = snprintf(fullpath, PATH_MAX, "%s/%s", stream->path_name,
1415                                 stream->channel_name);
1416                 if (ret < 0) {
1417                         goto error;
1418                 }
1419                 ret = open(fullpath, O_RDONLY);
1420                 if (ret < 0) {
1421                         PERROR("Relay opening metadata file");
1422                         goto error;
1423                 }
1424                 stream->read_fd = ret;
1425         }
1426
1427         reply.len = htobe64(len);
1428         data = zmalloc(len);
1429         if (!data) {
1430                 PERROR("viewer metadata zmalloc");
1431                 goto error;
1432         }
1433
1434         read_len = lttng_read(stream->read_fd, data, len);
1435         if (read_len < len) {
1436                 PERROR("Relay reading metadata file");
1437                 goto error;
1438         }
1439         stream->ctf_trace->metadata_sent += read_len;
1440         reply.status = htobe32(VIEWER_METADATA_OK);
1441         goto send_reply;
1442
1443 error:
1444         reply.status = htobe32(VIEWER_METADATA_ERR);
1445
1446 send_reply:
1447         health_code_update();
1448         ret = send_response(cmd->sock, &reply, sizeof(reply));
1449         if (ret < 0) {
1450                 goto end_unlock;
1451         }
1452         health_code_update();
1453
1454         if (len > 0) {
1455                 ret = send_response(cmd->sock, data, len);
1456                 if (ret < 0) {
1457                         goto end_unlock;
1458                 }
1459         }
1460
1461         DBG("Sent %" PRIu64 " bytes of metadata for stream %" PRIu64, len,
1462                         be64toh(request.stream_id));
1463
1464         DBG("Metadata sent");
1465
1466 end_unlock:
1467         free(data);
1468         rcu_read_unlock();
1469 end:
1470         return ret;
1471 }
1472
1473 /*
1474  * live_relay_unknown_command: send -1 if received unknown command
1475  */
1476 static
1477 void live_relay_unknown_command(struct relay_command *cmd)
1478 {
1479         struct lttcomm_relayd_generic_reply reply;
1480
1481         reply.ret_code = htobe32(LTTNG_ERR_UNK);
1482         (void) send_response(cmd->sock, &reply, sizeof(reply));
1483 }
1484
1485 /*
1486  * Process the commands received on the control socket
1487  */
1488 static
1489 int process_control(struct lttng_viewer_cmd *recv_hdr,
1490                 struct relay_command *cmd, struct lttng_ht *sessions_ht)
1491 {
1492         int ret = 0;
1493         uint32_t msg_value;
1494
1495         assert(recv_hdr);
1496         assert(cmd);
1497         assert(sessions_ht);
1498
1499         msg_value = be32toh(recv_hdr->cmd);
1500
1501         /*
1502          * Make sure we've done the version check before any command other then a
1503          * new client connection.
1504          */
1505         if (msg_value != VIEWER_CONNECT && !cmd->version_check_done) {
1506                 ERR("Viewer cmd value %" PRIu32 " before version check", msg_value);
1507                 ret = -1;
1508                 goto end;
1509         }
1510
1511         switch (msg_value) {
1512         case VIEWER_CONNECT:
1513                 ret = viewer_connect(cmd);
1514                 break;
1515         case VIEWER_LIST_SESSIONS:
1516                 ret = viewer_list_sessions(cmd, sessions_ht);
1517                 break;
1518         case VIEWER_ATTACH_SESSION:
1519                 ret = viewer_attach_session(cmd, sessions_ht);
1520                 break;
1521         case VIEWER_GET_NEXT_INDEX:
1522                 ret = viewer_get_next_index(cmd, sessions_ht);
1523                 break;
1524         case VIEWER_GET_PACKET:
1525                 ret = viewer_get_packet(cmd, sessions_ht);
1526                 break;
1527         case VIEWER_GET_METADATA:
1528                 ret = viewer_get_metadata(cmd);
1529                 break;
1530         case VIEWER_GET_NEW_STREAMS:
1531                 ret = viewer_get_new_streams(cmd, sessions_ht);
1532                 break;
1533         default:
1534                 ERR("Received unknown viewer command (%u)", be32toh(recv_hdr->cmd));
1535                 live_relay_unknown_command(cmd);
1536                 ret = -1;
1537                 goto end;
1538         }
1539
1540 end:
1541         return ret;
1542 }
1543
1544 static
1545 void cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1546 {
1547         int ret;
1548
1549         assert(events);
1550
1551         lttng_poll_del(events, pollfd);
1552
1553         ret = close(pollfd);
1554         if (ret < 0) {
1555                 ERR("Closing pollfd %d", pollfd);
1556         }
1557 }
1558
1559 /*
1560  * Create and add connection to the given hash table.
1561  *
1562  * Return poll add value or else -1 on error.
1563  */
1564 static
1565 int add_connection(int fd, struct lttng_poll_event *events,
1566                 struct lttng_ht *relay_connections_ht)
1567 {
1568         int ret;
1569         struct relay_command *relay_connection;
1570
1571         assert(events);
1572         assert(relay_connections_ht);
1573
1574         relay_connection = zmalloc(sizeof(struct relay_command));
1575         if (relay_connection == NULL) {
1576                 PERROR("Relay command zmalloc");
1577                 goto error;
1578         }
1579
1580         ret = lttng_read(fd, relay_connection, sizeof(*relay_connection));
1581         if (ret < sizeof(*relay_connection)) {
1582                 PERROR("read relay cmd pipe");
1583                 goto error_read;
1584         }
1585
1586         lttng_ht_node_init_ulong(&relay_connection->sock_n,
1587                         (unsigned long) relay_connection->sock->fd);
1588         rcu_read_lock();
1589         lttng_ht_add_unique_ulong(relay_connections_ht,
1590                         &relay_connection->sock_n);
1591         rcu_read_unlock();
1592
1593         return lttng_poll_add(events, relay_connection->sock->fd,
1594                         LPOLLIN | LPOLLRDHUP);
1595
1596 error_read:
1597         free(relay_connection);
1598 error:
1599         return -1;
1600 }
1601
1602 static
1603 void deferred_free_connection(struct rcu_head *head)
1604 {
1605         struct relay_command *relay_connection =
1606                 caa_container_of(head, struct relay_command, rcu_node);
1607
1608         if (relay_connection->session &&
1609                         relay_connection->session->viewer_attached > 0) {
1610                 relay_connection->session->viewer_attached--;
1611         }
1612         lttcomm_destroy_sock(relay_connection->sock);
1613         free(relay_connection);
1614 }
1615
1616 /*
1617  * Delete all streams for a specific session ID.
1618  */
1619 static
1620 void viewer_del_streams(uint64_t session_id)
1621 {
1622         struct relay_viewer_stream *stream;
1623         struct lttng_ht_iter iter;
1624
1625         rcu_read_lock();
1626         cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream,
1627                         stream_n.node) {
1628                 health_code_update();
1629
1630                 if (stream->session_id != session_id) {
1631                         continue;
1632                 }
1633
1634                 viewer_stream_delete(stream);
1635                 assert(stream->ctf_trace);
1636
1637                 if (stream->metadata_flag) {
1638                         /*
1639                          * The metadata viewer stream is destroyed once the refcount on the
1640                          * ctf trace goes to 0 in the destroy stream function thus there is
1641                          * no explicit call to that function here.
1642                          */
1643                         stream->ctf_trace->metadata_sent = 0;
1644                         stream->ctf_trace->viewer_metadata_stream = NULL;
1645                 } else {
1646                         viewer_stream_destroy(stream);
1647                 }
1648         }
1649         rcu_read_unlock();
1650 }
1651
1652 /*
1653  * Delete and free a connection.
1654  *
1655  * RCU read side lock MUST be acquired.
1656  */
1657 static
1658 void del_connection(struct lttng_ht *relay_connections_ht,
1659                 struct lttng_ht_iter *iter, struct relay_command *relay_connection)
1660 {
1661         int ret;
1662
1663         assert(relay_connections_ht);
1664         assert(iter);
1665         assert(relay_connection);
1666
1667         DBG("Cleaning connection of session ID %" PRIu64,
1668                         relay_connection->session_id);
1669
1670         ret = lttng_ht_del(relay_connections_ht, iter);
1671         assert(!ret);
1672
1673         viewer_del_streams(relay_connection->session_id);
1674
1675         call_rcu(&relay_connection->rcu_node, deferred_free_connection);
1676 }
1677
1678 /*
1679  * This thread does the actual work
1680  */
1681 static
1682 void *thread_worker(void *data)
1683 {
1684         int ret, err = -1;
1685         uint32_t nb_fd;
1686         struct relay_command *relay_connection;
1687         struct lttng_poll_event events;
1688         struct lttng_ht *relay_connections_ht;
1689         struct lttng_ht_node_ulong *node;
1690         struct lttng_ht_iter iter;
1691         struct lttng_viewer_cmd recv_hdr;
1692         struct relay_local_data *relay_ctx = (struct relay_local_data *) data;
1693         struct lttng_ht *sessions_ht = relay_ctx->sessions_ht;
1694
1695         DBG("[thread] Live viewer relay worker started");
1696
1697         rcu_register_thread();
1698
1699         health_register(health_relayd, HEALTH_RELAYD_TYPE_LIVE_WORKER);
1700
1701         if (testpoint(relayd_thread_live_worker)) {
1702                 goto error_testpoint;
1703         }
1704
1705         /* table of connections indexed on socket */
1706         relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1707         if (!relay_connections_ht) {
1708                 goto relay_connections_ht_error;
1709         }
1710
1711         ret = create_thread_poll_set(&events, 2);
1712         if (ret < 0) {
1713                 goto error_poll_create;
1714         }
1715
1716         ret = lttng_poll_add(&events, live_relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1717         if (ret < 0) {
1718                 goto error;
1719         }
1720
1721 restart:
1722         while (1) {
1723                 int i;
1724
1725                 health_code_update();
1726
1727                 /* Infinite blocking call, waiting for transmission */
1728                 DBG3("Relayd live viewer worker thread polling...");
1729                 health_poll_entry();
1730                 ret = lttng_poll_wait(&events, -1);
1731                 health_poll_exit();
1732                 if (ret < 0) {
1733                         /*
1734                          * Restart interrupted system call.
1735                          */
1736                         if (errno == EINTR) {
1737                                 goto restart;
1738                         }
1739                         goto error;
1740                 }
1741
1742                 nb_fd = ret;
1743
1744                 /*
1745                  * Process control. The control connection is prioritised so we don't
1746                  * starve it with high throughput tracing data on the data
1747                  * connection.
1748                  */
1749                 for (i = 0; i < nb_fd; i++) {
1750                         /* Fetch once the poll data */
1751                         uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1752                         int pollfd = LTTNG_POLL_GETFD(&events, i);
1753
1754                         health_code_update();
1755
1756                         /* Thread quit pipe has been closed. Killing thread. */
1757                         ret = check_thread_quit_pipe(pollfd, revents);
1758                         if (ret) {
1759                                 err = 0;
1760                                 goto exit;
1761                         }
1762
1763                         /* Inspect the relay cmd pipe for new connection */
1764                         if (pollfd == live_relay_cmd_pipe[0]) {
1765                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1766                                         ERR("Relay live pipe error");
1767                                         goto error;
1768                                 } else if (revents & LPOLLIN) {
1769                                         DBG("Relay live viewer command received");
1770                                         ret = add_connection(live_relay_cmd_pipe[0],
1771                                                         &events, relay_connections_ht);
1772                                         if (ret < 0) {
1773                                                 goto error;
1774                                         }
1775                                 }
1776                         } else if (revents) {
1777                                 rcu_read_lock();
1778                                 lttng_ht_lookup(relay_connections_ht,
1779                                                 (void *)((unsigned long) pollfd), &iter);
1780                                 node = lttng_ht_iter_get_node_ulong(&iter);
1781                                 if (node == NULL) {
1782                                         DBG2("Relay viewer sock %d not found", pollfd);
1783                                         rcu_read_unlock();
1784                                         goto error;
1785                                 }
1786                                 relay_connection = caa_container_of(node, struct relay_command,
1787                                                 sock_n);
1788
1789                                 if (revents & (LPOLLERR)) {
1790                                         cleanup_poll_connection(&events, pollfd);
1791                                         del_connection(relay_connections_ht, &iter,
1792                                                         relay_connection);
1793                                 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1794                                         DBG("Viewer socket %d hung up", pollfd);
1795                                         cleanup_poll_connection(&events, pollfd);
1796                                         del_connection(relay_connections_ht, &iter,
1797                                                         relay_connection);
1798                                 } else if (revents & LPOLLIN) {
1799                                         ret = relay_connection->sock->ops->recvmsg(
1800                                                         relay_connection->sock, &recv_hdr,
1801                                                         sizeof(struct lttng_viewer_cmd),
1802                                                         0);
1803                                         /* connection closed */
1804                                         if (ret <= 0) {
1805                                                 cleanup_poll_connection(&events, pollfd);
1806                                                 del_connection(relay_connections_ht, &iter,
1807                                                                 relay_connection);
1808                                                 DBG("Viewer control connection closed with %d",
1809                                                                 pollfd);
1810                                         } else {
1811                                                 if (relay_connection->session) {
1812                                                         DBG2("Relay viewer worker receiving data for "
1813                                                                         "session: %" PRIu64,
1814                                                                         relay_connection->session->id);
1815                                                 }
1816                                                 ret = process_control(&recv_hdr, relay_connection,
1817                                                                 sessions_ht);
1818                                                 if (ret < 0) {
1819                                                         /* Clear the session on error. */
1820                                                         cleanup_poll_connection(&events, pollfd);
1821                                                         del_connection(relay_connections_ht, &iter,
1822                                                                         relay_connection);
1823                                                         DBG("Viewer connection closed with %d", pollfd);
1824                                                 }
1825                                         }
1826                                 }
1827                                 rcu_read_unlock();
1828                         }
1829                 }
1830         }
1831
1832 exit:
1833 error:
1834         lttng_poll_clean(&events);
1835
1836         /* empty the hash table and free the memory */
1837         rcu_read_lock();
1838         cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1839                 health_code_update();
1840
1841                 node = lttng_ht_iter_get_node_ulong(&iter);
1842                 if (!node) {
1843                         continue;
1844                 }
1845
1846                 relay_connection = caa_container_of(node, struct relay_command,
1847                                 sock_n);
1848                 del_connection(relay_connections_ht, &iter, relay_connection);
1849         }
1850         rcu_read_unlock();
1851 error_poll_create:
1852         lttng_ht_destroy(relay_connections_ht);
1853 relay_connections_ht_error:
1854         /* Close relay cmd pipes */
1855         utils_close_pipe(live_relay_cmd_pipe);
1856         if (err) {
1857                 DBG("Viewer worker thread exited with error");
1858         }
1859         DBG("Viewer worker thread cleanup complete");
1860 error_testpoint:
1861         if (err) {
1862                 health_error();
1863                 ERR("Health error occurred in %s", __func__);
1864         }
1865         health_unregister(health_relayd);
1866         stop_threads();
1867         rcu_unregister_thread();
1868         return NULL;
1869 }
1870
1871 /*
1872  * Create the relay command pipe to wake thread_manage_apps.
1873  * Closed in cleanup().
1874  */
1875 static int create_relay_cmd_pipe(void)
1876 {
1877         int ret;
1878
1879         ret = utils_create_pipe_cloexec(live_relay_cmd_pipe);
1880
1881         return ret;
1882 }
1883
1884 void live_stop_threads(void)
1885 {
1886         int ret;
1887         void *status;
1888
1889         stop_threads();
1890
1891         ret = pthread_join(live_listener_thread, &status);
1892         if (ret != 0) {
1893                 PERROR("pthread_join live listener");
1894                 goto error;     /* join error, exit without cleanup */
1895         }
1896
1897         ret = pthread_join(live_worker_thread, &status);
1898         if (ret != 0) {
1899                 PERROR("pthread_join live worker");
1900                 goto error;     /* join error, exit without cleanup */
1901         }
1902
1903         ret = pthread_join(live_dispatcher_thread, &status);
1904         if (ret != 0) {
1905                 PERROR("pthread_join live dispatcher");
1906                 goto error;     /* join error, exit without cleanup */
1907         }
1908
1909         cleanup();
1910
1911 error:
1912         return;
1913 }
1914
1915 /*
1916  * main
1917  */
1918 int live_start_threads(struct lttng_uri *uri,
1919                 struct relay_local_data *relay_ctx)
1920 {
1921         int ret = 0;
1922         void *status;
1923         int is_root;
1924
1925         assert(uri);
1926         live_uri = uri;
1927
1928         /* Check if daemon is UID = 0 */
1929         is_root = !getuid();
1930
1931         if (!is_root) {
1932                 if (live_uri->port < 1024) {
1933                         ERR("Need to be root to use ports < 1024");
1934                         ret = -1;
1935                         goto exit;
1936                 }
1937         }
1938
1939         /* Setup the thread apps communication pipe. */
1940         if ((ret = create_relay_cmd_pipe()) < 0) {
1941                 goto exit;
1942         }
1943
1944         /* Init relay command queue. */
1945         cds_wfq_init(&viewer_cmd_queue.queue);
1946
1947         /* Set up max poll set size */
1948         lttng_poll_set_max_size();
1949
1950         /* Setup the dispatcher thread */
1951         ret = pthread_create(&live_dispatcher_thread, NULL,
1952                         thread_dispatcher, (void *) NULL);
1953         if (ret != 0) {
1954                 PERROR("pthread_create viewer dispatcher");
1955                 goto exit_dispatcher;
1956         }
1957
1958         /* Setup the worker thread */
1959         ret = pthread_create(&live_worker_thread, NULL,
1960                         thread_worker, relay_ctx);
1961         if (ret != 0) {
1962                 PERROR("pthread_create viewer worker");
1963                 goto exit_worker;
1964         }
1965
1966         /* Setup the listener thread */
1967         ret = pthread_create(&live_listener_thread, NULL,
1968                         thread_listener, (void *) NULL);
1969         if (ret != 0) {
1970                 PERROR("pthread_create viewer listener");
1971                 goto exit_listener;
1972         }
1973
1974         ret = 0;
1975         goto end;
1976
1977 exit_listener:
1978         ret = pthread_join(live_listener_thread, &status);
1979         if (ret != 0) {
1980                 PERROR("pthread_join live listener");
1981                 goto error;     /* join error, exit without cleanup */
1982         }
1983
1984 exit_worker:
1985         ret = pthread_join(live_worker_thread, &status);
1986         if (ret != 0) {
1987                 PERROR("pthread_join live worker");
1988                 goto error;     /* join error, exit without cleanup */
1989         }
1990
1991 exit_dispatcher:
1992         ret = pthread_join(live_dispatcher_thread, &status);
1993         if (ret != 0) {
1994                 PERROR("pthread_join live dispatcher");
1995                 goto error;     /* join error, exit without cleanup */
1996         }
1997
1998 exit:
1999         cleanup();
2000
2001 end:
2002 error:
2003         return ret;
2004 }