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) {
  69                 fprintf(stderr, "[error] Reading event failed.\n");
  70                 return ret;
  71         }
  72         return 0;
  73 }
  74
  75 /*
  76  * returns true if a < b, false otherwise.
  77  */
  78 static int stream_compare(void *a, void *b)
  79 {
  80         struct ctf_file_stream *s_a = a, *s_b = b;
  81
  82         if (s_a->parent.real_timestamp < s_b->parent.real_timestamp)
  83                 return 1;
  84         else
  85                 return 0;
  86 }
  87
  88 void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
  89 {
  90         if (!iter_pos)
  91                 return;
  92
  93         if (iter_pos->type == BT_SEEK_RESTORE && iter_pos->u.restore) {
  94                 if (iter_pos->u.restore->stream_saved_pos) {
  95                         g_array_free(
  96                                 iter_pos->u.restore->stream_saved_pos,
  97                                 TRUE);
  98                 }
  99                 g_free(iter_pos->u.restore);
 100         }
 101         g_free(iter_pos);
 102 }
 103
 104 /*
 105  * seek_file_stream_by_timestamp
 106  *
 107  * Browse a filestream by index, if an index contains the timestamp passed in
 108  * argument, seek inside the corresponding packet it until we find the event we
 109  * are looking for (either the exact timestamp or the event just after the
 110  * timestamp).
 111  *
 112  * Return 0 if the seek succeded, EOF if we didn't find any packet
 113  * containing the timestamp, or a positive integer for error.
 114  *
 115  * TODO: this should be turned into a binary search! It is currently
 116  * doing a linear search in the packets. This is a O(n) operation on a
 117  * very frequent code path.
 118  */
 119 static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
 120                 uint64_t timestamp)
 121 {
 122         struct ctf_stream_pos *stream_pos;
 123         struct packet_index *index;
 124         int i, ret;
 125
 126         stream_pos = &cfs->pos;
 127         for (i = 0; i < stream_pos->packet_real_index->len; i++) {
 128                 index = &g_array_index(stream_pos->packet_real_index,
 129                                 struct packet_index, i);
 130                 if (index->timestamp_end < timestamp)
 131                         continue;
 132
 133                 stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
 134                 do {
 135                         ret = stream_read_event(cfs);
 136                 } while (cfs->parent.real_timestamp < timestamp && ret == 0);
 137
 138                 /* Can return either EOF, 0, or error (> 0). */
 139                 return ret;
 140         }
 141         /*
 142          * Cannot find the timestamp within the stream packets, return
 143          * EOF.
 144          */
 145         return EOF;
 146 }
 147
 148 /*
 149  * seek_ctf_trace_by_timestamp : for each file stream, seek to the event with
 150  * the corresponding timestamp
 151  *
 152  * Return 0 on success.
 153  * If the timestamp is not part of any file stream, return EOF to inform the
 154  * user the timestamp is out of the scope.
 155  * On other errors, return positive value.
 156  */
 157 static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin,
 158                 uint64_t timestamp, struct ptr_heap *stream_heap)
 159 {
 160         int i, j, ret;
 161         int found = 0;
 162
 163         /* for each stream_class */
 164         for (i = 0; i < tin->streams->len; i++) {
 165                 struct ctf_stream_declaration *stream_class;
 166
 167                 stream_class = g_ptr_array_index(tin->streams, i);
 168                 if (!stream_class)
 169                         continue;
 170                 /* for each file_stream */
 171                 for (j = 0; j < stream_class->streams->len; j++) {
 172                         struct ctf_stream_definition *stream;
 173                         struct ctf_file_stream *cfs;
 174
 175                         stream = g_ptr_array_index(stream_class->streams, j);
 176                         if (!stream)
 177                                 continue;
 178                         cfs = container_of(stream, struct ctf_file_stream,
 179                                         parent);
 180                         ret = seek_file_stream_by_timestamp(cfs, timestamp);
 181                         if (ret == 0) {
 182                                 /* Add to heap */
 183                                 ret = heap_insert(stream_heap, cfs);
 184                                 if (ret) {
 185                                         /* Return positive error. */
 186                                         return -ret;
 187                                 }
 188                                 found = 1;
 189                         } else if (ret > 0) {
 190                                 /*
 191                                  * Error in seek (not EOF), failure.
 192                                  */
 193                                 return ret;
 194                         }
 195                         /* on EOF just do not put stream into heap. */
 196                 }
 197         }
 198
 199         return found ? 0 : EOF;
 200 }
 201
 202 /*
 203  * Find timestamp of last event in the stream.
 204  *
 205  * Return value: 0 if OK, positive error value on error, EOF if no
 206  * events were found.
 207  */
 208 static int find_max_timestamp_ctf_file_stream(struct ctf_file_stream *cfs,
 209                 uint64_t *timestamp_end)
 210 {
 211         int ret, count = 0, i;
 212         uint64_t timestamp = 0;
 213         struct ctf_stream_pos *stream_pos;
 214
 215         stream_pos = &cfs->pos;
 216         /*
 217          * We start by the last packet, and iterate backwards until we
 218          * either find at least one event, or we reach the first packet
 219          * (some packets can be empty).
 220          */
 221         for (i = stream_pos->packet_real_index->len - 1; i >= 0; i--) {
 222                 stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
 223                 count = 0;
 224                 /* read each event until we reach the end of the stream */
 225                 do {
 226                         ret = stream_read_event(cfs);
 227                         if (ret == 0) {
 228                                 count++;
 229                                 timestamp = cfs->parent.real_timestamp;
 230                         }
 231                 } while (ret == 0);
 232
 233                 /* Error */
 234                 if (ret > 0)
 235                         goto end;
 236                 assert(ret == EOF);
 237                 if (count)
 238                         break;
 239         }
 240
 241         if (count) {
 242                 *timestamp_end = timestamp;
 243                 ret = 0;
 244         } else {
 245                 /* Return EOF if no events were found */
 246                 ret = EOF;
 247         }
 248 end:
 249         return ret;
 250 }
 251
 252 /*
 253  * Find the stream within a stream class that contains the event with
 254  * the largest timestamp, and save that timestamp.
 255  *
 256  * Return 0 if OK, EOF if no events were found in the streams, or
 257  * positive value on error.
 258  */
 259 static int find_max_timestamp_ctf_stream_class(
 260                 struct ctf_stream_declaration *stream_class,
 261                 struct ctf_file_stream **cfsp,
 262                 uint64_t *max_timestamp)
 263 {
 264         int ret = EOF, i, found = 0;
 265
 266         for (i = 0; i < stream_class->streams->len; i++) {
 267                 struct ctf_stream_definition *stream;
 268                 struct ctf_file_stream *cfs;
 269                 uint64_t current_max_ts = 0;
 270
 271                 stream = g_ptr_array_index(stream_class->streams, i);
 272                 if (!stream)
 273                         continue;
 274                 cfs = container_of(stream, struct ctf_file_stream, parent);
 275                 ret = find_max_timestamp_ctf_file_stream(cfs, &current_max_ts);
 276                 if (ret == EOF)
 277                         continue;
 278                 if (ret != 0)
 279                         break;
 280                 if (current_max_ts >= *max_timestamp) {
 281                         *max_timestamp = current_max_ts;
 282                         *cfsp = cfs;
 283                         found = 1;
 284                 }
 285         }
 286         assert(ret >= 0 || ret == EOF);
 287         if (found) {
 288                 return 0;
 289         }
 290         return ret;
 291 }
 292
 293 /*
 294  * seek_last_ctf_trace_collection: seek trace collection to last event.
 295  *
 296  * Return 0 if OK, EOF if no events were found, or positive error value
 297  * on error.
 298  */
 299 static int seek_last_ctf_trace_collection(struct trace_collection *tc,
 300                 struct ctf_file_stream **cfsp)
 301 {
 302         int i, j, ret;
 303         int found = 0;
 304         uint64_t max_timestamp = 0;
 305
 306         if (!tc)
 307                 return 1;
 308
 309         /* For each trace in the trace_collection */
 310         for (i = 0; i < tc->array->len; i++) {
 311                 struct ctf_trace *tin;
 312                 struct trace_descriptor *td_read;
 313
 314                 td_read = g_ptr_array_index(tc->array, i);
 315                 if (!td_read)
 316                         continue;
 317                 tin = container_of(td_read, struct ctf_trace, parent);
 318                 /* For each stream_class in the trace */
 319                 for (j = 0; j < tin->streams->len; j++) {
 320                         struct ctf_stream_declaration *stream_class;
 321
 322                         stream_class = g_ptr_array_index(tin->streams, j);
 323                         if (!stream_class)
 324                                 continue;
 325                         ret = find_max_timestamp_ctf_stream_class(stream_class,
 326                                         cfsp, &max_timestamp);
 327                         if (ret > 0)
 328                                 goto end;
 329                         if (ret == 0)
 330                                 found = 1;
 331                         assert(ret == EOF || ret == 0);
 332                 }
 333         }
 334         /*
 335          * Now we know in which file stream the last event is located,
 336          * and we know its timestamp.
 337          */
 338         if (!found) {
 339                 ret = EOF;
 340         } else {
 341                 ret = seek_file_stream_by_timestamp(*cfsp, max_timestamp);
 342                 assert(ret == 0);
 343         }
 344 end:
 345         return ret;
 346 }
 347
 348 int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 349 {
 350         struct trace_collection *tc;
 351         int i, ret;
 352
 353         if (!iter || !iter_pos)
 354                 return -EINVAL;
 355
 356         switch (iter_pos->type) {
 357         case BT_SEEK_RESTORE:
 358                 if (!iter_pos->u.restore)
 359                         return -EINVAL;
 360
 361                 heap_free(iter->stream_heap);
 362                 ret = heap_init(iter->stream_heap, 0, stream_compare);
 363                 if (ret < 0)
 364                         goto error_heap_init;
 365
 366                 for (i = 0; i < iter_pos->u.restore->stream_saved_pos->len;
 367                                 i++) {
 368                         struct stream_saved_pos *saved_pos;
 369                         struct ctf_stream_pos *stream_pos;
 370                         struct ctf_stream_definition *stream;
 371
 372                         saved_pos = &g_array_index(
 373                                         iter_pos->u.restore->stream_saved_pos,
 374                                         struct stream_saved_pos, i);
 375                         stream = &saved_pos->file_stream->parent;
 376                         stream_pos = &saved_pos->file_stream->pos;
 377
 378                         stream_pos->packet_seek(&stream_pos->parent,
 379                                         saved_pos->cur_index, SEEK_SET);
 380
 381                         /*
 382                          * the timestamp needs to be restored after
 383                          * packet_seek, because this function resets
 384                          * the timestamp to the beginning of the packet
 385                          */
 386                         stream->real_timestamp = saved_pos->current_real_timestamp;
 387                         stream->cycles_timestamp = saved_pos->current_cycles_timestamp;
 388                         stream_pos->offset = saved_pos->offset;
 389                         stream_pos->last_offset = LAST_OFFSET_POISON;
 390
 391                         stream->prev_real_timestamp = 0;
 392                         stream->prev_real_timestamp_end = 0;
 393                         stream->prev_cycles_timestamp = 0;
 394                         stream->prev_cycles_timestamp_end = 0;
 395
 396                         printf_debug("restored to cur_index = %" PRId64 " and "
 397                                 "offset = %" PRId64 ", timestamp = %" PRIu64 "\n",
 398                                 stream_pos->cur_index,
 399                                 stream_pos->offset, stream->real_timestamp);
 400
 401                         ret = stream_read_event(saved_pos->file_stream);
 402                         if (ret != 0) {
 403                                 goto error;
 404                         }
 405
 406                         /* Add to heap */
 407                         ret = heap_insert(iter->stream_heap,
 408                                         saved_pos->file_stream);
 409                         if (ret)
 410                                 goto error;
 411                 }
 412                 return 0;
 413         case BT_SEEK_TIME:
 414                 tc = iter->ctx->tc;
 415
 416                 heap_free(iter->stream_heap);
 417                 ret = heap_init(iter->stream_heap, 0, stream_compare);
 418                 if (ret < 0)
 419                         goto error_heap_init;
 420
 421                 /* for each trace in the trace_collection */
 422                 for (i = 0; i < tc->array->len; i++) {
 423                         struct ctf_trace *tin;
 424                         struct trace_descriptor *td_read;
 425
 426                         td_read = g_ptr_array_index(tc->array, i);
 427                         if (!td_read)
 428                                 continue;
 429                         tin = container_of(td_read, struct ctf_trace, parent);
 430
 431                         ret = seek_ctf_trace_by_timestamp(tin,
 432                                         iter_pos->u.seek_time,
 433                                         iter->stream_heap);
 434                         /*
 435                          * Positive errors are failure. Negative value
 436                          * is EOF (for which we continue with other
 437                          * traces). 0 is success. Note: on EOF, it just
 438                          * means that no stream has been added to the
 439                          * iterator for that trace, which is fine.
 440                          */
 441                         if (ret != 0 && ret != EOF)
 442                                 goto error;
 443                 }
 444                 return 0;
 445         case BT_SEEK_BEGIN:
 446                 tc = iter->ctx->tc;
 447                 heap_free(iter->stream_heap);
 448                 ret = heap_init(iter->stream_heap, 0, stream_compare);
 449                 if (ret < 0)
 450                         goto error_heap_init;
 451
 452                 for (i = 0; i < tc->array->len; i++) {
 453                         struct ctf_trace *tin;
 454                         struct trace_descriptor *td_read;
 455                         int stream_id;
 456
 457                         td_read = g_ptr_array_index(tc->array, i);
 458                         if (!td_read)
 459                                 continue;
 460                         tin = container_of(td_read, struct ctf_trace, parent);
 461
 462                         /* Populate heap with each stream */
 463                         for (stream_id = 0; stream_id < tin->streams->len;
 464                                         stream_id++) {
 465                                 struct ctf_stream_declaration *stream;
 466                                 int filenr;
 467
 468                                 stream = g_ptr_array_index(tin->streams,
 469                                                 stream_id);
 470                                 if (!stream)
 471                                         continue;
 472                                 for (filenr = 0; filenr < stream->streams->len;
 473                                                 filenr++) {
 474                                         struct ctf_file_stream *file_stream;
 475                                         file_stream = g_ptr_array_index(
 476                                                         stream->streams,
 477                                                         filenr);
 478                                         if (!file_stream)
 479                                                 continue;
 480                                         ret = babeltrace_filestream_seek(
 481                                                         file_stream, iter_pos,
 482                                                         stream_id);
 483                                         if (ret != 0 && ret != EOF) {
 484                                                 goto error;
 485                                         }
 486                                         if (ret == EOF) {
 487                                                 /* Do not add EOF streams */
 488                                                 continue;
 489                                         }
 490                                         ret = heap_insert(iter->stream_heap, file_stream);
 491                                         if (ret)
 492                                                 goto error;
 493                                 }
 494                         }
 495                 }
 496                 break;
 497         case BT_SEEK_LAST:
 498         {
 499                 struct ctf_file_stream *cfs = NULL;
 500
 501                 tc = iter->ctx->tc;
 502                 ret = seek_last_ctf_trace_collection(tc, &cfs);
 503                 if (ret != 0 || !cfs)
 504                         goto error;
 505                 /* remove all streams from the heap */
 506                 heap_free(iter->stream_heap);
 507                 /* Create a new empty heap */
 508                 ret = heap_init(iter->stream_heap, 0, stream_compare);
 509                 if (ret < 0)
 510                         goto error;
 511                 /* Insert the stream that contains the last event */
 512                 ret = heap_insert(iter->stream_heap, cfs);
 513                 if (ret)
 514                         goto error;
 515                 break;
 516         }
 517         default:
 518                 /* not implemented */
 519                 return -EINVAL;
 520         }
 521
 522         return 0;
 523
 524 error:
 525         heap_free(iter->stream_heap);
 526 error_heap_init:
 527         if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
 528                 heap_free(iter->stream_heap);
 529                 g_free(iter->stream_heap);
 530                 iter->stream_heap = NULL;
 531                 ret = -ENOMEM;
 532         }
 533
 534         return ret;
 535 }
 536
 537 struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
 538 {
 539         struct bt_iter_pos *pos;
 540         struct trace_collection *tc;
 541         struct ctf_file_stream *file_stream = NULL, *removed;
 542         struct ptr_heap iter_heap_copy;
 543         int ret;
 544
 545         if (!iter)
 546                 return NULL;
 547
 548         tc = iter->ctx->tc;
 549         pos = g_new0(struct bt_iter_pos, 1);
 550         pos->type = BT_SEEK_RESTORE;
 551         pos->u.restore = g_new0(struct bt_saved_pos, 1);
 552         pos->u.restore->tc = tc;
 553         pos->u.restore->stream_saved_pos = g_array_new(FALSE, TRUE,
 554                         sizeof(struct stream_saved_pos));
 555         if (!pos->u.restore->stream_saved_pos)
 556                 goto error;
 557
 558         ret = heap_copy(&iter_heap_copy, iter->stream_heap);
 559         if (ret < 0)
 560                 goto error_heap;
 561
 562         /* iterate over each stream in the heap */
 563         file_stream = heap_maximum(&iter_heap_copy);
 564         while (file_stream != NULL) {
 565                 struct stream_saved_pos saved_pos;
 566
 567                 assert(file_stream->pos.last_offset != LAST_OFFSET_POISON);
 568                 saved_pos.offset = file_stream->pos.last_offset;
 569                 saved_pos.file_stream = file_stream;
 570                 saved_pos.cur_index = file_stream->pos.cur_index;
 571
 572                 saved_pos.current_real_timestamp = file_stream->parent.real_timestamp;
 573                 saved_pos.current_cycles_timestamp = file_stream->parent.cycles_timestamp;
 574
 575                 g_array_append_val(
 576                                 pos->u.restore->stream_saved_pos,
 577                                 saved_pos);
 578
 579                 printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
 580                                 "offset : %zd, "
 581                                 "timestamp = %" PRIu64 "\n",
 582                                 file_stream->parent.stream_id,
 583                                 saved_pos.cur_index, saved_pos.offset,
 584                                 saved_pos.current_real_timestamp);
 585
 586                 /* remove the stream from the heap copy */
 587                 removed = heap_remove(&iter_heap_copy);
 588                 assert(removed == file_stream);
 589
 590                 file_stream = heap_maximum(&iter_heap_copy);
 591         }
 592         heap_free(&iter_heap_copy);
 593         return pos;
 594
 595 error_heap:
 596         g_array_free(pos->u.restore->stream_saved_pos, TRUE);
 597 error:
 598         g_free(pos);
 599         return NULL;
 600 }
 601
 602 struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *iter,
 603                 uint64_t timestamp)
 604 {
 605         struct bt_iter_pos *pos;
 606
 607         if (!iter)
 608                 return NULL;
 609
 610         pos = g_new0(struct bt_iter_pos, 1);
 611         pos->type = BT_SEEK_TIME;
 612         pos->u.seek_time = timestamp;
 613         return pos;
 614 }
 615
 616 /*
 617  * babeltrace_filestream_seek: seek a filestream to given position.
 618  *
 619  * The stream_id parameter is only useful for BT_SEEK_RESTORE.
 620  */
 621 static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
 622                 const struct bt_iter_pos *begin_pos,
 623                 unsigned long stream_id)
 624 {
 625         int ret = 0;
 626
 627         if (!file_stream || !begin_pos)
 628                 return -EINVAL;
 629
 630         switch (begin_pos->type) {
 631         case BT_SEEK_CUR:
 632                 /*
 633                  * just insert into the heap we should already know
 634                  * the timestamps
 635                  */
 636                 break;
 637         case BT_SEEK_BEGIN:
 638                 file_stream->pos.packet_seek(&file_stream->pos.parent,
 639                                 0, SEEK_SET);
 640                 ret = stream_read_event(file_stream);
 641                 break;
 642         case BT_SEEK_TIME:
 643         case BT_SEEK_RESTORE:
 644         default:
 645                 assert(0); /* Not yet defined */
 646         }
 647
 648         return ret;
 649 }
 650
 651 int bt_iter_init(struct bt_iter *iter,
 652                 struct bt_context *ctx,
 653                 const struct bt_iter_pos *begin_pos,
 654                 const struct bt_iter_pos *end_pos)
 655 {
 656         int i, stream_id;
 657         int ret = 0;
 658
 659         if (!iter || !ctx)
 660                 return -EINVAL;
 661
 662         if (ctx->current_iterator) {
 663                 ret = -1;
 664                 goto error_ctx;
 665         }
 666
 667         iter->stream_heap = g_new(struct ptr_heap, 1);
 668         iter->end_pos = end_pos;
 669         bt_context_get(ctx);
 670         iter->ctx = ctx;
 671
 672         ret = heap_init(iter->stream_heap, 0, stream_compare);
 673         if (ret < 0)
 674                 goto error_heap_init;
 675
 676         for (i = 0; i < ctx->tc->array->len; i++) {
 677                 struct ctf_trace *tin;
 678                 struct trace_descriptor *td_read;
 679
 680                 td_read = g_ptr_array_index(ctx->tc->array, i);
 681                 if (!td_read)
 682                         continue;
 683                 tin = container_of(td_read, struct ctf_trace, parent);
 684
 685                 /* Populate heap with each stream */
 686                 for (stream_id = 0; stream_id < tin->streams->len;
 687                                 stream_id++) {
 688                         struct ctf_stream_declaration *stream;
 689                         int filenr;
 690
 691                         stream = g_ptr_array_index(tin->streams, stream_id);
 692                         if (!stream)
 693                                 continue;
 694                         for (filenr = 0; filenr < stream->streams->len;
 695                                         filenr++) {
 696                                 struct ctf_file_stream *file_stream;
 697
 698                                 file_stream = g_ptr_array_index(stream->streams,
 699                                                 filenr);
 700                                 if (!file_stream)
 701                                         continue;
 702                                 if (begin_pos) {
 703                                         ret = babeltrace_filestream_seek(
 704                                                         file_stream,
 705                                                         begin_pos,
 706                                                         stream_id);
 707                                 } else {
 708                                         struct bt_iter_pos pos;
 709                                         pos.type = BT_SEEK_BEGIN;
 710                                         ret = babeltrace_filestream_seek(
 711                                                         file_stream, &pos,
 712                                                         stream_id);
 713                                 }
 714                                 if (ret == EOF) {
 715                                         ret = 0;
 716                                         continue;
 717                                 } else if (ret) {
 718                                         goto error;
 719                                 }
 720                                 /* Add to heap */
 721                                 ret = heap_insert(iter->stream_heap, file_stream);
 722                                 if (ret)
 723                                         goto error;
 724                         }
 725                 }
 726         }
 727
 728         ctx->current_iterator = iter;
 729         return 0;
 730
 731 error:
 732         heap_free(iter->stream_heap);
 733 error_heap_init:
 734         g_free(iter->stream_heap);
 735         iter->stream_heap = NULL;
 736 error_ctx:
 737         return ret;
 738 }
 739
 740 struct bt_iter *bt_iter_create(struct bt_context *ctx,
 741                 const struct bt_iter_pos *begin_pos,
 742                 const struct bt_iter_pos *end_pos)
 743 {
 744         struct bt_iter *iter;
 745         int ret;
 746
 747         if (!ctx)
 748                 return NULL;
 749
 750         iter = g_new0(struct bt_iter, 1);
 751         ret = bt_iter_init(iter, ctx, begin_pos, end_pos);
 752         if (ret) {
 753                 g_free(iter);
 754                 return NULL;
 755         }
 756         return iter;
 757 }
 758
 759 void bt_iter_fini(struct bt_iter *iter)
 760 {
 761         assert(iter);
 762         if (iter->stream_heap) {
 763                 heap_free(iter->stream_heap);
 764                 g_free(iter->stream_heap);
 765         }
 766         iter->ctx->current_iterator = NULL;
 767         bt_context_put(iter->ctx);
 768 }
 769
 770 void bt_iter_destroy(struct bt_iter *iter)
 771 {
 772         assert(iter);
 773         bt_iter_fini(iter);
 774         g_free(iter);
 775 }
 776
 777 int bt_iter_next(struct bt_iter *iter)
 778 {
 779         struct ctf_file_stream *file_stream, *removed;
 780         int ret;
 781
 782         if (!iter)
 783                 return -EINVAL;
 784
 785         file_stream = heap_maximum(iter->stream_heap);
 786         if (!file_stream) {
 787                 /* end of file for all streams */
 788                 ret = 0;
 789                 goto end;
 790         }
 791
 792         ret = stream_read_event(file_stream);
 793         if (ret == EOF) {
 794                 removed = heap_remove(iter->stream_heap);
 795                 assert(removed == file_stream);
 796                 ret = 0;
 797                 goto end;
 798         } else if (ret) {
 799                 goto end;
 800         }
 801         /* Reinsert the file stream into the heap, and rebalance. */
 802         removed = heap_replace_max(iter->stream_heap, file_stream);
 803         assert(removed == file_stream);
 804
 805 end:
 806         return ret;
 807 }