formats/lttng-live/lttng-live-comm.c

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