lib/iterator.c

   1 /*
   2  * iterator.c
   3  *
   4  * Babeltrace Library
   5  *
   6  * Copyright 2010-2011 EfficiOS Inc. and Linux Foundation
   7  *
   8  * Author: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   9  *
  10  * Permission is hereby granted, free of charge, to any person obtaining a copy
  11  * of this software and associated documentation files (the "Software"), to deal
  12  * in the Software without restriction, including without limitation the rights
  13  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  14  * copies of the Software, and to permit persons to whom the Software is
  15  * furnished to do so, subject to the following conditions:
  16  *
  17  * The above copyright notice and this permission notice shall be included in
  18  * all copies or substantial portions of the Software.
  19  *
  20  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  21  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  22  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  23  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  24  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  25  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  26  * SOFTWARE.
  27  */
  28
  29 #include <stdlib.h>
  30 #include <babeltrace/babeltrace.h>
  31 #include <babeltrace/context.h>
  32 #include <babeltrace/context-internal.h>
  33 #include <babeltrace/iterator-internal.h>
  34 #include <babeltrace/iterator.h>
  35 #include <babeltrace/prio_heap.h>
  36 #include <babeltrace/ctf/metadata.h>
  37 #include <babeltrace/ctf/events.h>
  38 #include <inttypes.h>
  39
  40 static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
  41                 const struct bt_iter_pos *begin_pos,
  42                 unsigned long stream_id);
  43
  44 struct stream_saved_pos {
  45         /*
  46          * Use file_stream pointer to check if the trace collection we
  47          * restore to match the one we saved from, for each stream.
  48          */
  49         struct ctf_file_stream *file_stream;
  50         size_t cur_index;       /* current index in packet index */
  51         ssize_t offset;         /* offset from base, in bits. EOF for end of file. */
  52         uint64_t current_real_timestamp;
  53         uint64_t current_cycles_timestamp;
  54 };
  55
  56 struct bt_saved_pos {
  57         struct trace_collection *tc;
  58         GArray *stream_saved_pos;       /* Contains struct stream_saved_pos */
  59 };
  60
  61 static int stream_read_event(struct ctf_file_stream *sin)
  62 {
  63         int ret;
  64
  65         ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
  66         if (ret == EOF)
  67                 return EOF;
  68         else if (ret == EAGAIN)
  69                 /* Stream is inactive for now (live reading). */
  70                 return EAGAIN;
  71         else if (ret) {
  72                 fprintf(stderr, "[error] Reading event failed.\n");
  73                 return ret;
  74         }
  75         return 0;
  76 }
  77
  78 /*
  79  * Return true if a < b, false otherwise.
  80  * If time stamps are exactly the same, compare by stream path. This
  81  * ensures we get the same result between runs on the same trace
  82  * collection on different environments.
  83  * The result will be random for memory-mapped traces since there is no
  84  * fixed path leading to those (they have empty path string).
  85  */
  86 static int stream_compare(void *a, void *b)
  87 {
  88         struct ctf_file_stream *s_a = a, *s_b = b;
  89
  90         if (s_a->parent.real_timestamp < s_b->parent.real_timestamp) {
  91                 return 1;
  92         } else if (likely(s_a->parent.real_timestamp > s_b->parent.real_timestamp)) {
  93                 return 0;
  94         } else {
  95                 return strcmp(s_a->parent.path, s_b->parent.path);
  96         }
  97 }
  98
  99 void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
 100 {
 101         if (!iter_pos)
 102                 return;
 103
 104         if (iter_pos->type == BT_SEEK_RESTORE && iter_pos->u.restore) {
 105                 if (iter_pos->u.restore->stream_saved_pos) {
 106                         g_array_free(
 107                                 iter_pos->u.restore->stream_saved_pos,
 108                                 TRUE);
 109                 }
 110                 g_free(iter_pos->u.restore);
 111         }
 112         g_free(iter_pos);
 113 }
 114
 115 /*
 116  * seek_file_stream_by_timestamp
 117  *
 118  * Browse a filestream by index, if an index contains the timestamp passed in
 119  * argument, seek inside the corresponding packet it until we find the event we
 120  * are looking for (either the exact timestamp or the event just after the
 121  * timestamp).
 122  *
 123  * Return 0 if the seek succeded, EOF if we didn't find any packet
 124  * containing the timestamp, or a positive integer for error.
 125  *
 126  * TODO: this should be turned into a binary search! It is currently
 127  * doing a linear search in the packets. This is a O(n) operation on a
 128  * very frequent code path.
 129  */
 130 static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
 131                 uint64_t timestamp)
 132 {
 133         struct ctf_stream_pos *stream_pos;
 134         struct packet_index *index;
 135         int i, ret;
 136
 137         stream_pos = &cfs->pos;
 138         for (i = 0; i < stream_pos->packet_index->len; i++) {
 139                 index = &g_array_index(stream_pos->packet_index,
 140                                 struct packet_index, i);
 141                 if (index->ts_real.timestamp_end < timestamp)
 142                         continue;
 143
 144                 stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
 145                 do {
 146                         ret = stream_read_event(cfs);
 147                 } while (cfs->parent.real_timestamp < timestamp && ret == 0);
 148
 149                 /* Can return either EOF, 0, or error (> 0). */
 150                 return ret;
 151         }
 152         /*
 153          * Cannot find the timestamp within the stream packets, return
 154          * EOF.
 155          */
 156         return EOF;
 157 }
 158
 159 /*
 160  * seek_ctf_trace_by_timestamp : for each file stream, seek to the event with
 161  * the corresponding timestamp
 162  *
 163  * Return 0 on success.
 164  * If the timestamp is not part of any file stream, return EOF to inform the
 165  * user the timestamp is out of the scope.
 166  * On other errors, return positive value.
 167  */
 168 static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin,
 169                 uint64_t timestamp, struct ptr_heap *stream_heap)
 170 {
 171         int i, j, ret;
 172         int found = 0;
 173
 174         /* for each stream_class */
 175         for (i = 0; i < tin->streams->len; i++) {
 176                 struct ctf_stream_declaration *stream_class;
 177
 178                 stream_class = g_ptr_array_index(tin->streams, i);
 179                 if (!stream_class)
 180                         continue;
 181                 /* for each file_stream */
 182                 for (j = 0; j < stream_class->streams->len; j++) {
 183                         struct ctf_stream_definition *stream;
 184                         struct ctf_file_stream *cfs;
 185
 186                         stream = g_ptr_array_index(stream_class->streams, j);
 187                         if (!stream)
 188                                 continue;
 189                         cfs = container_of(stream, struct ctf_file_stream,
 190                                         parent);
 191                         ret = seek_file_stream_by_timestamp(cfs, timestamp);
 192                         if (ret == 0) {
 193                                 /* Add to heap */
 194                                 ret = bt_heap_insert(stream_heap, cfs);
 195                                 if (ret) {
 196                                         /* Return positive error. */
 197                                         return -ret;
 198                                 }
 199                                 found = 1;
 200                         } else if (ret > 0) {
 201                                 /*
 202                                  * Error in seek (not EOF), failure.
 203                                  */
 204                                 return ret;
 205                         }
 206                         /* on EOF just do not put stream into heap. */
 207                 }
 208         }
 209
 210         return found ? 0 : EOF;
 211 }
 212
 213 /*
 214  * Find timestamp of last event in the stream.
 215  *
 216  * Return value: 0 if OK, positive error value on error, EOF if no
 217  * events were found.
 218  */
 219 static int find_max_timestamp_ctf_file_stream(struct ctf_file_stream *cfs,
 220                 uint64_t *timestamp_end)
 221 {
 222         int ret, count = 0, i;
 223         uint64_t timestamp = 0;
 224         struct ctf_stream_pos *stream_pos;
 225
 226         stream_pos = &cfs->pos;
 227         /*
 228          * We start by the last packet, and iterate backwards until we
 229          * either find at least one event, or we reach the first packet
 230          * (some packets can be empty).
 231          */
 232         for (i = stream_pos->packet_index->len - 1; i >= 0; i--) {
 233                 stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
 234                 count = 0;
 235                 /* read each event until we reach the end of the stream */
 236                 do {
 237                         ret = stream_read_event(cfs);
 238                         if (ret == 0) {
 239                                 count++;
 240                                 timestamp = cfs->parent.real_timestamp;
 241                         }
 242                 } while (ret == 0);
 243
 244                 /* Error */
 245                 if (ret > 0)
 246                         goto end;
 247                 assert(ret == EOF);
 248                 if (count)
 249                         break;
 250         }
 251
 252         if (count) {
 253                 *timestamp_end = timestamp;
 254                 ret = 0;
 255         } else {
 256                 /* Return EOF if no events were found */
 257                 ret = EOF;
 258         }
 259 end:
 260         return ret;
 261 }
 262
 263 /*
 264  * Find the stream within a stream class that contains the event with
 265  * the largest timestamp, and save that timestamp.
 266  *
 267  * Return 0 if OK, EOF if no events were found in the streams, or
 268  * positive value on error.
 269  */
 270 static int find_max_timestamp_ctf_stream_class(
 271                 struct ctf_stream_declaration *stream_class,
 272                 struct ctf_file_stream **cfsp,
 273                 uint64_t *max_timestamp)
 274 {
 275         int ret = EOF, i, found = 0;
 276
 277         for (i = 0; i < stream_class->streams->len; i++) {
 278                 struct ctf_stream_definition *stream;
 279                 struct ctf_file_stream *cfs;
 280                 uint64_t current_max_ts = 0;
 281
 282                 stream = g_ptr_array_index(stream_class->streams, i);
 283                 if (!stream)
 284                         continue;
 285                 cfs = container_of(stream, struct ctf_file_stream, parent);
 286                 ret = find_max_timestamp_ctf_file_stream(cfs, &current_max_ts);
 287                 if (ret == EOF)
 288                         continue;
 289                 if (ret != 0)
 290                         break;
 291                 if (current_max_ts >= *max_timestamp) {
 292                         *max_timestamp = current_max_ts;
 293                         *cfsp = cfs;
 294                         found = 1;
 295                 }
 296         }
 297         assert(ret >= 0 || ret == EOF);
 298         if (found) {
 299                 return 0;
 300         }
 301         return ret;
 302 }
 303
 304 /*
 305  * seek_last_ctf_trace_collection: seek trace collection to last event.
 306  *
 307  * Return 0 if OK, EOF if no events were found, or positive error value
 308  * on error.
 309  */
 310 static int seek_last_ctf_trace_collection(struct trace_collection *tc,
 311                 struct ctf_file_stream **cfsp)
 312 {
 313         int i, j, ret;
 314         int found = 0;
 315         uint64_t max_timestamp = 0;
 316
 317         if (!tc)
 318                 return 1;
 319
 320         /* For each trace in the trace_collection */
 321         for (i = 0; i < tc->array->len; i++) {
 322                 struct ctf_trace *tin;
 323                 struct bt_trace_descriptor *td_read;
 324
 325                 td_read = g_ptr_array_index(tc->array, i);
 326                 if (!td_read)
 327                         continue;
 328                 tin = container_of(td_read, struct ctf_trace, parent);
 329                 /* For each stream_class in the trace */
 330                 for (j = 0; j < tin->streams->len; j++) {
 331                         struct ctf_stream_declaration *stream_class;
 332
 333                         stream_class = g_ptr_array_index(tin->streams, j);
 334                         if (!stream_class)
 335                                 continue;
 336                         ret = find_max_timestamp_ctf_stream_class(stream_class,
 337                                         cfsp, &max_timestamp);
 338                         if (ret > 0)
 339                                 goto end;
 340                         if (ret == 0)
 341                                 found = 1;
 342                         assert(ret == EOF || ret == 0);
 343                 }
 344         }
 345         /*
 346          * Now we know in which file stream the last event is located,
 347          * and we know its timestamp.
 348          */
 349         if (!found) {
 350                 ret = EOF;
 351         } else {
 352                 ret = seek_file_stream_by_timestamp(*cfsp, max_timestamp);
 353                 assert(ret == 0);
 354         }
 355 end:
 356         return ret;
 357 }
 358
 359 int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 360 {
 361         struct trace_collection *tc;
 362         int i, ret;
 363
 364         if (!iter || !iter_pos)
 365                 return -EINVAL;
 366
 367         switch (iter_pos->type) {
 368         case BT_SEEK_RESTORE:
 369                 if (!iter_pos->u.restore)
 370                         return -EINVAL;
 371
 372                 bt_heap_free(iter->stream_heap);
 373                 ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 374                 if (ret < 0)
 375                         goto error_heap_init;
 376
 377                 for (i = 0; i < iter_pos->u.restore->stream_saved_pos->len;
 378                                 i++) {
 379                         struct stream_saved_pos *saved_pos;
 380                         struct ctf_stream_pos *stream_pos;
 381                         struct ctf_stream_definition *stream;
 382
 383                         saved_pos = &g_array_index(
 384                                         iter_pos->u.restore->stream_saved_pos,
 385                                         struct stream_saved_pos, i);
 386                         stream = &saved_pos->file_stream->parent;
 387                         stream_pos = &saved_pos->file_stream->pos;
 388
 389                         stream_pos->packet_seek(&stream_pos->parent,
 390                                         saved_pos->cur_index, SEEK_SET);
 391
 392                         /*
 393                          * the timestamp needs to be restored after
 394                          * packet_seek, because this function resets
 395                          * the timestamp to the beginning of the packet
 396                          */
 397                         stream->real_timestamp = saved_pos->current_real_timestamp;
 398                         stream->cycles_timestamp = saved_pos->current_cycles_timestamp;
 399                         stream_pos->offset = saved_pos->offset;
 400                         stream_pos->last_offset = LAST_OFFSET_POISON;
 401
 402                         stream->current.real.begin = 0;
 403                         stream->current.real.end = 0;
 404                         stream->current.cycles.begin = 0;
 405                         stream->current.cycles.end = 0;
 406
 407                         stream->prev.real.begin = 0;
 408                         stream->prev.real.end = 0;
 409                         stream->prev.cycles.begin = 0;
 410                         stream->prev.cycles.end = 0;
 411
 412                         printf_debug("restored to cur_index = %" PRId64 " and "
 413                                 "offset = %" PRId64 ", timestamp = %" PRIu64 "\n",
 414                                 stream_pos->cur_index,
 415                                 stream_pos->offset, stream->real_timestamp);
 416
 417                         ret = stream_read_event(saved_pos->file_stream);
 418                         if (ret != 0) {
 419                                 goto error;
 420                         }
 421
 422                         /* Add to heap */
 423                         ret = bt_heap_insert(iter->stream_heap,
 424                                         saved_pos->file_stream);
 425                         if (ret)
 426                                 goto error;
 427                 }
 428                 return 0;
 429         case BT_SEEK_TIME:
 430                 tc = iter->ctx->tc;
 431
 432                 bt_heap_free(iter->stream_heap);
 433                 ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 434                 if (ret < 0)
 435                         goto error_heap_init;
 436
 437                 /* for each trace in the trace_collection */
 438                 for (i = 0; i < tc->array->len; i++) {
 439                         struct ctf_trace *tin;
 440                         struct bt_trace_descriptor *td_read;
 441
 442                         td_read = g_ptr_array_index(tc->array, i);
 443                         if (!td_read)
 444                                 continue;
 445                         tin = container_of(td_read, struct ctf_trace, parent);
 446
 447                         ret = seek_ctf_trace_by_timestamp(tin,
 448                                         iter_pos->u.seek_time,
 449                                         iter->stream_heap);
 450                         /*
 451                          * Positive errors are failure. Negative value
 452                          * is EOF (for which we continue with other
 453                          * traces). 0 is success. Note: on EOF, it just
 454                          * means that no stream has been added to the
 455                          * iterator for that trace, which is fine.
 456                          */
 457                         if (ret != 0 && ret != EOF)
 458                                 goto error;
 459                 }
 460                 return 0;
 461         case BT_SEEK_BEGIN:
 462                 tc = iter->ctx->tc;
 463                 bt_heap_free(iter->stream_heap);
 464                 ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 465                 if (ret < 0)
 466                         goto error_heap_init;
 467
 468                 for (i = 0; i < tc->array->len; i++) {
 469                         struct ctf_trace *tin;
 470                         struct bt_trace_descriptor *td_read;
 471                         int stream_id;
 472
 473                         td_read = g_ptr_array_index(tc->array, i);
 474                         if (!td_read)
 475                                 continue;
 476                         tin = container_of(td_read, struct ctf_trace, parent);
 477
 478                         /* Populate heap with each stream */
 479                         for (stream_id = 0; stream_id < tin->streams->len;
 480                                         stream_id++) {
 481                                 struct ctf_stream_declaration *stream;
 482                                 int filenr;
 483
 484                                 stream = g_ptr_array_index(tin->streams,
 485                                                 stream_id);
 486                                 if (!stream)
 487                                         continue;
 488                                 for (filenr = 0; filenr < stream->streams->len;
 489                                                 filenr++) {
 490                                         struct ctf_file_stream *file_stream;
 491                                         file_stream = g_ptr_array_index(
 492                                                         stream->streams,
 493                                                         filenr);
 494                                         if (!file_stream)
 495                                                 continue;
 496                                         ret = babeltrace_filestream_seek(
 497                                                         file_stream, iter_pos,
 498                                                         stream_id);
 499                                         if (ret != 0 && ret != EOF) {
 500                                                 goto error;
 501                                         }
 502                                         if (ret == EOF) {
 503                                                 /* Do not add EOF streams */
 504                                                 continue;
 505                                         }
 506                                         ret = bt_heap_insert(iter->stream_heap, file_stream);
 507                                         if (ret)
 508                                                 goto error;
 509                                 }
 510                         }
 511                 }
 512                 break;
 513         case BT_SEEK_LAST:
 514         {
 515                 struct ctf_file_stream *cfs = NULL;
 516
 517                 tc = iter->ctx->tc;
 518                 ret = seek_last_ctf_trace_collection(tc, &cfs);
 519                 if (ret != 0 || !cfs)
 520                         goto error;
 521                 /* remove all streams from the heap */
 522                 bt_heap_free(iter->stream_heap);
 523                 /* Create a new empty heap */
 524                 ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 525                 if (ret < 0)
 526                         goto error;
 527                 /* Insert the stream that contains the last event */
 528                 ret = bt_heap_insert(iter->stream_heap, cfs);
 529                 if (ret)
 530                         goto error;
 531                 break;
 532         }
 533         default:
 534                 /* not implemented */
 535                 return -EINVAL;
 536         }
 537
 538         return 0;
 539
 540 error:
 541         bt_heap_free(iter->stream_heap);
 542 error_heap_init:
 543         if (bt_heap_init(iter->stream_heap, 0, stream_compare) < 0) {
 544                 bt_heap_free(iter->stream_heap);
 545                 g_free(iter->stream_heap);
 546                 iter->stream_heap = NULL;
 547                 ret = -ENOMEM;
 548         }
 549
 550         return ret;
 551 }
 552
 553 struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
 554 {
 555         struct bt_iter_pos *pos;
 556         struct trace_collection *tc;
 557         struct ctf_file_stream *file_stream = NULL, *removed;
 558         struct ptr_heap iter_heap_copy;
 559         int ret;
 560
 561         if (!iter)
 562                 return NULL;
 563
 564         tc = iter->ctx->tc;
 565         pos = g_new0(struct bt_iter_pos, 1);
 566         pos->type = BT_SEEK_RESTORE;
 567         pos->u.restore = g_new0(struct bt_saved_pos, 1);
 568         pos->u.restore->tc = tc;
 569         pos->u.restore->stream_saved_pos = g_array_new(FALSE, TRUE,
 570                         sizeof(struct stream_saved_pos));
 571         if (!pos->u.restore->stream_saved_pos)
 572                 goto error;
 573
 574         ret = bt_heap_copy(&iter_heap_copy, iter->stream_heap);
 575         if (ret < 0)
 576                 goto error_heap;
 577
 578         /* iterate over each stream in the heap */
 579         file_stream = bt_heap_maximum(&iter_heap_copy);
 580         while (file_stream != NULL) {
 581                 struct stream_saved_pos saved_pos;
 582
 583                 assert(file_stream->pos.last_offset != LAST_OFFSET_POISON);
 584                 saved_pos.offset = file_stream->pos.last_offset;
 585                 saved_pos.file_stream = file_stream;
 586                 saved_pos.cur_index = file_stream->pos.cur_index;
 587
 588                 saved_pos.current_real_timestamp = file_stream->parent.real_timestamp;
 589                 saved_pos.current_cycles_timestamp = file_stream->parent.cycles_timestamp;
 590
 591                 g_array_append_val(
 592                                 pos->u.restore->stream_saved_pos,
 593                                 saved_pos);
 594
 595                 printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
 596                                 "offset : %zd, "
 597                                 "timestamp = %" PRIu64 "\n",
 598                                 file_stream->parent.stream_id,
 599                                 saved_pos.cur_index, saved_pos.offset,
 600                                 saved_pos.current_real_timestamp);
 601
 602                 /* remove the stream from the heap copy */
 603                 removed = bt_heap_remove(&iter_heap_copy);
 604                 assert(removed == file_stream);
 605
 606                 file_stream = bt_heap_maximum(&iter_heap_copy);
 607         }
 608         bt_heap_free(&iter_heap_copy);
 609         return pos;
 610
 611 error_heap:
 612         g_array_free(pos->u.restore->stream_saved_pos, TRUE);
 613 error:
 614         g_free(pos);
 615         return NULL;
 616 }
 617
 618 struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *iter,
 619                 uint64_t timestamp)
 620 {
 621         struct bt_iter_pos *pos;
 622
 623         if (!iter)
 624                 return NULL;
 625
 626         pos = g_new0(struct bt_iter_pos, 1);
 627         pos->type = BT_SEEK_TIME;
 628         pos->u.seek_time = timestamp;
 629         return pos;
 630 }
 631
 632 /*
 633  * babeltrace_filestream_seek: seek a filestream to given position.
 634  *
 635  * The stream_id parameter is only useful for BT_SEEK_RESTORE.
 636  */
 637 static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
 638                 const struct bt_iter_pos *begin_pos,
 639                 unsigned long stream_id)
 640 {
 641         int ret = 0;
 642
 643         if (!file_stream || !begin_pos)
 644                 return -EINVAL;
 645
 646         switch (begin_pos->type) {
 647         case BT_SEEK_CUR:
 648                 /*
 649                  * just insert into the heap we should already know
 650                  * the timestamps
 651                  */
 652                 break;
 653         case BT_SEEK_BEGIN:
 654                 file_stream->pos.packet_seek(&file_stream->pos.parent,
 655                                 0, SEEK_SET);
 656                 ret = stream_read_event(file_stream);
 657                 break;
 658         case BT_SEEK_TIME:
 659         case BT_SEEK_RESTORE:
 660         default:
 661                 assert(0); /* Not yet defined */
 662         }
 663
 664         return ret;
 665 }
 666
 667 int bt_iter_add_trace(struct bt_iter *iter,
 668                 struct bt_trace_descriptor *td_read)
 669 {
 670         struct ctf_trace *tin;
 671         int stream_id, ret = 0;
 672
 673         tin = container_of(td_read, struct ctf_trace, parent);
 674
 675         /* Populate heap with each stream */
 676         for (stream_id = 0; stream_id < tin->streams->len;
 677                         stream_id++) {
 678                 struct ctf_stream_declaration *stream;
 679                 int filenr;
 680
 681                 stream = g_ptr_array_index(tin->streams, stream_id);
 682                 if (!stream)
 683                         continue;
 684                 for (filenr = 0; filenr < stream->streams->len;
 685                                 filenr++) {
 686                         struct ctf_file_stream *file_stream;
 687                         struct bt_iter_pos pos;
 688
 689                         file_stream = g_ptr_array_index(stream->streams,
 690                                         filenr);
 691                         if (!file_stream)
 692                                 continue;
 693
 694                         pos.type = BT_SEEK_BEGIN;
 695                         ret = babeltrace_filestream_seek(file_stream,
 696                                         &pos, stream_id);
 697
 698                         if (ret == EOF) {
 699                                 ret = 0;
 700                                 continue;
 701                         } else if (ret != 0 && ret != EAGAIN) {
 702                                 goto error;
 703                         }
 704                         /* Add to heap */
 705                         ret = bt_heap_insert(iter->stream_heap, file_stream);
 706                         if (ret)
 707                                 goto error;
 708                 }
 709         }
 710
 711 error:
 712         return ret;
 713 }
 714
 715 int bt_iter_init(struct bt_iter *iter,
 716                 struct bt_context *ctx,
 717                 const struct bt_iter_pos *begin_pos,
 718                 const struct bt_iter_pos *end_pos)
 719 {
 720         int i;
 721         int ret = 0;
 722
 723         if (!iter || !ctx)
 724                 return -EINVAL;
 725
 726         if (ctx->current_iterator) {
 727                 ret = -1;
 728                 goto error_ctx;
 729         }
 730
 731         iter->stream_heap = g_new(struct ptr_heap, 1);
 732         iter->end_pos = end_pos;
 733         bt_context_get(ctx);
 734         iter->ctx = ctx;
 735
 736         ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 737         if (ret < 0)
 738                 goto error_heap_init;
 739
 740         for (i = 0; i < ctx->tc->array->len; i++) {
 741                 struct bt_trace_descriptor *td_read;
 742
 743                 td_read = g_ptr_array_index(ctx->tc->array, i);
 744                 if (!td_read)
 745                         continue;
 746                 ret = bt_iter_add_trace(iter, td_read);
 747                 if (ret < 0)
 748                         goto error;
 749         }
 750
 751         ctx->current_iterator = iter;
 752         if (begin_pos && begin_pos->type != BT_SEEK_BEGIN) {
 753                 ret = bt_iter_set_pos(iter, begin_pos);
 754         }
 755
 756         return ret;
 757
 758 error:
 759         bt_heap_free(iter->stream_heap);
 760 error_heap_init:
 761         g_free(iter->stream_heap);
 762         iter->stream_heap = NULL;
 763 error_ctx:
 764         return ret;
 765 }
 766
 767 struct bt_iter *bt_iter_create(struct bt_context *ctx,
 768                 const struct bt_iter_pos *begin_pos,
 769                 const struct bt_iter_pos *end_pos)
 770 {
 771         struct bt_iter *iter;
 772         int ret;
 773
 774         if (!ctx)
 775                 return NULL;
 776
 777         iter = g_new0(struct bt_iter, 1);
 778         ret = bt_iter_init(iter, ctx, begin_pos, end_pos);
 779         if (ret) {
 780                 g_free(iter);
 781                 return NULL;
 782         }
 783         return iter;
 784 }
 785
 786 void bt_iter_fini(struct bt_iter *iter)
 787 {
 788         assert(iter);
 789         if (iter->stream_heap) {
 790                 bt_heap_free(iter->stream_heap);
 791                 g_free(iter->stream_heap);
 792         }
 793         iter->ctx->current_iterator = NULL;
 794         bt_context_put(iter->ctx);
 795 }
 796
 797 void bt_iter_destroy(struct bt_iter *iter)
 798 {
 799         assert(iter);
 800         bt_iter_fini(iter);
 801         g_free(iter);
 802 }
 803
 804 int bt_iter_next(struct bt_iter *iter)
 805 {
 806         struct ctf_file_stream *file_stream, *removed;
 807         int ret;
 808
 809         if (!iter)
 810                 return -EINVAL;
 811
 812         file_stream = bt_heap_maximum(iter->stream_heap);
 813         if (!file_stream) {
 814                 /* end of file for all streams */
 815                 ret = 0;
 816                 goto end;
 817         }
 818
 819         ret = stream_read_event(file_stream);
 820         if (ret == EOF) {
 821                 removed = bt_heap_remove(iter->stream_heap);
 822                 assert(removed == file_stream);
 823                 ret = 0;
 824                 goto end;
 825         } else if (ret == EAGAIN) {
 826                 /*
 827                  * Live streaming: the stream is inactive for now, we
 828                  * just updated the timestamp_end to skip over this
 829                  * stream up to a certain point in time.
 830                  *
 831                  * Since we can't guarantee that a stream will ever have
 832                  * any activity, we can't rely on the fact that
 833                  * bt_iter_next will be called for each stream and deal
 834                  * with inactive streams. So instead, we return 0 here
 835                  * to the caller and let the read API handle the
 836                  * retry case.
 837                  */
 838                 ret = 0;
 839                 goto reinsert;
 840         } else if (ret) {
 841                 goto end;
 842         }
 843
 844 reinsert:
 845         /* Reinsert the file stream into the heap, and rebalance. */
 846         removed = bt_heap_replace_max(iter->stream_heap, file_stream);
 847         assert(removed == file_stream);
 848 end:
 849         return ret;
 850 }