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