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