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