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