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