formats/lttng-live/lttng-live-functions.c

   1 /*
   2  * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
   3  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a copy
   6  * of this software and associated documentation files (the "Software"), to deal
   7  * in the Software without restriction, including without limitation the rights
   8  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9  * copies of the Software, and to permit persons to whom the Software is
  10  * furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  21  * SOFTWARE.
  22  */
  23
  24 #include <sys/socket.h>
  25 #include <sys/types.h>
  26 #include <netinet/in.h>
  27 #include <netdb.h>
  28 #include <stdio.h>
  29 #include <string.h>
  30 #include <stdlib.h>
  31 #include <unistd.h>
  32 #include <errno.h>
  33 #include <inttypes.h>
  34 #include <fcntl.h>
  35 #include <sys/mman.h>
  36
  37 #include <babeltrace/ctf/ctf-index.h>
  38
  39 #include <babeltrace/babeltrace.h>
  40 #include <babeltrace/ctf/events.h>
  41 #include <babeltrace/ctf/callbacks.h>
  42 #include <babeltrace/ctf/iterator.h>
  43
  44 /* for packet_index */
  45 #include <babeltrace/ctf/types.h>
  46
  47 #include <babeltrace/ctf/metadata.h>
  48 #include <babeltrace/ctf-text/types.h>
  49 #include <babeltrace/ctf/events-internal.h>
  50 #include <formats/ctf/events-private.h>
  51
  52 #include "lttng-live-functions.h"
  53 #include "lttng-viewer-abi.h"
  54
  55 /*
  56  * Memory allocation zeroed
  57  */
  58 #define zmalloc(x) calloc(1, x)
  59
  60 #ifndef max_t
  61 #define max_t(type, a, b)       \
  62         ((type) (a) > (type) (b) ? (type) (a) : (type) (b))
  63 #endif
  64
  65 int lttng_live_connect_viewer(struct lttng_live_ctx *ctx, char *hostname,
  66                 int port)
  67 {
  68         struct hostent *host;
  69         struct sockaddr_in server_addr;
  70         int ret;
  71
  72         host = gethostbyname(hostname);
  73         if (!host) {
  74                 ret = -1;
  75                 goto end;
  76         }
  77
  78         if ((ctx->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
  79                 perror("Socket");
  80                 ret = -1;
  81                 goto end;
  82         }
  83
  84         server_addr.sin_family = AF_INET;
  85         server_addr.sin_port = htons(port);
  86         server_addr.sin_addr = *((struct in_addr *) host->h_addr);
  87         bzero(&(server_addr.sin_zero), 8);
  88
  89         if (connect(ctx->control_sock, (struct sockaddr *) &server_addr,
  90                                 sizeof(struct sockaddr)) == -1) {
  91                 perror("Connect");
  92                 ret = -1;
  93                 goto end;
  94         }
  95
  96         ret = 0;
  97
  98 end:
  99         return ret;
 100 }
 101
 102 int lttng_live_establish_connection(struct lttng_live_ctx *ctx)
 103 {
 104         struct lttng_viewer_cmd cmd;
 105         struct lttng_viewer_connect connect;
 106         int ret;
 107         ssize_t ret_len;
 108
 109         cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
 110         cmd.data_size = sizeof(connect);
 111         cmd.cmd_version = 0;
 112
 113         connect.viewer_session_id = -1ULL;      /* will be set on recv */
 114         connect.major = htobe32(LTTNG_LIVE_MAJOR);
 115         connect.minor = htobe32(LTTNG_LIVE_MINOR);
 116         connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
 117
 118         do {
 119                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 120         } while (ret_len < 0 && errno == EINTR);
 121         if (ret_len < 0) {
 122                 fprintf(stderr, "[error] Error sending cmd\n");
 123                 ret = ret_len;
 124                 goto error;
 125         }
 126         assert(ret_len == sizeof(cmd));
 127
 128         do {
 129                 ret_len = send(ctx->control_sock, &connect, sizeof(connect), 0);
 130         } while (ret_len < 0 && errno == EINTR);
 131         if (ret_len < 0) {
 132                 fprintf(stderr, "[error] Error sending version\n");
 133                 ret = ret_len;
 134                 goto error;
 135         }
 136         assert(ret_len == sizeof(connect));
 137
 138         do {
 139                 ret_len = recv(ctx->control_sock, &connect, sizeof(connect), 0);
 140         } while (ret_len < 0 && errno == EINTR);
 141         if (ret_len < 0) {
 142                 fprintf(stderr, "[error] Error receiving version\n");
 143                 ret = ret_len;
 144                 goto error;
 145         }
 146         assert(ret_len == sizeof(connect));
 147
 148         printf_verbose("Received viewer session ID : %" PRIu64 "\n",
 149                         be64toh(connect.viewer_session_id));
 150         printf_verbose("Relayd version : %u.%u\n", be32toh(connect.major),
 151                         be32toh(connect.minor));
 152
 153         ret = 0;
 154
 155 error:
 156         return ret;
 157 }
 158
 159 int lttng_live_list_sessions(struct lttng_live_ctx *ctx, const char *path)
 160 {
 161         struct lttng_viewer_cmd cmd;
 162         struct lttng_viewer_list_sessions list;
 163         struct lttng_viewer_session lsession;
 164         int i, ret;
 165         ssize_t ret_len;
 166         int sessions_count;
 167
 168         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 169         cmd.data_size = 0;
 170         cmd.cmd_version = 0;
 171
 172         do {
 173                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 174         } while (ret_len < 0 && errno == EINTR);
 175         if (ret_len < 0) {
 176                 fprintf(stderr, "[error] Error sending cmd\n");
 177                 ret = ret_len;
 178                 goto error;
 179         }
 180         assert(ret_len == sizeof(cmd));
 181
 182         do {
 183                 ret_len = recv(ctx->control_sock, &list, sizeof(list), 0);
 184         } while (ret_len < 0 && errno == EINTR);
 185         if (ret_len < 0) {
 186                 fprintf(stderr, "[error] Error receiving session list\n");
 187                 ret = ret_len;
 188                 goto error;
 189         }
 190         assert(ret_len == sizeof(list));
 191
 192         sessions_count = be32toh(list.sessions_count);
 193         fprintf(stdout, "%u active session(s)%c\n", sessions_count,
 194                         sessions_count > 0 ? ':' : ' ');
 195         for (i = 0; i < sessions_count; i++) {
 196                 do {
 197                         ret_len = recv(ctx->control_sock, &lsession, sizeof(lsession), 0);
 198                 } while (ret_len < 0 && errno == EINTR);
 199                 if (ret_len < 0) {
 200                         fprintf(stderr, "[error] Error receiving session\n");
 201                         ret = ret_len;
 202                         goto error;
 203                 }
 204                 assert(ret_len == sizeof(lsession));
 205                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 206                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 207
 208                 fprintf(stdout, "%s/%" PRIu64 " : %s on host %s (timer = %u, "
 209                                 "%u stream(s), %u client(s) connected)\n",
 210                                 path, be64toh(lsession.id),
 211                                 lsession.session_name, lsession.hostname,
 212                                 be32toh(lsession.live_timer),
 213                                 be32toh(lsession.streams),
 214                                 be32toh(lsession.clients));
 215         }
 216
 217         ret = 0;
 218
 219 error:
 220         return ret;
 221 }
 222
 223 int lttng_live_ctf_trace_assign(struct lttng_live_viewer_stream *stream,
 224                 uint64_t ctf_trace_id)
 225 {
 226         struct lttng_live_ctf_trace *trace;
 227         int ret = 0;
 228
 229         trace = g_hash_table_lookup(stream->session->ctf_traces,
 230                         (gpointer) ctf_trace_id);
 231         if (!trace) {
 232                 trace = g_new0(struct lttng_live_ctf_trace, 1);
 233                 trace->ctf_trace_id = ctf_trace_id;
 234                 trace->streams = g_ptr_array_new();
 235                 g_hash_table_insert(stream->session->ctf_traces,
 236                                 (gpointer) ctf_trace_id,
 237                                 trace);
 238         }
 239         if (stream->metadata_flag)
 240                 trace->metadata_stream = stream;
 241
 242         stream->ctf_trace = trace;
 243         g_ptr_array_add(trace->streams, stream);
 244
 245         return ret;
 246 }
 247
 248 int lttng_live_attach_session(struct lttng_live_ctx *ctx, uint64_t id)
 249 {
 250         struct lttng_viewer_cmd cmd;
 251         struct lttng_viewer_attach_session_request rq;
 252         struct lttng_viewer_attach_session_response rp;
 253         struct lttng_viewer_stream stream;
 254         int ret, i;
 255         ssize_t ret_len;
 256
 257         cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 258         cmd.data_size = sizeof(rq);
 259         cmd.cmd_version = 0;
 260
 261         memset(&rq, 0, sizeof(rq));
 262         rq.session_id = htobe64(id);
 263         // TODO: add cmd line parameter to select seek beginning
 264         // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
 265         rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
 266
 267         do {
 268                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 269         } while (ret_len < 0 && errno == EINTR);
 270         if (ret_len < 0) {
 271                 fprintf(stderr, "[error] Error sending cmd\n");
 272                 ret = ret_len;
 273                 goto error;
 274         }
 275         assert(ret_len == sizeof(cmd));
 276
 277         do {
 278                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
 279         } while (ret_len < 0 && errno == EINTR);
 280         if (ret_len < 0) {
 281                 fprintf(stderr, "[error] Error sending attach request\n");
 282                 ret = ret_len;
 283                 goto error;
 284         }
 285         assert(ret_len == sizeof(rq));
 286
 287         do {
 288                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
 289         } while (ret_len < 0 && errno == EINTR);
 290         if (ret_len < 0) {
 291                 fprintf(stderr, "[error] Error receiving attach response\n");
 292                 ret = ret_len;
 293                 goto error;
 294         }
 295         assert(ret_len == sizeof(rp));
 296
 297         switch(be32toh(rp.status)) {
 298         case LTTNG_VIEWER_ATTACH_OK:
 299                 break;
 300         case LTTNG_VIEWER_ATTACH_UNK:
 301                 ret = -LTTNG_VIEWER_ATTACH_UNK;
 302                 goto end;
 303         case LTTNG_VIEWER_ATTACH_ALREADY:
 304                 fprintf(stderr, "[error] Already a viewer attached\n");
 305                 ret = -1;
 306                 goto end;
 307         case LTTNG_VIEWER_ATTACH_NOT_LIVE:
 308                 fprintf(stderr, "[error] Not a live session\n");
 309                 ret = -1;
 310                 goto end;
 311         case LTTNG_VIEWER_ATTACH_SEEK_ERR:
 312                 fprintf(stderr, "[error] Wrong seek parameter\n");
 313                 ret = -1;
 314                 goto end;
 315         default:
 316                 fprintf(stderr, "[error] Unknown attach return code %u\n",
 317                                 be32toh(rp.status));
 318                 ret = -1;
 319                 goto end;
 320         }
 321         if (be32toh(rp.status) != LTTNG_VIEWER_ATTACH_OK) {
 322                 ret = -1;
 323                 goto end;
 324         }
 325
 326         ctx->session->stream_count = be32toh(rp.streams_count);
 327         /*
 328          * When the session is created but not started, we do an active wait
 329          * until it starts. It allows the viewer to start processing the trace
 330          * as soon as the session starts.
 331          */
 332         if (ctx->session->stream_count == 0) {
 333                 ret = 0;
 334                 goto end;
 335         }
 336         printf_verbose("Waiting for %" PRIu64 " streams:\n",
 337                 ctx->session->stream_count);
 338         ctx->session->streams = g_new0(struct lttng_live_viewer_stream,
 339                         ctx->session->stream_count);
 340         for (i = 0; i < be32toh(rp.streams_count); i++) {
 341                 do {
 342                         ret_len = recv(ctx->control_sock, &stream, sizeof(stream), 0);
 343                 } while (ret_len < 0 && errno == EINTR);
 344                 if (ret_len < 0) {
 345                         fprintf(stderr, "[error] Error receiving stream\n");
 346                         ret = ret_len;
 347                         goto error;
 348                 }
 349                 assert(ret_len == sizeof(stream));
 350                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
 351                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 352
 353                 printf_verbose("    stream %" PRIu64 " : %s/%s\n",
 354                                 be64toh(stream.id), stream.path_name,
 355                                 stream.channel_name);
 356                 ctx->session->streams[i].id = be64toh(stream.id);
 357                 ctx->session->streams[i].session = ctx->session;
 358
 359                 ctx->session->streams[i].first_read = 1;
 360                 ctx->session->streams[i].mmap_size = 0;
 361
 362                 if (be32toh(stream.metadata_flag)) {
 363                         char *path;
 364
 365                         path = strdup(LTTNG_METADATA_PATH_TEMPLATE);
 366                         if (!path) {
 367                                 perror("strdup");
 368                                 ret = -1;
 369                                 goto error;
 370                         }
 371                         if (!mkdtemp(path)) {
 372                                 perror("mkdtemp");
 373                                 free(path);
 374                                 ret = -1;
 375                                 goto error;
 376                         }
 377                         ctx->session->streams[i].metadata_flag = 1;
 378                         snprintf(ctx->session->streams[i].path,
 379                                         sizeof(ctx->session->streams[i].path),
 380                                         "%s/%s", path,
 381                                         stream.channel_name);
 382                         ret = open(ctx->session->streams[i].path,
 383                                         O_WRONLY | O_CREAT | O_TRUNC,
 384                                         S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
 385                         if (ret < 0) {
 386                                 perror("open");
 387                                 free(path);
 388                                 goto error;
 389                         }
 390                         ctx->session->streams[i].fd = ret;
 391                         free(path);
 392                 }
 393                 ret = lttng_live_ctf_trace_assign(&ctx->session->streams[i],
 394                                 be64toh(stream.ctf_trace_id));
 395                 if (ret < 0) {
 396                         goto error;
 397                 }
 398
 399         }
 400         ret = 0;
 401
 402 end:
 403 error:
 404         return ret;
 405 }
 406
 407 static
 408 int get_data_packet(struct lttng_live_ctx *ctx,
 409                 struct ctf_stream_pos *pos,
 410                 struct lttng_live_viewer_stream *stream, uint64_t offset,
 411                 uint64_t len)
 412 {
 413         struct lttng_viewer_cmd cmd;
 414         struct lttng_viewer_get_packet rq;
 415         struct lttng_viewer_trace_packet rp;
 416         ssize_t ret_len;
 417         int ret;
 418
 419         cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
 420         cmd.data_size = sizeof(rq);
 421         cmd.cmd_version = 0;
 422
 423         memset(&rq, 0, sizeof(rq));
 424         rq.stream_id = htobe64(stream->id);
 425         /* Already in big endian. */
 426         rq.offset = offset;
 427         rq.len = htobe32(len);
 428
 429         do {
 430                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 431         } while (ret_len < 0 && errno == EINTR);
 432         if (ret_len < 0) {
 433                 fprintf(stderr, "[error] Error sending cmd\n");
 434                 ret = ret_len;
 435                 goto error;
 436         }
 437         assert(ret_len == sizeof(cmd));
 438
 439         do {
 440                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
 441         } while (ret_len < 0 && errno == EINTR);
 442         if (ret_len < 0) {
 443                 fprintf(stderr, "[error] Error sending get_data_packet request\n");
 444                 ret = ret_len;
 445                 goto error;
 446         }
 447         assert(ret_len == sizeof(rq));
 448
 449         do {
 450                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
 451         } while (ret_len < 0 && errno == EINTR);
 452         if (ret_len < 0) {
 453                 fprintf(stderr, "[error] Error receiving data response\n");
 454                 ret = ret_len;
 455                 goto error;
 456         }
 457         if (ret_len != sizeof(rp)) {
 458                 fprintf(stderr, "[error] get_data_packet: expected %" PRId64
 459                                 ", received %" PRId64 "\n", ret_len,
 460                                 sizeof(rp));
 461                 ret = -1;
 462                 goto error;
 463         }
 464
 465         rp.flags = be32toh(rp.flags);
 466
 467         switch (be32toh(rp.status)) {
 468         case LTTNG_VIEWER_GET_PACKET_OK:
 469                 len = be32toh(rp.len);
 470                 printf_verbose("get_data_packet: Ok, packet size : %" PRIu64
 471                                 "\n", len);
 472                 break;
 473         case LTTNG_VIEWER_GET_PACKET_RETRY:
 474                 printf_verbose("get_data_packet: retry\n");
 475                 ret = -1;
 476                 goto end;
 477         case LTTNG_VIEWER_GET_PACKET_ERR:
 478                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 479                         printf_verbose("get_data_packet: new metadata needed\n");
 480                         ret = 0;
 481                         goto end;
 482                 }
 483                 fprintf(stderr, "[error] get_data_packet: error\n");
 484                 ret = -1;
 485                 goto end;
 486         case LTTNG_VIEWER_GET_PACKET_EOF:
 487                 ret = -2;
 488                 goto error;
 489         default:
 490                 printf_verbose("get_data_packet: unknown\n");
 491                 assert(0);
 492                 ret = -1;
 493                 goto end;
 494         }
 495
 496         if (len <= 0) {
 497                 ret = -1;
 498                 goto end;
 499         }
 500
 501         if (len > stream->mmap_size) {
 502                 uint64_t new_size;
 503
 504                 new_size = max_t(uint64_t, len, stream->mmap_size << 1);
 505                 if (pos->base_mma) {
 506                         /* unmap old base */
 507                         ret = munmap_align(pos->base_mma);
 508                         if (ret) {
 509                                 fprintf(stderr, "[error] Unable to unmap old base: %s.\n",
 510                                         strerror(errno));
 511                                 ret = -1;
 512                                 goto error;
 513                         }
 514                         pos->base_mma = NULL;
 515                 }
 516                 pos->base_mma = mmap_align(new_size,
 517                                 PROT_READ | PROT_WRITE,
 518                                 MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
 519                 if (pos->base_mma == MAP_FAILED) {
 520                         fprintf(stderr, "[error] mmap error %s.\n",
 521                                 strerror(errno));
 522                         pos->base_mma = NULL;
 523                         ret = -1;
 524                         goto error;
 525                 }
 526
 527                 stream->mmap_size = new_size;
 528                 printf_verbose("Expanding stream mmap size to %" PRIu64 " bytes\n",
 529                                 stream->mmap_size);
 530         }
 531
 532         do {
 533                 ret_len = recv(ctx->control_sock,
 534                         mmap_align_addr(pos->base_mma), len,
 535                         MSG_WAITALL);
 536         } while (ret_len < 0 && errno == EINTR);
 537         if (ret_len < 0) {
 538                 fprintf(stderr, "[error] Error receiving trace packet\n");
 539                 ret = ret_len;
 540                 goto error;
 541         }
 542         assert(ret_len == len);
 543
 544 end:
 545 error:
 546         return ret;
 547 }
 548
 549 /*
 550  * Return number of metadata bytes written or a negative value on error.
 551  */
 552 static
 553 int get_new_metadata(struct lttng_live_ctx *ctx,
 554                 struct lttng_live_viewer_stream *viewer_stream,
 555                 uint64_t *metadata_len)
 556 {
 557         uint64_t len = 0;
 558         int ret;
 559         struct lttng_viewer_cmd cmd;
 560         struct lttng_viewer_get_metadata rq;
 561         struct lttng_viewer_metadata_packet rp;
 562         struct lttng_live_viewer_stream *metadata_stream;
 563         char *data = NULL;
 564         ssize_t ret_len;
 565
 566         cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
 567         cmd.data_size = sizeof(rq);
 568         cmd.cmd_version = 0;
 569
 570         metadata_stream = viewer_stream->ctf_trace->metadata_stream;
 571         rq.stream_id = htobe64(metadata_stream->id);
 572
 573         do {
 574                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 575         } while (ret_len < 0 && errno == EINTR);
 576         if (ret_len < 0) {
 577                 fprintf(stderr, "[error] Error sending cmd\n");
 578                 ret = ret_len;
 579                 goto error;
 580         }
 581         assert(ret_len == sizeof(cmd));
 582
 583         do {
 584                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
 585         } while (ret_len < 0 && errno == EINTR);
 586         if (ret_len < 0) {
 587                 fprintf(stderr, "[error] Error sending get_metadata request\n");
 588                 ret = ret_len;
 589                 goto error;
 590         }
 591         assert(ret_len == sizeof(rq));
 592
 593         do {
 594                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
 595         } while (ret_len < 0 && errno == EINTR);
 596         if (ret_len < 0) {
 597                 fprintf(stderr, "[error] Error receiving metadata response\n");
 598                 ret = ret_len;
 599                 goto error;
 600         }
 601         assert(ret_len == sizeof(rp));
 602
 603         switch (be32toh(rp.status)) {
 604                 case LTTNG_VIEWER_METADATA_OK:
 605                         printf_verbose("get_metadata : OK\n");
 606                         break;
 607                 case LTTNG_VIEWER_NO_NEW_METADATA:
 608                         printf_verbose("get_metadata : NO NEW\n");
 609                         ret = -1;
 610                         goto end;
 611                 case LTTNG_VIEWER_METADATA_ERR:
 612                         printf_verbose("get_metadata : ERR\n");
 613                         ret = -1;
 614                         goto end;
 615                 default:
 616                         printf_verbose("get_metadata : UNKNOWN\n");
 617                         ret = -1;
 618                         goto end;
 619         }
 620
 621         len = be64toh(rp.len);
 622         printf_verbose("Writing %" PRIu64" bytes to metadata\n", len);
 623         if (len <= 0) {
 624                 ret = -1;
 625                 goto end;
 626         }
 627
 628         data = zmalloc(len);
 629         if (!data) {
 630                 perror("relay data zmalloc");
 631                 ret = -1;
 632                 goto error;
 633         }
 634         do {
 635                 ret_len = recv(ctx->control_sock, data, len, MSG_WAITALL);
 636         } while (ret_len < 0 && errno == EINTR);
 637         if (ret_len < 0) {
 638                 fprintf(stderr, "[error] Error receiving trace packet\n");
 639                 ret = ret_len;
 640                 free(data);
 641                 goto error;
 642         }
 643         assert(ret_len == len);
 644
 645         do {
 646                 ret_len = write(metadata_stream->fd, data, len);
 647         } while (ret_len < 0 && errno == EINTR);
 648         if (ret_len < 0) {
 649                 free(data);
 650                 ret = ret_len;
 651                 goto error;
 652         }
 653         assert(ret_len == len);
 654
 655         free(data);
 656
 657         *metadata_len = len;
 658         ret = 0;
 659 end:
 660 error:
 661         return ret;
 662 }
 663
 664 /*
 665  * Get one index for a stream.
 666  *
 667  * Returns 0 on success or a negative value on error.
 668  */
 669 static
 670 int get_next_index(struct lttng_live_ctx *ctx,
 671                 struct lttng_live_viewer_stream *viewer_stream,
 672                 struct packet_index *index)
 673 {
 674         struct lttng_viewer_cmd cmd;
 675         struct lttng_viewer_get_next_index rq;
 676         struct lttng_viewer_index rp;
 677         int ret;
 678         uint64_t metadata_len;
 679         ssize_t ret_len;
 680
 681         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
 682         cmd.data_size = sizeof(rq);
 683         cmd.cmd_version = 0;
 684
 685         memset(&rq, 0, sizeof(rq));
 686         rq.stream_id = htobe64(viewer_stream->id);
 687
 688 retry:
 689         do {
 690                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 691         } while (ret_len < 0 && errno == EINTR);
 692         if (ret_len < 0) {
 693                 fprintf(stderr, "[error] Error sending cmd\n");
 694                 ret = ret_len;
 695                 goto error;
 696         }
 697         assert(ret_len == sizeof(cmd));
 698
 699         do {
 700                 ret_len = send(ctx->control_sock, &rq, sizeof(rq), 0);
 701         } while (ret_len < 0 && errno == EINTR);
 702         if (ret_len < 0) {
 703                 fprintf(stderr, "[error] Error sending get_next_index request\n");
 704                 ret = ret_len;
 705                 goto error;
 706         }
 707         assert(ret_len == sizeof(rq));
 708
 709         do {
 710                 ret_len = recv(ctx->control_sock, &rp, sizeof(rp), 0);
 711         } while (ret_len < 0 && errno == EINTR);
 712         if (ret_len < 0) {
 713                 fprintf(stderr, "[error] Error receiving index response\n");
 714                 ret = ret_len;
 715                 goto error;
 716         }
 717         assert(ret_len == sizeof(rp));
 718
 719         rp.flags = be32toh(rp.flags);
 720
 721         switch (be32toh(rp.status)) {
 722         case LTTNG_VIEWER_INDEX_INACTIVE:
 723                 printf_verbose("get_next_index: inactive\n");
 724                 memset(index, 0, sizeof(struct packet_index));
 725                 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
 726                 break;
 727         case LTTNG_VIEWER_INDEX_OK:
 728                 printf_verbose("get_next_index: Ok, need metadata update : %u\n",
 729                                 rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA);
 730                 index->offset = be64toh(rp.offset);
 731                 index->packet_size = be64toh(rp.packet_size);
 732                 index->content_size = be64toh(rp.content_size);
 733                 index->ts_cycles.timestamp_begin = be64toh(rp.timestamp_begin);
 734                 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
 735                 index->events_discarded = be64toh(rp.events_discarded);
 736
 737                 if (rp.flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
 738                         printf_verbose("get_next_index: new metadata needed\n");
 739                         ret = get_new_metadata(ctx, viewer_stream,
 740                                         &metadata_len);
 741                         if (ret < 0) {
 742                                 goto error;
 743                         }
 744                 }
 745                 break;
 746         case LTTNG_VIEWER_INDEX_RETRY:
 747                 printf_verbose("get_next_index: retry\n");
 748                 sleep(1);
 749                 goto retry;
 750         case LTTNG_VIEWER_INDEX_HUP:
 751                 printf_verbose("get_next_index: stream hung up\n");
 752                 viewer_stream->id = -1ULL;
 753                 viewer_stream->fd = -1;
 754                 index->offset = EOF;
 755                 break;
 756         case LTTNG_VIEWER_INDEX_ERR:
 757                 fprintf(stderr, "[error] get_next_index: error\n");
 758                 ret = -1;
 759                 goto error;
 760         default:
 761                 fprintf(stderr, "[error] get_next_index: unkwown value\n");
 762                 ret = -1;
 763                 goto error;
 764         }
 765
 766         ret = 0;
 767
 768 error:
 769         return ret;
 770 }
 771
 772 void ctf_live_packet_seek(struct bt_stream_pos *stream_pos, size_t index,
 773                 int whence)
 774 {
 775         struct ctf_stream_pos *pos;
 776         struct ctf_file_stream *file_stream;
 777         struct packet_index *prev_index = NULL, *cur_index;
 778         struct lttng_live_viewer_stream *viewer_stream;
 779         struct lttng_live_session *session;
 780         int ret;
 781
 782 retry:
 783         pos = ctf_pos(stream_pos);
 784         file_stream = container_of(pos, struct ctf_file_stream, pos);
 785         viewer_stream = (struct lttng_live_viewer_stream *) pos->priv;
 786         session = viewer_stream->session;
 787
 788         switch (pos->packet_index->len) {
 789         case 0:
 790                 g_array_set_size(pos->packet_index, 1);
 791                 cur_index = &g_array_index(pos->packet_index,
 792                                 struct packet_index, 0);
 793                 break;
 794         case 1:
 795                 g_array_set_size(pos->packet_index, 2);
 796                 prev_index = &g_array_index(pos->packet_index,
 797                                 struct packet_index, 0);
 798                 cur_index = &g_array_index(pos->packet_index,
 799                                 struct packet_index, 1);
 800                 break;
 801         case 2:
 802                 g_array_index(pos->packet_index,
 803                         struct packet_index, 0) =
 804                                 g_array_index(pos->packet_index,
 805                                         struct packet_index, 1);
 806                 prev_index = &g_array_index(pos->packet_index,
 807                                 struct packet_index, 0);
 808                 cur_index = &g_array_index(pos->packet_index,
 809                                 struct packet_index, 1);
 810                 break;
 811         default:
 812                 abort();
 813                 break;
 814         }
 815         printf_verbose("get_next_index for stream %" PRIu64 "\n", viewer_stream->id);
 816         ret = get_next_index(session->ctx, viewer_stream, cur_index);
 817         if (ret < 0) {
 818                 pos->offset = EOF;
 819                 fprintf(stderr, "[error] get_next_index failed\n");
 820                 return;
 821         }
 822
 823         pos->packet_size = cur_index->packet_size;
 824         pos->content_size = cur_index->content_size;
 825         pos->mmap_base_offset = 0;
 826         if (cur_index->offset == EOF) {
 827                 pos->offset = EOF;
 828         } else {
 829                 pos->offset = 0;
 830         }
 831
 832         if (cur_index->content_size == 0) {
 833                 file_stream->parent.cycles_timestamp =
 834                                 cur_index->ts_cycles.timestamp_end;
 835                 file_stream->parent.real_timestamp = ctf_get_real_timestamp(
 836                                 &file_stream->parent,
 837                                 cur_index->ts_cycles.timestamp_end);
 838         } else {
 839                 /* Convert the timestamps and append to the real_index. */
 840                 cur_index->ts_real.timestamp_begin = ctf_get_real_timestamp(
 841                                 &file_stream->parent,
 842                                 cur_index->ts_cycles.timestamp_begin);
 843                 cur_index->ts_real.timestamp_end = ctf_get_real_timestamp(
 844                                 &file_stream->parent,
 845                                 cur_index->ts_cycles.timestamp_end);
 846
 847                 ctf_update_current_packet_index(&file_stream->parent,
 848                                 prev_index, cur_index);
 849
 850                 file_stream->parent.cycles_timestamp =
 851                                 cur_index->ts_cycles.timestamp_begin;
 852                 file_stream->parent.real_timestamp =
 853                                 cur_index->ts_real.timestamp_begin;
 854         }
 855
 856         if (pos->packet_size == 0 || pos->offset == EOF) {
 857                 goto end;
 858         }
 859
 860         printf_verbose("get_data_packet for stream %" PRIu64 "\n",
 861                         viewer_stream->id);
 862         ret = get_data_packet(session->ctx, pos, viewer_stream,
 863                         be64toh(cur_index->offset),
 864                         cur_index->packet_size / CHAR_BIT);
 865         if (ret == -2) {
 866                 goto retry;
 867         } else if (ret < 0) {
 868                 pos->offset = EOF;
 869                 fprintf(stderr, "[error] get_data_packet failed\n");
 870                 return;
 871         }
 872
 873         printf_verbose("Index received : packet_size : %" PRIu64
 874                         ", offset %" PRIu64 ", content_size %" PRIu64
 875                         ", timestamp_end : %" PRIu64 "\n",
 876                         cur_index->packet_size, cur_index->offset,
 877                         cur_index->content_size,
 878                         cur_index->ts_cycles.timestamp_end);
 879
 880         /* update trace_packet_header and stream_packet_context */
 881         if (pos->prot != PROT_WRITE && file_stream->parent.trace_packet_header) {
 882                 /* Read packet header */
 883                 ret = generic_rw(&pos->parent, &file_stream->parent.trace_packet_header->p);
 884                 if (ret) {
 885                         pos->offset = EOF;
 886                         fprintf(stderr, "[error] trace packet header read failed\n");
 887                         goto end;
 888                 }
 889         }
 890         if (pos->prot != PROT_WRITE && file_stream->parent.stream_packet_context) {
 891                 /* Read packet context */
 892                 ret = generic_rw(&pos->parent, &file_stream->parent.stream_packet_context->p);
 893                 if (ret) {
 894                         pos->offset = EOF;
 895                         fprintf(stderr, "[error] stream packet context read failed\n");
 896                         goto end;
 897                 }
 898         }
 899         pos->data_offset = pos->offset;
 900
 901 end:
 902         return;
 903 }
 904
 905 static int del_traces(gpointer key, gpointer value, gpointer user_data)
 906 {
 907         struct bt_context *bt_ctx = user_data;
 908         struct lttng_live_ctf_trace *trace = value;
 909         int ret;
 910
 911         ret = bt_context_remove_trace(bt_ctx, trace->trace_id);
 912         if (ret < 0)
 913                 fprintf(stderr, "[error] removing trace from context\n");
 914
 915         /* remove the key/value pair from the HT. */
 916         return 1;
 917 }
 918
 919 static void add_traces(gpointer key, gpointer value, gpointer user_data)
 920 {
 921         int i, ret, total_metadata = 0;
 922         uint64_t metadata_len;
 923         struct bt_context *bt_ctx = user_data;
 924         struct lttng_live_ctf_trace *trace = value;
 925         struct lttng_live_viewer_stream *stream;
 926         struct bt_mmap_stream *new_mmap_stream;
 927         struct bt_mmap_stream_list mmap_list;
 928         struct lttng_live_ctx *ctx = NULL;
 929
 930         BT_INIT_LIST_HEAD(&mmap_list.head);
 931
 932         for (i = 0; i < trace->streams->len; i++) {
 933                 stream = g_ptr_array_index(trace->streams, i);
 934                 ctx = stream->session->ctx;
 935
 936                 if (!stream->metadata_flag) {
 937                         new_mmap_stream = zmalloc(sizeof(struct bt_mmap_stream));
 938                         new_mmap_stream->priv = (void *) stream;
 939                         new_mmap_stream->fd = -1;
 940                         bt_list_add(&new_mmap_stream->list, &mmap_list.head);
 941                 } else {
 942                         /* Get all possible metadata before starting */
 943                         do {
 944                                 ret = get_new_metadata(ctx, stream,
 945                                                 &metadata_len);
 946                                 if (ret == 0) {
 947                                         total_metadata += metadata_len;
 948                                 }
 949                         } while (ret == 0 || total_metadata == 0);
 950                         trace->metadata_fp = fopen(stream->path, "r");
 951                 }
 952         }
 953
 954         if (!trace->metadata_fp) {
 955                 fprintf(stderr, "[error] No metadata stream opened\n");
 956                 goto end_free;
 957         }
 958
 959         ret = bt_context_add_trace(bt_ctx, NULL, "ctf",
 960                         ctf_live_packet_seek, &mmap_list, trace->metadata_fp);
 961         if (ret < 0) {
 962                 fprintf(stderr, "[error] Error adding trace\n");
 963                 goto end_free;
 964         }
 965         trace->trace_id = ret;
 966
 967         goto end;
 968
 969 end_free:
 970         bt_context_put(bt_ctx);
 971 end:
 972         return;
 973 }
 974
 975 int lttng_live_create_viewer_session(struct lttng_live_ctx *ctx)
 976 {
 977         struct lttng_viewer_cmd cmd;
 978         struct lttng_viewer_create_session_response resp;
 979         int ret;
 980         ssize_t ret_len;
 981
 982         cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
 983         cmd.data_size = 0;
 984         cmd.cmd_version = 0;
 985
 986         do {
 987                 ret_len = send(ctx->control_sock, &cmd, sizeof(cmd), 0);
 988         } while (ret_len < 0 && errno == EINTR);
 989         if (ret_len < 0) {
 990                 fprintf(stderr, "[error] Error sending cmd\n");
 991                 ret = ret_len;
 992                 goto error;
 993         }
 994         assert(ret_len == sizeof(cmd));
 995
 996         do {
 997                 ret_len = recv(ctx->control_sock, &resp, sizeof(resp), 0);
 998         } while (ret_len < 0 && errno == EINTR);
 999         if (ret_len < 0) {
1000                 fprintf(stderr, "[error] Error receiving create session reply\n");
1001                 ret = ret_len;
1002                 goto error;
1003         }
1004         assert(ret_len == sizeof(resp));
1005
1006         if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
1007                 fprintf(stderr, "[error] Error creating viewer session\n");
1008                 ret = -1;
1009                 goto error;
1010         }
1011         ret = 0;
1012
1013 error:
1014         return ret;
1015 }
1016
1017 void lttng_live_read(struct lttng_live_ctx *ctx, uint64_t session_id)
1018 {
1019         int ret, active_session = 0;
1020         struct bt_context *bt_ctx;
1021         struct bt_ctf_iter *iter;
1022         const struct bt_ctf_event *event;
1023         struct bt_iter_pos begin_pos;
1024         struct bt_trace_descriptor *td_write;
1025         struct bt_format *fmt_write;
1026         struct ctf_text_stream_pos *sout;
1027
1028         bt_ctx = bt_context_create();
1029         if (!bt_ctx) {
1030                 fprintf(stderr, "[error] bt_context_create allocation\n");
1031                 goto end;
1032         }
1033
1034         fmt_write = bt_lookup_format(g_quark_from_static_string("text"));
1035         if (!fmt_write) {
1036                 fprintf(stderr, "[error] ctf-text error\n");
1037                 goto end;
1038         }
1039
1040         td_write = fmt_write->open_trace(NULL, O_RDWR, NULL, NULL);
1041         if (!td_write) {
1042                 fprintf(stderr, "[error] Error opening output trace\n");
1043                 goto end_free;
1044         }
1045
1046         sout = container_of(td_write, struct ctf_text_stream_pos,
1047                         trace_descriptor);
1048         if (!sout->parent.event_cb)
1049                 goto end_free;
1050
1051         ret = lttng_live_create_viewer_session(ctx);
1052         if (ret < 0) {
1053                 goto end_free;
1054         }
1055
1056         /*
1057          * As long as the session is active, we try to reattach to it,
1058          * even if all the streams get closed.
1059          */
1060         do {
1061                 int flags;
1062
1063                 do {
1064                         ret = lttng_live_attach_session(ctx, session_id);
1065                         printf_verbose("Attaching session returns %d\n", ret);
1066                         if (ret < 0) {
1067                                 if (ret == -LTTNG_VIEWER_ATTACH_UNK) {
1068                                         if (active_session)
1069                                                 goto end_free;
1070                                         fprintf(stderr, "[error] Unknown "
1071                                                         "session ID\n");
1072                                 }
1073                                 goto end_free;
1074                         } else {
1075                                 active_session = 1;
1076                         }
1077                 } while (ctx->session->stream_count == 0);
1078
1079                 g_hash_table_foreach(ctx->session->ctf_traces, add_traces, bt_ctx);
1080
1081                 begin_pos.type = BT_SEEK_BEGIN;
1082                 iter = bt_ctf_iter_create(bt_ctx, &begin_pos, NULL);
1083                 if (!iter) {
1084                         fprintf(stderr, "[error] Iterator creation error\n");
1085                         goto end;
1086                 }
1087                 for (;;) {
1088                         event = bt_ctf_iter_read_event_flags(iter, &flags);
1089                         if (!(flags & BT_ITER_FLAG_RETRY)) {
1090                                 if (!event) {
1091                                         /* End of trace */
1092                                         break;
1093                                 }
1094                                 ret = sout->parent.event_cb(&sout->parent,
1095                                                 event->parent->stream);
1096                                 if (ret) {
1097                                         fprintf(stderr, "[error] Writing "
1098                                                         "event failed.\n");
1099                                         goto end_free;
1100                                 }
1101                         }
1102                         ret = bt_iter_next(bt_ctf_get_iter(iter));
1103                         if (ret < 0) {
1104                                 goto end_free;
1105                         }
1106                 }
1107                 bt_ctf_iter_destroy(iter);
1108                 g_hash_table_foreach_remove(ctx->session->ctf_traces, del_traces, bt_ctx);
1109         } while (active_session);
1110
1111 end_free:
1112         bt_context_put(bt_ctx);
1113 end:
1114         return;
1115 }