src/lttng-live-comm.c

   1 /*
   2  * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
   3  *                      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 #include <sys/socket.h>
  25 #include <sys/types.h>
  26 #include <netinet/in.h>
  27 #include <netdb.h>
  28 #include <stdio.h>
  29 #include <string.h>
  30 #include <stdlib.h>
  31 #include <unistd.h>
  32 #include <errno.h>
  33 #include <inttypes.h>
  34 #include <fcntl.h>
  35 #include <sys/mman.h>
  36 #include <poll.h>
  37
  38 #include <babeltrace/ctf/ctf-index.h>
  39
  40 #include <babeltrace/babeltrace.h>
  41 #include <babeltrace/ctf/events.h>
  42 #include <babeltrace/ctf/callbacks.h>
  43 #include <babeltrace/ctf/iterator.h>
  44
  45 /* for packet_index */
  46 #include <babeltrace/ctf/types.h>
  47
  48 #include <babeltrace/ctf/metadata.h>
  49 #include <babeltrace/ctf-text/types.h>
  50 #include <babeltrace/ctf/events-internal.h>
  51 #include <babeltrace/endian.h>
  52 #include <babeltrace/compat/memstream.h>
  53 /*
  54 #include <formats/ctf/events-private.h>
  55 replaced with
  56 */
  57 #include "network-live.h"
  58 #include "lttngtop.h"
  59
  60 #include "lttng-live-comm.h"
  61 #include "lttng-viewer-abi.h"
  62
  63 #define ACTIVE_POLL_DELAY       100     /* ms */
  64
  65 /*
  66  * Memory allocation zeroed
  67  */
  68 #define zmalloc(x) calloc(1, x)
  69
  70 #ifndef max_t
  71 #define max_t(type, a, b)       \
  72         ((type) (a) > (type) (b) ? (type) (a) : (type) (b))
  73 #endif
  74
  75 static void ctf_live_packet_seek(struct bt_stream_pos *stream_pos,
  76                 size_t index, int whence);
  77 static void add_traces(gpointer key, gpointer value, gpointer user_data);
  78 static int del_traces(gpointer key, gpointer value, gpointer user_data);
  79 static int get_new_metadata(struct lttng_live_ctx *ctx,
  80                 struct lttng_live_viewer_stream *viewer_stream,
  81                 char **metadata_buf);
  82
  83 static
  84 ssize_t lttng_live_recv(int fd, void *buf, size_t len)
  85 {
  86         ssize_t ret;
  87         size_t copied = 0, to_copy = len;
  88
  89         do {
  90                 ret = recv(fd, buf + copied, to_copy, 0);
  91                 if (ret > 0) {
  92                         assert(ret <= to_copy);
  93                         copied += ret;
  94                         to_copy -= ret;
  95                 }
  96         } while ((ret > 0 && to_copy > 0)
  97                 || (ret < 0 && errno == EINTR));
  98         if (ret > 0)
  99                 ret = copied;
 100         /* ret = 0 means orderly shutdown, ret < 0 is error. */
 101         return ret;
 102 }
 103
 104 static
 105 ssize_t lttng_live_send(int fd, const void *buf, size_t len)
 106 {
 107         ssize_t ret;
 108
 109         do {
 110                 ret = send(fd, buf, len, MSG_NOSIGNAL);
 111         } while (ret < 0 && errno == EINTR);
 112         return ret;
 113 }
 114
 115 int lttng_live_connect_viewer(struct lttng_live_ctx *ctx)
 116 {
 117         struct hostent *host;
 118         struct sockaddr_in server_addr;
 119         int ret;
 120
 121         if (lttng_live_should_quit()) {
 122                 ret = -1;
 123                 goto end;
 124         }
 125
 126         host = gethostbyname(ctx->relay_hostname);
 127         if (!host) {
 128                 fprintf(stderr, "[error] Cannot lookup hostname %s\n",
 129                         ctx->relay_hostname);
 130                 goto error;
 131         }
 132
 133         if ((ctx->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
 134                 perror("Socket");
 135                 goto error;
 136         }
 137
 138         server_addr.sin_family = AF_INET;
 139         server_addr.sin_port = htons(ctx->port);
 140         server_addr.sin_addr = *((struct in_addr *) host->h_addr);
 141         bzero(&(server_addr.sin_zero), 8);
 142
 143         if (connect(ctx->control_sock, (struct sockaddr *) &server_addr,
 144                                 sizeof(struct sockaddr)) == -1) {
 145                 perror("Connect");
 146                 goto error;
 147         }
 148
 149         ret = 0;
 150
 151 end:
 152         return ret;
 153
 154 error:
 155         fprintf(stderr, "[error] Connection failed\n");
 156         return -1;
 157 }
 158
 159 int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
 160 {
 161         struct lttng_viewer_cmd cmd;
 162         struct lttng_viewer_connect connect;
 163         int ret;
 164         ssize_t ret_len;
 165
 166         if (lttng_live_should_quit()) {
 167                 ret = -1;
 168                 goto end;
 169         }
 170
 171         cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
 172         cmd.data_size = sizeof(connect);
 173         cmd.cmd_version = 0;
 174
 175         connect.viewer_session_id = -1ULL;      /* will be set on recv */
 176         connect.major = htobe32(LTTNG_LIVE_MAJOR);
 177         connect.minor = htobe32(LTTNG_LIVE_MINOR);
 178         connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
 179
 180         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 181         if (ret_len < 0) {
 182                 perror("[error] Error sending cmd");
 183                 goto error;
 184         }
 185         assert(ret_len == sizeof(cmd));
 186
 187         ret_len = lttng_live_send(ctx->control_sock, &connect, sizeof(connect));
 188         if (ret_len < 0) {
 189                 perror("[error] Error sending version");
 190                 goto error;
 191         }
 192         assert(ret_len == sizeof(connect));
 193
 194         ret_len = lttng_live_recv(ctx->control_sock, &connect, sizeof(connect));
 195         if (ret_len == 0) {
 196                 fprintf(stderr, "[error] Remote side has closed connection\n");
 197                 goto error;
 198         }
 199         if (ret_len < 0) {
 200                 perror("[error] Error receiving version");
 201                 goto error;
 202         }
 203         assert(ret_len == sizeof(connect));
 204
 205         printf_verbose("Received viewer session ID : %" PRIu64 "\n",
 206                         be64toh(connect.viewer_session_id));
 207         printf_verbose("Relayd version : %u.%u\n", be32toh(connect.major),
 208                         be32toh(connect.minor));
 209
 210         if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) {
 211                 fprintf(stderr, "[error] Incompatible lttng-relayd protocol\n");
 212                 goto error;
 213         }
 214         /* Use the smallest protocol version implemented. */
 215         if (LTTNG_LIVE_MINOR > be32toh(connect.minor)) {
 216                 ctx->minor =  be32toh(connect.minor);
 217         } else {
 218                 ctx->minor =  LTTNG_LIVE_MINOR;
 219         }
 220         ctx->major = LTTNG_LIVE_MAJOR;
 221         ret = 0;
 222 end:
 223         return ret;
 224
 225 error:
 226         fprintf(stderr, "[error] Unable to establish connection\n");
 227         return -1;
 228 }
 229
 230 static
 231 void free_session_list(GPtrArray *session_list)
 232 {
 233         int i;
 234         struct lttng_live_relay_session *relay_session;
 235
 236         for (i = 0; i < session_list->len; i++) {
 237                 relay_session = g_ptr_array_index(session_list, i);
 238                 free(relay_session->name);
 239                 free(relay_session->hostname);
 240         }
 241         g_ptr_array_free(session_list, TRUE);
 242 }
 243
 244 static
 245 void print_session_list(GPtrArray *session_list, const char *path)
 246 {
 247         int i;
 248         struct lttng_live_relay_session *relay_session;
 249
 250         for (i = 0; i < session_list->len; i++) {
 251                 relay_session = g_ptr_array_index(session_list, i);
 252                 fprintf(stdout, "%s/host/%s/%s (timer = %u, "
 253                                 "%u stream(s), %u client(s) connected)\n",
 254                                 path, relay_session->hostname,
 255                                 relay_session->name, relay_session->timer,
 256                                 relay_session->streams, relay_session->clients);
 257         }
 258 }
 259
 260 static
 261 void update_session_list(GPtrArray *session_list, char *hostname,
 262                 char *session_name, uint32_t streams, uint32_t clients,
 263                 uint32_t timer)
 264 {
 265         int i, found = 0;
 266         struct lttng_live_relay_session *relay_session;
 267
 268         for (i = 0; i < session_list->len; i++) {
 269                 relay_session = g_ptr_array_index(session_list, i);
 270                 if ((strncmp(relay_session->hostname, hostname, NAME_MAX) == 0) &&
 271                                 strncmp(relay_session->name,
 272                                         session_name, NAME_MAX) == 0) {
 273                         relay_session->streams += streams;
 274                         if (relay_session->clients < clients)
 275                                 relay_session->clients = clients;
 276                         found = 1;
 277                         break;
 278                 }
 279         }
 280         if (found)
 281                 return;
 282
 283         relay_session = g_new0(struct lttng_live_relay_session, 1);
 284         relay_session->hostname = strndup(hostname, NAME_MAX);
 285         relay_session->name = strndup(session_name, NAME_MAX);
 286         relay_session->clients = clients;
 287         relay_session->streams = streams;
 288         relay_session->timer = timer;
 289         g_ptr_array_add(session_list, relay_session);
 290 }
 291
 292 int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
 293 {
 294         struct lttng_viewer_cmd cmd;
 295         struct lttng_viewer_list_sessions list;
 296         struct lttng_viewer_session lsession;
 297         int i, ret, sessions_count, print_list = 0;
 298         ssize_t ret_len;
 299         uint64_t session_id;
 300         GPtrArray *session_list = NULL;
 301
 302         if (lttng_live_should_quit()) {
 303                 ret = -1;
 304                 goto end;
 305         }
 306
 307         if (strlen(ctx->session_name) == 0) {
 308                 print_list = 1;
 309                 session_list = g_ptr_array_new();
 310         }
 311
 312         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 313         cmd.data_size = 0;
 314         cmd.cmd_version = 0;
 315
 316         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 317         if (ret_len < 0) {
 318                 perror("[error] Error sending cmd");
 319                 goto error;
 320         }
 321         assert(ret_len == sizeof(cmd));
 322
 323         ret_len = lttng_live_recv(ctx->control_sock, &list, sizeof(list));
 324         if (ret_len == 0) {
 325                 fprintf(stderr, "[error] Remote side has closed connection\n");
 326                 goto error;
 327         }
 328         if (ret_len < 0) {
 329                 perror("[error] Error receiving session list");
 330                 goto error;
 331         }
 332         assert(ret_len == sizeof(list));
 333
 334         sessions_count = be32toh(list.sessions_count);
 335         for (i = 0; i < sessions_count; i++) {
 336                 ret_len = lttng_live_recv(ctx->control_sock, &lsession, sizeof(lsession));
 337                 if (ret_len == 0) {
 338                         fprintf(stderr, "[error] Remote side has closed connection\n");
 339                         goto error;
 340                 }
 341                 if (ret_len < 0) {
 342                         perror("[error] Error receiving session");
 343                         goto error;
 344                 }
 345                 assert(ret_len == sizeof(lsession));
 346                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 347                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 348                 session_id = be64toh(lsession.id);
 349
 350                 if (print_list) {
 351                         update_session_list(session_list,
 352                                         lsession.hostname,
 353                                         lsession.session_name,
 354                                         be32toh(lsession.streams),
 355                                         be32toh(lsession.clients),
 356                                         be32toh(lsession.live_timer));
 357                 } else {
 358                         if ((strncmp(lsession.session_name, ctx->session_name,
 359                                 NAME_MAX) == 0) && (strncmp(lsession.hostname,
 360                                         ctx->traced_hostname, NAME_MAX) == 0)) {
 361                                 printf_verbose("Reading from session %" PRIu64 "\n",
 362                                                 session_id);
 363                                 g_array_append_val(ctx->session_ids,
 364                                                 session_id);
 365                         }
 366                 }
 367         }
 368
 369         if (print_list) {
 370                 print_session_list(session_list, path);
 371                 free_session_list(session_list);
 372         }
 373         ret = 0;
 374 end:
 375         return ret;
 376
 377 error:
 378         fprintf(stderr, "[error] Unable to list sessions\n");
 379         return -1;
 380 }
 381
 382 int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream,
 383                 uint64_t ctf_trace_id)
 384 {
 385         struct lttng_live_ctf_trace *trace;
 386         int ret = 0;
 387
 388         trace = g_hash_table_lookup(stream->session->ctf_traces,
 389                         (gpointer) ctf_trace_id);
 390         if (!trace) {
 391                 trace = g_new0(struct lttng_live_ctf_trace, 1);
 392                 trace->ctf_trace_id = ctf_trace_id;
 393                 trace->streams = g_ptr_array_new();
 394                 g_hash_table_insert(stream->session->ctf_traces,
 395                                 (gpointer) ctf_trace_id,
 396                                 trace);
 397         }
 398         if (stream->metadata_flag)
 399                 trace->metadata_stream = stream;
 400
 401         stream->ctf_trace = trace;
 402         g_ptr_array_add(trace->streams, stream);
 403
 404         return ret;
 405 }
 406
 407 static
 408 int open_metadata_fp_write(struct lttng_live_viewer_stream *stream,
 409                 char **metadata_buf, size_t *size)
 410 {
 411         int ret = 0;
 412
 413         stream->metadata_fp_write =
 414                 babeltrace_open_memstream(metadata_buf, size);
 415         if (!stream->metadata_fp_write) {
 416                 perror("Metadata open_memstream");
 417                 ret = -1;
 418         }
 419
 420         return ret;
 421 }
 422
 423 int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 424 {
 425         struct lttng_viewer_cmd cmd;
 426         struct lttng_viewer_attach_session_request rq;
 427         struct lttng_viewer_attach_session_response rp;
 428         struct lttng_viewer_stream stream;
 429         int ret, i;
 430         ssize_t ret_len;
 431
 432         if (lttng_live_should_quit()) {
 433                 ret = -1;
 434                 goto end;
 435         }
 436
 437         cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 438         cmd.data_size = sizeof(rq);
 439         cmd.cmd_version = 0;
 440
 441         memset(&rq, 0, sizeof(rq));
 442         rq.session_id = htobe64(id);
 443         // TODO: add cmd line parameter to select seek beginning
 444         // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
 445         rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
 446
 447         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 448         if (ret_len < 0) {
 449                 perror("[error] Error sending cmd");
 450                 goto error;
 451         }
 452         assert(ret_len == sizeof(cmd));
 453
 454         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 455         if (ret_len < 0) {
 456                 perror("[error] Error sending attach request");
 457                 goto error;
 458         }
 459         assert(ret_len == sizeof(rq));
 460
 461         ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 462         if (ret_len == 0) {
 463                 fprintf(stderr, "[error] Remote side has closed connection\n");
 464                 goto error;
 465         }
 466         if (ret_len < 0) {
 467                 perror("[error] Error receiving attach response");
 468                 goto error;
 469         }
 470         assert(ret_len == sizeof(rp));
 471
 472         switch(be32toh(rp.status)) {
 473         case LTTNG_VIEWER_ATTACH_OK:
 474                 break;
 475         case LTTNG_VIEWER_ATTACH_UNK:
 476                 ret = -LTTNG_VIEWER_ATTACH_UNK;
 477                 goto end;
 478         case LTTNG_VIEWER_ATTACH_ALREADY:
 479                 fprintf(stderr, "[error] There is already a viewer attached to this session\n");
 480                 goto error;
 481         case LTTNG_VIEWER_ATTACH_NOT_LIVE:
 482                 fprintf(stderr, "[error] Not a live session\n");
 483                 goto error;
 484         case LTTNG_VIEWER_ATTACH_SEEK_ERR:
 485                 fprintf(stderr, "[error] Wrong seek parameter\n");
 486                 goto error;
 487         default:
 488                 fprintf(stderr, "[error] Unknown attach return code %u\n",
 489                                 be32toh(rp.status));
 490                 goto error;
 491         }
 492         if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
 493                 goto error;
 494         }
 495
 496         ctx->session->stream_count += be32toh(rp.streams_count);
 497         /*
 498          * When the session is created but not started, we do an active wait
 499          * until it starts. It allows the viewer to start processing the trace
 500          * as soon as the session starts.
 501          */
 502         if (ctx->session->stream_count == 0) {
 503                 ret = 0;
 504                 goto end;
 505         }
 506         printf_verbose("Waiting for %" PRIu64 " streams:\n",
 507                 ctx->session->stream_count);
 508         ctx->session->streams = g_new0(struct lttng_live_viewer_stream,
 509                         ctx->session->stream_count);
 510         for (i = 0; i < be32toh(rp.streams_count); i++) {
 511                 ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream));
 512                 if (ret_len == 0) {
 513                         fprintf(stderr, "[error] Remote side has closed connection\n");
 514                         goto error;
 515                 }
 516                 if (ret_len < 0) {
 517                         perror("[error] Error receiving stream");
 518                         goto error;
 519                 }
 520                 assert(ret_len == sizeof(stream));
 521                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
 522                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 523
 524                 printf_verbose("    stream %" PRIu64 " : %s/%s\n",
 525                                 be64toh(stream.id), stream.path_name,
 526                                 stream.channel_name);
 527                 ctx->session->streams[i].id = be64toh(stream.id);
 528                 ctx->session->streams[i].session = ctx->session;
 529
 530                 ctx->session->streams[i].mmap_size = 0;
 531                 ctx->session->streams[i].ctf_stream_id = -1ULL;
 532
 533                 if (be32toh(stream.metadata_flag)) {
 534                         ctx->session->streams[i].metadata_flag = 1;
 535                 }
 536                 ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i],
 537                                 be64toh(stream.ctf_trace_id));
 538                 if (ret < 0) {
 539                         goto error;
 540                 }
 541
 542         }
 543         ret = 0;
 544 end:
 545         return ret;
 546
 547 error:
 548         return -1;
 549 }
 550
 551 /*
 552  * Ask the relay for new streams.
 553  *
 554  * Returns the number of new streams received or a negative value on error.
 555  */
 556 static
 557 int ask_new_streams(struct lttng_live_ctx *ctx)
 558 {
 559         int i, ret = 0, nb_streams = 0;
 560         uint64_t id;
 561
 562 restart:
 563         for (i = 0; i < ctx->session_ids->len; i++) {
 564                 id = g_array_index(ctx->session_ids, uint64_t, i);
 565                 ret = lttng_live_get_new_streams(ctx, id);
 566                 printf_verbose("Asking for new streams returns %d\n", ret);
 567                 if (ret < 0) {
 568                         if (lttng_live_should_quit()) {
 569                                 goto end;
 570                         }
 571                         if (ret == -LTTNG_VIEWER_NEW_STREAMS_HUP) {
 572                                 printf_verbose("Session %" PRIu64 " closed\n",
 573                                                 id);
 574                                 /*
 575                                  * The streams have already been closed during
 576                                  * the reading, so we only need to get rid of
 577                                  * the trace in our internal table of sessions.
 578                                  */
 579                                 g_array_remove_index(ctx->session_ids, i);
 580                                 /*
 581                                  * We can't continue iterating on the g_array
 582                                  * after a remove, we have to start again.
 583                                  */
 584                                 goto restart;
 585                         } else {
 586                                 ret = -1;
 587                                 goto end;
 588                         }
 589                 } else {
 590                         nb_streams += ret;
 591                 }
 592         }
 593         ret = nb_streams;
 594
 595 end:
 596         return ret;
 597 }
 598
 599 static
 600 int append_metadata(struct lttng_live_ctx *ctx,
 601                 struct lttng_live_viewer_stream *viewer_stream)
 602 {
 603         int ret;
 604         struct lttng_live_viewer_stream *metadata;
 605         char *metadata_buf = NULL;
 606
 607         printf_verbose("get_next_index: new metadata needed\n");
 608         ret = get_new_metadata(ctx, viewer_stream, &metadata_buf);
 609         if (ret < 0) {
 610                 free(metadata_buf);
 611                 goto error;
 612         }
 613
 614         metadata = viewer_stream->ctf_trace->metadata_stream;
 615         metadata->ctf_trace->metadata_fp =
 616                 babeltrace_fmemopen(metadata_buf,
 617                                 metadata->metadata_len, "rb");
 618         if (!metadata->ctf_trace->metadata_fp) {
 619                 perror("Metadata fmemopen");
 620                 free(metadata_buf);
 621                 ret = -1;
 622                 goto error;
 623         }
 624         ret = ctf_append_trace_metadata(
 625                         viewer_stream->ctf_trace->handle->td,
 626                         metadata->ctf_trace->metadata_fp);
 627         /* We accept empty metadata packets */
 628         if (ret != 0 && ret != -ENOENT) {
 629                 fprintf(stderr, "[error] Appending metadata\n");
 630                 goto error;
 631         }
 632         ret = 0;
 633
 634 error:
 635         return ret;
 636 }
 637
 638 static
 639 int get_data_packet(struct lttng_live_ctx *ctx,
 640                 struct ctf_stream_pos *pos,
 641                 struct lttng_live_viewer_stream *stream, uint64_t offset,
 642                 uint64_t len)
 643 {
 644         struct lttng_viewer_cmd cmd;
 645         struct lttng_viewer_get_packet rq;
 646         struct lttng_viewer_trace_packet rp;
 647         ssize_t ret_len;
 648         int ret;
 649
 650 retry:
 651         if (lttng_live_should_quit()) {
 652                 ret = -1;
 653                 goto end;
 654         }
 655         cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
 656         cmd.data_size = sizeof(rq);
 657         cmd.cmd_version = 0;
 658
 659         memset(&rq, 0, sizeof(rq));
 660         rq.stream_id = htobe64(stream->id);
 661         rq.offset = htobe64(offset);
 662         rq.len = htobe32(len);
 663
 664         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 665         if (ret_len < 0) {
 666                 perror("[error] Error sending cmd");
 667                 goto error;
 668         }
 669         assert(ret_len == sizeof(cmd));
 670
 671         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 672         if (ret_len < 0) {
 673                 perror("[error] Error sending get_data_packet request");
 674                 goto error;
 675         }
 676         assert(ret_len == sizeof(rq));
 677
 678         ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 679         if (ret_len == 0) {
 680                 fprintf(stderr, "[error] Remote side has closed connection\n");
 681                 goto error;
 682         }
 683         if (ret_len < 0) {
 684                 perror("[error] Error receiving data response");
 685                 goto error;
 686         }
 687         if (ret_len != sizeof(rp)) {
 688                 fprintf(stderr, "[error] get_data_packet: expected %" PRId64
 689                                 ", received %" PRId64 "\n", sizeof(rp),
 690                                 ret_len);
 691                 goto error;
 692         }
 693
 694         rp.flags = be32toh(rp.flags);
 695
 696         switch (be32toh(rp.status)) {
 697         case LTTNG_VIEWER_GET_PACKET_OK:
 698                 len = be32toh(rp.len);
 699                 printf_verbose("get_data_packet: Ok, packet size : %" PRIu64
 700                                 "\n", len);
 701                 break;
 702         case LTTNG_VIEWER_GET_PACKET_RETRY:
 703                 /* Unimplemented by relay daemon */
 704                 printf_verbose("get_data_packet: retry\n");
 705                 goto error;
 706         case LTTNG_VIEWER_GET_PACKET_ERR:
 707                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 708                         printf_verbose("get_data_packet: new metadata needed\n");
 709                         ret = append_metadata(ctx, stream);
 710                         if (ret)
 711                                 goto error;
 712                 }
 713                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
 714                         printf_verbose("get_data_packet: new streams needed\n");
 715                         ret = ask_new_streams(ctx);
 716                         if (ret < 0)
 717                                 goto error;
 718                         else if (ret > 0)
 719                                 g_hash_table_foreach(ctx->session->ctf_traces,
 720                                                 add_traces, ctx->bt_ctx);
 721                 }
 722                 if (rp.flags & (LTTNG_VIEWER_FLAG_NEW_METADATA
 723                                 | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
 724                         goto retry;
 725                 }
 726                 fprintf(stderr, "[error] get_data_packet: error\n");
 727                 goto error;
 728         case LTTNG_VIEWER_GET_PACKET_EOF:
 729                 ret = -2;
 730                 goto end;
 731         default:
 732                 printf_verbose("get_data_packet: unknown\n");
 733                 goto error;
 734         }
 735
 736         if (len == 0) {
 737                 goto error;
 738         }
 739
 740         if (len > stream->mmap_size) {
 741                 uint64_t new_size;
 742
 743                 new_size = max_t(uint64_t, len, stream->mmap_size << 1);
 744                 if (pos->base_mma) {
 745                         /* unmap old base */
 746                         ret = munmap_align(pos->base_mma);
 747                         if (ret) {
 748                                 perror("[error] Unable to unmap old base");
 749                                 goto error;
 750                         }
 751                         pos->base_mma = NULL;
 752                 }
 753                 pos->base_mma = mmap_align(new_size,
 754                                 PROT_READ | PROT_WRITE,
 755                                 MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
 756                 if (pos->base_mma == MAP_FAILED) {
 757                         perror("[error] mmap error");
 758                         pos->base_mma = NULL;
 759                         goto error;
 760                 }
 761
 762                 stream->mmap_size = new_size;
 763                 printf_verbose("Expanding stream mmap size to %" PRIu64 " bytes\n",
 764                                 stream->mmap_size);
 765         }
 766
 767         ret_len = lttng_live_recv(ctx->control_sock,
 768                         mmap_align_addr(pos->base_mma), len);
 769         if (ret_len == 0) {
 770                 fprintf(stderr, "[error] Remote side has closed connection\n");
 771                 goto error;
 772         }
 773         if (ret_len < 0) {
 774                 perror("[error] Error receiving trace packet");
 775                 goto error;
 776         }
 777         assert(ret_len == len);
 778         ret = 0;
 779 end:
 780         return ret;
 781
 782 error:
 783         return -1;
 784 }
 785
 786 static
 787 int get_one_metadata_packet(struct lttng_live_ctx *ctx,
 788                 struct lttng_live_viewer_stream *metadata_stream)
 789 {
 790         uint64_t len = 0;
 791         int ret;
 792         struct lttng_viewer_cmd cmd;
 793         struct lttng_viewer_get_metadata rq;
 794         struct lttng_viewer_metadata_packet rp;
 795         char *data = NULL;
 796         ssize_t ret_len;
 797
 798         if (lttng_live_should_quit()) {
 799                 ret = -1;
 800                 goto end;
 801         }
 802
 803         rq.stream_id = htobe64(metadata_stream->id);
 804         cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
 805         cmd.data_size = sizeof(rq);
 806         cmd.cmd_version = 0;
 807
 808         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 809         if (ret_len < 0) {
 810                 perror("[error] Error sending cmd");
 811                 goto error;
 812         }
 813         assert(ret_len == sizeof(cmd));
 814
 815         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 816         if (ret_len < 0) {
 817                 perror("[error] Error sending get_metadata request");
 818                 goto error;
 819         }
 820         assert(ret_len == sizeof(rq));
 821
 822         ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 823         if (ret_len == 0) {
 824                 fprintf(stderr, "[error] Remote side has closed connection\n");
 825                 goto error;
 826         }
 827         if (ret_len < 0) {
 828                 perror("[error] Error receiving metadata response");
 829                 goto error;
 830         }
 831         assert(ret_len == sizeof(rp));
 832
 833         switch (be32toh(rp.status)) {
 834                 case LTTNG_VIEWER_METADATA_OK:
 835                         printf_verbose("get_metadata : OK\n");
 836                         break;
 837                 case LTTNG_VIEWER_NO_NEW_METADATA:
 838                         printf_verbose("get_metadata : NO NEW\n");
 839                         ret = 0;
 840                         goto end;
 841                 case LTTNG_VIEWER_METADATA_ERR:
 842                         printf_verbose("get_metadata : ERR\n");
 843                         goto error;
 844                 default:
 845                         printf_verbose("get_metadata : UNKNOWN\n");
 846                         goto error;
 847         }
 848
 849         len = be64toh(rp.len);
 850         printf_verbose("Writing %" PRIu64" bytes to metadata\n", len);
 851         if (len <= 0) {
 852                 goto error;
 853         }
 854
 855         data = zmalloc(len);
 856         if (!data) {
 857                 perror("relay data zmalloc");
 858                 goto error;
 859         }
 860         ret_len = lttng_live_recv(ctx->control_sock, data, len);
 861         if (ret_len == 0) {
 862                 fprintf(stderr, "[error] Remote side has closed connection\n");
 863                 goto error_free_data;
 864         }
 865         if (ret_len < 0) {
 866                 perror("[error] Error receiving trace packet");
 867                 goto error_free_data;
 868         }
 869         assert(ret_len == len);
 870
 871         do {
 872                 ret_len = fwrite(data, 1, len,
 873                                 metadata_stream->metadata_fp_write);
 874         } while (ret_len < 0 && errno == EINTR);
 875         if (ret_len < 0) {
 876                 fprintf(stderr, "[error] Writing in the metadata fp\n");
 877                 goto error_free_data;
 878         }
 879         assert(ret_len == len);
 880         metadata_stream->metadata_len += len;
 881         free(data);
 882         ret = len;
 883 end:
 884         return ret;
 885
 886 error_free_data:
 887         free(data);
 888 error:
 889         return -1;
 890 }
 891
 892 /*
 893  * Return 0 on success, a negative value on error.
 894  */
 895 static
 896 int get_new_metadata(struct lttng_live_ctx *ctx,
 897                 struct lttng_live_viewer_stream *viewer_stream,
 898                 char **metadata_buf)
 899 {
 900         int ret = 0;
 901         struct lttng_live_viewer_stream *metadata_stream;
 902         size_t size, len_read = 0;
 903
 904         metadata_stream = viewer_stream->ctf_trace->metadata_stream;
 905         if (!metadata_stream) {
 906                 fprintf(stderr, "[error] No metadata stream\n");
 907                 ret = -1;
 908                 goto error;
 909         }
 910         metadata_stream->metadata_len = 0;
 911         ret = open_metadata_fp_write(metadata_stream, metadata_buf, &size);
 912         if (ret < 0) {
 913                 goto error;
 914         }
 915
 916         do {
 917                 /*
 918                  * get_one_metadata_packet returns the number of bytes
 919                  * received, 0 when we have received everything, a
 920                  * negative value on error.
 921                  */
 922                 ret = get_one_metadata_packet(ctx, metadata_stream);
 923                 if (ret > 0) {
 924                         len_read += ret;
 925                 }
 926                 if (!len_read) {
 927                         (void) poll(NULL, 0, ACTIVE_POLL_DELAY);
 928                 }
 929         } while (ret > 0 || !len_read);
 930
 931         if (fclose(metadata_stream->metadata_fp_write))
 932                 perror("fclose");
 933         metadata_stream->metadata_fp_write = NULL;
 934
 935 error:
 936         return ret;
 937 }
 938
 939 /*
 940  * Assign the fields from a lttng_viewer_index to a packet_index.
 941  */
 942 static
 943 void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
 944                 struct packet_index *pindex)
 945 {
 946         assert(lindex);
 947         assert(pindex);
 948
 949         pindex->offset = be64toh(lindex->offset);
 950         pindex->packet_size = be64toh(lindex->packet_size);
 951         pindex->content_size = be64toh(lindex->content_size);
 952         pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
 953         pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
 954         pindex->events_discarded = be64toh(lindex->events_discarded);
 955 }
 956
 957 /*
 958  * Get one index for a stream.
 959  *
 960  * Returns 0 on success or a negative value on error.
 961  */
 962 static
 963 int get_next_index(struct lttng_live_ctx *ctx,
 964                 struct lttng_live_viewer_stream *viewer_stream,
 965                 struct packet_index *index, uint64_t *stream_id)
 966 {
 967         struct lttng_viewer_cmd cmd;
 968         struct lttng_viewer_get_next_index rq;
 969         int ret;
 970         ssize_t ret_len;
 971         struct lttng_viewer_index *rp = &viewer_stream->current_index;
 972
 973         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
 974         cmd.data_size = sizeof(rq);
 975         cmd.cmd_version = 0;
 976
 977         memset(&rq, 0, sizeof(rq));
 978         rq.stream_id = htobe64(viewer_stream->id);
 979
 980 retry:
 981         if (lttng_live_should_quit()) {
 982                 ret = -1;
 983                 goto end;
 984         }
 985         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 986         if (ret_len < 0) {
 987                 perror("[error] Error sending cmd");
 988                 goto error;
 989         }
 990         assert(ret_len == sizeof(cmd));
 991
 992         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 993         if (ret_len < 0) {
 994                 perror("[error] Error sending get_next_index request");
 995                 goto error;
 996         }
 997         assert(ret_len == sizeof(rq));
 998
 999         ret_len = lttng_live_recv(ctx->control_sock, rp, sizeof(*rp));
1000         if (ret_len == 0) {
1001                 fprintf(stderr, "[error] Remote side has closed connection\n");
1002                 goto error;
1003         }
1004         if (ret_len < 0) {
1005                 perror("[error] Error receiving index response");
1006                 goto error;
1007         }
1008         assert(ret_len == sizeof(*rp));
1009
1010         rp->flags = be32toh(rp->flags);
1011
1012         switch (be32toh(rp->status)) {
1013         case LTTNG_VIEWER_INDEX_INACTIVE:
1014                 printf_verbose("get_next_index: inactive\n");
1015                 memset(index, 0, sizeof(struct packet_index));
1016                 index->ts_cycles.timestamp_end = be64toh(rp->timestamp_end);
1017                 *stream_id = be64toh(rp->stream_id);
1018                 break;
1019         case LTTNG_VIEWER_INDEX_OK:
1020                 printf_verbose("get_next_index: Ok, need metadata update : %u\n",
1021                                 rp->flags & LTTNG_VIEWER_FLAG_NEW_METADATA);
1022                 lttng_index_to_packet_index(rp, index);
1023                 *stream_id = be64toh(rp->stream_id);
1024                 viewer_stream->data_pending = 1;
1025
1026                 if (rp->flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1027                         ret = append_metadata(ctx, viewer_stream);
1028                         if (ret)
1029                                 goto error;
1030                 }
1031                 if (rp->flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1032                         printf_verbose("get_next_index: need new streams\n");
1033                         ret = ask_new_streams(ctx);
1034                         if (ret < 0)
1035                                 goto error;
1036                         else if (ret > 0)
1037                                 g_hash_table_foreach(ctx->session->ctf_traces,
1038                                                 add_traces, ctx->bt_ctx);
1039                 }
1040                 break;
1041         case LTTNG_VIEWER_INDEX_RETRY:
1042                 printf_verbose("get_next_index: retry\n");
1043                 (void) poll(NULL, 0, ACTIVE_POLL_DELAY);
1044                 goto retry;
1045         case LTTNG_VIEWER_INDEX_HUP:
1046                 printf_verbose("get_next_index: stream hung up\n");
1047                 viewer_stream->id = -1ULL;
1048                 index->offset = EOF;
1049                 ctx->session->stream_count--;
1050                 break;
1051         case LTTNG_VIEWER_INDEX_ERR:
1052                 fprintf(stderr, "[error] get_next_index: error\n");
1053                 goto error;
1054         default:
1055                 fprintf(stderr, "[error] get_next_index: unkwown value\n");
1056                 goto error;
1057         }
1058         ret = 0;
1059 end:
1060         return ret;
1061
1062 error:
1063         return -1;
1064 }
1065
1066 static
1067 void read_packet_header(struct ctf_stream_pos *pos,
1068                 struct ctf_file_stream *file_stream)
1069 {
1070         int ret;
1071
1072         /* update trace_packet_header and stream_packet_context */
1073         if (!(pos->prot & PROT_WRITE) &&
1074                 file_stream->parent.trace_packet_header) {
1075                 /* Read packet header */
1076                 ret = generic_rw(&pos->parent,
1077                                 &file_stream->parent.trace_packet_header->p);
1078                 if (ret) {
1079                         pos->offset = EOF;
1080                         fprintf(stderr, "[error] trace packet "
1081                                         "header read failed\n");
1082                         goto end;
1083                 }
1084         }
1085         if (!(pos->prot & PROT_WRITE) &&
1086                 file_stream->parent.stream_packet_context) {
1087                 /* Read packet context */
1088                 ret = generic_rw(&pos->parent,
1089                                 &file_stream->parent.stream_packet_context->p);
1090                 if (ret) {
1091                         pos->offset = EOF;
1092                         fprintf(stderr, "[error] stream packet "
1093                                         "context read failed\n");
1094                         goto end;
1095                 }
1096         }
1097         pos->data_offset = pos->offset;
1098
1099 end:
1100         return;
1101 }
1102
1103 /*
1104  * Handle the seek parameters.
1105  * Returns 0 if the packet_seek can continue, a positive value to
1106  * cleanly exit the packet_seek, a negative value on error.
1107  */
1108 static
1109 int handle_seek_position(size_t index, int whence,
1110                 struct lttng_live_viewer_stream *viewer_stream,
1111                 struct ctf_stream_pos *pos,
1112                 struct ctf_file_stream *file_stream)
1113 {
1114         int ret;
1115
1116         switch (whence) {
1117         case SEEK_CUR:
1118                 ret = 0;
1119                 goto end;
1120         case SEEK_SET:
1121                 /*
1122                  * We only allow to seek to 0.
1123                  */
1124                 if (index != 0) {
1125                         fprintf(stderr, "[error] Arbitrary seek in lttng-live "
1126                                         "trace not supported\n");
1127                         pos->offset = EOF;
1128                         ret = -1;
1129                         goto end;
1130                 }
1131
1132                 ret = 0;
1133                 goto end;
1134
1135         default:
1136                 fprintf(stderr, "[error] Invalid seek parameter\n");
1137                 assert(0);
1138         }
1139
1140 end:
1141         return ret;
1142 }
1143
1144 static
1145 void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
1146                 int whence)
1147 {
1148         struct ctf_stream_pos *pos;
1149         struct ctf_file_stream *file_stream;
1150         struct packet_index *prev_index = NULL, *cur_index;
1151         struct lttng_live_viewer_stream *viewer_stream;
1152         struct lttng_live_session *session;
1153         uint64_t stream_id = -1ULL;
1154         int ret;
1155
1156         pos = ctf_pos(stream_pos);
1157         file_stream = container_of(pos, struct ctf_file_stream, pos);
1158         viewer_stream = (struct lttng_live_viewer_stream *) pos->priv;
1159         session = viewer_stream->session;
1160
1161         ret = handle_seek_position(index, whence, viewer_stream, pos,
1162                         file_stream);
1163         if (ret != 0) {
1164                 return;
1165         }
1166
1167 retry:
1168         switch (pos->packet_index->len) {
1169         case 0:
1170                 g_array_set_size(pos->packet_index, 1);
1171                 cur_index = &g_array_index(pos->packet_index,
1172                                 struct packet_index, 0);
1173                 break;
1174         case 1:
1175                 g_array_set_size(pos->packet_index, 2);
1176                 prev_index = &g_array_index(pos->packet_index,
1177                                 struct packet_index, 0);
1178                 cur_index = &g_array_index(pos->packet_index,
1179                                 struct packet_index, 1);
1180                 break;
1181         case 2:
1182                 g_array_index(pos->packet_index,
1183                         struct packet_index, 0) =
1184                                 g_array_index(pos->packet_index,
1185                                         struct packet_index, 1);
1186                 prev_index = &g_array_index(pos->packet_index,
1187                                 struct packet_index, 0);
1188                 cur_index = &g_array_index(pos->packet_index,
1189                                 struct packet_index, 1);
1190                 break;
1191         default:
1192                 abort();
1193                 break;
1194         }
1195
1196         if (viewer_stream->data_pending) {
1197                 lttng_index_to_packet_index(&viewer_stream->current_index, cur_index);
1198         } else {
1199                 printf_verbose("get_next_index for stream %" PRIu64 "\n", viewer_stream->id);
1200                 ret = get_next_index(session->ctx, viewer_stream, cur_index, &stream_id);
1201                 if (ret < 0) {
1202                         pos->offset = EOF;
1203                         if (!lttng_live_should_quit()) {
1204                                 fprintf(stderr, "[error] get_next_index failed\n");
1205                         }
1206                         return;
1207                 }
1208                 printf_verbose("Index received : packet_size : %" PRIu64
1209                                 ", offset %" PRIu64 ", content_size %" PRIu64
1210                                 ", timestamp_end : %" PRIu64 "\n",
1211                                 cur_index->packet_size, cur_index->offset,
1212                                 cur_index->content_size,
1213                                 cur_index->ts_cycles.timestamp_end);
1214
1215         }
1216
1217         /*
1218          * On the first time we receive an index, the stream_id needs to
1219          * be set for the stream in order to use it, we don't want any
1220          * data at this stage.
1221          */
1222         if (file_stream->parent.stream_id == -1ULL) {
1223                 /*
1224                  * Warning: with lttng-tools < 2.4.2, the beacon does not
1225                  * contain the real stream ID, it is memset to 0, so this
1226                  * might create a problem when a session has multiple
1227                  * channels. We can't detect it at this stage, lttng-tools
1228                  * has to be upgraded to fix this problem.
1229                  */
1230                 printf_verbose("Assigning stream_id %" PRIu64 "\n",
1231                                 stream_id);
1232                 file_stream->parent.stream_id = stream_id;
1233                 viewer_stream->ctf_stream_id = stream_id;
1234
1235                 return;
1236         }
1237
1238         pos->packet_size = cur_index->packet_size;
1239         pos->content_size = cur_index->content_size;
1240         pos->mmap_base_offset = 0;
1241         if (cur_index->offset == EOF) {
1242                 pos->offset = EOF;
1243         } else {
1244                 pos->offset = 0;
1245         }
1246
1247         if (cur_index->content_size == 0) {
1248                 if (file_stream->parent.stream_class) {
1249                         file_stream->parent.cycles_timestamp =
1250                                 cur_index->ts_cycles.timestamp_end;
1251                         file_stream->parent.real_timestamp = ctf_get_real_timestamp(
1252                                         &file_stream->parent,
1253                                         cur_index->ts_cycles.timestamp_end);
1254                 }
1255         } else {
1256                 if (file_stream->parent.stream_class) {
1257                         /* Convert the timestamps and append to the real_index. */
1258                         cur_index->ts_real.timestamp_begin = ctf_get_real_timestamp(
1259                                         &file_stream->parent,
1260                                         cur_index->ts_cycles.timestamp_begin);
1261                         cur_index->ts_real.timestamp_end = ctf_get_real_timestamp(
1262                                         &file_stream->parent,
1263                                         cur_index->ts_cycles.timestamp_end);
1264                 }
1265
1266                 ctf_update_current_packet_index(&file_stream->parent,
1267                                 prev_index, cur_index);
1268
1269                 file_stream->parent.cycles_timestamp =
1270                                 cur_index->ts_cycles.timestamp_begin;
1271                 file_stream->parent.real_timestamp =
1272                                 cur_index->ts_real.timestamp_begin;
1273         }
1274
1275         if (pos->packet_size == 0 || pos->offset == EOF) {
1276                 goto end;
1277         }
1278
1279         printf_verbose("get_data_packet for stream %" PRIu64 "\n",
1280                         viewer_stream->id);
1281         ret = get_data_packet(session->ctx, pos, viewer_stream,
1282                         cur_index->offset,
1283                         cur_index->packet_size / CHAR_BIT);
1284         if (ret == -2) {
1285                 goto retry;
1286         } else if (ret < 0) {
1287                 pos->offset = EOF;
1288                 if (!lttng_live_should_quit()) {
1289                         fprintf(stderr, "[error] get_data_packet failed\n");
1290                 }
1291                 return;
1292         }
1293         viewer_stream->data_pending = 0;
1294
1295         read_packet_header(pos, file_stream);
1296
1297 end:
1298         return;
1299 }
1300
1301 int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx)
1302 {
1303         struct lttng_viewer_cmd cmd;
1304         struct lttng_viewer_create_session_response resp;
1305         int ret;
1306         ssize_t ret_len;
1307
1308         if (lttng_live_should_quit()) {
1309                 ret = -1;
1310                 goto end;
1311         }
1312
1313         cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
1314         cmd.data_size = 0;
1315         cmd.cmd_version = 0;
1316
1317         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
1318         if (ret_len < 0) {
1319                 perror("[error] Error sending cmd");
1320                 goto error;
1321         }
1322         assert(ret_len == sizeof(cmd));
1323
1324         ret_len = lttng_live_recv(ctx->control_sock, &resp, sizeof(resp));
1325         if (ret_len == 0) {
1326                 fprintf(stderr, "[error] Remote side has closed connection\n");
1327                 goto error;
1328         }
1329         if (ret_len < 0) {
1330                 perror("[error] Error receiving create session reply");
1331                 goto error;
1332         }
1333         assert(ret_len == sizeof(resp));
1334
1335         if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
1336                 fprintf(stderr, "[error] Error creating viewer session\n");
1337                 goto error;
1338         }
1339         ret = 0;
1340 end:
1341         return ret;
1342
1343 error:
1344         return -1;
1345 }
1346
1347 static
1348 int del_traces(gpointer key, gpointer value, gpointer user_data)
1349 {
1350         struct bt_context *bt_ctx = user_data;
1351         struct lttng_live_ctf_trace *trace = value;
1352         int ret;
1353
1354         ret = bt_context_remove_trace(bt_ctx, trace->trace_id);
1355         if (ret < 0)
1356                 fprintf(stderr, "[error] removing trace from context\n");
1357
1358         /* remove the key/value pair from the HT. */
1359         return 1;
1360 }
1361
1362 static
1363 void add_traces(gpointer key, gpointer value, gpointer user_data)
1364 {
1365         int i, ret;
1366         struct bt_context *bt_ctx = user_data;
1367         struct lttng_live_ctf_trace *trace = value;
1368         struct lttng_live_viewer_stream *stream;
1369         struct bt_mmap_stream *new_mmap_stream;
1370         struct bt_mmap_stream_list mmap_list;
1371         struct lttng_live_ctx *ctx = NULL;
1372         struct bt_trace_descriptor *td;
1373         struct bt_trace_handle *handle;
1374
1375         /*
1376          * We don't know how many streams we will receive for a trace, so
1377          * once we are done receiving the traces, we add all the traces
1378          * received to the bt_context.
1379          * We can receive streams during the attach command or the
1380          * get_new_streams, so we have to make sure not to add multiple
1381          * times the same traces.
1382          * If a trace is already in the context, we just skip this function.
1383          */
1384         if (trace->in_use)
1385                 return;
1386
1387         BT_INIT_LIST_HEAD(&mmap_list.head);
1388
1389         for (i = 0; i < trace->streams->len; i++) {
1390                 stream = g_ptr_array_index(trace->streams, i);
1391                 ctx = stream->session->ctx;
1392
1393                 if (!stream->metadata_flag) {
1394                         new_mmap_stream = zmalloc(sizeof(struct bt_mmap_stream));
1395                         new_mmap_stream->priv = (void *) stream;
1396                         new_mmap_stream->fd = -1;
1397                         bt_list_add(&new_mmap_stream->list, &mmap_list.head);
1398                 } else {
1399                         char *metadata_buf = NULL;
1400
1401                         /* Get all possible metadata before starting */
1402                         ret = get_new_metadata(ctx, stream, &metadata_buf);
1403                         if (ret) {
1404                                 free(metadata_buf);
1405                                 goto end_free;
1406                         }
1407                         if (!stream->metadata_len) {
1408                                 fprintf(stderr, "[error] empty metadata\n");
1409                                 ret = -1;
1410                                 free(metadata_buf);
1411                                 goto end_free;
1412                         }
1413
1414                         trace->metadata_fp = babeltrace_fmemopen(metadata_buf,
1415                                         stream->metadata_len, "rb");
1416                         if (!trace->metadata_fp) {
1417                                 perror("Metadata fmemopen2");
1418                                 ret = -1;
1419                                 free(metadata_buf);
1420                                 goto end_free;
1421                         }
1422                 }
1423         }
1424
1425         if (!trace->metadata_fp) {
1426                 fprintf(stderr, "[error] No metadata stream opened\n");
1427                 goto end_free;
1428         }
1429
1430         ret = bt_context_add_trace(bt_ctx, NULL, "ctf",
1431                         ctf_live_packet_seek, &mmap_list, trace->metadata_fp);
1432         if (ret < 0) {
1433                 fprintf(stderr, "[error] Error adding trace\n");
1434                 goto end_free;
1435         }
1436         trace->metadata_stream->metadata_len = 0;
1437
1438         handle = (struct bt_trace_handle *) g_hash_table_lookup(
1439                         bt_ctx->trace_handles,
1440                         (gpointer) (unsigned long) ret);
1441         td = handle->td;
1442         trace->handle = handle;
1443         if (bt_ctx->current_iterator) {
1444                 bt_iter_add_trace(bt_ctx->current_iterator, td);
1445         }
1446
1447         trace->trace_id = ret;
1448         trace->in_use = 1;
1449
1450         goto end;
1451
1452 end_free:
1453         bt_context_put(bt_ctx);
1454 end:
1455         return;
1456 }
1457
1458 /*
1459  * Request new streams for a session.
1460  * Returns the number of streams received or a negative value on error.
1461  */
1462 int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
1463 {
1464         struct lttng_viewer_cmd cmd;
1465         struct lttng_viewer_new_streams_request rq;
1466         struct lttng_viewer_new_streams_response rp;
1467         struct lttng_viewer_stream stream;
1468         int ret, i, nb_streams = 0;
1469         ssize_t ret_len;
1470         uint32_t stream_count;
1471
1472         if (lttng_live_should_quit()) {
1473                 ret = -1;
1474                 goto end;
1475         }
1476
1477         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1478         cmd.data_size = sizeof(rq);
1479         cmd.cmd_version = 0;
1480
1481         memset(&rq, 0, sizeof(rq));
1482         rq.session_id = htobe64(id);
1483
1484         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
1485         if (ret_len < 0) {
1486                 perror("[error] Error sending cmd");
1487                 goto error;
1488         }
1489         assert(ret_len == sizeof(cmd));
1490
1491         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
1492         if (ret_len < 0) {
1493                 perror("[error] Error sending get_new_streams request");
1494                 goto error;
1495         }
1496         assert(ret_len == sizeof(rq));
1497
1498         ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
1499         if (ret_len == 0) {
1500                 fprintf(stderr, "[error] Remote side has closed connection\n");
1501                 goto error;
1502         }
1503         if (ret_len < 0) {
1504                 perror("[error] Error receiving get_new_streams response");
1505                 goto error;
1506         }
1507         assert(ret_len == sizeof(rp));
1508
1509         switch(be32toh(rp.status)) {
1510         case LTTNG_VIEWER_NEW_STREAMS_OK:
1511                 break;
1512         case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1513                 ret = 0;
1514                 goto end;
1515         case LTTNG_VIEWER_NEW_STREAMS_HUP:
1516                 ret = -LTTNG_VIEWER_NEW_STREAMS_HUP;
1517                 goto end;
1518         case LTTNG_VIEWER_NEW_STREAMS_ERR:
1519                 fprintf(stderr, "[error] get_new_streams error\n");
1520                 goto error;
1521         default:
1522                 fprintf(stderr, "[error] Unknown return code %u\n",
1523                                 be32toh(rp.status));
1524                 goto error;
1525         }
1526
1527         stream_count = be32toh(rp.streams_count);
1528         ctx->session->stream_count += stream_count;
1529         /*
1530          * When the session is created but not started, we do an active wait
1531          * until it starts. It allows the viewer to start processing the trace
1532          * as soon as the session starts.
1533          */
1534         if (ctx->session->stream_count == 0) {
1535                 ret = 0;
1536                 goto end;
1537         }
1538         printf_verbose("Waiting for %" PRIu64 " streams:\n",
1539                 ctx->session->stream_count);
1540         ctx->session->streams = g_new0(struct lttng_live_viewer_stream,
1541                         ctx->session->stream_count);
1542         for (i = 0; i < stream_count; i++) {
1543                 ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream));
1544                 if (ret_len == 0) {
1545                         fprintf(stderr, "[error] Remote side has closed connection\n");
1546                         goto error;
1547                 }
1548                 if (ret_len < 0) {
1549                         perror("[error] Error receiving stream");
1550                         goto error;
1551                 }
1552                 assert(ret_len == sizeof(stream));
1553                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
1554                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
1555
1556                 printf_verbose("    stream %" PRIu64 " : %s/%s\n",
1557                                 be64toh(stream.id), stream.path_name,
1558                                 stream.channel_name);
1559                 ctx->session->streams[i].id = be64toh(stream.id);
1560                 ctx->session->streams[i].session = ctx->session;
1561
1562                 ctx->session->streams[i].mmap_size = 0;
1563                 ctx->session->streams[i].ctf_stream_id = -1ULL;
1564
1565                 if (be32toh(stream.metadata_flag)) {
1566                         ctx->session->streams[i].metadata_flag = 1;
1567                 }
1568                 ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i],
1569                                 be64toh(stream.ctf_trace_id));
1570                 if (ret < 0) {
1571                         goto error;
1572                 }
1573                 nb_streams++;
1574
1575         }
1576         ret = nb_streams;
1577 end:
1578         return ret;
1579
1580 error:
1581         return -1;
1582 }
1583
1584 int lttng_live_read(struct lttng_live_ctx *ctx)
1585 {
1586         int ret, i;
1587 #if 0
1588         struct bt_ctf_iter *iter;
1589         const struct bt_ctf_event *event;
1590         struct bt_iter_pos begin_pos;
1591         struct bt_trace_descriptor *td_write;
1592         struct bt_format *fmt_write;
1593         struct ctf_text_stream_pos *sout;
1594 #endif
1595         uint64_t id;
1596
1597         ctx->bt_ctx = bt_context_create();
1598         if (!ctx->bt_ctx) {
1599                 fprintf(stderr, "[error] bt_context_create allocation\n");
1600                 goto end;
1601         }
1602
1603 #if 0
1604         fmt_write = bt_lookup_format(g_quark_from_static_string("text"));
1605         if (!fmt_write) {
1606                 fprintf(stderr, "[error] ctf-text error\n");
1607                 goto end;
1608         }
1609
1610         td_write = fmt_write->open_trace(NULL, O_RDWR, NULL, NULL);
1611         if (!td_write) {
1612                 fprintf(stderr, "[error] Error opening output trace\n");
1613                 goto end_free;
1614         }
1615
1616         sout = container_of(td_write, struct ctf_text_stream_pos,
1617                         trace_descriptor);
1618         if (!sout->parent.event_cb) {
1619                 goto end_free;
1620         }
1621 #endif
1622
1623         ret = lttng_live_create_viewer_session(ctx);
1624         if (ret < 0) {
1625                 goto end_free;
1626         }
1627
1628         for (i = 0; i < ctx->session_ids->len; i++) {
1629                 id = g_array_index(ctx->session_ids, uint64_t, i);
1630                 printf_verbose("Attaching to session %lu\n", id);
1631                 ret = lttng_live_attach_session(ctx, id);
1632                 printf_verbose("Attaching session returns %d\n", ret);
1633                 if (ret < 0) {
1634                         if (ret == -LTTNG_VIEWER_ATTACH_UNK) {
1635                                 fprintf(stderr, "[error] Unknown session ID\n");
1636                         }
1637                         goto end_free;
1638                 }
1639         }
1640
1641         /*
1642          * As long as the session is active, we try to get new streams.
1643          */
1644 #if 0
1645         for (;;) {
1646                 int flags;
1647 #endif
1648                 if (lttng_live_should_quit()) {
1649                         ret = 0;
1650                         goto end_free;
1651                 }
1652
1653                 while (!ctx->session->stream_count) {
1654                         if (lttng_live_should_quit()
1655                                         || ctx->session_ids->len == 0) {
1656                                 ret = 0;
1657                                 goto end_free;
1658                         }
1659                         ret = ask_new_streams(ctx);
1660                         if (ret < 0) {
1661                                 goto end_free;
1662                         }
1663                         if (!ctx->session->stream_count) {
1664                                 (void) poll(NULL, 0, ACTIVE_POLL_DELAY);
1665                         }
1666                 }
1667
1668                 g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
1669                                 ctx->bt_ctx);
1670
1671 #if 0
1672                 begin_pos.type = BT_SEEK_BEGIN;
1673                 iter = bt_ctf_iter_create(ctx->bt_ctx, &begin_pos, NULL);
1674                 if (!iter) {
1675                         if (lttng_live_should_quit()) {
1676                                 ret = 0;
1677                                 goto end;
1678                         }
1679                         fprintf(stderr, "[error] Iterator creation error\n");
1680                         goto end;
1681                 }
1682
1683                 for (;;) {
1684                         if (lttng_live_should_quit()) {
1685                                 ret = 0;
1686                                 goto end_free;
1687                         }
1688                         event = bt_ctf_iter_read_event_flags(iter, &flags);
1689                         if (!(flags & BT_ITER_FLAG_RETRY)) {
1690                                 if (!event) {
1691                                         /* End of trace */
1692                                         break;
1693                                 }
1694                                 ret = sout->parent.event_cb(&sout->parent,
1695                                                 event->parent->stream);
1696                                 if (ret) {
1697                                         fprintf(stderr, "[error] Writing "
1698                                                         "event failed.\n");
1699                                         goto end_free;
1700                                 }
1701                         }
1702                         ret = bt_iter_next(bt_ctf_get_iter(iter));
1703                         if (ret < 0) {
1704                                 goto end_free;
1705                         }
1706                 }
1707                 bt_ctf_iter_destroy(iter);
1708 #endif
1709                 ret = check_requirements(ctx->bt_ctx);
1710                 if (ret < 0 && !valid_trace) {
1711                         fprintf(stderr, "[error] some mandatory contexts "
1712                                         "were missing, exiting.\n");
1713                         goto end;
1714                 }
1715
1716                 if (!opt_textdump) {
1717 #ifdef HAVE_LIBNCURSES
1718                         pthread_create(&display_thread, NULL, ncurses_display,
1719                                         (void *) NULL);
1720                         pthread_create(&timer_thread, NULL, refresh_thread,
1721                                         (void *) NULL);
1722 #else
1723                         printf("Ncurses support not compiled, please install "
1724                                         "the missing dependencies and recompile\n");
1725                         goto end_free;
1726 #endif
1727                 }
1728                 iter_trace(ctx->bt_ctx);
1729                 g_hash_table_foreach_remove(ctx->session->ctf_traces,
1730                                 del_traces, ctx->bt_ctx);
1731                 ctx->session->stream_count = 0;
1732 #if 0
1733         }
1734 #endif
1735
1736 end_free:
1737         bt_context_put(ctx->bt_ctx);
1738 end:
1739         ret = 0;
1740         if (lttng_live_should_quit()) {
1741                 ret = 0;
1742         }
1743         return ret;
1744 }