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