src/liblttng-live.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
  37 #include <babeltrace/ctf/ctf-index.h>
  38
  39 #include <babeltrace/babeltrace.h>
  40 #include <babeltrace/ctf/events.h>
  41 #include <babeltrace/ctf/callbacks.h>
  42 #include <babeltrace/ctf/iterator.h>
  43
  44 /* for packet_index */
  45 #include <babeltrace/ctf/types.h>
  46
  47 #include <babeltrace/ctf/metadata.h>
  48 #include <babeltrace/ctf-text/types.h>
  49 #include <babeltrace/ctf/events-internal.h>
  50 /*
  51 #include <formats/ctf/events-private.h>
  52 replaced with
  53 */
  54 #include "network-live.h"
  55 #include "lttngtop.h"
  56
  57 #include "liblttng-live.h"
  58 #include "lttng-viewer-abi.h"
  59
  60 /*
  61  * Memory allocation zeroed
  62  */
  63 #define zmalloc(x) calloc(1, x)
  64
  65 #ifndef max_t
  66 #define max_t(type, a, b)       \
  67         ((type) (a) > (type) (b) ? (type) (a) : (type) (b))
  68 #endif
  69
  70 static void ctf_live_packet_seek(struct bt_stream_pos *stream_pos,
  71                 size_t index, int whence);
  72 static void add_traces(gpointer key, gpointer value, gpointer user_data);
  73 static int del_traces(gpointer key, gpointer value, gpointer user_data);
  74
  75 int lttng_live_connect_viewer(struct lttng_live_ctx *ctx)
  76 {
  77         struct hostent *host;
  78         struct sockaddr_in server_addr;
  79         int ret;
  80
  81         host = gethostbyname(ctx->relay_hostname);
  82         if (!host) {
  83                 ret = -1;
  84                 goto end;
  85         }
  86
  87         if ((ctx->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
  88                 perror("Socket");
  89                 ret = -1;
  90                 goto end;
  91         }
  92
  93         server_addr.sin_family = AF_INET;
  94         server_addr.sin_port = htons(ctx->port);
  95         server_addr.sin_addr = *((struct in_addr *) host->h_addr);
  96         bzero(&(server_addr.sin_zero), 8);
  97
  98         if (connect(ctx->control_sock, (struct sockaddr *) &server_addr,
  99                                 sizeof(struct sockaddr)) == -1) {
 100                 perror("Connect");
 101                 ret = -1;
 102                 goto end;
 103         }
 104
 105         ret = 0;
 106
 107 end:
 108         return ret;
 109 }
 110
 111 int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
 112 {
 113         struct lttng_viewer_cmd cmd;
 114         struct lttng_viewer_connect connect;
 115         int ret;
 116         ssize_t ret_len;
 117
 118         cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
 119         cmd.data_size = sizeof(connect);
 120         cmd.cmd_version = 0;
 121
 122         connect.viewer_session_id = -1ULL;      /* will be set on recv */
 123         connect.major = htobe32(LTTNG_LIVE_MAJOR);
 124         connect.minor = htobe32(LTTNG_LIVE_MINOR);
 125         connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
 126
 127         do {
 128                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 129         } while (ret_len < 0 && errno == EINTR);
 130         if (ret_len < 0) {
 131                 fprintf(stderr, "[error] Error sending cmd\n");
 132                 ret = ret_len;
 133                 goto error;
 134         }
 135         assert(ret_len == sizeof(cmd));
 136
 137         do {
 138                 ret_len = send(ctx->control_sock, &connect, sizeof(connect), 0);
 139         } while (ret_len < 0 && errno == EINTR);
 140         if (ret_len < 0) {
 141                 fprintf(stderr, "[error] Error sending version\n");
 142                 ret = ret_len;
 143                 goto error;
 144         }
 145         assert(ret_len == sizeof(connect));
 146
 147         do {
 148                 ret_len = recv(ctx->control_sock, &connect, sizeof(connect), 0);
 149         } while (ret_len < 0 && errno == EINTR);
 150         if (ret_len < 0) {
 151                 fprintf(stderr, "[error] Error receiving version\n");
 152                 ret = ret_len;
 153                 goto error;
 154         }
 155         assert(ret_len == sizeof(connect));
 156
 157         printf_verbose("Received viewer session ID : %" PRIu64 "\n",
 158                         be64toh(connect.viewer_session_id));
 159         printf_verbose("Relayd version : %u.%u\n", be32toh(connect.major),
 160                         be32toh(connect.minor));
 161
 162         ret = 0;
 163
 164 error:
 165         return ret;
 166 }
 167
 168 static
 169 void free_session_list(GPtrArray *session_list)
 170 {
 171         int i;
 172         struct lttng_live_relay_session *relay_session;
 173
 174         for (i = 0; i < session_list->len; i++) {
 175                 relay_session = g_ptr_array_index(session_list, i);
 176                 free(relay_session->name);
 177                 free(relay_session->hostname);
 178         }
 179         g_ptr_array_free(session_list, TRUE);
 180 }
 181
 182 static
 183 void print_session_list(GPtrArray *session_list, const char *path)
 184 {
 185         int i;
 186         struct lttng_live_relay_session *relay_session;
 187
 188         for (i = 0; i < session_list->len; i++) {
 189                 relay_session = g_ptr_array_index(session_list, i);
 190                 fprintf(stdout, "%s/host/%s/%s (timer = %u, "
 191                                 "%u stream(s), %u client(s) connected)\n",
 192                                 path, relay_session->hostname,
 193                                 relay_session->name, relay_session->timer,
 194                                 relay_session->streams, relay_session->clients);
 195         }
 196 }
 197
 198 static
 199 void update_session_list(GPtrArray *session_list, char *hostname,
 200                 char *session_name, uint32_t streams, uint32_t clients,
 201                 uint32_t timer)
 202 {
 203         int i, found = 0;
 204         struct lttng_live_relay_session *relay_session;
 205
 206         for (i = 0; i < session_list->len; i++) {
 207                 relay_session = g_ptr_array_index(session_list, i);
 208                 if ((strncmp(relay_session->hostname, hostname, NAME_MAX) == 0) &&
 209                                 strncmp(relay_session->name,
 210                                         session_name, NAME_MAX) == 0) {
 211                         relay_session->streams += streams;
 212                         if (relay_session->clients < clients)
 213                                 relay_session->clients = clients;
 214                         found = 1;
 215                         break;
 216                 }
 217         }
 218         if (found)
 219                 return;
 220
 221         relay_session = g_new0(struct lttng_live_relay_session, 1);
 222         relay_session->hostname = strndup(hostname, NAME_MAX);
 223         relay_session->name = strndup(session_name, NAME_MAX);
 224         relay_session->clients = clients;
 225         relay_session->streams = streams;
 226         relay_session->timer = timer;
 227         g_ptr_array_add(session_list, relay_session);
 228 }
 229
 230 int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
 231 {
 232         struct lttng_viewer_cmd cmd;
 233         struct lttng_viewer_list_sessions list;
 234         struct lttng_viewer_session lsession;
 235         int i, ret, sessions_count, print_list = 0;
 236         ssize_t ret_len;
 237         uint64_t session_id;
 238         GPtrArray *session_list = NULL;
 239
 240         if (strlen(ctx->session_name) == 0) {
 241                 print_list = 1;
 242                 session_list = g_ptr_array_new();
 243         }
 244
 245         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 246         cmd.data_size = 0;
 247         cmd.cmd_version = 0;
 248
 249         do {
 250                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 251         } while (ret_len < 0 && errno == EINTR);
 252         if (ret_len < 0) {
 253                 fprintf(stderr, "[error] Error sending cmd\n");
 254                 ret = ret_len;
 255                 goto error;
 256         }
 257         assert(ret_len == sizeof(cmd));
 258
 259         do {
 260                 ret_len = recv(ctx->control_sock, &list, sizeof(list), 0);
 261         } while (ret_len < 0 && errno == EINTR);
 262         if (ret_len < 0) {
 263                 fprintf(stderr, "[error] Error receiving session list\n");
 264                 ret = ret_len;
 265                 goto error;
 266         }
 267         assert(ret_len == sizeof(list));
 268
 269         sessions_count = be32toh(list.sessions_count);
 270         for (i = 0; i < sessions_count; i++) {
 271                 do {
 272                         ret_len = recv(ctx->control_sock, &lsession, sizeof(lsession), 0);
 273                 } while (ret_len < 0 && errno == EINTR);
 274                 if (ret_len < 0) {
 275                         fprintf(stderr, "[error] Error receiving session\n");
 276                         ret = ret_len;
 277                         goto error;
 278                 }
 279                 assert(ret_len == sizeof(lsession));
 280                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 281                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 282                 session_id = be64toh(lsession.id);
 283
 284                 if (print_list) {
 285                         update_session_list(session_list,
 286                                         lsession.hostname,
 287                                         lsession.session_name,
 288                                         be32toh(lsession.streams),
 289                                         be32toh(lsession.clients),
 290                                         be32toh(lsession.live_timer));
 291                 } else {
 292                         if ((strncmp(lsession.session_name, ctx->session_name,
 293                                 NAME_MAX) == 0) && (strncmp(lsession.hostname,
 294                                         ctx->traced_hostname, NAME_MAX) == 0)) {
 295                                 printf_verbose("Reading from session %" PRIu64 "\n",
 296                                                 session_id);
 297                                 g_array_append_val(ctx->session_ids,
 298                                                 session_id);
 299                         }
 300                 }
 301         }
 302
 303         if (print_list) {
 304                 print_session_list(session_list, path);
 305                 free_session_list(session_list);
 306         }
 307
 308         ret = 0;
 309
 310 error:
 311         return ret;
 312 }
 313
 314 int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream,
 315                 uint64_t ctf_trace_id)
 316 {
 317         struct lttng_live_ctf_trace *trace;
 318         int ret = 0;
 319
 320         trace = g_hash_table_lookup(stream->session->ctf_traces,
 321                         (gpointer) ctf_trace_id);
 322         if (!trace) {
 323                 trace = g_new0(struct lttng_live_ctf_trace, 1);
 324                 trace->ctf_trace_id = ctf_trace_id;
 325                 trace->streams = g_ptr_array_new();
 326                 g_hash_table_insert(stream->session->ctf_traces,
 327                                 (gpointer) ctf_trace_id,
 328                                 trace);
 329         }
 330         if (stream->metadata_flag)
 331                 trace->metadata_stream = stream;
 332
 333         stream->ctf_trace = trace;
 334         g_ptr_array_add(trace->streams, stream);
 335
 336         return ret;
 337 }
 338
 339 int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 340 {
 341         struct lttng_viewer_cmd cmd;
 342         struct lttng_viewer_attach_session_request rq;
 343         struct lttng_viewer_attach_session_response rp;
 344         struct lttng_viewer_stream stream;
 345         int ret, i;
 346         ssize_t ret_len;
 347
 348         cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 349         cmd.data_size = sizeof(rq);
 350         cmd.cmd_version = 0;
 351
 352         memset(&rq, 0, sizeof(rq));
 353         rq.session_id = htobe64(id);
 354         // TODO: add cmd line parameter to select seek beginning
 355         // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
 356         rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
 357
 358         do {
 359                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 360         } while (ret_len < 0 && errno == EINTR);
 361         if (ret_len < 0) {
 362                 fprintf(stderr, "[error] Error sending cmd\n");
 363                 ret = ret_len;
 364                 goto error;
 365         }
 366         assert(ret_len == sizeof(cmd));
 367
 368         do {
 369                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
 370         } while (ret_len < 0 && errno == EINTR);
 371         if (ret_len < 0) {
 372                 fprintf(stderr, "[error] Error sending attach request\n");
 373                 ret = ret_len;
 374                 goto error;
 375         }
 376         assert(ret_len == sizeof(rq));
 377
 378         do {
 379                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
 380         } while (ret_len < 0 && errno == EINTR);
 381         if (ret_len < 0) {
 382                 fprintf(stderr, "[error] Error receiving attach response\n");
 383                 ret = ret_len;
 384                 goto error;
 385         }
 386         assert(ret_len == sizeof(rp));
 387
 388         switch(be32toh(rp.status)) {
 389         case LTTNG_VIEWER_ATTACH_OK:
 390                 break;
 391         case LTTNG_VIEWER_ATTACH_UNK:
 392                 ret = -LTTNG_VIEWER_ATTACH_UNK;
 393                 goto end;
 394         case LTTNG_VIEWER_ATTACH_ALREADY:
 395                 fprintf(stderr, "[error] Already a viewer attached\n");
 396                 ret = -1;
 397                 goto end;
 398         case LTTNG_VIEWER_ATTACH_NOT_LIVE:
 399                 fprintf(stderr, "[error] Not a live session\n");
 400                 ret = -1;
 401                 goto end;
 402         case LTTNG_VIEWER_ATTACH_SEEK_ERR:
 403                 fprintf(stderr, "[error] Wrong seek parameter\n");
 404                 ret = -1;
 405                 goto end;
 406         default:
 407                 fprintf(stderr, "[error] Unknown attach return code %u\n",
 408                                 be32toh(rp.status));
 409                 ret = -1;
 410                 goto end;
 411         }
 412         if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
 413                 ret = -1;
 414                 goto end;
 415         }
 416
 417         ctx->session->stream_count += be32toh(rp.streams_count);
 418         /*
 419          * When the session is created but not started, we do an active wait
 420          * until it starts. It allows the viewer to start processing the trace
 421          * as soon as the session starts.
 422          */
 423         if (ctx->session->stream_count == 0) {
 424                 ret = 0;
 425                 goto end;
 426         }
 427         printf_verbose("Waiting for %" PRIu64 " streams:\n",
 428                 ctx->session->stream_count);
 429         ctx->session->streams = g_new0(struct lttng_live_viewer_stream,
 430                         ctx->session->stream_count);
 431         for (i = 0; i < be32toh(rp.streams_count); i++) {
 432                 do {
 433                         ret_len = recv(ctx->control_sock, &stream, sizeof(stream), 0);
 434                 } while (ret_len < 0 && errno == EINTR);
 435                 if (ret_len < 0) {
 436                         fprintf(stderr, "[error] Error receiving stream\n");
 437                         ret = ret_len;
 438                         goto error;
 439                 }
 440                 assert(ret_len == sizeof(stream));
 441                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
 442                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 443
 444                 printf_verbose("    stream %" PRIu64 " : %s/%s\n",
 445                                 be64toh(stream.id), stream.path_name,
 446                                 stream.channel_name);
 447                 ctx->session->streams[i].id = be64toh(stream.id);
 448                 ctx->session->streams[i].session = ctx->session;
 449
 450                 ctx->session->streams[i].first_read = 1;
 451                 ctx->session->streams[i].mmap_size = 0;
 452
 453                 if (be32toh(stream.metadata_flag)) {
 454                         char *path;
 455
 456                         path = strdup(LTTNG_METADATA_PATH_TEMPLATE);
 457                         if (!path) {
 458                                 perror("strdup");
 459                                 ret = -1;
 460                                 goto error;
 461                         }
 462                         if (!mkdtemp(path)) {
 463                                 perror("mkdtemp");
 464                                 free(path);
 465                                 ret = -1;
 466                                 goto error;
 467                         }
 468                         ctx->session->streams[i].metadata_flag = 1;
 469                         snprintf(ctx->session->streams[i].path,
 470                                         sizeof(ctx->session->streams[i].path),
 471                                         "%s/%s", path,
 472                                         stream.channel_name);
 473                         ret = open(ctx->session->streams[i].path,
 474                                         O_WRONLY | O_CREAT | O_TRUNC,
 475                                         S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
 476                         if (ret < 0) {
 477                                 perror("open");
 478                                 free(path);
 479                                 goto error;
 480                         }
 481                         ctx->session->streams[i].fd = ret;
 482                         free(path);
 483                 }
 484                 ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i],
 485                                 be64toh(stream.ctf_trace_id));
 486                 if (ret < 0) {
 487                         goto error;
 488                 }
 489
 490         }
 491         ret = 0;
 492
 493 end:
 494 error:
 495         return ret;
 496 }
 497
 498 static
 499 int ask_new_streams(struct lttng_live_ctx *ctx)
 500 {
 501         int i, ret = 0;
 502         uint64_t id;
 503
 504 restart:
 505         for (i = 0; i < ctx->session_ids->len; i++) {
 506                 id = g_array_index(ctx->session_ids, uint64_t, i);
 507                 ret = lttng_live_get_new_streams(ctx, id);
 508                 printf_verbose("Asking for new streams returns %d\n",
 509                                 ret);
 510                 if (ret < 0) {
 511                         if (ret == -LTTNG_VIEWER_NEW_STREAMS_HUP) {
 512                                 printf_verbose("Session %" PRIu64 " closed\n",
 513                                                 id);
 514                                 /*
 515                                  * The streams have already been closed during
 516                                  * the reading, so we only need to get rid of
 517                                  * the trace in our internal table of sessions.
 518                                  */
 519                                 g_array_remove_index(ctx->session_ids, i);
 520                                 /*
 521                                  * We can't continue iterating on the g_array
 522                                  * after a remove, we have to start again.
 523                                  */
 524                                 goto restart;
 525                         } else {
 526                                 ret = -1;
 527                                 goto end;
 528                         }
 529                 }
 530         }
 531
 532 end:
 533         return ret;
 534 }
 535
 536 static
 537 int get_data_packet(struct lttng_live_ctx *ctx,
 538                 struct ctf_stream_pos *pos,
 539                 struct lttng_live_viewer_stream *stream, uint64_t offset,
 540                 uint64_t len)
 541 {
 542         struct lttng_viewer_cmd cmd;
 543         struct lttng_viewer_get_packet rq;
 544         struct lttng_viewer_trace_packet rp;
 545         ssize_t ret_len;
 546         int ret;
 547
 548         cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
 549         cmd.data_size = sizeof(rq);
 550         cmd.cmd_version = 0;
 551
 552         memset(&rq, 0, sizeof(rq));
 553         rq.stream_id = htobe64(stream->id);
 554         /* Already in big endian. */
 555         rq.offset = offset;
 556         rq.len = htobe32(len);
 557
 558         do {
 559                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 560         } while (ret_len < 0 && errno == EINTR);
 561         if (ret_len < 0) {
 562                 fprintf(stderr, "[error] Error sending cmd\n");
 563                 ret = ret_len;
 564                 goto error;
 565         }
 566         assert(ret_len == sizeof(cmd));
 567
 568         do {
 569                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
 570         } while (ret_len < 0 && errno == EINTR);
 571         if (ret_len < 0) {
 572                 fprintf(stderr, "[error] Error sending get_data_packet request\n");
 573                 ret = ret_len;
 574                 goto error;
 575         }
 576         assert(ret_len == sizeof(rq));
 577
 578         do {
 579                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
 580         } while (ret_len < 0 && errno == EINTR);
 581         if (ret_len < 0) {
 582                 fprintf(stderr, "[error] Error receiving data response\n");
 583                 ret = ret_len;
 584                 goto error;
 585         }
 586         if (ret_len != sizeof(rp)) {
 587                 fprintf(stderr, "[error] get_data_packet: expected %" PRId64
 588                                 ", received %" PRId64 "\n", ret_len,
 589                                 sizeof(rp));
 590                 ret = -1;
 591                 goto error;
 592         }
 593
 594         rp.flags = be32toh(rp.flags);
 595
 596         switch (be32toh(rp.status)) {
 597         case LTTNG_VIEWER_GET_PACKET_OK:
 598                 len = be32toh(rp.len);
 599                 printf_verbose("get_data_packet: Ok, packet size : %" PRIu64
 600                                 "\n", len);
 601                 break;
 602         case LTTNG_VIEWER_GET_PACKET_RETRY:
 603                 printf_verbose("get_data_packet: retry\n");
 604                 ret = -1;
 605                 goto end;
 606         case LTTNG_VIEWER_GET_PACKET_ERR:
 607                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 608                         printf_verbose("get_data_packet: new metadata needed\n");
 609                         ret = 0;
 610                         goto end;
 611                 }
 612                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
 613                         ret = ask_new_streams(ctx);
 614                         if (ret < 0)
 615                                 goto error;
 616                         g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
 617                                         ctx->bt_ctx);
 618                 }
 619                 fprintf(stderr, "[error] get_data_packet: error\n");
 620                 ret = -1;
 621                 goto end;
 622         case LTTNG_VIEWER_GET_PACKET_EOF:
 623                 ret = -2;
 624                 goto error;
 625         default:
 626                 printf_verbose("get_data_packet: unknown\n");
 627                 assert(0);
 628                 ret = -1;
 629                 goto end;
 630         }
 631
 632         if (len <= 0) {
 633                 ret = -1;
 634                 goto end;
 635         }
 636
 637         if (len > stream->mmap_size) {
 638                 uint64_t new_size;
 639
 640                 new_size = max_t(uint64_t, len, stream->mmap_size << 1);
 641                 if (pos->base_mma) {
 642                         /* unmap old base */
 643                         ret = munmap_align(pos->base_mma);
 644                         if (ret) {
 645                                 fprintf(stderr, "[error] Unable to unmap old base: %s.\n",
 646                                         strerror(errno));
 647                                 ret = -1;
 648                                 goto error;
 649                         }
 650                         pos->base_mma = NULL;
 651                 }
 652                 pos->base_mma = mmap_align(new_size,
 653                                 PROT_READ | PROT_WRITE,
 654                                 MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
 655                 if (pos->base_mma == MAP_FAILED) {
 656                         fprintf(stderr, "[error] mmap error %s.\n",
 657                                 strerror(errno));
 658                         pos->base_mma = NULL;
 659                         ret = -1;
 660                         goto error;
 661                 }
 662
 663                 stream->mmap_size = new_size;
 664                 printf_verbose("Expanding stream mmap size to %" PRIu64 " bytes\n",
 665                                 stream->mmap_size);
 666         }
 667
 668         do {
 669                 ret_len = recv(ctx->control_sock,
 670                         mmap_align_addr(pos->base_mma), len,
 671                         MSG_WAITALL);
 672         } while (ret_len < 0 && errno == EINTR);
 673         if (ret_len < 0) {
 674                 fprintf(stderr, "[error] Error receiving trace packet\n");
 675                 ret = ret_len;
 676                 goto error;
 677         }
 678         assert(ret_len == len);
 679
 680 end:
 681 error:
 682         return ret;
 683 }
 684
 685 /*
 686  * Return number of metadata bytes written or a negative value on error.
 687  */
 688 static
 689 int get_new_metadata(struct lttng_live_ctx *ctx,
 690                 struct lttng_live_viewer_stream *viewer_stream,
 691                 uint64_t *metadata_len)
 692 {
 693         uint64_t len = 0;
 694         int ret;
 695         struct lttng_viewer_cmd cmd;
 696         struct lttng_viewer_get_metadata rq;
 697         struct lttng_viewer_metadata_packet rp;
 698         struct lttng_live_viewer_stream *metadata_stream;
 699         char *data = NULL;
 700         ssize_t ret_len;
 701
 702         cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
 703         cmd.data_size = sizeof(rq);
 704         cmd.cmd_version = 0;
 705
 706         metadata_stream = viewer_stream->ctf_trace->metadata_stream;
 707         rq.stream_id = htobe64(metadata_stream->id);
 708
 709         do {
 710                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 711         } while (ret_len < 0 && errno == EINTR);
 712         if (ret_len < 0) {
 713                 fprintf(stderr, "[error] Error sending cmd\n");
 714                 ret = ret_len;
 715                 goto error;
 716         }
 717         assert(ret_len == sizeof(cmd));
 718
 719         do {
 720                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
 721         } while (ret_len < 0 && errno == EINTR);
 722         if (ret_len < 0) {
 723                 fprintf(stderr, "[error] Error sending get_metadata request\n");
 724                 ret = ret_len;
 725                 goto error;
 726         }
 727         assert(ret_len == sizeof(rq));
 728
 729         do {
 730                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
 731         } while (ret_len < 0 && errno == EINTR);
 732         if (ret_len < 0) {
 733                 fprintf(stderr, "[error] Error receiving metadata response\n");
 734                 ret = ret_len;
 735                 goto error;
 736         }
 737         assert(ret_len == sizeof(rp));
 738
 739         switch (be32toh(rp.status)) {
 740                 case LTTNG_VIEWER_METADATA_OK:
 741                         printf_verbose("get_metadata : OK\n");
 742                         break;
 743                 case LTTNG_VIEWER_NO_NEW_METADATA:
 744                         printf_verbose("get_metadata : NO NEW\n");
 745                         ret = -1;
 746                         goto end;
 747                 case LTTNG_VIEWER_METADATA_ERR:
 748                         printf_verbose("get_metadata : ERR\n");
 749                         ret = -1;
 750                         goto end;
 751                 default:
 752                         printf_verbose("get_metadata : UNKNOWN\n");
 753                         ret = -1;
 754                         goto end;
 755         }
 756
 757         len = be64toh(rp.len);
 758         printf_verbose("Writing %" PRIu64" bytes to metadata\n", len);
 759         if (len <= 0) {
 760                 ret = -1;
 761                 goto end;
 762         }
 763
 764         data = zmalloc(len);
 765         if (!data) {
 766                 perror("relay data zmalloc");
 767                 ret = -1;
 768                 goto error;
 769         }
 770         do {
 771                 ret_len = recv(ctx->control_sock, data, len, MSG_WAITALL);
 772         } while (ret_len < 0 && errno == EINTR);
 773         if (ret_len < 0) {
 774                 fprintf(stderr, "[error] Error receiving trace packet\n");
 775                 ret = ret_len;
 776                 free(data);
 777                 goto error;
 778         }
 779         assert(ret_len == len);
 780
 781         do {
 782                 ret_len = write(metadata_stream->fd, data, len);
 783         } while (ret_len < 0 && errno == EINTR);
 784         if (ret_len < 0) {
 785                 free(data);
 786                 ret = ret_len;
 787                 goto error;
 788         }
 789         assert(ret_len == len);
 790
 791         free(data);
 792
 793         *metadata_len = len;
 794         ret = 0;
 795 end:
 796 error:
 797         return ret;
 798 }
 799
 800 /*
 801  * Get one index for a stream.
 802  *
 803  * Returns 0 on success or a negative value on error.
 804  */
 805 static
 806 int get_next_index(struct lttng_live_ctx *ctx,
 807                 struct lttng_live_viewer_stream *viewer_stream,
 808                 struct packet_index *index)
 809 {
 810         struct lttng_viewer_cmd cmd;
 811         struct lttng_viewer_get_next_index rq;
 812         struct lttng_viewer_index rp;
 813         int ret;
 814         uint64_t metadata_len;
 815         ssize_t ret_len;
 816
 817         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
 818         cmd.data_size = sizeof(rq);
 819         cmd.cmd_version = 0;
 820
 821         memset(&rq, 0, sizeof(rq));
 822         rq.stream_id = htobe64(viewer_stream->id);
 823
 824 retry:
 825         do {
 826                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 827         } while (ret_len < 0 && errno == EINTR);
 828         if (ret_len < 0) {
 829                 fprintf(stderr, "[error] Error sending cmd\n");
 830                 ret = ret_len;
 831                 goto error;
 832         }
 833         assert(ret_len == sizeof(cmd));
 834
 835         do {
 836                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
 837         } while (ret_len < 0 && errno == EINTR);
 838         if (ret_len < 0) {
 839                 fprintf(stderr, "[error] Error sending get_next_index request\n");
 840                 ret = ret_len;
 841                 goto error;
 842         }
 843         assert(ret_len == sizeof(rq));
 844
 845         do {
 846                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
 847         } while (ret_len < 0 && errno == EINTR);
 848         if (ret_len < 0) {
 849                 fprintf(stderr, "[error] Error receiving index response\n");
 850                 ret = ret_len;
 851                 goto error;
 852         }
 853         assert(ret_len == sizeof(rp));
 854
 855         rp.flags = be32toh(rp.flags);
 856
 857         switch (be32toh(rp.status)) {
 858         case LTTNG_VIEWER_INDEX_INACTIVE:
 859                 printf_verbose("get_next_index: inactive\n");
 860                 memset(index, 0, sizeof(struct packet_index));
 861                 index->timestamp_end = be64toh(rp.timestamp_end);
 862                 break;
 863         case LTTNG_VIEWER_INDEX_OK:
 864                 printf_verbose("get_next_index: Ok, need metadata update : %u\n",
 865                                 rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA);
 866                 index->offset = be64toh(rp.offset);
 867                 index->packet_size = be64toh(rp.packet_size);
 868                 index->content_size = be64toh(rp.content_size);
 869                 index->timestamp_begin = be64toh(rp.timestamp_begin);
 870                 index->timestamp_end = be64toh(rp.timestamp_end);
 871                 index->events_discarded = be64toh(rp.events_discarded);
 872
 873                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 874                         printf_verbose("get_next_index: new metadata needed\n");
 875                         ret = get_new_metadata(ctx, viewer_stream,
 876                                         &metadata_len);
 877                         if (ret < 0) {
 878                                 goto error;
 879                         }
 880                 }
 881                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
 882                         ret = ask_new_streams(ctx);
 883                         if (ret < 0)
 884                                 goto error;
 885                         g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
 886                                         ctx->bt_ctx);
 887                 }
 888                 break;
 889         case LTTNG_VIEWER_INDEX_RETRY:
 890                 printf_verbose("get_next_index: retry\n");
 891                 sleep(0.1);
 892                 goto retry;
 893         case LTTNG_VIEWER_INDEX_HUP:
 894                 printf_verbose("get_next_index: stream hung up\n");
 895                 viewer_stream->id = -1ULL;
 896                 viewer_stream->fd = -1;
 897                 index->offset = EOF;
 898                 ctx->session->stream_count--;
 899                 break;
 900         case LTTNG_VIEWER_INDEX_ERR:
 901                 fprintf(stderr, "[error] get_next_index: error\n");
 902                 ret = -1;
 903                 goto error;
 904         default:
 905                 fprintf(stderr, "[error] get_next_index: unkwown value\n");
 906                 ret = -1;
 907                 goto error;
 908         }
 909
 910         ret = 0;
 911
 912 error:
 913         return ret;
 914 }
 915
 916 static
 917 void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
 918                 int whence)
 919 {
 920         struct ctf_stream_pos *pos;
 921         struct ctf_file_stream *file_stream;
 922         struct packet_index packet_index;
 923         struct lttng_live_viewer_stream *viewer_stream;
 924         struct lttng_live_session *session;
 925         int ret;
 926
 927 retry:
 928         pos = ctf_pos(stream_pos);
 929         file_stream = container_of(pos, struct ctf_file_stream, pos);
 930         viewer_stream = (struct lttng_live_viewer_stream *) pos->priv;
 931         session = viewer_stream->session;
 932
 933         printf_verbose("get_next_index for stream %" PRIu64 "\n", viewer_stream->id);
 934         ret = get_next_index(session->ctx, viewer_stream, &packet_index);
 935         if (ret < 0) {
 936                 pos->offset = EOF;
 937                 fprintf(stderr, "[error] get_next_index failed\n");
 938                 return;
 939         }
 940
 941         pos->packet_size = packet_index.packet_size;
 942         pos->content_size = packet_index.content_size;
 943         pos->mmap_base_offset = 0;
 944         if (packet_index.offset == EOF) {
 945                 pos->offset = EOF;
 946         } else {
 947                 pos->offset = 0;
 948         }
 949
 950         if (packet_index.content_size == 0) {
 951                 file_stream->parent.cycles_timestamp = packet_index.timestamp_end;
 952                 file_stream->parent.real_timestamp = ctf_get_real_timestamp(
 953                                 &file_stream->parent, packet_index.timestamp_end);
 954         } else {
 955                 file_stream->parent.cycles_timestamp = packet_index.timestamp_begin;
 956                 file_stream->parent.real_timestamp = ctf_get_real_timestamp(
 957                                 &file_stream->parent, packet_index.timestamp_begin);
 958         }
 959
 960         if (pos->packet_size == 0 || pos->offset == EOF) {
 961                 goto end;
 962         }
 963
 964         printf_verbose("get_data_packet for stream %" PRIu64 "\n",
 965                         viewer_stream->id);
 966         ret = get_data_packet(session->ctx, pos, viewer_stream,
 967                         be64toh(packet_index.offset),
 968                         packet_index.packet_size / CHAR_BIT);
 969         if (ret == -2) {
 970                 goto retry;
 971         } else if (ret < 0) {
 972                 pos->offset = EOF;
 973                 fprintf(stderr, "[error] get_data_packet failed\n");
 974                 return;
 975         }
 976
 977         printf_verbose("Index received : packet_size : %" PRIu64
 978                         ", offset %" PRIu64 ", content_size %" PRIu64
 979                         ", timestamp_end : %" PRIu64 "\n",
 980                         packet_index.packet_size, packet_index.offset,
 981                         packet_index.content_size, packet_index.timestamp_end);
 982
 983         /* update trace_packet_header and stream_packet_context */
 984         if (pos->prot != PROT_WRITE && file_stream->parent.trace_packet_header) {
 985                 /* Read packet header */
 986                 ret = generic_rw(&pos->parent, &file_stream->parent.trace_packet_header->p);
 987                 if (ret) {
 988                         pos->offset = EOF;
 989                         fprintf(stderr, "[error] trace packet header read failed\n");
 990                         goto end;
 991                 }
 992         }
 993         if (pos->prot != PROT_WRITE && file_stream->parent.stream_packet_context) {
 994                 /* Read packet context */
 995                 ret = generic_rw(&pos->parent, &file_stream->parent.stream_packet_context->p);
 996                 if (ret) {
 997                         pos->offset = EOF;
 998                         fprintf(stderr, "[error] stream packet context read failed\n");
 999                         goto end;
1000                 }
1001         }
1002         pos->data_offset = pos->offset;
1003
1004 end:
1005         return;
1006 }
1007
1008 int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx)
1009 {
1010         struct lttng_viewer_cmd cmd;
1011         struct lttng_viewer_create_session_response resp;
1012         int ret;
1013         ssize_t ret_len;
1014
1015         cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
1016         cmd.data_size = 0;
1017         cmd.cmd_version = 0;
1018
1019         do {
1020                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
1021         } while (ret_len < 0 && errno == EINTR);
1022         if (ret_len < 0) {
1023                 fprintf(stderr, "[error] Error sending cmd\n");
1024                 ret = ret_len;
1025                 goto error;
1026         }
1027         assert(ret_len == sizeof(cmd));
1028
1029         do {
1030                 ret_len = recv(ctx->control_sock, &resp, sizeof(resp), 0);
1031         } while (ret_len < 0 && errno == EINTR);
1032         if (ret_len < 0) {
1033                 fprintf(stderr, "[error] Error receiving create session reply\n");
1034                 ret = ret_len;
1035                 goto error;
1036         }
1037         assert(ret_len == sizeof(resp));
1038
1039         if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
1040                 fprintf(stderr, "[error] Error creating viewer session\n");
1041                 ret = -1;
1042                 goto error;
1043         }
1044         ret = 0;
1045
1046 error:
1047         return ret;
1048 }
1049
1050 static
1051 int del_traces(gpointer key, gpointer value, gpointer user_data)
1052 {
1053         struct bt_context *bt_ctx = user_data;
1054         struct lttng_live_ctf_trace *trace = value;
1055         int ret;
1056
1057         ret = bt_context_remove_trace(bt_ctx, trace->trace_id);
1058         if (ret < 0)
1059                 fprintf(stderr, "[error] removing trace from context\n");
1060
1061         /* remove the key/value pair from the HT. */
1062         return 1;
1063 }
1064
1065 static
1066 void add_traces(gpointer key, gpointer value, gpointer user_data)
1067 {
1068         int i, ret, total_metadata = 0;
1069         uint64_t metadata_len;
1070         struct bt_context *bt_ctx = user_data;
1071         struct lttng_live_ctf_trace *trace = value;
1072         struct lttng_live_viewer_stream *stream;
1073         struct bt_mmap_stream *new_mmap_stream;
1074         struct bt_mmap_stream_list mmap_list;
1075         struct lttng_live_ctx *ctx = NULL;
1076
1077         /*
1078          * We don't know how many streams we will receive for a trace, so
1079          * once we are done receiving the traces, we add all the traces
1080          * received to the bt_context.
1081          * We can receive streams during the attach command or the
1082          * get_new_streams, so we have to make sure not to add multiple
1083          * times the same traces.
1084          * If a trace is already in the context, we just skip this function.
1085          */
1086         if (trace->in_use)
1087                 return;
1088
1089         BT_INIT_LIST_HEAD(&mmap_list.head);
1090
1091         for (i = 0; i < trace->streams->len; i++) {
1092                 stream = g_ptr_array_index(trace->streams, i);
1093                 ctx = stream->session->ctx;
1094
1095                 if (!stream->metadata_flag) {
1096                         new_mmap_stream = zmalloc(sizeof(struct bt_mmap_stream));
1097                         new_mmap_stream->priv = (void *) stream;
1098                         new_mmap_stream->fd = -1;
1099                         bt_list_add(&new_mmap_stream->list, &mmap_list.head);
1100                 } else {
1101                         /* Get all possible metadata before starting */
1102                         do {
1103                                 ret = get_new_metadata(ctx, stream,
1104                                                 &metadata_len);
1105                                 if (ret == 0) {
1106                                         total_metadata += metadata_len;
1107                                 }
1108                         } while (ret == 0 || total_metadata == 0);
1109                         trace->metadata_fp = fopen(stream->path, "r");
1110                 }
1111         }
1112
1113         if (!trace->metadata_fp) {
1114                 fprintf(stderr, "[error] No metadata stream opened\n");
1115                 goto end_free;
1116         }
1117
1118         ret = bt_context_add_trace(bt_ctx, NULL, "ctf",
1119                         ctf_live_packet_seek, &mmap_list, trace->metadata_fp);
1120         if (ret < 0) {
1121                 fprintf(stderr, "[error] Error adding trace\n");
1122                 goto end_free;
1123         }
1124
1125         if (bt_ctx->current_iterator) {
1126                 struct bt_trace_descriptor *td;
1127                 struct bt_trace_handle *handle;
1128
1129                 handle = (struct bt_trace_handle *) g_hash_table_lookup(
1130                                 bt_ctx->trace_handles,
1131                                 (gpointer) (unsigned long) ret);
1132                 td = handle->td;
1133                 bt_iter_add_trace(bt_ctx->current_iterator, td);
1134         }
1135
1136         trace->trace_id = ret;
1137         trace->in_use = 1;
1138
1139         goto end;
1140
1141 end_free:
1142         bt_context_put(bt_ctx);
1143 end:
1144         return;
1145 }
1146
1147 int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
1148 {
1149         struct lttng_viewer_cmd cmd;
1150         struct lttng_viewer_new_streams_request rq;
1151         struct lttng_viewer_new_streams_response rp;
1152         struct lttng_viewer_stream stream;
1153         int ret, i;
1154         ssize_t ret_len;
1155         uint32_t stream_count;
1156
1157         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1158         cmd.data_size = sizeof(rq);
1159         cmd.cmd_version = 0;
1160
1161         memset(&rq, 0, sizeof(rq));
1162         rq.session_id = htobe64(id);
1163
1164         do {
1165                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
1166         } while (ret_len < 0 && errno == EINTR);
1167         if (ret_len < 0) {
1168                 fprintf(stderr, "[error] Error sending cmd\n");
1169                 ret = ret_len;
1170                 goto error;
1171         }
1172         assert(ret_len == sizeof(cmd));
1173
1174         do {
1175                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
1176         } while (ret_len < 0 && errno == EINTR);
1177         if (ret_len < 0) {
1178                 fprintf(stderr, "[error] Error sending get_new_streams request\n");
1179                 ret = ret_len;
1180                 goto error;
1181         }
1182         assert(ret_len == sizeof(rq));
1183
1184         do {
1185                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
1186         } while (ret_len < 0 && errno == EINTR);
1187         if (ret_len < 0) {
1188                 fprintf(stderr, "[error] Error receiving get_new_streams response\n");
1189                 ret = ret_len;
1190                 goto error;
1191         }
1192         assert(ret_len == sizeof(rp));
1193
1194         switch(be32toh(rp.status)) {
1195         case LTTNG_VIEWER_NEW_STREAMS_OK:
1196                 break;
1197         case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1198                 ret = 0;
1199                 goto end;
1200         case LTTNG_VIEWER_NEW_STREAMS_HUP:
1201                 ret = -LTTNG_VIEWER_NEW_STREAMS_HUP;
1202                 goto end;
1203         case LTTNG_VIEWER_NEW_STREAMS_ERR:
1204                 fprintf(stderr, "[error] get_new_streams error\n");
1205                 ret = -1;
1206                 goto end;
1207         default:
1208                 fprintf(stderr, "[error] Unknown return code %u\n",
1209                                 be32toh(rp.status));
1210                 ret = -1;
1211                 goto end;
1212         }
1213
1214         stream_count = be32toh(rp.streams_count);
1215         ctx->session->stream_count += stream_count;
1216         /*
1217          * When the session is created but not started, we do an active wait
1218          * until it starts. It allows the viewer to start processing the trace
1219          * as soon as the session starts.
1220          */
1221         if (ctx->session->stream_count == 0) {
1222                 ret = 0;
1223                 goto end;
1224         }
1225         printf_verbose("Waiting for %" PRIu64 " streams:\n",
1226                 ctx->session->stream_count);
1227         ctx->session->streams = g_new0(struct lttng_live_viewer_stream,
1228                         ctx->session->stream_count);
1229         for (i = 0; i < stream_count; i++) {
1230                 do {
1231                         ret_len = recv(ctx->control_sock, &stream, sizeof(stream), 0);
1232                 } while (ret_len < 0 && errno == EINTR);
1233                 if (ret_len < 0) {
1234                         fprintf(stderr, "[error] Error receiving stream\n");
1235                         ret = ret_len;
1236                         goto error;
1237                 }
1238                 assert(ret_len == sizeof(stream));
1239                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
1240                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
1241
1242                 printf_verbose("    stream %" PRIu64 " : %s/%s\n",
1243                                 be64toh(stream.id), stream.path_name,
1244                                 stream.channel_name);
1245                 ctx->session->streams[i].id = be64toh(stream.id);
1246                 ctx->session->streams[i].session = ctx->session;
1247
1248                 ctx->session->streams[i].first_read = 1;
1249                 ctx->session->streams[i].mmap_size = 0;
1250
1251                 if (be32toh(stream.metadata_flag)) {
1252                         char *path;
1253
1254                         path = strdup(LTTNG_METADATA_PATH_TEMPLATE);
1255                         if (!path) {
1256                                 perror("strdup");
1257                                 ret = -1;
1258                                 goto error;
1259                         }
1260                         if (!mkdtemp(path)) {
1261                                 perror("mkdtemp");
1262                                 free(path);
1263                                 ret = -1;
1264                                 goto error;
1265                         }
1266                         ctx->session->streams[i].metadata_flag = 1;
1267                         snprintf(ctx->session->streams[i].path,
1268                                         sizeof(ctx->session->streams[i].path),
1269                                         "%s/%s", path,
1270                                         stream.channel_name);
1271                         ret = open(ctx->session->streams[i].path,
1272                                         O_WRONLY | O_CREAT | O_TRUNC,
1273                                         S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
1274                         if (ret < 0) {
1275                                 perror("open");
1276                                 free(path);
1277                                 goto error;
1278                         }
1279                         ctx->session->streams[i].fd = ret;
1280                         free(path);
1281                 }
1282                 ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i],
1283                                 be64toh(stream.ctf_trace_id));
1284                 if (ret < 0) {
1285                         goto error;
1286                 }
1287
1288         }
1289         ret = 0;
1290
1291 end:
1292 error:
1293         return ret;
1294 }
1295
1296 void lttng_live_read(struct lttng_live_ctx *ctx)
1297 {
1298         int ret, i;
1299 #if 0
1300         struct bt_ctf_iter *iter;
1301         const struct bt_ctf_event *event;
1302         struct bt_iter_pos begin_pos;
1303         struct bt_trace_descriptor *td_write;
1304         struct bt_format *fmt_write;
1305         struct ctf_text_stream_pos *sout;
1306 #endif
1307         uint64_t id;
1308
1309         ctx->bt_ctx = bt_context_create();
1310         if (!ctx->bt_ctx) {
1311                 fprintf(stderr, "[error] bt_context_create allocation\n");
1312                 goto end;
1313         }
1314
1315 #if 0
1316         fmt_write = bt_lookup_format(g_quark_from_static_string("text"));
1317         if (!fmt_write) {
1318                 fprintf(stderr, "[error] ctf-text error\n");
1319                 goto end;
1320         }
1321
1322         td_write = fmt_write->open_trace(NULL, O_RDWR, NULL, NULL);
1323         if (!td_write) {
1324                 fprintf(stderr, "[error] Error opening output trace\n");
1325                 goto end_free;
1326         }
1327
1328         sout = container_of(td_write, struct ctf_text_stream_pos,
1329                         trace_descriptor);
1330         if (!sout->parent.event_cb)
1331                 goto end_free;
1332 #endif
1333
1334         ret = lttng_live_create_viewer_session(ctx);
1335         if (ret < 0) {
1336                 goto end_free;
1337         }
1338
1339         for (i = 0; i < ctx->session_ids->len; i++) {
1340                 id = g_array_index(ctx->session_ids, uint64_t, i);
1341                 printf_verbose("Attaching to session %lu\n", id);
1342                 ret = lttng_live_attach_session(ctx, id);
1343                 printf_verbose("Attaching session returns %d\n", ret);
1344                 if (ret < 0) {
1345                         if (ret == -LTTNG_VIEWER_ATTACH_UNK) {
1346                                 fprintf(stderr, "[error] Unknown session ID\n");
1347                         }
1348                         goto end_free;
1349                 }
1350         }
1351
1352         /*
1353          * As long as the session is active, we try to get new streams.
1354          */
1355 #if 0
1356         for (;;) {
1357                 int flags;
1358 #endif
1359
1360                 while (!ctx->session->stream_count) {
1361                         if (ctx->session_ids->len == 0)
1362                                 goto end_free;
1363                         ret = ask_new_streams(ctx);
1364                         if (ret < 0)
1365                                 goto end_free;
1366                 }
1367
1368                 g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
1369                                 ctx->bt_ctx);
1370
1371 #if 0
1372                 begin_pos.type = BT_SEEK_BEGIN;
1373                 iter = bt_ctf_iter_create(ctx->bt_ctx, &begin_pos, NULL);
1374                 if (!iter) {
1375                         fprintf(stderr, "[error] Iterator creation error\n");
1376                         goto end;
1377                 }
1378
1379                 for (;;) {
1380                         event = bt_ctf_iter_read_event_flags(iter, &flags);
1381                         if (!(flags & BT_ITER_FLAG_RETRY)) {
1382                                 if (!event) {
1383                                         /* End of trace */
1384                                         break;
1385                                 }
1386                                 ret = sout->parent.event_cb(&sout->parent,
1387                                                 event->parent->stream);
1388                                 if (ret) {
1389                                         fprintf(stderr, "[error] Writing "
1390                                                         "event failed.\n");
1391                                         goto end_free;
1392                                 }
1393                         }
1394                         ret = bt_iter_next(bt_ctf_get_iter(iter));
1395                         if (ret < 0) {
1396                                 goto end_free;
1397                         }
1398                 }
1399                 bt_ctf_iter_destroy(iter);
1400 #endif
1401                 ret = check_requirements(ctx->bt_ctx);
1402                 if (ret < 0) {
1403                         fprintf(stderr, "[error] some mandatory contexts "
1404                                         "were missing, exiting.\n");
1405                         goto end;
1406                 }
1407
1408                 if (!opt_textdump) {
1409                         pthread_create(&display_thread, NULL, ncurses_display,
1410                                         (void *) NULL);
1411                         pthread_create(&timer_thread, NULL, refresh_thread,
1412                                         (void *) NULL);
1413                 }
1414                 iter_trace(ctx->bt_ctx);
1415                 g_hash_table_foreach_remove(ctx->session->ctf_traces,
1416                                 del_traces, ctx->bt_ctx);
1417                 ctx->session->stream_count = 0;
1418 #if 0
1419         }
1420 #endif
1421
1422 end_free:
1423         bt_context_put(ctx->bt_ctx);
1424 end:
1425         return;
1426 }