src/plugins/ctf/lttng-live/viewer-connection.c

   1 /*
   2  * Copyright 2019 - Francis Deslauriers <francis.deslauriers@efficios.com>
   3  * Copyright 2016 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a copy
   6  * of this software and associated documentation files (the "Software"), to deal
   7  * in the Software without restriction, including without limitation the rights
   8  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9  * copies of the Software, and to permit persons to whom the Software is
  10  * furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  21  * SOFTWARE.
  22  */
  23
  24 #define BT_COMP_LOG_SELF_COMP (viewer_connection->self_comp)
  25 #define BT_LOG_OUTPUT_LEVEL (viewer_connection->log_level)
  26 #define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE/VIEWER"
  27 #include "logging/comp-logging.h"
  28
  29 #include <stdio.h>
  30 #include <stdint.h>
  31 #include <stdlib.h>
  32 #include <stdbool.h>
  33 #include <unistd.h>
  34 #include <glib.h>
  35 #include <inttypes.h>
  36 #include <sys/types.h>
  37 #include <fcntl.h>
  38
  39 #include "compat/socket.h"
  40 #include "compat/endian.h"
  41 #include "compat/compiler.h"
  42 #include "common/common.h"
  43 #include <babeltrace2/babeltrace.h>
  44
  45 #include "lttng-live.h"
  46 #include "viewer-connection.h"
  47 #include "lttng-viewer-abi.h"
  48 #include "data-stream.h"
  49 #include "metadata.h"
  50
  51 static
  52 ssize_t lttng_live_recv(struct live_viewer_connection *viewer_connection,
  53                 void *buf, size_t len)
  54 {
  55         ssize_t ret;
  56         size_t copied = 0, to_copy = len;
  57         struct lttng_live_msg_iter *lttng_live_msg_iter =
  58                 viewer_connection->lttng_live_msg_iter;
  59         BT_SOCKET sock = viewer_connection->control_sock;
  60
  61         do {
  62                 ret = bt_socket_recv(sock, buf + copied, to_copy, 0);
  63                 if (ret > 0) {
  64                         BT_ASSERT(ret <= to_copy);
  65                         copied += ret;
  66                         to_copy -= ret;
  67                 }
  68                 if (ret == BT_SOCKET_ERROR && bt_socket_interrupted()) {
  69                         if (!viewer_connection->in_query &&
  70                                         lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
  71                                 break;
  72                         } else {
  73                                 continue;
  74                         }
  75                 }
  76         } while (ret > 0 && to_copy > 0);
  77
  78         if (ret > 0) {
  79                 ret = copied;
  80         }
  81
  82         /* ret = 0 means orderly shutdown, ret == BT_SOCKET_ERROR is error. */
  83         return ret;
  84 }
  85
  86 static
  87 ssize_t lttng_live_send(struct live_viewer_connection *viewer_connection,
  88                 const void *buf, size_t len)
  89 {
  90         struct lttng_live_msg_iter *lttng_live_msg_iter =
  91                 viewer_connection->lttng_live_msg_iter;
  92         BT_SOCKET sock = viewer_connection->control_sock;
  93         ssize_t ret;
  94
  95         for (;;) {
  96                 ret = bt_socket_send_nosigpipe(sock, buf, len);
  97                 if (ret == BT_SOCKET_ERROR && bt_socket_interrupted()) {
  98                         if (!viewer_connection->in_query &&
  99                                         lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
 100                                 break;
 101                         } else {
 102                                 continue;
 103                         }
 104                 } else {
 105                         break;
 106                 }
 107         }
 108         return ret;
 109 }
 110
 111 static
 112 int parse_url(struct live_viewer_connection *viewer_connection)
 113 {
 114         char error_buf[256] = { 0 };
 115         struct bt_common_lttng_live_url_parts lttng_live_url_parts = { 0 };
 116         int ret = -1;
 117         const char *path = viewer_connection->url->str;
 118
 119         if (!path) {
 120                 goto end;
 121         }
 122
 123         lttng_live_url_parts = bt_common_parse_lttng_live_url(path, error_buf,
 124                 sizeof(error_buf));
 125         if (!lttng_live_url_parts.proto) {
 126                 BT_COMP_LOGW("Invalid LTTng live URL format: %s", error_buf);
 127                 goto end;
 128         }
 129
 130         viewer_connection->relay_hostname = lttng_live_url_parts.hostname;
 131         lttng_live_url_parts.hostname = NULL;
 132
 133         if (lttng_live_url_parts.port >= 0) {
 134                 viewer_connection->port = lttng_live_url_parts.port;
 135         } else {
 136                 viewer_connection->port = LTTNG_DEFAULT_NETWORK_VIEWER_PORT;
 137         }
 138
 139         viewer_connection->target_hostname = lttng_live_url_parts.target_hostname;
 140         lttng_live_url_parts.target_hostname = NULL;
 141
 142         if (lttng_live_url_parts.session_name) {
 143                 viewer_connection->session_name = lttng_live_url_parts.session_name;
 144                 lttng_live_url_parts.session_name = NULL;
 145         }
 146
 147         BT_COMP_LOGI("Connecting to hostname : %s, port : %d, "
 148                 "target hostname : %s, session name : %s, proto : %s",
 149                 viewer_connection->relay_hostname->str,
 150                 viewer_connection->port,
 151                 !viewer_connection->target_hostname ?
 152                 "<none>" : viewer_connection->target_hostname->str,
 153                 !viewer_connection->session_name ?
 154                 "<none>" : viewer_connection->session_name->str,
 155                 lttng_live_url_parts.proto->str);
 156         ret = 0;
 157
 158 end:
 159         bt_common_destroy_lttng_live_url_parts(&lttng_live_url_parts);
 160         return ret;
 161 }
 162
 163 static
 164 int lttng_live_handshake(struct live_viewer_connection *viewer_connection)
 165 {
 166         struct lttng_viewer_cmd cmd;
 167         struct lttng_viewer_connect connect;
 168         const size_t cmd_buf_len = sizeof(cmd) + sizeof(connect);
 169         char cmd_buf[cmd_buf_len];
 170         int ret;
 171         ssize_t ret_len;
 172
 173         cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
 174         cmd.data_size = htobe64((uint64_t) sizeof(connect));
 175         cmd.cmd_version = htobe32(0);
 176
 177         connect.viewer_session_id = -1ULL;      /* will be set on recv */
 178         connect.major = htobe32(LTTNG_LIVE_MAJOR);
 179         connect.minor = htobe32(LTTNG_LIVE_MINOR);
 180         connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
 181
 182         /*
 183          * Merge the cmd and connection request to prevent a write-write
 184          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
 185          * second write to be performed quickly in presence of Nagle's algorithm
 186          */
 187         memcpy(cmd_buf, &cmd, sizeof(cmd));
 188         memcpy(cmd_buf + sizeof(cmd), &connect, sizeof(connect));
 189         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
 190         if (ret_len == BT_SOCKET_ERROR) {
 191                 BT_COMP_LOGE("Error sending version: %s", bt_socket_errormsg());
 192                 goto error;
 193         }
 194
 195         BT_ASSERT(ret_len == cmd_buf_len);
 196
 197         ret_len = lttng_live_recv(viewer_connection, &connect, sizeof(connect));
 198         if (ret_len == 0) {
 199                 BT_COMP_LOGI("Remote side has closed connection");
 200                 goto error;
 201         }
 202         if (ret_len == BT_SOCKET_ERROR) {
 203                 BT_COMP_LOGE("Error receiving version: %s", bt_socket_errormsg());
 204                 goto error;
 205         }
 206         BT_ASSERT(ret_len == sizeof(connect));
 207
 208         BT_COMP_LOGI("Received viewer session ID : %" PRIu64,
 209                         (uint64_t) be64toh(connect.viewer_session_id));
 210         BT_COMP_LOGI("Relayd version : %u.%u", be32toh(connect.major),
 211                         be32toh(connect.minor));
 212
 213         if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) {
 214                 BT_COMP_LOGE("Incompatible lttng-relayd protocol");
 215                 goto error;
 216         }
 217         /* Use the smallest protocol version implemented. */
 218         if (LTTNG_LIVE_MINOR > be32toh(connect.minor)) {
 219                 viewer_connection->minor =  be32toh(connect.minor);
 220         } else {
 221                 viewer_connection->minor =  LTTNG_LIVE_MINOR;
 222         }
 223         viewer_connection->major = LTTNG_LIVE_MAJOR;
 224         ret = 0;
 225         return ret;
 226
 227 error:
 228         BT_COMP_LOGE("Unable to establish connection");
 229         return -1;
 230 }
 231
 232 static
 233 int lttng_live_connect_viewer(struct live_viewer_connection *viewer_connection)
 234 {
 235         struct hostent *host;
 236         struct sockaddr_in server_addr;
 237         int ret;
 238
 239         if (parse_url(viewer_connection)) {
 240                 goto error;
 241         }
 242
 243         host = gethostbyname(viewer_connection->relay_hostname->str);
 244         if (!host) {
 245                 BT_COMP_LOGE("Cannot lookup hostname %s",
 246                         viewer_connection->relay_hostname->str);
 247                 goto error;
 248         }
 249
 250         if ((viewer_connection->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == BT_INVALID_SOCKET) {
 251                 BT_COMP_LOGE("Socket creation failed: %s", bt_socket_errormsg());
 252                 goto error;
 253         }
 254
 255         server_addr.sin_family = AF_INET;
 256         server_addr.sin_port = htons(viewer_connection->port);
 257         server_addr.sin_addr = *((struct in_addr *) host->h_addr);
 258         memset(&(server_addr.sin_zero), 0, 8);
 259
 260         if (connect(viewer_connection->control_sock, (struct sockaddr *) &server_addr,
 261                                 sizeof(struct sockaddr)) == BT_SOCKET_ERROR) {
 262                 BT_COMP_LOGE("Connection failed: %s", bt_socket_errormsg());
 263                 goto error;
 264         }
 265         if (lttng_live_handshake(viewer_connection)) {
 266                 goto error;
 267         }
 268
 269         ret = 0;
 270
 271         return ret;
 272
 273 error:
 274         if (viewer_connection->control_sock != BT_INVALID_SOCKET) {
 275                 if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
 276                         BT_COMP_LOGE("Close: %s", bt_socket_errormsg());
 277                 }
 278         }
 279         viewer_connection->control_sock = BT_INVALID_SOCKET;
 280         return -1;
 281 }
 282
 283 static
 284 void lttng_live_disconnect_viewer(
 285                 struct live_viewer_connection *viewer_connection)
 286 {
 287         if (viewer_connection->control_sock == BT_INVALID_SOCKET) {
 288                 return;
 289         }
 290         if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
 291                 BT_COMP_LOGE("Close: %s", bt_socket_errormsg());
 292                 viewer_connection->control_sock = BT_INVALID_SOCKET;
 293         }
 294 }
 295
 296 static
 297 void connection_release(bt_object *obj)
 298 {
 299         struct live_viewer_connection *conn =
 300                 container_of(obj, struct live_viewer_connection, obj);
 301
 302         live_viewer_connection_destroy(conn);
 303 }
 304
 305 static
 306 int list_update_session(bt_value *results,
 307                 const struct lttng_viewer_session *session,
 308                 bool *_found, struct live_viewer_connection *viewer_connection)
 309 {
 310         int ret = 0;
 311         uint64_t i, len;
 312         bt_value *map = NULL;
 313         bt_value *hostname = NULL;
 314         bt_value *session_name = NULL;
 315         bt_value *btval = NULL;
 316         bool found = false;
 317
 318         len = bt_value_array_get_length(results);
 319         for (i = 0; i < len; i++) {
 320                 const char *hostname_str = NULL;
 321                 const char *session_name_str = NULL;
 322
 323                 map = bt_value_array_borrow_element_by_index(results, i);
 324                 hostname = bt_value_map_borrow_entry_value(map, "target-hostname");
 325                 if (!hostname) {
 326                         BT_COMP_LOGE_STR("Error borrowing \"target-hostname\" entry.");
 327                         ret = -1;
 328                         goto end;
 329                 }
 330                 session_name = bt_value_map_borrow_entry_value(map, "session-name");
 331                 if (!session_name) {
 332                         BT_COMP_LOGE_STR("Error borrowing \"session-name\" entry.");
 333                         ret = -1;
 334                         goto end;
 335                 }
 336                 hostname_str = bt_value_string_get(hostname);
 337                 session_name_str = bt_value_string_get(session_name);
 338
 339                 if (strcmp(session->hostname, hostname_str) == 0
 340                                 && strcmp(session->session_name, session_name_str) == 0) {
 341                         int64_t val;
 342                         uint32_t streams = be32toh(session->streams);
 343                         uint32_t clients = be32toh(session->clients);
 344
 345                         found = true;
 346
 347                         btval = bt_value_map_borrow_entry_value(map, "stream-count");
 348                         if (!btval) {
 349                                 BT_COMP_LOGE_STR("Error borrowing \"stream-count\" entry.");
 350                                 ret = -1;
 351                                 goto end;
 352                         }
 353                         val = bt_value_integer_unsigned_get(btval);
 354                         /* sum */
 355                         val += streams;
 356                         bt_value_integer_unsigned_set(btval, val);
 357
 358                         btval = bt_value_map_borrow_entry_value(map, "client-count");
 359                         if (!btval) {
 360                                 BT_COMP_LOGE_STR("Error borrowing \"client-count\" entry.");
 361                                 ret = -1;
 362                                 goto end;
 363                         }
 364                         val = bt_value_integer_unsigned_get(btval);
 365                         /* max */
 366                         val = bt_max_t(int64_t, clients, val);
 367                         bt_value_integer_unsigned_set(btval, val);
 368                 }
 369
 370                 if (found) {
 371                         break;
 372                 }
 373         }
 374 end:
 375         *_found = found;
 376         return ret;
 377 }
 378
 379 static
 380 int list_append_session(bt_value *results,
 381                 GString *base_url,
 382                 const struct lttng_viewer_session *session,
 383                 struct live_viewer_connection *viewer_connection)
 384 {
 385         int ret = 0;
 386         bt_value_map_insert_entry_status insert_status;
 387         bt_value_array_append_element_status append_status;
 388         bt_value *map = NULL;
 389         GString *url = NULL;
 390         bool found = false;
 391
 392         /*
 393          * If the session already exists, add the stream count to it,
 394          * and do max of client counts.
 395          */
 396         ret = list_update_session(results, session, &found, viewer_connection);
 397         if (ret || found) {
 398                 goto end;
 399         }
 400
 401         map = bt_value_map_create();
 402         if (!map) {
 403                 BT_COMP_LOGE_STR("Error creating map value.");
 404                 ret = -1;
 405                 goto end;
 406         }
 407
 408         if (base_url->len < 1) {
 409                 BT_COMP_LOGE_STR("Error: base_url length smaller than 1.");
 410                 ret = -1;
 411                 goto end;
 412         }
 413         /*
 414          * key = "url",
 415          * value = <string>,
 416          */
 417         url = g_string_new(base_url->str);
 418         g_string_append(url, "/host/");
 419         g_string_append(url, session->hostname);
 420         g_string_append_c(url, '/');
 421         g_string_append(url, session->session_name);
 422
 423         insert_status = bt_value_map_insert_string_entry(map, "url", url->str);
 424         if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 425                 BT_COMP_LOGE_STR("Error inserting \"url\" entry.");
 426                 ret = -1;
 427                 goto end;
 428         }
 429
 430         /*
 431          * key = "target-hostname",
 432          * value = <string>,
 433          */
 434         insert_status = bt_value_map_insert_string_entry(map, "target-hostname",
 435                 session->hostname);
 436         if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 437                 BT_COMP_LOGE_STR("Error inserting \"target-hostname\" entry.");
 438                 ret = -1;
 439                 goto end;
 440         }
 441
 442         /*
 443          * key = "session-name",
 444          * value = <string>,
 445          */
 446         insert_status = bt_value_map_insert_string_entry(map, "session-name",
 447                 session->session_name);
 448         if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 449                 BT_COMP_LOGE_STR("Error inserting \"session-name\" entry.");
 450                 ret = -1;
 451                 goto end;
 452         }
 453
 454         /*
 455          * key = "timer-us",
 456          * value = <integer>,
 457          */
 458         {
 459                 uint32_t live_timer = be32toh(session->live_timer);
 460
 461                 insert_status = bt_value_map_insert_unsigned_integer_entry(
 462                         map, "timer-us", live_timer);
 463                 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 464                         BT_COMP_LOGE_STR("Error inserting \"timer-us\" entry.");
 465                         ret = -1;
 466                         goto end;
 467                 }
 468         }
 469
 470         /*
 471          * key = "stream-count",
 472          * value = <integer>,
 473          */
 474         {
 475                 uint32_t streams = be32toh(session->streams);
 476
 477                 insert_status = bt_value_map_insert_unsigned_integer_entry(map,
 478                         "stream-count", streams);
 479                 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 480                         BT_COMP_LOGE_STR("Error inserting \"stream-count\" entry.");
 481                         ret = -1;
 482                         goto end;
 483                 }
 484         }
 485
 486         /*
 487          * key = "client-count",
 488          * value = <integer>,
 489          */
 490         {
 491                 uint32_t clients = be32toh(session->clients);
 492
 493                 insert_status = bt_value_map_insert_unsigned_integer_entry(map,
 494                         "client-count", clients);
 495                 if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 496                         BT_COMP_LOGE_STR("Error inserting \"client-count\" entry.");
 497                         ret = -1;
 498                         goto end;
 499                 }
 500         }
 501
 502         append_status = bt_value_array_append_element(results, map);
 503         if (append_status != BT_VALUE_ARRAY_APPEND_ELEMENT_STATUS_OK) {
 504                 BT_COMP_LOGE_STR("Error appending map to results.");
 505                 ret = -1;
 506         }
 507
 508 end:
 509         if (url) {
 510                 g_string_free(url, true);
 511         }
 512         BT_VALUE_PUT_REF_AND_RESET(map);
 513         return ret;
 514 }
 515
 516 /*
 517  * Data structure returned:
 518  *
 519  * {
 520  *   <array> = {
 521  *     [n] = {
 522  *       <map> = {
 523  *         {
 524  *           key = "url",
 525  *           value = <string>,
 526  *         },
 527  *         {
 528  *           key = "target-hostname",
 529  *           value = <string>,
 530  *         },
 531  *         {
 532  *           key = "session-name",
 533  *           value = <string>,
 534  *         },
 535  *         {
 536  *           key = "timer-us",
 537  *           value = <integer>,
 538  *         },
 539  *         {
 540  *           key = "stream-count",
 541  *           value = <integer>,
 542  *         },
 543  *         {
 544  *           key = "client-count",
 545  *           value = <integer>,
 546  *         },
 547  *       },
 548  *     }
 549  *   }
 550  */
 551
 552 BT_HIDDEN
 553 bt_component_class_query_method_status live_viewer_connection_list_sessions(
 554                 struct live_viewer_connection *viewer_connection,
 555                 const bt_value **user_result)
 556 {
 557         bt_component_class_query_method_status status =
 558                 BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
 559         bt_value *result = NULL;
 560         struct lttng_viewer_cmd cmd;
 561         struct lttng_viewer_list_sessions list;
 562         uint32_t i, sessions_count;
 563         ssize_t ret_len;
 564
 565         result = bt_value_array_create();
 566         if (!result) {
 567                 BT_COMP_LOGE("Error creating array");
 568                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
 569                 goto error;
 570         }
 571
 572         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 573         cmd.data_size = htobe64((uint64_t) 0);
 574         cmd.cmd_version = htobe32(0);
 575
 576         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 577         if (ret_len == BT_SOCKET_ERROR) {
 578                 BT_COMP_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 579                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 580                 goto error;
 581         }
 582         BT_ASSERT(ret_len == sizeof(cmd));
 583
 584         ret_len = lttng_live_recv(viewer_connection, &list, sizeof(list));
 585         if (ret_len == 0) {
 586                 BT_COMP_LOGI("Remote side has closed connection");
 587                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 588                 goto error;
 589         }
 590         if (ret_len == BT_SOCKET_ERROR) {
 591                 BT_COMP_LOGE("Error receiving session list: %s", bt_socket_errormsg());
 592                 status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 593                 goto error;
 594         }
 595         BT_ASSERT(ret_len == sizeof(list));
 596
 597         sessions_count = be32toh(list.sessions_count);
 598         for (i = 0; i < sessions_count; i++) {
 599                 struct lttng_viewer_session lsession;
 600
 601                 ret_len = lttng_live_recv(viewer_connection, &lsession,
 602                         sizeof(lsession));
 603                 if (ret_len == 0) {
 604                         BT_COMP_LOGI("Remote side has closed connection");
 605                         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 606                         goto error;
 607                 }
 608                 if (ret_len == BT_SOCKET_ERROR) {
 609                         BT_COMP_LOGE("Error receiving session: %s", bt_socket_errormsg());
 610                         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 611                         goto error;
 612                 }
 613                 BT_ASSERT(ret_len == sizeof(lsession));
 614                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 615                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 616                 if (list_append_session(result, viewer_connection->url,
 617                                 &lsession, viewer_connection)) {
 618                         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 619                         goto error;
 620                 }
 621         }
 622
 623         *user_result = result;
 624         goto end;
 625 error:
 626         BT_VALUE_PUT_REF_AND_RESET(result);
 627 end:
 628         return status;
 629 }
 630
 631 static
 632 int lttng_live_query_session_ids(struct lttng_live_msg_iter *lttng_live_msg_iter)
 633 {
 634         struct lttng_viewer_cmd cmd;
 635         struct lttng_viewer_list_sessions list;
 636         struct lttng_viewer_session lsession;
 637         uint32_t i, sessions_count;
 638         ssize_t ret_len;
 639         uint64_t session_id;
 640         struct live_viewer_connection *viewer_connection =
 641                         lttng_live_msg_iter->viewer_connection;
 642
 643         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 644         cmd.data_size = htobe64((uint64_t) 0);
 645         cmd.cmd_version = htobe32(0);
 646
 647         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 648         if (ret_len == BT_SOCKET_ERROR) {
 649                 BT_COMP_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 650                 goto error;
 651         }
 652         BT_ASSERT(ret_len == sizeof(cmd));
 653
 654         ret_len = lttng_live_recv(viewer_connection, &list, sizeof(list));
 655         if (ret_len == 0) {
 656                 BT_COMP_LOGI("Remote side has closed connection");
 657                 goto error;
 658         }
 659         if (ret_len == BT_SOCKET_ERROR) {
 660                 BT_COMP_LOGE("Error receiving session list: %s", bt_socket_errormsg());
 661                 goto error;
 662         }
 663         BT_ASSERT(ret_len == sizeof(list));
 664
 665         sessions_count = be32toh(list.sessions_count);
 666         for (i = 0; i < sessions_count; i++) {
 667                 ret_len = lttng_live_recv(viewer_connection,
 668                                 &lsession, sizeof(lsession));
 669                 if (ret_len == 0) {
 670                         BT_COMP_LOGI("Remote side has closed connection");
 671                         goto error;
 672                 }
 673                 if (ret_len == BT_SOCKET_ERROR) {
 674                         BT_COMP_LOGE("Error receiving session: %s", bt_socket_errormsg());
 675                         goto error;
 676                 }
 677                 BT_ASSERT(ret_len == sizeof(lsession));
 678                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 679                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 680                 session_id = be64toh(lsession.id);
 681
 682                 BT_COMP_LOGI("Adding session %" PRIu64 " hostname: %s session_name: %s",
 683                         session_id, lsession.hostname, lsession.session_name);
 684
 685                 if ((strncmp(lsession.session_name,
 686                         viewer_connection->session_name->str,
 687                         LTTNG_VIEWER_NAME_MAX) == 0) && (strncmp(lsession.hostname,
 688                                 viewer_connection->target_hostname->str,
 689                                 LTTNG_VIEWER_HOST_NAME_MAX) == 0)) {
 690                         if (lttng_live_add_session(lttng_live_msg_iter, session_id,
 691                                         lsession.hostname,
 692                                         lsession.session_name)) {
 693                                 goto error;
 694                         }
 695                 }
 696         }
 697
 698         return 0;
 699
 700 error:
 701         BT_COMP_LOGE("Unable to query session ids");
 702         return -1;
 703 }
 704
 705 BT_HIDDEN
 706 int lttng_live_create_viewer_session(
 707                 struct lttng_live_msg_iter *lttng_live_msg_iter)
 708 {
 709         struct lttng_viewer_cmd cmd;
 710         struct lttng_viewer_create_session_response resp;
 711         ssize_t ret_len;
 712         struct live_viewer_connection *viewer_connection =
 713                         lttng_live_msg_iter->viewer_connection;
 714
 715         cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
 716         cmd.data_size = htobe64((uint64_t) 0);
 717         cmd.cmd_version = htobe32(0);
 718
 719         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 720         if (ret_len == BT_SOCKET_ERROR) {
 721                 BT_COMP_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 722                 goto error;
 723         }
 724         BT_ASSERT(ret_len == sizeof(cmd));
 725
 726         ret_len = lttng_live_recv(viewer_connection, &resp, sizeof(resp));
 727         if (ret_len == 0) {
 728                 BT_COMP_LOGI("Remote side has closed connection");
 729                 goto error;
 730         }
 731         if (ret_len == BT_SOCKET_ERROR) {
 732                 BT_COMP_LOGE("Error receiving create session reply: %s", bt_socket_errormsg());
 733                 goto error;
 734         }
 735         BT_ASSERT(ret_len == sizeof(resp));
 736
 737         if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
 738                 BT_COMP_LOGE("Error creating viewer session");
 739                 goto error;
 740         }
 741         if (lttng_live_query_session_ids(lttng_live_msg_iter)) {
 742                 goto error;
 743         }
 744
 745         return 0;
 746
 747 error:
 748         return -1;
 749 }
 750
 751 static
 752 int receive_streams(struct lttng_live_session *session,
 753                 uint32_t stream_count)
 754 {
 755         ssize_t ret_len;
 756         uint32_t i;
 757         struct lttng_live_msg_iter *lttng_live_msg_iter =
 758                 session->lttng_live_msg_iter;
 759         struct live_viewer_connection *viewer_connection =
 760                 lttng_live_msg_iter->viewer_connection;
 761
 762         BT_COMP_LOGI("Getting %" PRIu32 " new streams:", stream_count);
 763         for (i = 0; i < stream_count; i++) {
 764                 struct lttng_viewer_stream stream;
 765                 struct lttng_live_stream_iterator *live_stream;
 766                 uint64_t stream_id;
 767                 uint64_t ctf_trace_id;
 768
 769                 ret_len = lttng_live_recv(viewer_connection, &stream, sizeof(stream));
 770                 if (ret_len == 0) {
 771                         BT_COMP_LOGI("Remote side has closed connection");
 772                         goto error;
 773                 }
 774                 if (ret_len == BT_SOCKET_ERROR) {
 775                         BT_COMP_LOGE("Error receiving stream");
 776                         goto error;
 777                 }
 778                 BT_ASSERT(ret_len == sizeof(stream));
 779                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
 780                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 781                 stream_id = be64toh(stream.id);
 782                 ctf_trace_id = be64toh(stream.ctf_trace_id);
 783
 784                 if (stream.metadata_flag) {
 785                         BT_COMP_LOGI("    metadata stream %" PRIu64 " : %s/%s",
 786                                 stream_id, stream.path_name, stream.channel_name);
 787                         if (lttng_live_metadata_create_stream(session,
 788                                         ctf_trace_id, stream_id,
 789                                         stream.path_name)) {
 790                                 BT_COMP_LOGE("Error creating metadata stream");
 791
 792                                 goto error;
 793                         }
 794                         session->lazy_stream_msg_init = true;
 795                 } else {
 796                         BT_COMP_LOGI("    stream %" PRIu64 " : %s/%s",
 797                                 stream_id, stream.path_name, stream.channel_name);
 798                         live_stream = lttng_live_stream_iterator_create(session,
 799                                 ctf_trace_id, stream_id);
 800                         if (!live_stream) {
 801                                 BT_COMP_LOGE("Error creating streamn");
 802                                 goto error;
 803                         }
 804                 }
 805         }
 806         return 0;
 807
 808 error:
 809         return -1;
 810 }
 811
 812 BT_HIDDEN
 813 enum lttng_live_attach_session_status lttng_live_attach_session(
 814                 struct lttng_live_session *session)
 815 {
 816         struct lttng_viewer_cmd cmd;
 817         enum lttng_live_attach_session_status attach_status;
 818         struct lttng_viewer_attach_session_request rq;
 819         struct lttng_viewer_attach_session_response rp;
 820         struct lttng_live_msg_iter *lttng_live_msg_iter =
 821                 session->lttng_live_msg_iter;
 822         struct live_viewer_connection *viewer_connection =
 823                 lttng_live_msg_iter->viewer_connection;
 824         uint64_t session_id = session->id;
 825         uint32_t streams_count;
 826         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
 827         char cmd_buf[cmd_buf_len];
 828         ssize_t ret_len;
 829
 830         cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 831         cmd.data_size = htobe64((uint64_t) sizeof(rq));
 832         cmd.cmd_version = htobe32(0);
 833
 834         memset(&rq, 0, sizeof(rq));
 835         rq.session_id = htobe64(session_id);
 836         // TODO: add cmd line parameter to select seek beginning
 837         // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
 838         rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
 839
 840         /*
 841          * Merge the cmd and connection request to prevent a write-write
 842          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
 843          * second write to be performed quickly in presence of Nagle's algorithm.
 844          */
 845         memcpy(cmd_buf, &cmd, sizeof(cmd));
 846         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
 847         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
 848         if (ret_len == BT_SOCKET_ERROR) {
 849                 BT_COMP_LOGE("Error sending attach request: %s", bt_socket_errormsg());
 850                 goto error;
 851         }
 852
 853         BT_ASSERT(ret_len == cmd_buf_len);
 854         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
 855         if (ret_len == 0) {
 856                 BT_COMP_LOGI("Remote side has closed connection");
 857                 goto error;
 858         }
 859         if (ret_len == BT_SOCKET_ERROR) {
 860                 BT_COMP_LOGE("Error receiving attach response: %s", bt_socket_errormsg());
 861                 goto error;
 862         }
 863         BT_ASSERT(ret_len == sizeof(rp));
 864
 865         streams_count = be32toh(rp.streams_count);
 866         switch(be32toh(rp.status)) {
 867         case LTTNG_VIEWER_ATTACH_OK:
 868                 break;
 869         case LTTNG_VIEWER_ATTACH_UNK:
 870                 BT_COMP_LOGW("Session id %" PRIu64 " is unknown", session_id);
 871                 goto error;
 872         case LTTNG_VIEWER_ATTACH_ALREADY:
 873                 BT_COMP_LOGW("There is already a viewer attached to this session");
 874                 goto error;
 875         case LTTNG_VIEWER_ATTACH_NOT_LIVE:
 876                 BT_COMP_LOGW("Not a live session");
 877                 goto error;
 878         case LTTNG_VIEWER_ATTACH_SEEK_ERR:
 879                 BT_COMP_LOGE("Wrong seek parameter");
 880                 goto error;
 881         default:
 882                 BT_COMP_LOGE("Unknown attach return code %u", be32toh(rp.status));
 883                 goto error;
 884         }
 885
 886         /* We receive the initial list of streams. */
 887         if (receive_streams(session, streams_count)) {
 888                 goto error;
 889         }
 890
 891         session->attached = true;
 892         session->new_streams_needed = false;
 893
 894         attach_status = LTTNG_LIVE_ATTACH_SESSION_STATUS_OK;
 895         goto end;
 896
 897 error:
 898         attach_status = LTTNG_LIVE_ATTACH_SESSION_STATUS_ERROR;
 899
 900 end:
 901         return attach_status;
 902 }
 903
 904 BT_HIDDEN
 905 int lttng_live_detach_session(struct lttng_live_session *session)
 906 {
 907         struct lttng_viewer_cmd cmd;
 908         struct lttng_viewer_detach_session_request rq;
 909         struct lttng_viewer_detach_session_response rp;
 910         ssize_t ret_len;
 911         struct lttng_live_msg_iter *lttng_live_msg_iter =
 912                 session->lttng_live_msg_iter;
 913         struct live_viewer_connection *viewer_connection =
 914                 lttng_live_msg_iter->viewer_connection;
 915         uint64_t session_id = session->id;
 916         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
 917         char cmd_buf[cmd_buf_len];
 918
 919         if (!session->attached) {
 920                 return 0;
 921         }
 922
 923         cmd.cmd = htobe32(LTTNG_VIEWER_DETACH_SESSION);
 924         cmd.data_size = htobe64((uint64_t) sizeof(rq));
 925         cmd.cmd_version = htobe32(0);
 926
 927         memset(&rq, 0, sizeof(rq));
 928         rq.session_id = htobe64(session_id);
 929
 930         /*
 931          * Merge the cmd and connection request to prevent a write-write
 932          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
 933          * second write to be performed quickly in presence of Nagle's algorithm.
 934          */
 935         memcpy(cmd_buf, &cmd, sizeof(cmd));
 936         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
 937         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
 938         if (ret_len == BT_SOCKET_ERROR) {
 939                 BT_COMP_LOGE("Error sending detach request: %s", bt_socket_errormsg());
 940                 goto error;
 941         }
 942
 943         BT_ASSERT(ret_len == cmd_buf_len);
 944         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
 945         if (ret_len == 0) {
 946                 BT_COMP_LOGI("Remote side has closed connection");
 947                 goto error;
 948         }
 949         if (ret_len == BT_SOCKET_ERROR) {
 950                 BT_COMP_LOGE("Error receiving detach response: %s", bt_socket_errormsg());
 951                 goto error;
 952         }
 953         BT_ASSERT(ret_len == sizeof(rp));
 954
 955         switch(be32toh(rp.status)) {
 956         case LTTNG_VIEWER_DETACH_SESSION_OK:
 957                 break;
 958         case LTTNG_VIEWER_DETACH_SESSION_UNK:
 959                 BT_COMP_LOGW("Session id %" PRIu64 " is unknown", session_id);
 960                 goto error;
 961         case LTTNG_VIEWER_DETACH_SESSION_ERR:
 962                 BT_COMP_LOGW("Error detaching session id %" PRIu64 "", session_id);
 963                 goto error;
 964         default:
 965                 BT_COMP_LOGE("Unknown detach return code %u", be32toh(rp.status));
 966                 goto error;
 967         }
 968
 969         session->attached = false;
 970
 971         return 0;
 972
 973 error:
 974         return -1;
 975 }
 976
 977 BT_HIDDEN
 978 enum lttng_live_get_one_metadata_status lttng_live_get_one_metadata_packet(
 979                 struct lttng_live_trace *trace, FILE *fp, size_t *reply_len)
 980 {
 981         uint64_t len = 0;
 982         enum lttng_live_get_one_metadata_status metadata_status;
 983         struct lttng_viewer_cmd cmd;
 984         struct lttng_viewer_get_metadata rq;
 985         struct lttng_viewer_metadata_packet rp;
 986         char *data = NULL;
 987         ssize_t ret_len;
 988         struct lttng_live_session *session = trace->session;
 989         struct lttng_live_msg_iter *lttng_live_msg_iter =
 990                 session->lttng_live_msg_iter;
 991         struct lttng_live_metadata *metadata = trace->metadata;
 992         struct live_viewer_connection *viewer_connection =
 993                 lttng_live_msg_iter->viewer_connection;
 994         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
 995         char cmd_buf[cmd_buf_len];
 996
 997         rq.stream_id = htobe64(metadata->stream_id);
 998         cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
 999         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1000         cmd.cmd_version = htobe32(0);
1001
1002         /*
1003          * Merge the cmd and connection request to prevent a write-write
1004          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1005          * second write to be performed quickly in presence of Nagle's algorithm.
1006          */
1007         memcpy(cmd_buf, &cmd, sizeof(cmd));
1008         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1009         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1010         if (ret_len == BT_SOCKET_ERROR) {
1011                 BT_COMP_LOGE("Error sending get_metadata request: %s", bt_socket_errormsg());
1012                 goto error;
1013         }
1014
1015         BT_ASSERT(ret_len == cmd_buf_len);
1016         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1017         if (ret_len == 0) {
1018                 BT_COMP_LOGI("Remote side has closed connection");
1019                 goto error;
1020         }
1021         if (ret_len == BT_SOCKET_ERROR) {
1022                 BT_COMP_LOGE("Error receiving get_metadata response: %s", bt_socket_errormsg());
1023                 goto error;
1024         }
1025         BT_ASSERT(ret_len == sizeof(rp));
1026
1027         switch (be32toh(rp.status)) {
1028                 case LTTNG_VIEWER_METADATA_OK:
1029                         BT_COMP_LOGD("Received get_metadata response : OK");
1030                         break;
1031                 case LTTNG_VIEWER_NO_NEW_METADATA:
1032                         BT_COMP_LOGD("Received get_metadata response: no new");
1033                         metadata_status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_END;
1034                         goto end;
1035                 case LTTNG_VIEWER_METADATA_ERR:
1036                         /*
1037                          * The Relayd cannot find this stream id. Maybe its
1038                          * gone already. This can happen in short lived UST app
1039                          * in a per-pid session.
1040                          */
1041                         BT_COMP_LOGD("Received get_metadata response: error");
1042                         metadata_status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_CLOSED;
1043                         goto end;
1044                 default:
1045                         BT_COMP_LOGD("get_metadata : UNKNOWN");
1046                         goto error;
1047         }
1048
1049         len = be64toh(rp.len);
1050         BT_COMP_LOGD("Writing %" PRIu64" bytes to metadata", len);
1051         if (len <= 0) {
1052                 goto error;
1053         }
1054
1055         data = calloc(1, len);
1056         if (!data) {
1057                 BT_COMP_LOGE("relay data calloc: %s", strerror(errno));
1058                 goto error;
1059         }
1060         ret_len = lttng_live_recv(viewer_connection, data, len);
1061         if (ret_len == 0) {
1062                 BT_COMP_LOGI("Remote side has closed connection");
1063                 goto error_free_data;
1064         }
1065         if (ret_len == BT_SOCKET_ERROR) {
1066                 BT_COMP_LOGE("Error receiving trace packet: %s", bt_socket_errormsg());
1067                 goto error_free_data;
1068         }
1069         BT_ASSERT(ret_len == len);
1070
1071         do {
1072                 ret_len = fwrite(data, 1, len, fp);
1073         } while (ret_len < 0 && errno == EINTR);
1074         if (ret_len < 0) {
1075                 BT_COMP_LOGE("Writing in the metadata fp");
1076                 goto error_free_data;
1077         }
1078         BT_ASSERT(ret_len == len);
1079         free(data);
1080         *reply_len = len;
1081         metadata_status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_OK;
1082
1083         goto end;
1084
1085 error_free_data:
1086         free(data);
1087 error:
1088         metadata_status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1089
1090 end:
1091         return metadata_status;
1092 }
1093
1094 /*
1095  * Assign the fields from a lttng_viewer_index to a packet_index.
1096  */
1097 static
1098 void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
1099                 struct packet_index *pindex)
1100 {
1101         BT_ASSERT(lindex);
1102         BT_ASSERT(pindex);
1103
1104         pindex->offset = be64toh(lindex->offset);
1105         pindex->packet_size = be64toh(lindex->packet_size);
1106         pindex->content_size = be64toh(lindex->content_size);
1107         pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
1108         pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
1109         pindex->events_discarded = be64toh(lindex->events_discarded);
1110 }
1111
1112 BT_HIDDEN
1113 enum lttng_live_iterator_status lttng_live_get_next_index(
1114                 struct lttng_live_msg_iter *lttng_live_msg_iter,
1115                 struct lttng_live_stream_iterator *stream,
1116                 struct packet_index *index)
1117 {
1118         struct lttng_viewer_cmd cmd;
1119         struct lttng_viewer_get_next_index rq;
1120         struct lttng_viewer_index rp;
1121         enum lttng_live_iterator_status retstatus = LTTNG_LIVE_ITERATOR_STATUS_OK;
1122         struct live_viewer_connection *viewer_connection =
1123                         lttng_live_msg_iter->viewer_connection;
1124         struct lttng_live_trace *trace = stream->trace;
1125         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1126         char cmd_buf[cmd_buf_len];
1127         uint32_t flags, status;
1128         ssize_t ret_len;
1129
1130         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1131         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1132         cmd.cmd_version = htobe32(0);
1133
1134
1135         memset(&rq, 0, sizeof(rq));
1136         rq.stream_id = htobe64(stream->viewer_stream_id);
1137
1138         /*
1139          * Merge the cmd and connection request to prevent a write-write
1140          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1141          * second write to be performed quickly in presence of Nagle's algorithm.
1142          */
1143         memcpy(cmd_buf, &cmd, sizeof(cmd));
1144         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1145         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1146         if (ret_len == BT_SOCKET_ERROR) {
1147                 BT_COMP_LOGE("Error sending get_next_index request: %s",
1148                                 bt_socket_errormsg());
1149                 goto error;
1150         }
1151
1152         BT_ASSERT(ret_len == cmd_buf_len);
1153         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1154         if (ret_len == 0) {
1155                 BT_COMP_LOGI("Remote side has closed connection");
1156                 goto error;
1157         }
1158         if (ret_len == BT_SOCKET_ERROR) {
1159                 BT_COMP_LOGE("Error receiving get_next_index response: %s",
1160                                 bt_socket_errormsg());
1161                 goto error;
1162         }
1163         BT_ASSERT(ret_len == sizeof(rp));
1164
1165         flags = be32toh(rp.flags);
1166         status = be32toh(rp.status);
1167
1168         switch (status) {
1169         case LTTNG_VIEWER_INDEX_INACTIVE:
1170         {
1171                 uint64_t ctf_stream_class_id;
1172
1173                 BT_COMP_LOGD("get_next_index: inactive");
1174                 memset(index, 0, sizeof(struct packet_index));
1175                 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
1176                 stream->current_inactivity_ts = index->ts_cycles.timestamp_end;
1177                 ctf_stream_class_id = be64toh(rp.stream_id);
1178                 if (stream->ctf_stream_class_id != -1ULL) {
1179                         BT_ASSERT(stream->ctf_stream_class_id ==
1180                                 ctf_stream_class_id);
1181                 } else {
1182                         stream->ctf_stream_class_id = ctf_stream_class_id;
1183                 }
1184                 stream->state = LTTNG_LIVE_STREAM_QUIESCENT;
1185                 break;
1186         }
1187         case LTTNG_VIEWER_INDEX_OK:
1188         {
1189                 uint64_t ctf_stream_class_id;
1190
1191                 BT_COMP_LOGD("get_next_index: OK");
1192                 lttng_index_to_packet_index(&rp, index);
1193                 ctf_stream_class_id = be64toh(rp.stream_id);
1194                 if (stream->ctf_stream_class_id != -1ULL) {
1195                         BT_ASSERT(stream->ctf_stream_class_id ==
1196                                 ctf_stream_class_id);
1197                 } else {
1198                         stream->ctf_stream_class_id = ctf_stream_class_id;
1199                 }
1200
1201                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_DATA;
1202
1203                 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1204                         BT_COMP_LOGD("get_next_index: new metadata needed");
1205                         trace->new_metadata_needed = true;
1206                 }
1207                 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1208                         BT_COMP_LOGD("get_next_index: new streams needed");
1209                         lttng_live_need_new_streams(lttng_live_msg_iter);
1210                 }
1211                 break;
1212         }
1213         case LTTNG_VIEWER_INDEX_RETRY:
1214                 BT_COMP_LOGD("get_next_index: retry");
1215                 memset(index, 0, sizeof(struct packet_index));
1216                 retstatus = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1217                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1218                 goto end;
1219         case LTTNG_VIEWER_INDEX_HUP:
1220                 BT_COMP_LOGD("get_next_index: stream hung up");
1221                 memset(index, 0, sizeof(struct packet_index));
1222                 index->offset = EOF;
1223                 retstatus = LTTNG_LIVE_ITERATOR_STATUS_END;
1224                 stream->state = LTTNG_LIVE_STREAM_EOF;
1225                 stream->has_stream_hung_up = true;
1226                 break;
1227         case LTTNG_VIEWER_INDEX_ERR:
1228                 BT_COMP_LOGE("get_next_index: error");
1229                 memset(index, 0, sizeof(struct packet_index));
1230                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1231                 goto error;
1232         default:
1233                 BT_COMP_LOGE("get_next_index: unknown value");
1234                 memset(index, 0, sizeof(struct packet_index));
1235                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1236                 goto error;
1237         }
1238 end:
1239         return retstatus;
1240
1241 error:
1242         if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
1243                 retstatus = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1244         } else {
1245                 retstatus = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1246         }
1247         return retstatus;
1248 }
1249
1250 BT_HIDDEN
1251 enum bt_msg_iter_medium_status lttng_live_get_stream_bytes(
1252                 struct lttng_live_msg_iter *lttng_live_msg_iter,
1253                 struct lttng_live_stream_iterator *stream, uint8_t *buf,
1254                 uint64_t offset, uint64_t req_len, uint64_t *recv_len)
1255 {
1256         enum bt_msg_iter_medium_status retstatus = BT_MSG_ITER_MEDIUM_STATUS_OK;
1257         struct lttng_viewer_trace_packet rp;
1258         struct lttng_viewer_cmd cmd;
1259         struct lttng_viewer_get_packet rq;
1260         struct live_viewer_connection *viewer_connection =
1261                         lttng_live_msg_iter->viewer_connection;
1262         struct lttng_live_trace *trace = stream->trace;
1263         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1264         char cmd_buf[cmd_buf_len];
1265         uint32_t flags, status;
1266         ssize_t ret_len;
1267
1268         BT_COMP_LOGD("lttng_live_get_stream_bytes: offset=%" PRIu64 ", req_len=%" PRIu64,
1269                         offset, req_len);
1270         cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
1271         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1272         cmd.cmd_version = htobe32(0);
1273
1274         memset(&rq, 0, sizeof(rq));
1275         rq.stream_id = htobe64(stream->viewer_stream_id);
1276         rq.offset = htobe64(offset);
1277         rq.len = htobe32(req_len);
1278
1279         /*
1280          * Merge the cmd and connection request to prevent a write-write
1281          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1282          * second write to be performed quickly in presence of Nagle's algorithm.
1283          */
1284         memcpy(cmd_buf, &cmd, sizeof(cmd));
1285         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1286         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1287         if (ret_len == BT_SOCKET_ERROR) {
1288                 BT_COMP_LOGE("Error sending get_data request: %s", bt_socket_errormsg());
1289                 goto error;
1290         }
1291
1292         BT_ASSERT(ret_len == cmd_buf_len);
1293         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1294         if (ret_len == 0) {
1295                 BT_COMP_LOGI("Remote side has closed connection");
1296                 goto error;
1297         }
1298         if (ret_len == BT_SOCKET_ERROR) {
1299                 BT_COMP_LOGE("Error receiving get_data response: %s", bt_socket_errormsg());
1300                 goto error;
1301         }
1302         if (ret_len != sizeof(rp)) {
1303                 BT_COMP_LOGE("get_data_packet: expected %zu"
1304                                 ", received %zd", sizeof(rp),
1305                                 ret_len);
1306                 goto error;
1307         }
1308
1309         flags = be32toh(rp.flags);
1310         status = be32toh(rp.status);
1311
1312         switch (status) {
1313         case LTTNG_VIEWER_GET_PACKET_OK:
1314                 req_len = be32toh(rp.len);
1315                 BT_COMP_LOGD("get_data_packet: Ok, packet size : %" PRIu64 "", req_len);
1316                 break;
1317         case LTTNG_VIEWER_GET_PACKET_RETRY:
1318                 /* Unimplemented by relay daemon */
1319                 BT_COMP_LOGD("get_data_packet: retry");
1320                 retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1321                 goto end;
1322         case LTTNG_VIEWER_GET_PACKET_ERR:
1323                 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1324                         BT_COMP_LOGD("get_data_packet: new metadata needed, try again later");
1325                         trace->new_metadata_needed = true;
1326                 }
1327                 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1328                         BT_COMP_LOGD("get_data_packet: new streams needed, try again later");
1329                         lttng_live_need_new_streams(lttng_live_msg_iter);
1330                 }
1331                 if (flags & (LTTNG_VIEWER_FLAG_NEW_METADATA
1332                                 | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
1333                         retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1334                         goto end;
1335                 }
1336                 BT_COMP_LOGE("get_data_packet: error");
1337                 goto error;
1338         case LTTNG_VIEWER_GET_PACKET_EOF:
1339                 retstatus = BT_MSG_ITER_MEDIUM_STATUS_EOF;
1340                 goto end;
1341         default:
1342                 BT_COMP_LOGE("get_data_packet: unknown");
1343                 goto error;
1344         }
1345
1346         if (req_len == 0) {
1347                 goto error;
1348         }
1349
1350         ret_len = lttng_live_recv(viewer_connection, buf, req_len);
1351         if (ret_len == 0) {
1352                 BT_COMP_LOGI("Remote side has closed connection");
1353                 goto error;
1354         }
1355         if (ret_len == BT_SOCKET_ERROR) {
1356                 BT_COMP_LOGE("Error receiving trace packet: %s", bt_socket_errormsg());
1357                 goto error;
1358         }
1359         BT_ASSERT(ret_len == req_len);
1360         *recv_len = ret_len;
1361 end:
1362         return retstatus;
1363
1364 error:
1365         if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
1366                 retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1367         } else {
1368                 retstatus = BT_MSG_ITER_MEDIUM_STATUS_ERROR;
1369         }
1370         return retstatus;
1371 }
1372
1373 /*
1374  * Request new streams for a session.
1375  */
1376 BT_HIDDEN
1377 enum lttng_live_iterator_status lttng_live_get_new_streams(
1378                 struct lttng_live_session *session)
1379 {
1380         enum lttng_live_iterator_status status =
1381                 LTTNG_LIVE_ITERATOR_STATUS_OK;
1382         struct lttng_viewer_cmd cmd;
1383         struct lttng_viewer_new_streams_request rq;
1384         struct lttng_viewer_new_streams_response rp;
1385         struct lttng_live_msg_iter *lttng_live_msg_iter =
1386                 session->lttng_live_msg_iter;
1387         struct live_viewer_connection *viewer_connection =
1388                 lttng_live_msg_iter->viewer_connection;
1389         uint32_t streams_count;
1390         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1391         char cmd_buf[cmd_buf_len];
1392         ssize_t ret_len;
1393
1394         if (!session->new_streams_needed) {
1395                 return LTTNG_LIVE_ITERATOR_STATUS_OK;
1396         }
1397
1398         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1399         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1400         cmd.cmd_version = htobe32(0);
1401
1402         memset(&rq, 0, sizeof(rq));
1403         rq.session_id = htobe64(session->id);
1404
1405         /*
1406          * Merge the cmd and connection request to prevent a write-write
1407          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1408          * second write to be performed quickly in presence of Nagle's algorithm.
1409          */
1410         memcpy(cmd_buf, &cmd, sizeof(cmd));
1411         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1412         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1413         if (ret_len == BT_SOCKET_ERROR) {
1414                 BT_COMP_LOGE("Error sending get_new_streams request: %s",
1415                                 bt_socket_errormsg());
1416                 goto error;
1417         }
1418
1419         BT_ASSERT(ret_len == cmd_buf_len);
1420         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1421         if (ret_len == 0) {
1422                 BT_COMP_LOGI("Remote side has closed connection");
1423                 goto error;
1424         }
1425         if (ret_len == BT_SOCKET_ERROR) {
1426                 BT_COMP_LOGE("Error receiving get_new_streams response");
1427                 goto error;
1428         }
1429         BT_ASSERT(ret_len == sizeof(rp));
1430
1431         streams_count = be32toh(rp.streams_count);
1432
1433         switch(be32toh(rp.status)) {
1434         case LTTNG_VIEWER_NEW_STREAMS_OK:
1435                 session->new_streams_needed = false;
1436                 break;
1437         case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1438                 session->new_streams_needed = false;
1439                 goto end;
1440         case LTTNG_VIEWER_NEW_STREAMS_HUP:
1441                 session->new_streams_needed = false;
1442                 session->closed = true;
1443                 status = LTTNG_LIVE_ITERATOR_STATUS_END;
1444                 goto end;
1445         case LTTNG_VIEWER_NEW_STREAMS_ERR:
1446                 BT_COMP_LOGE("get_new_streams error");
1447                 goto error;
1448         default:
1449                 BT_COMP_LOGE("Unknown return code %u", be32toh(rp.status));
1450                 goto error;
1451         }
1452
1453         if (receive_streams(session, streams_count)) {
1454                 goto error;
1455         }
1456 end:
1457         return status;
1458
1459 error:
1460         if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
1461                 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1462         } else {
1463                 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1464         }
1465         return status;
1466 }
1467
1468 BT_HIDDEN
1469 struct live_viewer_connection *live_viewer_connection_create(
1470                 const char *url, bool in_query,
1471                 struct lttng_live_msg_iter *lttng_live_msg_iter,
1472                 bt_logging_level log_level)
1473 {
1474         struct live_viewer_connection *viewer_connection;
1475
1476         viewer_connection = g_new0(struct live_viewer_connection, 1);
1477
1478         if (bt_socket_init(log_level) != 0) {
1479                 goto error;
1480         }
1481
1482         viewer_connection->log_level = log_level;
1483
1484         if (lttng_live_msg_iter) {
1485                 viewer_connection->self_comp = lttng_live_msg_iter->self_comp;
1486         }
1487
1488         bt_object_init_shared(&viewer_connection->obj, connection_release);
1489         viewer_connection->control_sock = BT_INVALID_SOCKET;
1490         viewer_connection->port = -1;
1491         viewer_connection->in_query = in_query;
1492         viewer_connection->lttng_live_msg_iter = lttng_live_msg_iter;
1493         viewer_connection->url = g_string_new(url);
1494         if (!viewer_connection->url) {
1495                 goto error;
1496         }
1497
1498         BT_COMP_LOGI("Establishing connection to url \"%s\"...", url);
1499         if (lttng_live_connect_viewer(viewer_connection)) {
1500                 goto error_report;
1501         }
1502         BT_COMP_LOGI("Connection to url \"%s\" is established", url);
1503         return viewer_connection;
1504
1505 error_report:
1506         BT_COMP_LOGW("Failure to establish connection to url \"%s\"", url);
1507 error:
1508         if (viewer_connection) {
1509                 live_viewer_connection_destroy(viewer_connection);
1510         }
1511
1512         return NULL;
1513 }
1514
1515 BT_HIDDEN
1516 void live_viewer_connection_destroy(
1517                 struct live_viewer_connection *viewer_connection)
1518 {
1519         BT_COMP_LOGI("Closing connection to url \"%s\"", viewer_connection->url->str);
1520         lttng_live_disconnect_viewer(viewer_connection);
1521         if (viewer_connection->url) {
1522                 g_string_free(viewer_connection->url, true);
1523         }
1524         if (viewer_connection->relay_hostname) {
1525                 g_string_free(viewer_connection->relay_hostname, true);
1526         }
1527         if (viewer_connection->target_hostname) {
1528                 g_string_free(viewer_connection->target_hostname, true);
1529         }
1530         if (viewer_connection->session_name) {
1531                 g_string_free(viewer_connection->session_name, true);
1532         }
1533         g_free(viewer_connection);
1534
1535         bt_socket_fini();
1536 }