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