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