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_timestamp;
  45 };
  46
  47 struct bt_saved_pos {
  48         struct trace_collection *tc;
  49         GArray *stream_saved_pos;       /* Contains struct stream_saved_pos */
  50 };
  51
  52 static int stream_read_event(struct ctf_file_stream *sin)
  53 {
  54         int ret;
  55
  56         ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
  57         if (ret == EOF)
  58                 return EOF;
  59         else if (ret) {
  60                 fprintf(stderr, "[error] Reading event failed.\n");
  61                 return ret;
  62         }
  63         return 0;
  64 }
  65
  66 /*
  67  * returns true if a < b, false otherwise.
  68  */
  69 int stream_compare(void *a, void *b)
  70 {
  71         struct ctf_file_stream *s_a = a, *s_b = b;
  72
  73         if (s_a->parent.timestamp < s_b->parent.timestamp)
  74                 return 1;
  75         else
  76                 return 0;
  77 }
  78
  79 void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
  80 {
  81         if (!iter_pos)
  82                 return;
  83
  84         if (iter_pos->u.restore) {
  85                 if (iter_pos->u.restore->stream_saved_pos) {
  86                         g_array_free(
  87                                 iter_pos->u.restore->stream_saved_pos,
  88                                 TRUE);
  89                 }
  90                 g_free(iter_pos->u.restore);
  91         }
  92         g_free(iter_pos);
  93 }
  94
  95 /*
  96  * seek_file_stream_by_timestamp
  97  *
  98  * Browse a filestream by index, if an index contains the timestamp passed in
  99  * argument, seek inside the corresponding packet it until we find the event we
 100  * are looking for (either the exact timestamp or the event just after the
 101  * timestamp).
 102  *
 103  * Return 0 if the seek succeded and EOF if we didn't find any packet
 104  * containing the timestamp.
 105  */
 106 static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
 107                 uint64_t timestamp)
 108 {
 109         struct ctf_stream_pos *stream_pos;
 110         struct packet_index *index;
 111         int i, ret;
 112
 113         stream_pos = &cfs->pos;
 114         for (i = 0; i < stream_pos->packet_index->len; i++) {
 115                 index = &g_array_index(stream_pos->packet_index,
 116                                 struct packet_index, i);
 117                 if (index->timestamp_end <= timestamp)
 118                         continue;
 119
 120                 stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
 121                 do {
 122                         ret = stream_read_event(cfs);
 123                 } while (cfs->parent.timestamp < timestamp && ret >= 0);
 124
 125                 return 0;
 126         }
 127         return EOF;
 128 }
 129
 130 /*
 131  * seek_ctf_trace_by_timestamp : for each file stream, seek to the event with
 132  * the corresponding timestamp
 133  *
 134  * Return 0 on success.
 135  * If the timestamp is not part of any file stream, return EOF to inform the
 136  * user the timestamp is out of the scope
 137  */
 138 static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin,
 139                 uint64_t timestamp, struct ptr_heap *stream_heap)
 140 {
 141         int i, j, ret;
 142         int found = EOF;
 143
 144         /* for each stream_class */
 145         for (i = 0; i < tin->streams->len; i++) {
 146                 struct ctf_stream_declaration *stream_class;
 147
 148                 stream_class = g_ptr_array_index(tin->streams, i);
 149                 /* for each file_stream */
 150                 for (j = 0; j < stream_class->streams->len; j++) {
 151                         struct ctf_stream_definition *stream;
 152                         struct ctf_file_stream *cfs;
 153
 154                         stream = g_ptr_array_index(stream_class->streams, j);
 155                         cfs = container_of(stream, struct ctf_file_stream,
 156                                         parent);
 157                         ret = seek_file_stream_by_timestamp(cfs, timestamp);
 158                         if (ret == 0) {
 159                                 /* Add to heap */
 160                                 ret = heap_insert(stream_heap, cfs);
 161                                 if (ret)
 162                                         goto error;
 163                                 found = 0;
 164                         }
 165                 }
 166         }
 167
 168         return found;
 169
 170 error:
 171         return -2;
 172 }
 173
 174 int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 175 {
 176         struct trace_collection *tc;
 177         int i, ret;
 178
 179         switch (iter_pos->type) {
 180         case BT_SEEK_RESTORE:
 181                 if (!iter_pos->u.restore)
 182                         return -EINVAL;
 183
 184                 heap_free(iter->stream_heap);
 185                 ret = heap_init(iter->stream_heap, 0, stream_compare);
 186                 if (ret < 0)
 187                         goto error;
 188
 189                 for (i = 0; i < iter_pos->u.restore->stream_saved_pos->len;
 190                                 i++) {
 191                         struct stream_saved_pos *saved_pos;
 192                         struct ctf_stream_pos *stream_pos;
 193                         struct ctf_stream_definition *stream;
 194
 195                         saved_pos = &g_array_index(
 196                                         iter_pos->u.restore->stream_saved_pos,
 197                                         struct stream_saved_pos, i);
 198                         stream = &saved_pos->file_stream->parent;
 199                         stream_pos = &saved_pos->file_stream->pos;
 200
 201                         stream_pos->packet_seek(&stream_pos->parent,
 202                                         saved_pos->cur_index, SEEK_SET);
 203
 204                         /*
 205                          * the timestamp needs to be restored after
 206                          * packet_seek, because this function resets
 207                          * the timestamp to the beginning of the packet
 208                          */
 209                         stream->timestamp = saved_pos->current_timestamp;
 210                         stream_pos->offset = saved_pos->offset;
 211                         stream_pos->last_offset = saved_pos->offset;
 212
 213                         stream->prev_timestamp = 0;
 214                         stream->prev_timestamp_end = 0;
 215                         stream->consumed = 0;
 216
 217                         printf_debug("restored to cur_index = %zd and "
 218                                 "offset = %zd, timestamp = %" PRIu64 "\n",
 219                                 stream_pos->cur_index,
 220                                 stream_pos->offset, stream->timestamp);
 221
 222                         stream_read_event(saved_pos->file_stream);
 223
 224                         /* Add to heap */
 225                         ret = heap_insert(iter->stream_heap,
 226                                         saved_pos->file_stream);
 227                         if (ret)
 228                                 goto error;
 229                 }
 230         case BT_SEEK_TIME:
 231                 tc = iter->ctx->tc;
 232
 233                 heap_free(iter->stream_heap);
 234                 ret = heap_init(iter->stream_heap, 0, stream_compare);
 235                 if (ret < 0)
 236                         goto error;
 237
 238                 /* for each trace in the trace_collection */
 239                 for (i = 0; i < tc->array->len; i++) {
 240                         struct ctf_trace *tin;
 241                         struct trace_descriptor *td_read;
 242
 243                         td_read = g_ptr_array_index(tc->array, i);
 244                         tin = container_of(td_read, struct ctf_trace, parent);
 245
 246                         ret = seek_ctf_trace_by_timestamp(tin,
 247                                         iter_pos->u.seek_time,
 248                                         iter->stream_heap);
 249                         if (ret < 0)
 250                                 goto error;
 251                 }
 252                 return 0;
 253         case BT_SEEK_BEGIN:
 254                 tc = iter->ctx->tc;
 255                 for (i = 0; i < tc->array->len; i++) {
 256                         struct ctf_trace *tin;
 257                         struct trace_descriptor *td_read;
 258                         int stream_id;
 259
 260                         td_read = g_ptr_array_index(tc->array, i);
 261                         tin = container_of(td_read, struct ctf_trace, parent);
 262
 263                         /* Populate heap with each stream */
 264                         for (stream_id = 0; stream_id < tin->streams->len;
 265                                         stream_id++) {
 266                                 struct ctf_stream_declaration *stream;
 267                                 int filenr;
 268
 269                                 stream = g_ptr_array_index(tin->streams,
 270                                                 stream_id);
 271                                 if (!stream)
 272                                         continue;
 273                                 for (filenr = 0; filenr < stream->streams->len;
 274                                                 filenr++) {
 275                                         struct ctf_file_stream *file_stream;
 276                                         file_stream = g_ptr_array_index(
 277                                                         stream->streams,
 278                                                         filenr);
 279                                         ret = babeltrace_filestream_seek(
 280                                                         file_stream, iter_pos,
 281                                                         stream_id);
 282                                 }
 283                         }
 284                 }
 285                 break;
 286         default:
 287                 /* not implemented */
 288                 return -EINVAL;
 289         }
 290
 291         return 0;
 292
 293 error:
 294         heap_free(iter->stream_heap);
 295         if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
 296                 heap_free(iter->stream_heap);
 297                 g_free(iter->stream_heap);
 298                 iter->stream_heap = NULL;
 299                 ret = -ENOMEM;
 300         }
 301         return ret;
 302 }
 303
 304 struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
 305 {
 306         struct bt_iter_pos *pos;
 307         struct trace_collection *tc = iter->ctx->tc;
 308         int i, stream_class_id, stream_id;
 309
 310         pos = g_new0(struct bt_iter_pos, 1);
 311         pos->u.restore = g_new0(struct bt_saved_pos, 1);
 312         pos->u.restore->tc = tc;
 313         pos->u.restore->stream_saved_pos = g_array_new(FALSE, TRUE,
 314                         sizeof(struct stream_saved_pos));
 315         if (!pos->u.restore->stream_saved_pos)
 316                 goto error;
 317
 318         for (i = 0; i < tc->array->len; i++) {
 319                 struct ctf_trace *tin;
 320                 struct trace_descriptor *td_read;
 321
 322                 td_read = g_ptr_array_index(tc->array, i);
 323                 tin = container_of(td_read, struct ctf_trace, parent);
 324
 325                 for (stream_class_id = 0; stream_class_id < tin->streams->len;
 326                                 stream_class_id++) {
 327                         struct ctf_stream_declaration *stream_class;
 328
 329                         stream_class = g_ptr_array_index(tin->streams,
 330                                         stream_class_id);
 331                         for (stream_id = 0;
 332                                         stream_id < stream_class->streams->len;
 333                                         stream_id++) {
 334                                 struct ctf_stream_definition *stream;
 335                                 struct ctf_file_stream *cfs;
 336                                 struct stream_saved_pos saved_pos;
 337
 338                                 stream = g_ptr_array_index(
 339                                                 stream_class->streams,
 340                                                 stream_id);
 341                                 cfs = container_of(stream,
 342                                                 struct ctf_file_stream,
 343                                                 parent);
 344
 345                                 saved_pos.file_stream = cfs;
 346                                 saved_pos.cur_index = cfs->pos.cur_index;
 347
 348                                 /*
 349                                  * It is possible that an event was read during
 350                                  * the last restore, never consumed and its
 351                                  * position saved again.  For this case, we
 352                                  * need to check if the event really was
 353                                  * consumed by the caller otherwise it is lost.
 354                                  */
 355                                 if (stream->consumed)
 356                                         saved_pos.offset = cfs->pos.offset;
 357                                 else
 358                                         saved_pos.offset = cfs->pos.last_offset;
 359
 360                                 saved_pos.current_timestamp = stream->timestamp;
 361
 362                                 g_array_append_val(
 363                                         pos->u.restore->stream_saved_pos,
 364                                         saved_pos);
 365
 366                                 printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
 367                                         "offset : %zd, "
 368                                         "timestamp = %" PRIu64 "\n",
 369                                         stream->stream_id, saved_pos.cur_index,
 370                                         saved_pos.offset,
 371                                         saved_pos.current_timestamp);
 372                         }
 373                 }
 374         }
 375
 376         return pos;
 377
 378 error:
 379         return NULL;
 380 }
 381
 382 struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *iter,
 383                 uint64_t timestamp)
 384 {
 385         struct bt_iter_pos *pos;
 386
 387         pos = g_new0(struct bt_iter_pos, 1);
 388         pos->type = BT_SEEK_TIME;
 389         pos->u.seek_time = timestamp;
 390         return pos;
 391 }
 392
 393 /*
 394  * babeltrace_filestream_seek: seek a filestream to given position.
 395  *
 396  * The stream_id parameter is only useful for BT_SEEK_RESTORE.
 397  */
 398 static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
 399                 const struct bt_iter_pos *begin_pos,
 400                 unsigned long stream_id)
 401 {
 402         int ret = 0;
 403
 404         switch (begin_pos->type) {
 405         case BT_SEEK_CUR:
 406                 /*
 407                  * just insert into the heap we should already know
 408                  * the timestamps
 409                  */
 410                 break;
 411         case BT_SEEK_BEGIN:
 412                 file_stream->pos.packet_seek(&file_stream->pos.parent,
 413                                 0, SEEK_SET);
 414                 ret = stream_read_event(file_stream);
 415                 break;
 416         case BT_SEEK_TIME:
 417         case BT_SEEK_RESTORE:
 418         case BT_SEEK_END:
 419         default:
 420                 assert(0); /* Not yet defined */
 421         }
 422
 423         return ret;
 424 }
 425
 426 /*
 427  * bt_iter_seek: seek iterator to given position.
 428  */
 429 int bt_iter_seek(struct bt_iter *iter,
 430                 const struct bt_iter_pos *begin_pos)
 431 {
 432         int i, stream_id;
 433         int ret = 0;
 434         struct trace_collection *tc = iter->ctx->tc;
 435
 436         for (i = 0; i < tc->array->len; i++) {
 437                 struct ctf_trace *tin;
 438                 struct trace_descriptor *td_read;
 439
 440                 td_read = g_ptr_array_index(tc->array, i);
 441                 tin = container_of(td_read, struct ctf_trace, parent);
 442
 443                 /* Populate heap with each stream */
 444                 for (stream_id = 0; stream_id < tin->streams->len;
 445                                 stream_id++) {
 446                         struct ctf_stream_declaration *stream;
 447                         int filenr;
 448
 449                         stream = g_ptr_array_index(tin->streams, stream_id);
 450                         for (filenr = 0; filenr < stream->streams->len;
 451                                         filenr++) {
 452                                 struct ctf_file_stream *file_stream;
 453
 454                                 file_stream = g_ptr_array_index(stream->streams,
 455                                                 filenr);
 456                                 ret = babeltrace_filestream_seek(file_stream, begin_pos,
 457                                                 stream_id);
 458                                 if (ret < 0)
 459                                         goto end;
 460                         }
 461                 }
 462         }
 463 end:
 464         return ret;
 465 }
 466
 467 int bt_iter_init(struct bt_iter *iter,
 468                 struct bt_context *ctx,
 469                 const struct bt_iter_pos *begin_pos,
 470                 const struct bt_iter_pos *end_pos)
 471 {
 472         int i, stream_id;
 473         int ret = 0;
 474
 475         if (ctx->current_iterator) {
 476                 ret = -1;
 477                 goto error_ctx;
 478         }
 479
 480         iter->stream_heap = g_new(struct ptr_heap, 1);
 481         iter->end_pos = end_pos;
 482         bt_context_get(ctx);
 483         iter->ctx = ctx;
 484
 485         ret = heap_init(iter->stream_heap, 0, stream_compare);
 486         if (ret < 0)
 487                 goto error_heap_init;
 488
 489         for (i = 0; i < ctx->tc->array->len; i++) {
 490                 struct ctf_trace *tin;
 491                 struct trace_descriptor *td_read;
 492
 493                 td_read = g_ptr_array_index(ctx->tc->array, i);
 494                 tin = container_of(td_read, struct ctf_trace, parent);
 495
 496                 /* Populate heap with each stream */
 497                 for (stream_id = 0; stream_id < tin->streams->len;
 498                                 stream_id++) {
 499                         struct ctf_stream_declaration *stream;
 500                         int filenr;
 501
 502                         stream = g_ptr_array_index(tin->streams, stream_id);
 503                         if (!stream)
 504                                 continue;
 505                         for (filenr = 0; filenr < stream->streams->len;
 506                                         filenr++) {
 507                                 struct ctf_file_stream *file_stream;
 508
 509                                 file_stream = g_ptr_array_index(stream->streams,
 510                                                 filenr);
 511
 512                                 if (begin_pos) {
 513                                         ret = babeltrace_filestream_seek(
 514                                                         file_stream,
 515                                                         begin_pos,
 516                                                         stream_id);
 517                                 } else {
 518                                         struct bt_iter_pos pos;
 519                                         pos.type = BT_SEEK_BEGIN;
 520                                         ret = babeltrace_filestream_seek(
 521                                                         file_stream, &pos,
 522                                                         stream_id);
 523                                 }
 524                                 if (ret == EOF) {
 525                                         ret = 0;
 526                                         continue;
 527                                 } else if (ret) {
 528                                         goto error;
 529                                 }
 530                                 /* Add to heap */
 531                                 ret = heap_insert(iter->stream_heap, file_stream);
 532                                 if (ret)
 533                                         goto error;
 534                         }
 535                 }
 536         }
 537
 538         ctx->current_iterator = iter;
 539         return 0;
 540
 541 error:
 542         heap_free(iter->stream_heap);
 543 error_heap_init:
 544         g_free(iter->stream_heap);
 545 error_ctx:
 546         return ret;
 547 }
 548
 549 struct bt_iter *bt_iter_create(struct bt_context *ctx,
 550                 const struct bt_iter_pos *begin_pos,
 551                 const struct bt_iter_pos *end_pos)
 552 {
 553         struct bt_iter *iter;
 554         int ret;
 555
 556         iter = g_new0(struct bt_iter, 1);
 557         ret = bt_iter_init(iter, ctx, begin_pos, end_pos);
 558         if (ret) {
 559                 g_free(iter);
 560                 return NULL;
 561         }
 562         return iter;
 563 }
 564
 565 void bt_iter_fini(struct bt_iter *iter)
 566 {
 567         if (iter->stream_heap) {
 568                 heap_free(iter->stream_heap);
 569                 g_free(iter->stream_heap);
 570         }
 571         iter->ctx->current_iterator = NULL;
 572         bt_context_put(iter->ctx);
 573 }
 574
 575 void bt_iter_destroy(struct bt_iter *iter)
 576 {
 577         bt_iter_fini(iter);
 578         g_free(iter);
 579 }
 580
 581 int bt_iter_next(struct bt_iter *iter)
 582 {
 583         struct ctf_file_stream *file_stream, *removed;
 584         int ret;
 585
 586         file_stream = heap_maximum(iter->stream_heap);
 587         if (!file_stream) {
 588                 /* end of file for all streams */
 589                 ret = 0;
 590                 goto end;
 591         }
 592
 593         ret = stream_read_event(file_stream);
 594         if (ret == EOF) {
 595                 removed = heap_remove(iter->stream_heap);
 596                 assert(removed == file_stream);
 597                 ret = 0;
 598                 goto end;
 599         } else if (ret) {
 600                 goto end;
 601         }
 602         /* Reinsert the file stream into the heap, and rebalance. */
 603         removed = heap_replace_max(iter->stream_heap, file_stream);
 604         assert(removed == file_stream);
 605
 606 end:
 607         return ret;
 608 }