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