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_LOG_OUTPUT_LEVEL (viewer_connection->log_level)
  25 #define BT_LOG_TAG "PLUGIN/SRC.CTF.LTTNG-LIVE/VIEWER"
  26 #include "logging/log.h"
  27
  28 #include <stdio.h>
  29 #include <stdint.h>
  30 #include <stdlib.h>
  31 #include <stdbool.h>
  32 #include <unistd.h>
  33 #include <glib.h>
  34 #include <inttypes.h>
  35 #include <sys/types.h>
  36 #include <fcntl.h>
  37
  38 #include "compat/socket.h"
  39 #include "compat/endian.h"
  40 #include "compat/compiler.h"
  41 #include "common/common.h"
  42 #include <babeltrace2/babeltrace.h>
  43
  44 #include "lttng-live.h"
  45 #include "viewer-connection.h"
  46 #include "lttng-viewer-abi.h"
  47 #include "data-stream.h"
  48 #include "metadata.h"
  49
  50 static
  51 ssize_t lttng_live_recv(struct live_viewer_connection *viewer_connection,
  52                 void *buf, size_t len)
  53 {
  54         ssize_t ret;
  55         size_t copied = 0, to_copy = len;
  56         struct lttng_live_msg_iter *lttng_live_msg_iter =
  57                 viewer_connection->lttng_live_msg_iter;
  58         BT_SOCKET sock = viewer_connection->control_sock;
  59
  60         do {
  61                 ret = bt_socket_recv(sock, buf + copied, to_copy, 0);
  62                 if (ret > 0) {
  63                         BT_ASSERT(ret <= to_copy);
  64                         copied += ret;
  65                         to_copy -= ret;
  66                 }
  67                 if (ret == BT_SOCKET_ERROR && bt_socket_interrupted()) {
  68                         if (!viewer_connection->in_query &&
  69                                         lttng_live_graph_is_canceled(lttng_live_msg_iter->lttng_live_comp)) {
  70                                 break;
  71                         } else {
  72                                 continue;
  73                         }
  74                 }
  75         } while (ret > 0 && to_copy > 0);
  76         if (ret > 0)
  77                 ret = copied;
  78         /* ret = 0 means orderly shutdown, ret == BT_SOCKET_ERROR is error. */
  79         return ret;
  80 }
  81
  82 static
  83 ssize_t lttng_live_send(struct live_viewer_connection *viewer_connection,
  84                 const void *buf, size_t len)
  85 {
  86         struct lttng_live_msg_iter *lttng_live_msg_iter =
  87                 viewer_connection->lttng_live_msg_iter;
  88         BT_SOCKET sock = viewer_connection->control_sock;
  89         ssize_t ret;
  90
  91         for (;;) {
  92                 ret = bt_socket_send_nosigpipe(sock, buf, len);
  93                 if (ret == BT_SOCKET_ERROR && bt_socket_interrupted()) {
  94                         if (!viewer_connection->in_query &&
  95                                         lttng_live_graph_is_canceled(lttng_live_msg_iter->lttng_live_comp)) {
  96                                 break;
  97                         } else {
  98                                 continue;
  99                         }
 100                 } else {
 101                         break;
 102                 }
 103         }
 104         return ret;
 105 }
 106
 107 static
 108 int parse_url(struct live_viewer_connection *viewer_connection)
 109 {
 110         char error_buf[256] = { 0 };
 111         struct bt_common_lttng_live_url_parts lttng_live_url_parts = { 0 };
 112         int ret = -1;
 113         const char *path = viewer_connection->url->str;
 114
 115         if (!path) {
 116                 goto end;
 117         }
 118
 119         lttng_live_url_parts = bt_common_parse_lttng_live_url(path,
 120                         error_buf, sizeof(error_buf));
 121         if (!lttng_live_url_parts.proto) {
 122                 BT_LOGW("Invalid LTTng live URL format: %s", error_buf);
 123                 goto end;
 124         }
 125
 126         viewer_connection->relay_hostname =
 127                         lttng_live_url_parts.hostname;
 128         lttng_live_url_parts.hostname = NULL;
 129
 130         if (lttng_live_url_parts.port >= 0) {
 131                 viewer_connection->port = lttng_live_url_parts.port;
 132         } else {
 133                 viewer_connection->port = LTTNG_DEFAULT_NETWORK_VIEWER_PORT;
 134         }
 135
 136         viewer_connection->target_hostname =
 137                         lttng_live_url_parts.target_hostname;
 138         lttng_live_url_parts.target_hostname = NULL;
 139
 140         if (lttng_live_url_parts.session_name) {
 141                 viewer_connection->session_name =
 142                                 lttng_live_url_parts.session_name;
 143                 lttng_live_url_parts.session_name = NULL;
 144         }
 145
 146         BT_LOGI("Connecting to hostname : %s, port : %d, "
 147                         "target hostname : %s, session name : %s, "
 148                         "proto : %s",
 149                         viewer_connection->relay_hostname->str,
 150                         viewer_connection->port,
 151                         viewer_connection->target_hostname == NULL ?
 152                                 "<none>" : viewer_connection->target_hostname->str,
 153                         viewer_connection->session_name == NULL ?
 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_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_LOGI("Remote side has closed connection");
 200                 goto error;
 201         }
 202         if (ret_len == BT_SOCKET_ERROR) {
 203                 BT_LOGE("Error receiving version: %s", bt_socket_errormsg());
 204                 goto error;
 205         }
 206         BT_ASSERT(ret_len == sizeof(connect));
 207
 208         BT_LOGI("Received viewer session ID : %" PRIu64,
 209                         (uint64_t) be64toh(connect.viewer_session_id));
 210         BT_LOGI("Relayd version : %u.%u", be32toh(connect.major),
 211                         be32toh(connect.minor));
 212
 213         if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) {
 214                 BT_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_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_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_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_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_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_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         bt_value *map = NULL;
 312         bt_value *hostname = NULL;
 313         bt_value *session_name = NULL;
 314         bt_value *btval = NULL;
 315         int i, len;
 316         bool found = false;
 317
 318         len = bt_value_array_get_size(results);
 319         if (len < 0) {
 320                 BT_LOGE_STR("Error getting size of array.");
 321                 ret = -1;
 322                 goto end;
 323         }
 324         for (i = 0; i < len; i++) {
 325                 const char *hostname_str = NULL;
 326                 const char *session_name_str = NULL;
 327
 328                 map = bt_value_array_borrow_element_by_index(results, (size_t) i);
 329                 if (!map) {
 330                         BT_LOGE_STR("Error borrowing map.");
 331                         ret = -1;
 332                         goto end;
 333                 }
 334                 hostname = bt_value_map_borrow_entry_value(map, "target-hostname");
 335                 if (!hostname) {
 336                         BT_LOGE_STR("Error borrowing \"target-hostname\" entry.");
 337                         ret = -1;
 338                         goto end;
 339                 }
 340                 session_name = bt_value_map_borrow_entry_value(map, "session-name");
 341                 if (!session_name) {
 342                         BT_LOGE_STR("Error borrowing \"session-name\" entry.");
 343                         ret = -1;
 344                         goto end;
 345                 }
 346                 hostname_str = bt_value_string_get(hostname);
 347                 session_name_str = bt_value_string_get(session_name);
 348
 349                 if (!strcmp(session->hostname, hostname_str)
 350                                 && !strcmp(session->session_name,
 351                                         session_name_str)) {
 352                         int64_t val;
 353                         uint32_t streams = be32toh(session->streams);
 354                         uint32_t clients = be32toh(session->clients);
 355
 356                         found = true;
 357
 358                         btval = bt_value_map_borrow_entry_value(map, "stream-count");
 359                         if (!btval) {
 360                                 BT_LOGE_STR("Error borrowing \"stream-count\" entry.");
 361                                 ret = -1;
 362                                 goto end;
 363                         }
 364                         val = bt_value_signed_integer_get(btval);
 365                         /* sum */
 366                         val += streams;
 367                         bt_value_signed_integer_set(btval, val);
 368
 369                         btval = bt_value_map_borrow_entry_value(map, "client-count");
 370                         if (!btval) {
 371                                 BT_LOGE_STR("Error borrowing \"client-count\" entry.");
 372                                 ret = -1;
 373                                 goto end;
 374                         }
 375                         val = bt_value_signed_integer_get(btval);
 376                         /* max */
 377                         val = bt_max_t(int64_t, clients, val);
 378                         bt_value_signed_integer_set(btval, val);
 379                 }
 380
 381                 if (found) {
 382                         break;
 383                 }
 384         }
 385 end:
 386         *_found = found;
 387         return ret;
 388 }
 389
 390 static
 391 int list_append_session(bt_value *results,
 392                 GString *base_url,
 393                 const struct lttng_viewer_session *session,
 394                 struct live_viewer_connection *viewer_connection)
 395 {
 396         int ret = 0;
 397         bt_value_status ret_status;
 398         bt_value *map = NULL;
 399         GString *url = NULL;
 400         bool found = false;
 401
 402         /*
 403          * If the session already exists, add the stream count to it,
 404          * and do max of client counts.
 405          */
 406         ret = list_update_session(results, session, &found, viewer_connection);
 407         if (ret || found) {
 408                 goto end;
 409         }
 410
 411         map = bt_value_map_create();
 412         if (!map) {
 413                 BT_LOGE_STR("Error creating map value.");
 414                 ret = -1;
 415                 goto end;
 416         }
 417
 418         if (base_url->len < 1) {
 419                 BT_LOGE_STR("Error: base_url length smaller than 1.");
 420                 ret = -1;
 421                 goto end;
 422         }
 423         /*
 424          * key = "url",
 425          * value = <string>,
 426          */
 427         url = g_string_new(base_url->str);
 428         g_string_append(url, "/host/");
 429         g_string_append(url, session->hostname);
 430         g_string_append_c(url, '/');
 431         g_string_append(url, session->session_name);
 432
 433         ret_status = bt_value_map_insert_string_entry(map, "url", url->str);
 434         if (ret_status != BT_VALUE_STATUS_OK) {
 435                 BT_LOGE_STR("Error inserting \"url\" entry.");
 436                 ret = -1;
 437                 goto end;
 438         }
 439
 440         /*
 441          * key = "target-hostname",
 442          * value = <string>,
 443          */
 444         ret_status = bt_value_map_insert_string_entry(map, "target-hostname",
 445                 session->hostname);
 446         if (ret_status != BT_VALUE_STATUS_OK) {
 447                 BT_LOGE_STR("Error inserting \"target-hostname\" entry.");
 448                 ret = -1;
 449                 goto end;
 450         }
 451
 452         /*
 453          * key = "session-name",
 454          * value = <string>,
 455          */
 456         ret_status = bt_value_map_insert_string_entry(map, "session-name",
 457                 session->session_name);
 458         if (ret_status != BT_VALUE_STATUS_OK) {
 459                 BT_LOGE_STR("Error inserting \"session-name\" entry.");
 460                 ret = -1;
 461                 goto end;
 462         }
 463
 464         /*
 465          * key = "timer-us",
 466          * value = <integer>,
 467          */
 468         {
 469                 uint32_t live_timer = be32toh(session->live_timer);
 470
 471                 ret_status = bt_value_map_insert_signed_integer_entry(
 472                         map, "timer-us", live_timer);
 473                 if (ret_status != BT_VALUE_STATUS_OK) {
 474                         BT_LOGE_STR("Error inserting \"timer-us\" entry.");
 475                         ret = -1;
 476                         goto end;
 477                 }
 478         }
 479
 480         /*
 481          * key = "stream-count",
 482          * value = <integer>,
 483          */
 484         {
 485                 uint32_t streams = be32toh(session->streams);
 486
 487                 ret_status = bt_value_map_insert_signed_integer_entry(map,
 488                         "stream-count", streams);
 489                 if (ret_status != BT_VALUE_STATUS_OK) {
 490                         BT_LOGE_STR("Error inserting \"stream-count\" entry.");
 491                         ret = -1;
 492                         goto end;
 493                 }
 494         }
 495
 496         /*
 497          * key = "client-count",
 498          * value = <integer>,
 499          */
 500         {
 501                 uint32_t clients = be32toh(session->clients);
 502
 503                 ret_status = bt_value_map_insert_signed_integer_entry(map,
 504                         "client-count", clients);
 505                 if (ret_status != BT_VALUE_STATUS_OK) {
 506                         BT_LOGE_STR("Error inserting \"client-count\" entry.");
 507                         ret = -1;
 508                         goto end;
 509                 }
 510         }
 511
 512         ret_status = bt_value_array_append_element(results, map);
 513         if (ret_status != BT_VALUE_STATUS_OK) {
 514                 BT_LOGE_STR("Error appending map to results.");
 515                 ret = -1;
 516         }
 517
 518 end:
 519         if (url) {
 520                 g_string_free(url, true);
 521         }
 522         BT_VALUE_PUT_REF_AND_RESET(map);
 523         return ret;
 524 }
 525
 526 /*
 527  * Data structure returned:
 528  *
 529  * {
 530  *   <array> = {
 531  *     [n] = {
 532  *       <map> = {
 533  *         {
 534  *           key = "url",
 535  *           value = <string>,
 536  *         },
 537  *         {
 538  *           key = "target-hostname",
 539  *           value = <string>,
 540  *         },
 541  *         {
 542  *           key = "session-name",
 543  *           value = <string>,
 544  *         },
 545  *         {
 546  *           key = "timer-us",
 547  *           value = <integer>,
 548  *         },
 549  *         {
 550  *           key = "stream-count",
 551  *           value = <integer>,
 552  *         },
 553  *         {
 554  *           key = "client-count",
 555  *           value = <integer>,
 556  *         },
 557  *       },
 558  *     }
 559  *   }
 560  */
 561
 562 BT_HIDDEN
 563 bt_query_status live_viewer_connection_list_sessions(
 564                 struct live_viewer_connection *viewer_connection,
 565                 const bt_value **user_result)
 566 {
 567         bt_query_status status = BT_QUERY_STATUS_OK;
 568         bt_value *result = NULL;
 569         struct lttng_viewer_cmd cmd;
 570         struct lttng_viewer_list_sessions list;
 571         uint32_t i, sessions_count;
 572         ssize_t ret_len;
 573
 574         if (lttng_live_handshake(viewer_connection)) {
 575                 goto error;
 576         }
 577
 578         result = bt_value_array_create();
 579         if (!result) {
 580                 BT_LOGE("Error creating array");
 581                 status = BT_QUERY_STATUS_NOMEM;
 582                 goto error;
 583         }
 584
 585         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 586         cmd.data_size = htobe64((uint64_t) 0);
 587         cmd.cmd_version = htobe32(0);
 588
 589         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 590         if (ret_len == BT_SOCKET_ERROR) {
 591                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 592                 status = BT_QUERY_STATUS_ERROR;
 593                 goto error;
 594         }
 595         BT_ASSERT(ret_len == sizeof(cmd));
 596
 597         ret_len = lttng_live_recv(viewer_connection, &list, sizeof(list));
 598         if (ret_len == 0) {
 599                 BT_LOGI("Remote side has closed connection");
 600                 status = BT_QUERY_STATUS_ERROR;
 601                 goto error;
 602         }
 603         if (ret_len == BT_SOCKET_ERROR) {
 604                 BT_LOGE("Error receiving session list: %s", bt_socket_errormsg());
 605                 status = BT_QUERY_STATUS_ERROR;
 606                 goto error;
 607         }
 608         BT_ASSERT(ret_len == sizeof(list));
 609
 610         sessions_count = be32toh(list.sessions_count);
 611         for (i = 0; i < sessions_count; i++) {
 612                 struct lttng_viewer_session lsession;
 613
 614                 ret_len = lttng_live_recv(viewer_connection, &lsession,
 615                         sizeof(lsession));
 616                 if (ret_len == 0) {
 617                         BT_LOGI("Remote side has closed connection");
 618                         status = BT_QUERY_STATUS_ERROR;
 619                         goto error;
 620                 }
 621                 if (ret_len == BT_SOCKET_ERROR) {
 622                         BT_LOGE("Error receiving session: %s", bt_socket_errormsg());
 623                         status = BT_QUERY_STATUS_ERROR;
 624                         goto error;
 625                 }
 626                 BT_ASSERT(ret_len == sizeof(lsession));
 627                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 628                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 629                 if (list_append_session(result, viewer_connection->url,
 630                                 &lsession, viewer_connection)) {
 631                         status = BT_QUERY_STATUS_ERROR;
 632                         goto error;
 633                 }
 634         }
 635
 636         *user_result = result;
 637         goto end;
 638 error:
 639         BT_VALUE_PUT_REF_AND_RESET(result);
 640 end:
 641         return status;
 642 }
 643
 644 static
 645 int lttng_live_query_session_ids(struct lttng_live_msg_iter *lttng_live_msg_iter)
 646 {
 647         struct lttng_viewer_cmd cmd;
 648         struct lttng_viewer_list_sessions list;
 649         struct lttng_viewer_session lsession;
 650         uint32_t i, sessions_count;
 651         ssize_t ret_len;
 652         uint64_t session_id;
 653         struct live_viewer_connection *viewer_connection =
 654                         lttng_live_msg_iter->viewer_connection;
 655
 656         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 657         cmd.data_size = htobe64((uint64_t) 0);
 658         cmd.cmd_version = htobe32(0);
 659
 660         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 661         if (ret_len == BT_SOCKET_ERROR) {
 662                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 663                 goto error;
 664         }
 665         BT_ASSERT(ret_len == sizeof(cmd));
 666
 667         ret_len = lttng_live_recv(viewer_connection, &list, sizeof(list));
 668         if (ret_len == 0) {
 669                 BT_LOGI("Remote side has closed connection");
 670                 goto error;
 671         }
 672         if (ret_len == BT_SOCKET_ERROR) {
 673                 BT_LOGE("Error receiving session list: %s", bt_socket_errormsg());
 674                 goto error;
 675         }
 676         BT_ASSERT(ret_len == sizeof(list));
 677
 678         sessions_count = be32toh(list.sessions_count);
 679         for (i = 0; i < sessions_count; i++) {
 680                 ret_len = lttng_live_recv(viewer_connection,
 681                                 &lsession, sizeof(lsession));
 682                 if (ret_len == 0) {
 683                         BT_LOGI("Remote side has closed connection");
 684                         goto error;
 685                 }
 686                 if (ret_len == BT_SOCKET_ERROR) {
 687                         BT_LOGE("Error receiving session: %s", bt_socket_errormsg());
 688                         goto error;
 689                 }
 690                 BT_ASSERT(ret_len == sizeof(lsession));
 691                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 692                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 693                 session_id = be64toh(lsession.id);
 694
 695                 BT_LOGI("Adding session %" PRIu64 " hostname: %s session_name: %s",
 696                         session_id, lsession.hostname, lsession.session_name);
 697
 698                 if ((strncmp(lsession.session_name,
 699                         viewer_connection->session_name->str,
 700                         LTTNG_VIEWER_NAME_MAX) == 0) && (strncmp(lsession.hostname,
 701                                 viewer_connection->target_hostname->str,
 702                                 LTTNG_VIEWER_HOST_NAME_MAX) == 0)) {
 703                         if (lttng_live_add_session(lttng_live_msg_iter, session_id,
 704                                         lsession.hostname,
 705                                         lsession.session_name)) {
 706                                 goto error;
 707                         }
 708                 }
 709         }
 710
 711         return 0;
 712
 713 error:
 714         BT_LOGE("Unable to query session ids");
 715         return -1;
 716 }
 717
 718 BT_HIDDEN
 719 int lttng_live_create_viewer_session(
 720                 struct lttng_live_msg_iter *lttng_live_msg_iter)
 721 {
 722         struct lttng_viewer_cmd cmd;
 723         struct lttng_viewer_create_session_response resp;
 724         ssize_t ret_len;
 725         struct live_viewer_connection *viewer_connection =
 726                         lttng_live_msg_iter->viewer_connection;
 727
 728         cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
 729         cmd.data_size = htobe64((uint64_t) 0);
 730         cmd.cmd_version = htobe32(0);
 731
 732         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 733         if (ret_len == BT_SOCKET_ERROR) {
 734                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 735                 goto error;
 736         }
 737         BT_ASSERT(ret_len == sizeof(cmd));
 738
 739         ret_len = lttng_live_recv(viewer_connection, &resp, sizeof(resp));
 740         if (ret_len == 0) {
 741                 BT_LOGI("Remote side has closed connection");
 742                 goto error;
 743         }
 744         if (ret_len == BT_SOCKET_ERROR) {
 745                 BT_LOGE("Error receiving create session reply: %s", bt_socket_errormsg());
 746                 goto error;
 747         }
 748         BT_ASSERT(ret_len == sizeof(resp));
 749
 750         if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
 751                 BT_LOGE("Error creating viewer session");
 752                 goto error;
 753         }
 754         if (lttng_live_query_session_ids(lttng_live_msg_iter)) {
 755                 goto error;
 756         }
 757
 758         return 0;
 759
 760 error:
 761         return -1;
 762 }
 763
 764 static
 765 int receive_streams(struct lttng_live_session *session,
 766                 uint32_t stream_count)
 767 {
 768         ssize_t ret_len;
 769         uint32_t i;
 770         struct lttng_live_msg_iter *lttng_live_msg_iter =
 771                         session->lttng_live_msg_iter;
 772         struct live_viewer_connection *viewer_connection =
 773                         lttng_live_msg_iter->viewer_connection;
 774
 775         BT_LOGI("Getting %" PRIu32 " new streams:", stream_count);
 776         for (i = 0; i < stream_count; i++) {
 777                 struct lttng_viewer_stream stream;
 778                 struct lttng_live_stream_iterator *live_stream;
 779                 uint64_t stream_id;
 780                 uint64_t ctf_trace_id;
 781
 782                 ret_len = lttng_live_recv(viewer_connection, &stream, sizeof(stream));
 783                 if (ret_len == 0) {
 784                         BT_LOGI("Remote side has closed connection");
 785                         goto error;
 786                 }
 787                 if (ret_len == BT_SOCKET_ERROR) {
 788                         BT_LOGE("Error receiving stream");
 789                         goto error;
 790                 }
 791                 BT_ASSERT(ret_len == sizeof(stream));
 792                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
 793                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 794                 stream_id = be64toh(stream.id);
 795                 ctf_trace_id = be64toh(stream.ctf_trace_id);
 796
 797                 if (stream.metadata_flag) {
 798                         BT_LOGI("    metadata stream %" PRIu64 " : %s/%s",
 799                                         stream_id, stream.path_name,
 800                                         stream.channel_name);
 801                         if (lttng_live_metadata_create_stream(session,
 802                                         ctf_trace_id, stream_id,
 803                                         stream.path_name)) {
 804                                 BT_LOGE("Error creating metadata stream");
 805
 806                                 goto error;
 807                         }
 808                         session->lazy_stream_msg_init = true;
 809                 } else {
 810                         BT_LOGI("    stream %" PRIu64 " : %s/%s",
 811                                         stream_id, stream.path_name,
 812                                         stream.channel_name);
 813                         live_stream = lttng_live_stream_iterator_create(session,
 814                                 ctf_trace_id, stream_id);
 815                         if (!live_stream) {
 816                                 BT_LOGE("Error creating streamn");
 817                                 goto error;
 818                         }
 819                 }
 820         }
 821         return 0;
 822
 823 error:
 824         return -1;
 825 }
 826
 827 BT_HIDDEN
 828 int lttng_live_attach_session(struct lttng_live_session *session)
 829 {
 830         struct lttng_viewer_cmd cmd;
 831         struct lttng_viewer_attach_session_request rq;
 832         struct lttng_viewer_attach_session_response rp;
 833         ssize_t ret_len;
 834         struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
 835         struct live_viewer_connection *viewer_connection =
 836                         lttng_live_msg_iter->viewer_connection;
 837         uint64_t session_id = session->id;
 838         uint32_t streams_count;
 839         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
 840         char cmd_buf[cmd_buf_len];
 841
 842         cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 843         cmd.data_size = htobe64((uint64_t) sizeof(rq));
 844         cmd.cmd_version = htobe32(0);
 845
 846         memset(&rq, 0, sizeof(rq));
 847         rq.session_id = htobe64(session_id);
 848         // TODO: add cmd line parameter to select seek beginning
 849         // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
 850         rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
 851
 852         /*
 853          * Merge the cmd and connection request to prevent a write-write
 854          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
 855          * second write to be performed quickly in presence of Nagle's algorithm.
 856          */
 857         memcpy(cmd_buf, &cmd, sizeof(cmd));
 858         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
 859         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
 860         if (ret_len == BT_SOCKET_ERROR) {
 861                 BT_LOGE("Error sending attach request: %s", bt_socket_errormsg());
 862                 goto error;
 863         }
 864
 865         BT_ASSERT(ret_len == cmd_buf_len);
 866         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
 867         if (ret_len == 0) {
 868                 BT_LOGI("Remote side has closed connection");
 869                 goto error;
 870         }
 871         if (ret_len == BT_SOCKET_ERROR) {
 872                 BT_LOGE("Error receiving attach response: %s", bt_socket_errormsg());
 873                 goto error;
 874         }
 875         BT_ASSERT(ret_len == sizeof(rp));
 876
 877         streams_count = be32toh(rp.streams_count);
 878         switch(be32toh(rp.status)) {
 879         case LTTNG_VIEWER_ATTACH_OK:
 880                 break;
 881         case LTTNG_VIEWER_ATTACH_UNK:
 882                 BT_LOGW("Session id %" PRIu64 " is unknown", session_id);
 883                 goto error;
 884         case LTTNG_VIEWER_ATTACH_ALREADY:
 885                 BT_LOGW("There is already a viewer attached to this session");
 886                 goto error;
 887         case LTTNG_VIEWER_ATTACH_NOT_LIVE:
 888                 BT_LOGW("Not a live session");
 889                 goto error;
 890         case LTTNG_VIEWER_ATTACH_SEEK_ERR:
 891                 BT_LOGE("Wrong seek parameter");
 892                 goto error;
 893         default:
 894                 BT_LOGE("Unknown attach return code %u", be32toh(rp.status));
 895                 goto error;
 896         }
 897
 898         /* We receive the initial list of streams. */
 899         if (receive_streams(session, streams_count)) {
 900                 goto error;
 901         }
 902
 903         session->attached = true;
 904         session->new_streams_needed = false;
 905
 906         return 0;
 907
 908 error:
 909         return -1;
 910 }
 911
 912 BT_HIDDEN
 913 int lttng_live_detach_session(struct lttng_live_session *session)
 914 {
 915         struct lttng_viewer_cmd cmd;
 916         struct lttng_viewer_detach_session_request rq;
 917         struct lttng_viewer_detach_session_response rp;
 918         ssize_t ret_len;
 919         struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
 920         struct live_viewer_connection *viewer_connection =
 921                         lttng_live_msg_iter->viewer_connection;
 922         uint64_t session_id = session->id;
 923         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
 924         char cmd_buf[cmd_buf_len];
 925
 926         if (!session->attached) {
 927                 return 0;
 928         }
 929
 930         cmd.cmd = htobe32(LTTNG_VIEWER_DETACH_SESSION);
 931         cmd.data_size = htobe64((uint64_t) sizeof(rq));
 932         cmd.cmd_version = htobe32(0);
 933
 934         memset(&rq, 0, sizeof(rq));
 935         rq.session_id = htobe64(session_id);
 936
 937         /*
 938          * Merge the cmd and connection request to prevent a write-write
 939          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
 940          * second write to be performed quickly in presence of Nagle's algorithm.
 941          */
 942         memcpy(cmd_buf, &cmd, sizeof(cmd));
 943         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
 944         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
 945         if (ret_len == BT_SOCKET_ERROR) {
 946                 BT_LOGE("Error sending detach request: %s", bt_socket_errormsg());
 947                 goto error;
 948         }
 949
 950         BT_ASSERT(ret_len == cmd_buf_len);
 951         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
 952         if (ret_len == 0) {
 953                 BT_LOGI("Remote side has closed connection");
 954                 goto error;
 955         }
 956         if (ret_len == BT_SOCKET_ERROR) {
 957                 BT_LOGE("Error receiving detach response: %s", bt_socket_errormsg());
 958                 goto error;
 959         }
 960         BT_ASSERT(ret_len == sizeof(rp));
 961
 962         switch(be32toh(rp.status)) {
 963         case LTTNG_VIEWER_DETACH_SESSION_OK:
 964                 break;
 965         case LTTNG_VIEWER_DETACH_SESSION_UNK:
 966                 BT_LOGW("Session id %" PRIu64 " is unknown", session_id);
 967                 goto error;
 968         case LTTNG_VIEWER_DETACH_SESSION_ERR:
 969                 BT_LOGW("Error detaching session id %" PRIu64 "", session_id);
 970                 goto error;
 971         default:
 972                 BT_LOGE("Unknown detach return code %u", be32toh(rp.status));
 973                 goto error;
 974         }
 975
 976         session->attached = false;
 977
 978         return 0;
 979
 980 error:
 981         return -1;
 982 }
 983
 984 BT_HIDDEN
 985 ssize_t lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace,
 986                 FILE *fp)
 987 {
 988         uint64_t len = 0;
 989         int ret;
 990         struct lttng_viewer_cmd cmd;
 991         struct lttng_viewer_get_metadata rq;
 992         struct lttng_viewer_metadata_packet rp;
 993         char *data = NULL;
 994         ssize_t ret_len;
 995         struct lttng_live_session *session = trace->session;
 996         struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
 997         struct lttng_live_metadata *metadata = trace->metadata;
 998         struct live_viewer_connection *viewer_connection =
 999                         lttng_live_msg_iter->viewer_connection;
1000         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1001         char cmd_buf[cmd_buf_len];
1002
1003         rq.stream_id = htobe64(metadata->stream_id);
1004         cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
1005         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1006         cmd.cmd_version = htobe32(0);
1007
1008         /*
1009          * Merge the cmd and connection request to prevent a write-write
1010          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1011          * second write to be performed quickly in presence of Nagle's algorithm.
1012          */
1013         memcpy(cmd_buf, &cmd, sizeof(cmd));
1014         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1015         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1016         if (ret_len == BT_SOCKET_ERROR) {
1017                 BT_LOGE("Error sending get_metadata request: %s", bt_socket_errormsg());
1018                 goto error;
1019         }
1020
1021         BT_ASSERT(ret_len == cmd_buf_len);
1022         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1023         if (ret_len == 0) {
1024                 BT_LOGI("Remote side has closed connection");
1025                 goto error;
1026         }
1027         if (ret_len == BT_SOCKET_ERROR) {
1028                 BT_LOGE("Error receiving get_metadata response: %s", bt_socket_errormsg());
1029                 goto error;
1030         }
1031         BT_ASSERT(ret_len == sizeof(rp));
1032
1033         switch (be32toh(rp.status)) {
1034                 case LTTNG_VIEWER_METADATA_OK:
1035                         BT_LOGD("get_metadata : OK");
1036                         break;
1037                 case LTTNG_VIEWER_NO_NEW_METADATA:
1038                         BT_LOGD("get_metadata : NO NEW");
1039                         ret = 0;
1040                         goto end;
1041                 case LTTNG_VIEWER_METADATA_ERR:
1042                         BT_LOGD("get_metadata : ERR");
1043                         goto error;
1044                 default:
1045                         BT_LOGD("get_metadata : UNKNOWN");
1046                         goto error;
1047         }
1048
1049         len = be64toh(rp.len);
1050         BT_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_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_LOGI("Remote side has closed connection");
1063                 goto error_free_data;
1064         }
1065         if (ret_len == BT_SOCKET_ERROR) {
1066                 BT_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_LOGE("Writing in the metadata fp");
1076                 goto error_free_data;
1077         }
1078         BT_ASSERT(ret_len == len);
1079         free(data);
1080         ret = len;
1081 end:
1082         return ret;
1083
1084 error_free_data:
1085         free(data);
1086 error:
1087         return -1;
1088 }
1089
1090 /*
1091  * Assign the fields from a lttng_viewer_index to a packet_index.
1092  */
1093 static
1094 void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
1095                 struct packet_index *pindex)
1096 {
1097         BT_ASSERT(lindex);
1098         BT_ASSERT(pindex);
1099
1100         pindex->offset = be64toh(lindex->offset);
1101         pindex->packet_size = be64toh(lindex->packet_size);
1102         pindex->content_size = be64toh(lindex->content_size);
1103         pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
1104         pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
1105         pindex->events_discarded = be64toh(lindex->events_discarded);
1106 }
1107
1108 BT_HIDDEN
1109 enum lttng_live_iterator_status lttng_live_get_next_index(
1110                 struct lttng_live_msg_iter *lttng_live_msg_iter,
1111                 struct lttng_live_stream_iterator *stream,
1112                 struct packet_index *index)
1113 {
1114         struct lttng_viewer_cmd cmd;
1115         struct lttng_viewer_get_next_index rq;
1116         ssize_t ret_len;
1117         struct lttng_viewer_index rp;
1118         uint32_t flags, status;
1119         enum lttng_live_iterator_status retstatus =
1120                         LTTNG_LIVE_ITERATOR_STATUS_OK;
1121         struct live_viewer_connection *viewer_connection =
1122                         lttng_live_msg_iter->viewer_connection;
1123         struct lttng_live_trace *trace = stream->trace;
1124         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1125         char cmd_buf[cmd_buf_len];
1126         struct lttng_live_component *lttng_live =
1127                 lttng_live_msg_iter->lttng_live_comp;
1128
1129         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1130         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1131         cmd.cmd_version = htobe32(0);
1132
1133
1134         memset(&rq, 0, sizeof(rq));
1135         rq.stream_id = htobe64(stream->viewer_stream_id);
1136
1137         /*
1138          * Merge the cmd and connection request to prevent a write-write
1139          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1140          * second write to be performed quickly in presence of Nagle's algorithm.
1141          */
1142         memcpy(cmd_buf, &cmd, sizeof(cmd));
1143         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1144         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1145         if (ret_len == BT_SOCKET_ERROR) {
1146                 BT_LOGE("Error sending get_next_index request: %s",
1147                                 bt_socket_errormsg());
1148                 goto error;
1149         }
1150
1151         BT_ASSERT(ret_len == cmd_buf_len);
1152         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1153         if (ret_len == 0) {
1154                 BT_LOGI("Remote side has closed connection");
1155                 goto error;
1156         }
1157         if (ret_len == BT_SOCKET_ERROR) {
1158                 BT_LOGE("Error receiving get_next_index response: %s",
1159                                 bt_socket_errormsg());
1160                 goto error;
1161         }
1162         BT_ASSERT(ret_len == sizeof(rp));
1163
1164         flags = be32toh(rp.flags);
1165         status = be32toh(rp.status);
1166
1167         switch (status) {
1168         case LTTNG_VIEWER_INDEX_INACTIVE:
1169         {
1170                 uint64_t ctf_stream_class_id;
1171
1172                 BT_LOGD("get_next_index: inactive");
1173                 memset(index, 0, sizeof(struct packet_index));
1174                 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
1175                 stream->current_inactivity_ts = index->ts_cycles.timestamp_end;
1176                 ctf_stream_class_id = be64toh(rp.stream_id);
1177                 if (stream->ctf_stream_class_id != -1ULL) {
1178                         BT_ASSERT(stream->ctf_stream_class_id ==
1179                                 ctf_stream_class_id);
1180                 } else {
1181                         stream->ctf_stream_class_id = ctf_stream_class_id;
1182                 }
1183                 stream->state = LTTNG_LIVE_STREAM_QUIESCENT;
1184                 break;
1185         }
1186         case LTTNG_VIEWER_INDEX_OK:
1187         {
1188                 uint64_t ctf_stream_class_id;
1189
1190                 BT_LOGD("get_next_index: OK");
1191                 lttng_index_to_packet_index(&rp, index);
1192                 ctf_stream_class_id = be64toh(rp.stream_id);
1193                 if (stream->ctf_stream_class_id != -1ULL) {
1194                         BT_ASSERT(stream->ctf_stream_class_id ==
1195                                 ctf_stream_class_id);
1196                 } else {
1197                         stream->ctf_stream_class_id = ctf_stream_class_id;
1198                 }
1199
1200                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_DATA;
1201
1202                 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1203                         BT_LOGD("get_next_index: new metadata needed");
1204                         trace->new_metadata_needed = true;
1205                 }
1206                 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1207                         BT_LOGD("get_next_index: new streams needed");
1208                         lttng_live_need_new_streams(lttng_live_msg_iter);
1209                 }
1210                 break;
1211         }
1212         case LTTNG_VIEWER_INDEX_RETRY:
1213                 BT_LOGD("get_next_index: retry");
1214                 memset(index, 0, sizeof(struct packet_index));
1215                 retstatus = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1216                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1217                 goto end;
1218         case LTTNG_VIEWER_INDEX_HUP:
1219                 BT_LOGD("get_next_index: stream hung up");
1220                 memset(index, 0, sizeof(struct packet_index));
1221                 index->offset = EOF;
1222                 retstatus = LTTNG_LIVE_ITERATOR_STATUS_END;
1223                 stream->state = LTTNG_LIVE_STREAM_EOF;
1224                 break;
1225         case LTTNG_VIEWER_INDEX_ERR:
1226                 BT_LOGE("get_next_index: error");
1227                 memset(index, 0, sizeof(struct packet_index));
1228                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1229                 goto error;
1230         default:
1231                 BT_LOGE("get_next_index: unknown value");
1232                 memset(index, 0, sizeof(struct packet_index));
1233                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1234                 goto error;
1235         }
1236 end:
1237         return retstatus;
1238
1239 error:
1240         if (lttng_live_graph_is_canceled(lttng_live)) {
1241                 retstatus = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1242         } else {
1243                 retstatus = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1244         }
1245         return retstatus;
1246 }
1247
1248 BT_HIDDEN
1249 enum bt_msg_iter_medium_status lttng_live_get_stream_bytes(
1250                 struct lttng_live_msg_iter *lttng_live_msg_iter,
1251                 struct lttng_live_stream_iterator *stream, uint8_t *buf,
1252                 uint64_t offset, uint64_t req_len, uint64_t *recv_len)
1253 {
1254         enum bt_msg_iter_medium_status retstatus = BT_MSG_ITER_MEDIUM_STATUS_OK;
1255         struct lttng_viewer_cmd cmd;
1256         struct lttng_viewer_get_packet rq;
1257         struct lttng_viewer_trace_packet rp;
1258         ssize_t ret_len;
1259         uint32_t flags, status;
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         struct lttng_live_component *lttng_live =
1266                 lttng_live_msg_iter->lttng_live_comp;
1267
1268         BT_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_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_LOGI("Remote side has closed connection");
1296                 goto error;
1297         }
1298         if (ret_len == BT_SOCKET_ERROR) {
1299                 BT_LOGE("Error receiving get_data response: %s", bt_socket_errormsg());
1300                 goto error;
1301         }
1302         if (ret_len != sizeof(rp)) {
1303                 BT_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_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_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_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_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_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_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_LOGI("Remote side has closed connection");
1353                 goto error;
1354         }
1355         if (ret_len == BT_SOCKET_ERROR) {
1356                 BT_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)) {
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         ssize_t ret_len;
1386         struct lttng_live_msg_iter *lttng_live_msg_iter =
1387                 session->lttng_live_msg_iter;
1388         struct live_viewer_connection *viewer_connection =
1389                 lttng_live_msg_iter->viewer_connection;
1390         struct lttng_live_component *lttng_live =
1391                 lttng_live_msg_iter->lttng_live_comp;
1392         uint32_t streams_count;
1393         const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1394         char cmd_buf[cmd_buf_len];
1395
1396         if (!session->new_streams_needed) {
1397                 return LTTNG_LIVE_ITERATOR_STATUS_OK;
1398         }
1399
1400         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1401         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1402         cmd.cmd_version = htobe32(0);
1403
1404         memset(&rq, 0, sizeof(rq));
1405         rq.session_id = htobe64(session->id);
1406
1407         /*
1408          * Merge the cmd and connection request to prevent a write-write
1409          * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1410          * second write to be performed quickly in presence of Nagle's algorithm.
1411          */
1412         memcpy(cmd_buf, &cmd, sizeof(cmd));
1413         memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1414         ret_len = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1415         if (ret_len == BT_SOCKET_ERROR) {
1416                 BT_LOGE("Error sending get_new_streams request: %s",
1417                                 bt_socket_errormsg());
1418                 goto error;
1419         }
1420
1421         BT_ASSERT(ret_len == cmd_buf_len);
1422         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1423         if (ret_len == 0) {
1424                 BT_LOGI("Remote side has closed connection");
1425                 goto error;
1426         }
1427         if (ret_len == BT_SOCKET_ERROR) {
1428                 BT_LOGE("Error receiving get_new_streams response");
1429                 goto error;
1430         }
1431         BT_ASSERT(ret_len == sizeof(rp));
1432
1433         streams_count = be32toh(rp.streams_count);
1434
1435         switch(be32toh(rp.status)) {
1436         case LTTNG_VIEWER_NEW_STREAMS_OK:
1437                 session->new_streams_needed = false;
1438                 break;
1439         case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1440                 session->new_streams_needed = false;
1441                 goto end;
1442         case LTTNG_VIEWER_NEW_STREAMS_HUP:
1443                 session->new_streams_needed = false;
1444                 session->closed = true;
1445                 status = LTTNG_LIVE_ITERATOR_STATUS_END;
1446                 goto end;
1447         case LTTNG_VIEWER_NEW_STREAMS_ERR:
1448                 BT_LOGE("get_new_streams error");
1449                 goto error;
1450         default:
1451                 BT_LOGE("Unknown return code %u", be32toh(rp.status));
1452                 goto error;
1453         }
1454
1455         if (receive_streams(session, streams_count)) {
1456                 goto error;
1457         }
1458 end:
1459         return status;
1460
1461 error:
1462         if (lttng_live_graph_is_canceled(lttng_live)) {
1463                 status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1464         } else {
1465                 status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1466         }
1467         return status;
1468 }
1469
1470 BT_HIDDEN
1471 struct live_viewer_connection *live_viewer_connection_create(
1472                 const char *url, bool in_query,
1473                 struct lttng_live_msg_iter *lttng_live_msg_iter)
1474 {
1475         struct live_viewer_connection *viewer_connection;
1476
1477         viewer_connection = g_new0(struct live_viewer_connection, 1);
1478
1479         if (bt_socket_init() != 0) {
1480                 goto error;
1481         }
1482
1483         viewer_connection->log_level = lttng_live_msg_iter->log_level;
1484         bt_object_init_shared(&viewer_connection->obj, connection_release);
1485         viewer_connection->control_sock = BT_INVALID_SOCKET;
1486         viewer_connection->port = -1;
1487         viewer_connection->in_query = in_query;
1488         viewer_connection->lttng_live_msg_iter = lttng_live_msg_iter;
1489         viewer_connection->url = g_string_new(url);
1490         if (!viewer_connection->url) {
1491                 goto error;
1492         }
1493
1494         BT_LOGI("Establishing connection to url \"%s\"...", url);
1495         if (lttng_live_connect_viewer(viewer_connection)) {
1496                 goto error_report;
1497         }
1498         BT_LOGI("Connection to url \"%s\" is established", url);
1499         return viewer_connection;
1500
1501 error_report:
1502         BT_LOGW("Failure to establish connection to url \"%s\"", url);
1503 error:
1504         g_free(viewer_connection);
1505         return NULL;
1506 }
1507
1508 BT_HIDDEN
1509 void live_viewer_connection_destroy(
1510                 struct live_viewer_connection *viewer_connection)
1511 {
1512         BT_LOGI("Closing connection to url \"%s\"", viewer_connection->url->str);
1513         lttng_live_disconnect_viewer(viewer_connection);
1514         g_string_free(viewer_connection->url, true);
1515         if (viewer_connection->relay_hostname) {
1516                 g_string_free(viewer_connection->relay_hostname, true);
1517         }
1518         if (viewer_connection->target_hostname) {
1519                 g_string_free(viewer_connection->target_hostname, true);
1520         }
1521         if (viewer_connection->session_name) {
1522                 g_string_free(viewer_connection->session_name, true);
1523         }
1524         g_free(viewer_connection);
1525
1526         bt_socket_fini();
1527 }