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