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 /*
 535  * Ask the relay for new streams.
 536  *
 537  * Returns the number of new streams received or a negative value on error.
 538  */
 539 static
 540 int ask_new_streams(struct lttng_live_ctx *ctx)
 541 {
 542         int i, ret = 0, nb_streams = 0;
 543         uint64_t id;
 544
 545 restart:
 546         for (i = 0; i < ctx->session_ids->len; i++) {
 547                 id = g_array_index(ctx->session_ids, uint64_t, i);
 548                 ret = lttng_live_get_new_streams(ctx, id);
 549                 printf_verbose("Asking for new streams returns %d\n", ret);
 550                 if (ret < 0) {
 551                         if (lttng_live_should_quit()) {
 552                                 goto end;
 553                         }
 554                         if (ret == -LTTNG_VIEWER_NEW_STREAMS_HUP) {
 555                                 printf_verbose("Session %" PRIu64 " closed\n",
 556                                                 id);
 557                                 /*
 558                                  * The streams have already been closed during
 559                                  * the reading, so we only need to get rid of
 560                                  * the trace in our internal table of sessions.
 561                                  */
 562                                 g_array_remove_index(ctx->session_ids, i);
 563                                 /*
 564                                  * We can't continue iterating on the g_array
 565                                  * after a remove, we have to start again.
 566                                  */
 567                                 goto restart;
 568                         } else {
 569                                 ret = -1;
 570                                 goto end;
 571                         }
 572                 } else {
 573                         nb_streams += ret;
 574                 }
 575         }
 576         ret = nb_streams;
 577
 578 end:
 579         return ret;
 580 }
 581
 582 static
 583 int append_metadata(struct lttng_live_ctx *ctx,
 584                 struct lttng_live_viewer_stream *viewer_stream)
 585 {
 586         int ret;
 587         struct lttng_live_viewer_stream *metadata;
 588         char *metadata_buf = NULL;
 589
 590         printf_verbose("get_next_index: new metadata needed\n");
 591         ret = get_new_metadata(ctx, viewer_stream, &metadata_buf);
 592         if (ret < 0) {
 593                 free(metadata_buf);
 594                 goto error;
 595         }
 596
 597         metadata = viewer_stream->ctf_trace->metadata_stream;
 598         metadata->ctf_trace->metadata_fp =
 599                 babeltrace_fmemopen(metadata_buf,
 600                                 metadata->metadata_len, "rb");
 601         if (!metadata->ctf_trace->metadata_fp) {
 602                 perror("Metadata fmemopen");
 603                 free(metadata_buf);
 604                 ret = -1;
 605                 goto error;
 606         }
 607         ret = ctf_append_trace_metadata(
 608                         viewer_stream->ctf_trace->handle->td,
 609                         metadata->ctf_trace->metadata_fp);
 610         /* We accept empty metadata packets */
 611         if (ret != 0 && ret != -ENOENT) {
 612                 fprintf(stderr, "[error] Appending metadata\n");
 613                 goto error;
 614         }
 615         ret = 0;
 616
 617 error:
 618         return ret;
 619 }
 620
 621 static
 622 int get_data_packet(struct lttng_live_ctx *ctx,
 623                 struct ctf_stream_pos *pos,
 624                 struct lttng_live_viewer_stream *stream, uint64_t offset,
 625                 uint64_t len)
 626 {
 627         struct lttng_viewer_cmd cmd;
 628         struct lttng_viewer_get_packet rq;
 629         struct lttng_viewer_trace_packet rp;
 630         ssize_t ret_len;
 631         int ret;
 632
 633 retry:
 634         if (lttng_live_should_quit()) {
 635                 ret = -1;
 636                 goto end;
 637         }
 638         cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
 639         cmd.data_size = sizeof(rq);
 640         cmd.cmd_version = 0;
 641
 642         memset(&rq, 0, sizeof(rq));
 643         rq.stream_id = htobe64(stream->id);
 644         rq.offset = htobe64(offset);
 645         rq.len = htobe32(len);
 646
 647         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 648         if (ret_len < 0) {
 649                 perror("[error] Error sending cmd");
 650                 goto error;
 651         }
 652         assert(ret_len == sizeof(cmd));
 653
 654         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 655         if (ret_len < 0) {
 656                 perror("[error] Error sending get_data_packet request");
 657                 goto error;
 658         }
 659         assert(ret_len == sizeof(rq));
 660
 661         ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 662         if (ret_len == 0) {
 663                 fprintf(stderr, "[error] Remote side has closed connection\n");
 664                 goto error;
 665         }
 666         if (ret_len < 0) {
 667                 perror("[error] Error receiving data response");
 668                 goto error;
 669         }
 670         if (ret_len != sizeof(rp)) {
 671                 fprintf(stderr, "[error] get_data_packet: expected %" PRId64
 672                                 ", received %" PRId64 "\n", sizeof(rp),
 673                                 ret_len);
 674                 goto error;
 675         }
 676
 677         rp.flags = be32toh(rp.flags);
 678
 679         switch (be32toh(rp.status)) {
 680         case LTTNG_VIEWER_GET_PACKET_OK:
 681                 len = be32toh(rp.len);
 682                 printf_verbose("get_data_packet: Ok, packet size : %" PRIu64
 683                                 "\n", len);
 684                 break;
 685         case LTTNG_VIEWER_GET_PACKET_RETRY:
 686                 /* Unimplemented by relay daemon */
 687                 printf_verbose("get_data_packet: retry\n");
 688                 goto error;
 689         case LTTNG_VIEWER_GET_PACKET_ERR:
 690                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 691                         printf_verbose("get_data_packet: new metadata needed\n");
 692                         ret = append_metadata(ctx, stream);
 693                         if (ret)
 694                                 goto error;
 695                 }
 696                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
 697                         printf_verbose("get_data_packet: new streams needed\n");
 698                         ret = ask_new_streams(ctx);
 699                         if (ret < 0)
 700                                 goto error;
 701                         else if (ret > 0)
 702                                 g_hash_table_foreach(ctx->session->ctf_traces,
 703                                                 add_traces, ctx->bt_ctx);
 704                 }
 705                 if (rp.flags & (LTTNG_VIEWER_FLAG_NEW_METADATA
 706                                 | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
 707                         goto retry;
 708                 }
 709                 fprintf(stderr, "[error] get_data_packet: error\n");
 710                 goto error;
 711         case LTTNG_VIEWER_GET_PACKET_EOF:
 712                 ret = -2;
 713                 goto end;
 714         default:
 715                 printf_verbose("get_data_packet: unknown\n");
 716                 goto error;
 717         }
 718
 719         if (len == 0) {
 720                 goto error;
 721         }
 722
 723         if (len > stream->mmap_size) {
 724                 uint64_t new_size;
 725
 726                 new_size = max_t(uint64_t, len, stream->mmap_size << 1);
 727                 if (pos->base_mma) {
 728                         /* unmap old base */
 729                         ret = munmap_align(pos->base_mma);
 730                         if (ret) {
 731                                 perror("[error] Unable to unmap old base");
 732                                 goto error;
 733                         }
 734                         pos->base_mma = NULL;
 735                 }
 736                 pos->base_mma = mmap_align(new_size,
 737                                 PROT_READ | PROT_WRITE,
 738                                 MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
 739                 if (pos->base_mma == MAP_FAILED) {
 740                         perror("[error] mmap error");
 741                         pos->base_mma = NULL;
 742                         goto error;
 743                 }
 744
 745                 stream->mmap_size = new_size;
 746                 printf_verbose("Expanding stream mmap size to %" PRIu64 " bytes\n",
 747                                 stream->mmap_size);
 748         }
 749
 750         ret_len = lttng_live_recv(ctx->control_sock,
 751                         mmap_align_addr(pos->base_mma), len);
 752         if (ret_len == 0) {
 753                 fprintf(stderr, "[error] Remote side has closed connection\n");
 754                 goto error;
 755         }
 756         if (ret_len < 0) {
 757                 perror("[error] Error receiving trace packet");
 758                 goto error;
 759         }
 760         assert(ret_len == len);
 761         ret = 0;
 762 end:
 763         return ret;
 764
 765 error:
 766         return -1;
 767 }
 768
 769 static
 770 int get_one_metadata_packet(struct lttng_live_ctx *ctx,
 771                 struct lttng_live_viewer_stream *metadata_stream)
 772 {
 773         uint64_t len = 0;
 774         int ret;
 775         struct lttng_viewer_cmd cmd;
 776         struct lttng_viewer_get_metadata rq;
 777         struct lttng_viewer_metadata_packet rp;
 778         char *data = NULL;
 779         ssize_t ret_len;
 780
 781         if (lttng_live_should_quit()) {
 782                 ret = -1;
 783                 goto end;
 784         }
 785
 786         rq.stream_id = htobe64(metadata_stream->id);
 787         cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
 788         cmd.data_size = sizeof(rq);
 789         cmd.cmd_version = 0;
 790
 791         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 792         if (ret_len < 0) {
 793                 perror("[error] Error sending cmd");
 794                 goto error;
 795         }
 796         assert(ret_len == sizeof(cmd));
 797
 798         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 799         if (ret_len < 0) {
 800                 perror("[error] Error sending get_metadata request");
 801                 goto error;
 802         }
 803         assert(ret_len == sizeof(rq));
 804
 805         ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 806         if (ret_len == 0) {
 807                 fprintf(stderr, "[error] Remote side has closed connection\n");
 808                 goto error;
 809         }
 810         if (ret_len < 0) {
 811                 perror("[error] Error receiving metadata response");
 812                 goto error;
 813         }
 814         assert(ret_len == sizeof(rp));
 815
 816         switch (be32toh(rp.status)) {
 817                 case LTTNG_VIEWER_METADATA_OK:
 818                         printf_verbose("get_metadata : OK\n");
 819                         break;
 820                 case LTTNG_VIEWER_NO_NEW_METADATA:
 821                         printf_verbose("get_metadata : NO NEW\n");
 822                         ret = 0;
 823                         goto end;
 824                 case LTTNG_VIEWER_METADATA_ERR:
 825                         printf_verbose("get_metadata : ERR\n");
 826                         goto error;
 827                 default:
 828                         printf_verbose("get_metadata : UNKNOWN\n");
 829                         goto error;
 830         }
 831
 832         len = be64toh(rp.len);
 833         printf_verbose("Writing %" PRIu64" bytes to metadata\n", len);
 834         if (len <= 0) {
 835                 goto error;
 836         }
 837
 838         data = zmalloc(len);
 839         if (!data) {
 840                 perror("relay data zmalloc");
 841                 goto error;
 842         }
 843         ret_len = lttng_live_recv(ctx->control_sock, data, len);
 844         if (ret_len == 0) {
 845                 fprintf(stderr, "[error] Remote side has closed connection\n");
 846                 goto error_free_data;
 847         }
 848         if (ret_len < 0) {
 849                 perror("[error] Error receiving trace packet");
 850                 goto error_free_data;
 851         }
 852         assert(ret_len == len);
 853
 854         do {
 855                 ret_len = fwrite(data, 1, len,
 856                                 metadata_stream->metadata_fp_write);
 857         } while (ret_len < 0 && errno == EINTR);
 858         if (ret_len < 0) {
 859                 fprintf(stderr, "[error] Writing in the metadata fp\n");
 860                 goto error_free_data;
 861         }
 862         assert(ret_len == len);
 863         metadata_stream->metadata_len += len;
 864         free(data);
 865         ret = len;
 866 end:
 867         return ret;
 868
 869 error_free_data:
 870         free(data);
 871 error:
 872         return -1;
 873 }
 874
 875 /*
 876  * Return 0 on success, a negative value on error.
 877  */
 878 static
 879 int get_new_metadata(struct lttng_live_ctx *ctx,
 880                 struct lttng_live_viewer_stream *viewer_stream,
 881                 char **metadata_buf)
 882 {
 883         int ret = 0;
 884         struct lttng_live_viewer_stream *metadata_stream;
 885         size_t size, len_read = 0;
 886
 887         metadata_stream = viewer_stream->ctf_trace->metadata_stream;
 888         if (!metadata_stream) {
 889                 fprintf(stderr, "[error] No metadata stream\n");
 890                 ret = -1;
 891                 goto error;
 892         }
 893         metadata_stream->metadata_len = 0;
 894         ret = open_metadata_fp_write(metadata_stream, metadata_buf, &size);
 895         if (ret < 0) {
 896                 goto error;
 897         }
 898
 899         do {
 900                 /*
 901                  * get_one_metadata_packet returns the number of bytes
 902                  * received, 0 when we have received everything, a
 903                  * negative value on error.
 904                  */
 905                 ret = get_one_metadata_packet(ctx, metadata_stream);
 906                 if (ret > 0) {
 907                         len_read += ret;
 908                 }
 909                 if (!len_read) {
 910                         (void) poll(NULL, 0, ACTIVE_POLL_DELAY);
 911                 }
 912         } while (ret > 0 || !len_read);
 913
 914         if (fclose(metadata_stream->metadata_fp_write))
 915                 perror("fclose");
 916         metadata_stream->metadata_fp_write = NULL;
 917
 918 error:
 919         return ret;
 920 }
 921
 922 /*
 923  * Get one index for a stream.
 924  *
 925  * Returns 0 on success or a negative value on error.
 926  */
 927 static
 928 int get_next_index(struct lttng_live_ctx *ctx,
 929                 struct lttng_live_viewer_stream *viewer_stream,
 930                 struct packet_index *index)
 931 {
 932         struct lttng_viewer_cmd cmd;
 933         struct lttng_viewer_get_next_index rq;
 934         struct lttng_viewer_index rp;
 935         int ret;
 936         ssize_t ret_len;
 937
 938         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
 939         cmd.data_size = sizeof(rq);
 940         cmd.cmd_version = 0;
 941
 942         memset(&rq, 0, sizeof(rq));
 943         rq.stream_id = htobe64(viewer_stream->id);
 944
 945 retry:
 946         if (lttng_live_should_quit()) {
 947                 ret = -1;
 948                 goto end;
 949         }
 950         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
 951         if (ret_len < 0) {
 952                 perror("[error] Error sending cmd");
 953                 goto error;
 954         }
 955         assert(ret_len == sizeof(cmd));
 956
 957         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
 958         if (ret_len < 0) {
 959                 perror("[error] Error sending get_next_index request");
 960                 goto error;
 961         }
 962         assert(ret_len == sizeof(rq));
 963
 964         ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
 965         if (ret_len == 0) {
 966                 fprintf(stderr, "[error] Remote side has closed connection\n");
 967                 goto error;
 968         }
 969         if (ret_len < 0) {
 970                 perror("[error] Error receiving index response");
 971                 goto error;
 972         }
 973         assert(ret_len == sizeof(rp));
 974
 975         rp.flags = be32toh(rp.flags);
 976
 977         switch (be32toh(rp.status)) {
 978         case LTTNG_VIEWER_INDEX_INACTIVE:
 979                 printf_verbose("get_next_index: inactive\n");
 980                 memset(index, 0, sizeof(struct packet_index));
 981                 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
 982                 break;
 983         case LTTNG_VIEWER_INDEX_OK:
 984                 printf_verbose("get_next_index: Ok, need metadata update : %u\n",
 985                                 rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA);
 986                 index->offset = be64toh(rp.offset);
 987                 index->packet_size = be64toh(rp.packet_size);
 988                 index->content_size = be64toh(rp.content_size);
 989                 index->ts_cycles.timestamp_begin = be64toh(rp.timestamp_begin);
 990                 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
 991                 index->events_discarded = be64toh(rp.events_discarded);
 992
 993                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 994                         ret = append_metadata(ctx, viewer_stream);
 995                         if (ret)
 996                                 goto error;
 997                 }
 998                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
 999                         printf_verbose("get_next_index: need new streams\n");
1000                         ret = ask_new_streams(ctx);
1001                         if (ret < 0)
1002                                 goto error;
1003                         else if (ret > 0)
1004                                 g_hash_table_foreach(ctx->session->ctf_traces,
1005                                                 add_traces, ctx->bt_ctx);
1006                 }
1007                 break;
1008         case LTTNG_VIEWER_INDEX_RETRY:
1009                 printf_verbose("get_next_index: retry\n");
1010                 (void) poll(NULL, 0, ACTIVE_POLL_DELAY);
1011                 goto retry;
1012         case LTTNG_VIEWER_INDEX_HUP:
1013                 printf_verbose("get_next_index: stream hung up\n");
1014                 viewer_stream->id = -1ULL;
1015                 index->offset = EOF;
1016                 ctx->session->stream_count--;
1017                 break;
1018         case LTTNG_VIEWER_INDEX_ERR:
1019                 fprintf(stderr, "[error] get_next_index: error\n");
1020                 goto error;
1021         default:
1022                 fprintf(stderr, "[error] get_next_index: unkwown value\n");
1023                 goto error;
1024         }
1025         ret = 0;
1026 end:
1027         return ret;
1028
1029 error:
1030         return -1;
1031 }
1032
1033 static
1034 void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
1035                 int whence)
1036 {
1037         struct ctf_stream_pos *pos;
1038         struct ctf_file_stream *file_stream;
1039         struct packet_index *prev_index = NULL, *cur_index;
1040         struct lttng_live_viewer_stream *viewer_stream;
1041         struct lttng_live_session *session;
1042         int ret;
1043
1044 retry:
1045         pos = ctf_pos(stream_pos);
1046         file_stream = container_of(pos, struct ctf_file_stream, pos);
1047         viewer_stream = (struct lttng_live_viewer_stream *) pos->priv;
1048         session = viewer_stream->session;
1049
1050         switch (pos->packet_index->len) {
1051         case 0:
1052                 g_array_set_size(pos->packet_index, 1);
1053                 cur_index = &g_array_index(pos->packet_index,
1054                                 struct packet_index, 0);
1055                 break;
1056         case 1:
1057                 g_array_set_size(pos->packet_index, 2);
1058                 prev_index = &g_array_index(pos->packet_index,
1059                                 struct packet_index, 0);
1060                 cur_index = &g_array_index(pos->packet_index,
1061                                 struct packet_index, 1);
1062                 break;
1063         case 2:
1064                 g_array_index(pos->packet_index,
1065                         struct packet_index, 0) =
1066                                 g_array_index(pos->packet_index,
1067                                         struct packet_index, 1);
1068                 prev_index = &g_array_index(pos->packet_index,
1069                                 struct packet_index, 0);
1070                 cur_index = &g_array_index(pos->packet_index,
1071                                 struct packet_index, 1);
1072                 break;
1073         default:
1074                 abort();
1075                 break;
1076         }
1077         printf_verbose("get_next_index for stream %" PRIu64 "\n", viewer_stream->id);
1078         ret = get_next_index(session->ctx, viewer_stream, cur_index);
1079         if (ret < 0) {
1080                 pos->offset = EOF;
1081                 if (!lttng_live_should_quit()) {
1082                         fprintf(stderr, "[error] get_next_index failed\n");
1083                 }
1084                 return;
1085         }
1086
1087         pos->packet_size = cur_index->packet_size;
1088         pos->content_size = cur_index->content_size;
1089         pos->mmap_base_offset = 0;
1090         if (cur_index->offset == EOF) {
1091                 pos->offset = EOF;
1092         } else {
1093                 pos->offset = 0;
1094         }
1095
1096         if (cur_index->content_size == 0) {
1097                 file_stream->parent.cycles_timestamp =
1098                                 cur_index->ts_cycles.timestamp_end;
1099                 file_stream->parent.real_timestamp = ctf_get_real_timestamp(
1100                                 &file_stream->parent,
1101                                 cur_index->ts_cycles.timestamp_end);
1102         } else {
1103                 /* Convert the timestamps and append to the real_index. */
1104                 cur_index->ts_real.timestamp_begin = ctf_get_real_timestamp(
1105                                 &file_stream->parent,
1106                                 cur_index->ts_cycles.timestamp_begin);
1107                 cur_index->ts_real.timestamp_end = ctf_get_real_timestamp(
1108                                 &file_stream->parent,
1109                                 cur_index->ts_cycles.timestamp_end);
1110
1111                 ctf_update_current_packet_index(&file_stream->parent,
1112                                 prev_index, cur_index);
1113
1114                 file_stream->parent.cycles_timestamp =
1115                                 cur_index->ts_cycles.timestamp_begin;
1116                 file_stream->parent.real_timestamp =
1117                                 cur_index->ts_real.timestamp_begin;
1118         }
1119
1120         if (pos->packet_size == 0 || pos->offset == EOF) {
1121                 goto end;
1122         }
1123
1124         printf_verbose("get_data_packet for stream %" PRIu64 "\n",
1125                         viewer_stream->id);
1126         ret = get_data_packet(session->ctx, pos, viewer_stream,
1127                         cur_index->offset,
1128                         cur_index->packet_size / CHAR_BIT);
1129         if (ret == -2) {
1130                 goto retry;
1131         } else if (ret < 0) {
1132                 pos->offset = EOF;
1133                 if (!lttng_live_should_quit()) {
1134                         fprintf(stderr, "[error] get_data_packet failed\n");
1135                 }
1136                 return;
1137         }
1138
1139         printf_verbose("Index received : packet_size : %" PRIu64
1140                         ", offset %" PRIu64 ", content_size %" PRIu64
1141                         ", timestamp_end : %" PRIu64 "\n",
1142                         cur_index->packet_size, cur_index->offset,
1143                         cur_index->content_size,
1144                         cur_index->ts_cycles.timestamp_end);
1145
1146         /* update trace_packet_header and stream_packet_context */
1147         if (pos->prot != PROT_WRITE && file_stream->parent.trace_packet_header) {
1148                 /* Read packet header */
1149                 ret = generic_rw(&pos->parent, &file_stream->parent.trace_packet_header->p);
1150                 if (ret) {
1151                         pos->offset = EOF;
1152                         fprintf(stderr, "[error] trace packet header read failed\n");
1153                         goto end;
1154                 }
1155         }
1156         if (pos->prot != PROT_WRITE && file_stream->parent.stream_packet_context) {
1157                 /* Read packet context */
1158                 ret = generic_rw(&pos->parent, &file_stream->parent.stream_packet_context->p);
1159                 if (ret) {
1160                         pos->offset = EOF;
1161                         fprintf(stderr, "[error] stream packet context read failed\n");
1162                         goto end;
1163                 }
1164         }
1165         pos->data_offset = pos->offset;
1166
1167 end:
1168         return;
1169 }
1170
1171 int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx)
1172 {
1173         struct lttng_viewer_cmd cmd;
1174         struct lttng_viewer_create_session_response resp;
1175         int ret;
1176         ssize_t ret_len;
1177
1178         if (lttng_live_should_quit()) {
1179                 ret = -1;
1180                 goto end;
1181         }
1182
1183         cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
1184         cmd.data_size = 0;
1185         cmd.cmd_version = 0;
1186
1187         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
1188         if (ret_len < 0) {
1189                 perror("[error] Error sending cmd");
1190                 goto error;
1191         }
1192         assert(ret_len == sizeof(cmd));
1193
1194         ret_len = lttng_live_recv(ctx->control_sock, &resp, sizeof(resp));
1195         if (ret_len == 0) {
1196                 fprintf(stderr, "[error] Remote side has closed connection\n");
1197                 goto error;
1198         }
1199         if (ret_len < 0) {
1200                 perror("[error] Error receiving create session reply");
1201                 goto error;
1202         }
1203         assert(ret_len == sizeof(resp));
1204
1205         if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
1206                 fprintf(stderr, "[error] Error creating viewer session\n");
1207                 goto error;
1208         }
1209         ret = 0;
1210 end:
1211         return ret;
1212
1213 error:
1214         return -1;
1215 }
1216
1217 static
1218 int del_traces(gpointer key, gpointer value, gpointer user_data)
1219 {
1220         struct bt_context *bt_ctx = user_data;
1221         struct lttng_live_ctf_trace *trace = value;
1222         int ret;
1223
1224         ret = bt_context_remove_trace(bt_ctx, trace->trace_id);
1225         if (ret < 0)
1226                 fprintf(stderr, "[error] removing trace from context\n");
1227
1228         /* remove the key/value pair from the HT. */
1229         return 1;
1230 }
1231
1232 static
1233 void add_traces(gpointer key, gpointer value, gpointer user_data)
1234 {
1235         int i, ret;
1236         struct bt_context *bt_ctx = user_data;
1237         struct lttng_live_ctf_trace *trace = value;
1238         struct lttng_live_viewer_stream *stream;
1239         struct bt_mmap_stream *new_mmap_stream;
1240         struct bt_mmap_stream_list mmap_list;
1241         struct lttng_live_ctx *ctx = NULL;
1242         struct bt_trace_descriptor *td;
1243         struct bt_trace_handle *handle;
1244
1245         /*
1246          * We don't know how many streams we will receive for a trace, so
1247          * once we are done receiving the traces, we add all the traces
1248          * received to the bt_context.
1249          * We can receive streams during the attach command or the
1250          * get_new_streams, so we have to make sure not to add multiple
1251          * times the same traces.
1252          * If a trace is already in the context, we just skip this function.
1253          */
1254         if (trace->in_use)
1255                 return;
1256
1257         BT_INIT_LIST_HEAD(&mmap_list.head);
1258
1259         for (i = 0; i < trace->streams->len; i++) {
1260                 stream = g_ptr_array_index(trace->streams, i);
1261                 ctx = stream->session->ctx;
1262
1263                 if (!stream->metadata_flag) {
1264                         new_mmap_stream = zmalloc(sizeof(struct bt_mmap_stream));
1265                         new_mmap_stream->priv = (void *) stream;
1266                         new_mmap_stream->fd = -1;
1267                         bt_list_add(&new_mmap_stream->list, &mmap_list.head);
1268                 } else {
1269                         char *metadata_buf = NULL;
1270
1271                         /* Get all possible metadata before starting */
1272                         ret = get_new_metadata(ctx, stream, &metadata_buf);
1273                         if (ret) {
1274                                 free(metadata_buf);
1275                                 goto end_free;
1276                         }
1277                         if (!stream->metadata_len) {
1278                                 fprintf(stderr, "[error] empty metadata\n");
1279                                 ret = -1;
1280                                 free(metadata_buf);
1281                                 goto end_free;
1282                         }
1283
1284                         trace->metadata_fp = babeltrace_fmemopen(metadata_buf,
1285                                         stream->metadata_len, "rb");
1286                         if (!trace->metadata_fp) {
1287                                 perror("Metadata fmemopen");
1288                                 ret = -1;
1289                                 free(metadata_buf);
1290                                 goto end_free;
1291                         }
1292                 }
1293         }
1294
1295         if (!trace->metadata_fp) {
1296                 fprintf(stderr, "[error] No metadata stream opened\n");
1297                 goto end_free;
1298         }
1299
1300         ret = bt_context_add_trace(bt_ctx, NULL, "ctf",
1301                         ctf_live_packet_seek, &mmap_list, trace->metadata_fp);
1302         if (ret < 0) {
1303                 fprintf(stderr, "[error] Error adding trace\n");
1304                 goto end_free;
1305         }
1306         trace->metadata_stream->metadata_len = 0;
1307
1308         handle = (struct bt_trace_handle *) g_hash_table_lookup(
1309                         bt_ctx->trace_handles,
1310                         (gpointer) (unsigned long) ret);
1311         td = handle->td;
1312         trace->handle = handle;
1313         if (bt_ctx->current_iterator) {
1314                 bt_iter_add_trace(bt_ctx->current_iterator, td);
1315         }
1316
1317         trace->trace_id = ret;
1318         trace->in_use = 1;
1319
1320         goto end;
1321
1322 end_free:
1323         bt_context_put(bt_ctx);
1324 end:
1325         return;
1326 }
1327
1328 /*
1329  * Request new streams for a session.
1330  * Returns the number of streams received or a negative value on error.
1331  */
1332 int lttng_live_get_new_streams(struct lttng_live_ctx *ctx, uint64_t id)
1333 {
1334         struct lttng_viewer_cmd cmd;
1335         struct lttng_viewer_new_streams_request rq;
1336         struct lttng_viewer_new_streams_response rp;
1337         struct lttng_viewer_stream stream;
1338         int ret, i, nb_streams = 0;
1339         ssize_t ret_len;
1340         uint32_t stream_count;
1341
1342         if (lttng_live_should_quit()) {
1343                 ret = -1;
1344                 goto end;
1345         }
1346
1347         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1348         cmd.data_size = sizeof(rq);
1349         cmd.cmd_version = 0;
1350
1351         memset(&rq, 0, sizeof(rq));
1352         rq.session_id = htobe64(id);
1353
1354         ret_len = lttng_live_send(ctx->control_sock, &cmd, sizeof(cmd));
1355         if (ret_len < 0) {
1356                 perror("[error] Error sending cmd");
1357                 goto error;
1358         }
1359         assert(ret_len == sizeof(cmd));
1360
1361         ret_len = lttng_live_send(ctx->control_sock, &rq, sizeof(rq));
1362         if (ret_len < 0) {
1363                 perror("[error] Error sending get_new_streams request");
1364                 goto error;
1365         }
1366         assert(ret_len == sizeof(rq));
1367
1368         ret_len = lttng_live_recv(ctx->control_sock, &rp, sizeof(rp));
1369         if (ret_len == 0) {
1370                 fprintf(stderr, "[error] Remote side has closed connection\n");
1371                 goto error;
1372         }
1373         if (ret_len < 0) {
1374                 perror("[error] Error receiving get_new_streams response");
1375                 goto error;
1376         }
1377         assert(ret_len == sizeof(rp));
1378
1379         switch(be32toh(rp.status)) {
1380         case LTTNG_VIEWER_NEW_STREAMS_OK:
1381                 break;
1382         case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1383                 ret = 0;
1384                 goto end;
1385         case LTTNG_VIEWER_NEW_STREAMS_HUP:
1386                 ret = -LTTNG_VIEWER_NEW_STREAMS_HUP;
1387                 goto end;
1388         case LTTNG_VIEWER_NEW_STREAMS_ERR:
1389                 fprintf(stderr, "[error] get_new_streams error\n");
1390                 goto error;
1391         default:
1392                 fprintf(stderr, "[error] Unknown return code %u\n",
1393                                 be32toh(rp.status));
1394                 goto error;
1395         }
1396
1397         stream_count = be32toh(rp.streams_count);
1398         ctx->session->stream_count += stream_count;
1399         /*
1400          * When the session is created but not started, we do an active wait
1401          * until it starts. It allows the viewer to start processing the trace
1402          * as soon as the session starts.
1403          */
1404         if (ctx->session->stream_count == 0) {
1405                 ret = 0;
1406                 goto end;
1407         }
1408         printf_verbose("Waiting for %" PRIu64 " streams:\n",
1409                 ctx->session->stream_count);
1410         ctx->session->streams = g_new0(struct lttng_live_viewer_stream,
1411                         ctx->session->stream_count);
1412         for (i = 0; i < stream_count; i++) {
1413                 ret_len = lttng_live_recv(ctx->control_sock, &stream, sizeof(stream));
1414                 if (ret_len == 0) {
1415                         fprintf(stderr, "[error] Remote side has closed connection\n");
1416                         goto error;
1417                 }
1418                 if (ret_len < 0) {
1419                         perror("[error] Error receiving stream");
1420                         goto error;
1421                 }
1422                 assert(ret_len == sizeof(stream));
1423                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
1424                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
1425
1426                 printf_verbose("    stream %" PRIu64 " : %s/%s\n",
1427                                 be64toh(stream.id), stream.path_name,
1428                                 stream.channel_name);
1429                 ctx->session->streams[i].id = be64toh(stream.id);
1430                 ctx->session->streams[i].session = ctx->session;
1431
1432                 ctx->session->streams[i].first_read = 1;
1433                 ctx->session->streams[i].mmap_size = 0;
1434
1435                 if (be32toh(stream.metadata_flag)) {
1436                         ctx->session->streams[i].metadata_flag = 1;
1437                 }
1438                 ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i],
1439                                 be64toh(stream.ctf_trace_id));
1440                 if (ret < 0) {
1441                         goto error;
1442                 }
1443                 nb_streams++;
1444
1445         }
1446         ret = nb_streams;
1447 end:
1448         return ret;
1449
1450 error:
1451         return -1;
1452 }
1453
1454 int lttng_live_read(struct lttng_live_ctx *ctx)
1455 {
1456         int ret = -1;
1457         int i;
1458         struct bt_ctf_iter *iter;
1459         const struct bt_ctf_event *event;
1460         struct bt_iter_pos begin_pos;
1461         struct bt_trace_descriptor *td_write;
1462         struct bt_format *fmt_write;
1463         struct ctf_text_stream_pos *sout;
1464         uint64_t id;
1465
1466         ctx->bt_ctx = bt_context_create();
1467         if (!ctx->bt_ctx) {
1468                 fprintf(stderr, "[error] bt_context_create allocation\n");
1469                 goto end;
1470         }
1471
1472         fmt_write = bt_lookup_format(g_quark_from_static_string("text"));
1473         if (!fmt_write) {
1474                 fprintf(stderr, "[error] ctf-text error\n");
1475                 goto end;
1476         }
1477
1478         td_write = fmt_write->open_trace(NULL, O_RDWR, NULL, NULL);
1479         if (!td_write) {
1480                 fprintf(stderr, "[error] Error opening output trace\n");
1481                 goto end_free;
1482         }
1483
1484         sout = container_of(td_write, struct ctf_text_stream_pos,
1485                         trace_descriptor);
1486         if (!sout->parent.event_cb) {
1487                 goto end_free;
1488         }
1489
1490         ret = lttng_live_create_viewer_session(ctx);
1491         if (ret < 0) {
1492                 goto end_free;
1493         }
1494
1495         for (i = 0; i < ctx->session_ids->len; i++) {
1496                 id = g_array_index(ctx->session_ids, uint64_t, i);
1497                 printf_verbose("Attaching to session %lu\n", id);
1498                 ret = lttng_live_attach_session(ctx, id);
1499                 printf_verbose("Attaching session returns %d\n", ret);
1500                 if (ret < 0) {
1501                         if (ret == -LTTNG_VIEWER_ATTACH_UNK) {
1502                                 fprintf(stderr, "[error] Unknown session ID\n");
1503                         }
1504                         goto end_free;
1505                 }
1506         }
1507
1508         /*
1509          * As long as the session is active, we try to get new streams.
1510          */
1511         for (;;) {
1512                 int flags;
1513
1514                 if (lttng_live_should_quit()) {
1515                         ret = 0;
1516                         goto end_free;
1517                 }
1518
1519                 while (!ctx->session->stream_count) {
1520                         if (lttng_live_should_quit()
1521                                         || ctx->session_ids->len == 0) {
1522                                 ret = 0;
1523                                 goto end_free;
1524                         }
1525                         ret = ask_new_streams(ctx);
1526                         if (ret < 0) {
1527                                 goto end_free;
1528                         }
1529                         if (!ctx->session->stream_count) {
1530                                 (void) poll(NULL, 0, ACTIVE_POLL_DELAY);
1531                         }
1532                 }
1533
1534                 g_hash_table_foreach(ctx->session->ctf_traces, add_traces,
1535                                 ctx->bt_ctx);
1536
1537                 begin_pos.type = BT_SEEK_BEGIN;
1538                 iter = bt_ctf_iter_create(ctx->bt_ctx, &begin_pos, NULL);
1539                 if (!iter) {
1540                         if (lttng_live_should_quit()) {
1541                                 ret = 0;
1542                                 goto end;
1543                         }
1544                         fprintf(stderr, "[error] Iterator creation error\n");
1545                         goto end;
1546                 }
1547                 for (;;) {
1548                         if (lttng_live_should_quit()) {
1549                                 ret = 0;
1550                                 goto end_free;
1551                         }
1552                         event = bt_ctf_iter_read_event_flags(iter, &flags);
1553                         if (!(flags & BT_ITER_FLAG_RETRY)) {
1554                                 if (!event) {
1555                                         /* End of trace */
1556                                         break;
1557                                 }
1558                                 ret = sout->parent.event_cb(&sout->parent,
1559                                                 event->parent->stream);
1560                                 if (ret) {
1561                                         fprintf(stderr, "[error] Writing "
1562                                                         "event failed.\n");
1563                                         goto end_free;
1564                                 }
1565                         }
1566                         ret = bt_iter_next(bt_ctf_get_iter(iter));
1567                         if (ret < 0) {
1568                                 goto end_free;
1569                         }
1570                 }
1571                 bt_ctf_iter_destroy(iter);
1572                 g_hash_table_foreach_remove(ctx->session->ctf_traces,
1573                                 del_traces, ctx->bt_ctx);
1574                 ctx->session->stream_count = 0;
1575         }
1576
1577 end_free:
1578         bt_context_put(ctx->bt_ctx);
1579 end:
1580         if (lttng_live_should_quit()) {
1581                 ret = 0;
1582         }
1583         return ret;
1584 }