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