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