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->prev_real_timestamp = 0;
 403                         stream->prev_real_timestamp_end = 0;
 404                         stream->prev_cycles_timestamp = 0;
 405                         stream->prev_cycles_timestamp_end = 0;
 406
 407                         printf_debug("restored to cur_index = %" PRId64 " and "
 408                                 "offset = %" PRId64 ", timestamp = %" PRIu64 "\n",
 409                                 stream_pos->cur_index,
 410                                 stream_pos->offset, stream->real_timestamp);
 411
 412                         ret = stream_read_event(saved_pos->file_stream);
 413                         if (ret != 0) {
 414                                 goto error;
 415                         }
 416
 417                         /* Add to heap */
 418                         ret = bt_heap_insert(iter->stream_heap,
 419                                         saved_pos->file_stream);
 420                         if (ret)
 421                                 goto error;
 422                 }
 423                 return 0;
 424         case BT_SEEK_TIME:
 425                 tc = iter->ctx->tc;
 426
 427                 bt_heap_free(iter->stream_heap);
 428                 ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 429                 if (ret < 0)
 430                         goto error_heap_init;
 431
 432                 /* for each trace in the trace_collection */
 433                 for (i = 0; i < tc->array->len; i++) {
 434                         struct ctf_trace *tin;
 435                         struct bt_trace_descriptor *td_read;
 436
 437                         td_read = g_ptr_array_index(tc->array, i);
 438                         if (!td_read)
 439                                 continue;
 440                         tin = container_of(td_read, struct ctf_trace, parent);
 441
 442                         ret = seek_ctf_trace_by_timestamp(tin,
 443                                         iter_pos->u.seek_time,
 444                                         iter->stream_heap);
 445                         /*
 446                          * Positive errors are failure. Negative value
 447                          * is EOF (for which we continue with other
 448                          * traces). 0 is success. Note: on EOF, it just
 449                          * means that no stream has been added to the
 450                          * iterator for that trace, which is fine.
 451                          */
 452                         if (ret != 0 && ret != EOF)
 453                                 goto error;
 454                 }
 455                 return 0;
 456         case BT_SEEK_BEGIN:
 457                 tc = iter->ctx->tc;
 458                 bt_heap_free(iter->stream_heap);
 459                 ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 460                 if (ret < 0)
 461                         goto error_heap_init;
 462
 463                 for (i = 0; i < tc->array->len; i++) {
 464                         struct ctf_trace *tin;
 465                         struct bt_trace_descriptor *td_read;
 466                         int stream_id;
 467
 468                         td_read = g_ptr_array_index(tc->array, i);
 469                         if (!td_read)
 470                                 continue;
 471                         tin = container_of(td_read, struct ctf_trace, parent);
 472
 473                         /* Populate heap with each stream */
 474                         for (stream_id = 0; stream_id < tin->streams->len;
 475                                         stream_id++) {
 476                                 struct ctf_stream_declaration *stream;
 477                                 int filenr;
 478
 479                                 stream = g_ptr_array_index(tin->streams,
 480                                                 stream_id);
 481                                 if (!stream)
 482                                         continue;
 483                                 for (filenr = 0; filenr < stream->streams->len;
 484                                                 filenr++) {
 485                                         struct ctf_file_stream *file_stream;
 486                                         file_stream = g_ptr_array_index(
 487                                                         stream->streams,
 488                                                         filenr);
 489                                         if (!file_stream)
 490                                                 continue;
 491                                         ret = babeltrace_filestream_seek(
 492                                                         file_stream, iter_pos,
 493                                                         stream_id);
 494                                         if (ret != 0 && ret != EOF) {
 495                                                 goto error;
 496                                         }
 497                                         if (ret == EOF) {
 498                                                 /* Do not add EOF streams */
 499                                                 continue;
 500                                         }
 501                                         ret = bt_heap_insert(iter->stream_heap, file_stream);
 502                                         if (ret)
 503                                                 goto error;
 504                                 }
 505                         }
 506                 }
 507                 break;
 508         case BT_SEEK_LAST:
 509         {
 510                 struct ctf_file_stream *cfs = NULL;
 511
 512                 tc = iter->ctx->tc;
 513                 ret = seek_last_ctf_trace_collection(tc, &cfs);
 514                 if (ret != 0 || !cfs)
 515                         goto error;
 516                 /* remove all streams from the heap */
 517                 bt_heap_free(iter->stream_heap);
 518                 /* Create a new empty heap */
 519                 ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 520                 if (ret < 0)
 521                         goto error;
 522                 /* Insert the stream that contains the last event */
 523                 ret = bt_heap_insert(iter->stream_heap, cfs);
 524                 if (ret)
 525                         goto error;
 526                 break;
 527         }
 528         default:
 529                 /* not implemented */
 530                 return -EINVAL;
 531         }
 532
 533         return 0;
 534
 535 error:
 536         bt_heap_free(iter->stream_heap);
 537 error_heap_init:
 538         if (bt_heap_init(iter->stream_heap, 0, stream_compare) < 0) {
 539                 bt_heap_free(iter->stream_heap);
 540                 g_free(iter->stream_heap);
 541                 iter->stream_heap = NULL;
 542                 ret = -ENOMEM;
 543         }
 544
 545         return ret;
 546 }
 547
 548 struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
 549 {
 550         struct bt_iter_pos *pos;
 551         struct trace_collection *tc;
 552         struct ctf_file_stream *file_stream = NULL, *removed;
 553         struct ptr_heap iter_heap_copy;
 554         int ret;
 555
 556         if (!iter)
 557                 return NULL;
 558
 559         tc = iter->ctx->tc;
 560         pos = g_new0(struct bt_iter_pos, 1);
 561         pos->type = BT_SEEK_RESTORE;
 562         pos->u.restore = g_new0(struct bt_saved_pos, 1);
 563         pos->u.restore->tc = tc;
 564         pos->u.restore->stream_saved_pos = g_array_new(FALSE, TRUE,
 565                         sizeof(struct stream_saved_pos));
 566         if (!pos->u.restore->stream_saved_pos)
 567                 goto error;
 568
 569         ret = bt_heap_copy(&iter_heap_copy, iter->stream_heap);
 570         if (ret < 0)
 571                 goto error_heap;
 572
 573         /* iterate over each stream in the heap */
 574         file_stream = bt_heap_maximum(&iter_heap_copy);
 575         while (file_stream != NULL) {
 576                 struct stream_saved_pos saved_pos;
 577
 578                 assert(file_stream->pos.last_offset != LAST_OFFSET_POISON);
 579                 saved_pos.offset = file_stream->pos.last_offset;
 580                 saved_pos.file_stream = file_stream;
 581                 saved_pos.cur_index = file_stream->pos.cur_index;
 582
 583                 saved_pos.current_real_timestamp = file_stream->parent.real_timestamp;
 584                 saved_pos.current_cycles_timestamp = file_stream->parent.cycles_timestamp;
 585
 586                 g_array_append_val(
 587                                 pos->u.restore->stream_saved_pos,
 588                                 saved_pos);
 589
 590                 printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
 591                                 "offset : %zd, "
 592                                 "timestamp = %" PRIu64 "\n",
 593                                 file_stream->parent.stream_id,
 594                                 saved_pos.cur_index, saved_pos.offset,
 595                                 saved_pos.current_real_timestamp);
 596
 597                 /* remove the stream from the heap copy */
 598                 removed = bt_heap_remove(&iter_heap_copy);
 599                 assert(removed == file_stream);
 600
 601                 file_stream = bt_heap_maximum(&iter_heap_copy);
 602         }
 603         bt_heap_free(&iter_heap_copy);
 604         return pos;
 605
 606 error_heap:
 607         g_array_free(pos->u.restore->stream_saved_pos, TRUE);
 608 error:
 609         g_free(pos);
 610         return NULL;
 611 }
 612
 613 struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *iter,
 614                 uint64_t timestamp)
 615 {
 616         struct bt_iter_pos *pos;
 617
 618         if (!iter)
 619                 return NULL;
 620
 621         pos = g_new0(struct bt_iter_pos, 1);
 622         pos->type = BT_SEEK_TIME;
 623         pos->u.seek_time = timestamp;
 624         return pos;
 625 }
 626
 627 /*
 628  * babeltrace_filestream_seek: seek a filestream to given position.
 629  *
 630  * The stream_id parameter is only useful for BT_SEEK_RESTORE.
 631  */
 632 static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
 633                 const struct bt_iter_pos *begin_pos,
 634                 unsigned long stream_id)
 635 {
 636         int ret = 0;
 637
 638         if (!file_stream || !begin_pos)
 639                 return -EINVAL;
 640
 641         switch (begin_pos->type) {
 642         case BT_SEEK_CUR:
 643                 /*
 644                  * just insert into the heap we should already know
 645                  * the timestamps
 646                  */
 647                 break;
 648         case BT_SEEK_BEGIN:
 649                 file_stream->pos.packet_seek(&file_stream->pos.parent,
 650                                 0, SEEK_SET);
 651                 ret = stream_read_event(file_stream);
 652                 break;
 653         case BT_SEEK_TIME:
 654         case BT_SEEK_RESTORE:
 655         default:
 656                 assert(0); /* Not yet defined */
 657         }
 658
 659         return ret;
 660 }
 661
 662 int bt_iter_init(struct bt_iter *iter,
 663                 struct bt_context *ctx,
 664                 const struct bt_iter_pos *begin_pos,
 665                 const struct bt_iter_pos *end_pos)
 666 {
 667         int i, stream_id;
 668         int ret = 0;
 669
 670         if (!iter || !ctx)
 671                 return -EINVAL;
 672
 673         if (ctx->current_iterator) {
 674                 ret = -1;
 675                 goto error_ctx;
 676         }
 677
 678         iter->stream_heap = g_new(struct ptr_heap, 1);
 679         iter->end_pos = end_pos;
 680         bt_context_get(ctx);
 681         iter->ctx = ctx;
 682
 683         ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 684         if (ret < 0)
 685                 goto error_heap_init;
 686
 687         for (i = 0; i < ctx->tc->array->len; i++) {
 688                 struct ctf_trace *tin;
 689                 struct bt_trace_descriptor *td_read;
 690
 691                 td_read = g_ptr_array_index(ctx->tc->array, i);
 692                 if (!td_read)
 693                         continue;
 694                 tin = container_of(td_read, struct ctf_trace, parent);
 695
 696                 /* Populate heap with each stream */
 697                 for (stream_id = 0; stream_id < tin->streams->len;
 698                                 stream_id++) {
 699                         struct ctf_stream_declaration *stream;
 700                         int filenr;
 701
 702                         stream = g_ptr_array_index(tin->streams, stream_id);
 703                         if (!stream)
 704                                 continue;
 705                         for (filenr = 0; filenr < stream->streams->len;
 706                                         filenr++) {
 707                                 struct ctf_file_stream *file_stream;
 708                                 struct bt_iter_pos pos;
 709
 710                                 file_stream = g_ptr_array_index(stream->streams,
 711                                                 filenr);
 712                                 if (!file_stream)
 713                                         continue;
 714
 715                                 pos.type = BT_SEEK_BEGIN;
 716                                 ret = babeltrace_filestream_seek(file_stream,
 717                                         &pos, stream_id);
 718
 719                                 if (ret == EOF) {
 720                                         ret = 0;
 721                                         continue;
 722                                 } else if (ret != 0 && ret != EAGAIN) {
 723                                         goto error;
 724                                 }
 725                                 /* Add to heap */
 726                                 ret = bt_heap_insert(iter->stream_heap, file_stream);
 727                                 if (ret)
 728                                         goto error;
 729                         }
 730                 }
 731         }
 732
 733         ctx->current_iterator = iter;
 734         if (begin_pos && begin_pos->type != BT_SEEK_BEGIN) {
 735                 ret = bt_iter_set_pos(iter, begin_pos);
 736         }
 737
 738         return ret;
 739
 740 error:
 741         bt_heap_free(iter->stream_heap);
 742 error_heap_init:
 743         g_free(iter->stream_heap);
 744         iter->stream_heap = NULL;
 745 error_ctx:
 746         return ret;
 747 }
 748
 749 struct bt_iter *bt_iter_create(struct bt_context *ctx,
 750                 const struct bt_iter_pos *begin_pos,
 751                 const struct bt_iter_pos *end_pos)
 752 {
 753         struct bt_iter *iter;
 754         int ret;
 755
 756         if (!ctx)
 757                 return NULL;
 758
 759         iter = g_new0(struct bt_iter, 1);
 760         ret = bt_iter_init(iter, ctx, begin_pos, end_pos);
 761         if (ret) {
 762                 g_free(iter);
 763                 return NULL;
 764         }
 765         return iter;
 766 }
 767
 768 void bt_iter_fini(struct bt_iter *iter)
 769 {
 770         assert(iter);
 771         if (iter->stream_heap) {
 772                 bt_heap_free(iter->stream_heap);
 773                 g_free(iter->stream_heap);
 774         }
 775         iter->ctx->current_iterator = NULL;
 776         bt_context_put(iter->ctx);
 777 }
 778
 779 void bt_iter_destroy(struct bt_iter *iter)
 780 {
 781         assert(iter);
 782         bt_iter_fini(iter);
 783         g_free(iter);
 784 }
 785
 786 int bt_iter_next(struct bt_iter *iter)
 787 {
 788         struct ctf_file_stream *file_stream, *removed;
 789         int ret;
 790
 791         if (!iter)
 792                 return -EINVAL;
 793
 794         file_stream = bt_heap_maximum(iter->stream_heap);
 795         if (!file_stream) {
 796                 /* end of file for all streams */
 797                 ret = 0;
 798                 goto end;
 799         }
 800
 801         ret = stream_read_event(file_stream);
 802         if (ret == EOF) {
 803                 removed = bt_heap_remove(iter->stream_heap);
 804                 assert(removed == file_stream);
 805                 ret = 0;
 806                 goto end;
 807         } else if (ret == EAGAIN) {
 808                 /*
 809                  * Live streaming: the stream is inactive for now, we
 810                  * just updated the timestamp_end to skip over this
 811                  * stream up to a certain point in time.
 812                  *
 813                  * Since we can't guarantee that a stream will ever have
 814                  * any activity, we can't rely on the fact that
 815                  * bt_iter_next will be called for each stream and deal
 816                  * with inactive streams. So instead, we return 0 here
 817                  * to the caller and let the read API handle the
 818                  * retry case.
 819                  */
 820                 ret = 0;
 821                 goto reinsert;
 822         } else if (ret) {
 823                 goto end;
 824         }
 825
 826 reinsert:
 827         /* Reinsert the file stream into the heap, and rebalance. */
 828         removed = bt_heap_replace_max(iter->stream_heap, file_stream);
 829         assert(removed == file_stream);
 830 end:
 831         return ret;
 832 }