converter/babeltrace-lib.c

   1 /*
   2  * babeltrace-lib.c
   3  *
   4  * Babeltrace Trace Converter 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 <errno.h>
  23 #include <stdio.h>
  24 #include <inttypes.h>
  25 #include <babeltrace/babeltrace-internal.h>
  26 #include <babeltrace/format.h>
  27 #include <babeltrace/ctf/types.h>
  28 #include <babeltrace/ctf/metadata.h>
  29 #include <babeltrace/ctf-text/types.h>
  30 #include <babeltrace/prio_heap.h>
  31 #include <babeltrace/babeltrace.h>
  32 #include <babeltrace/types.h>
  33 #include <babeltrace/ctf/types.h>
  34 #include <babeltrace/ctf-ir/metadata.h>
  35 #include <stdarg.h>
  36
  37 struct stream_saved_pos {
  38         /*
  39          * Use file_stream pointer to check if the trace collection we
  40          * restore to match the one we saved from, for each stream.
  41          */
  42         struct ctf_file_stream *file_stream;
  43         size_t cur_index;       /* current index in packet index */
  44         ssize_t offset;         /* offset from base, in bits. EOF for end of file. */
  45 };
  46
  47 struct babeltrace_saved_pos {
  48         struct trace_collection *tc;
  49         GArray *stream_saved_pos;       /* Contains struct stream_saved_pos */
  50 };
  51
  52 struct bt_callback {
  53         int prio;               /* Callback order priority. Lower first. Dynamically assigned from dependency graph. */
  54         void *private_data;
  55         int flags;
  56         struct bt_dependencies *depends;
  57         struct bt_dependencies *weak_depends;
  58         struct bt_dependencies *provides;
  59         enum bt_cb_ret (*callback)(struct bt_ctf_data *ctf_data,
  60                                    void *private_data);
  61 };
  62
  63 struct bt_callback_chain {
  64         GArray *callback;       /* Array of struct bt_callback, ordered by priority */
  65 };
  66
  67 /*
  68  * per id callbacks need to be per stream class because event ID vs
  69  * event name mapping can vary from stream to stream.
  70  */
  71 struct bt_stream_callbacks {
  72         GArray *per_id_callbacks;       /* Array of struct bt_callback_chain */
  73 };
  74
  75 /*
  76  * struct babeltrace_iter: data structure representing an iterator on a trace
  77  * collection.
  78  */
  79 struct babeltrace_iter {
  80         struct ptr_heap *stream_heap;
  81         struct trace_collection *tc;
  82         struct trace_collection_pos *end_pos;
  83         GArray *callbacks;                              /* Array of struct bt_stream_callbacks */
  84         struct bt_callback_chain main_callbacks;        /* For all events */
  85         /*
  86          * Flag indicating if dependency graph needs to be recalculated.
  87          * Set by babeltrace_iter_add_callback(), and checked (and
  88          * cleared) by upon entry into babeltrace_iter_read_event().
  89          * babeltrace_iter_read_event() is responsible for calling dep
  90          * graph calculation if it sees this flag set.
  91          */
  92         int recalculate_dep_graph;
  93         /*
  94          * Array of pointers to struct bt_dependencies, for garbage
  95          * collection. We're not using a linked list here because each
  96          * struct bt_dependencies can belong to more than one
  97          * babeltrace_iter.
  98          */
  99         GPtrArray *dep_gc;
 100 };
 101
 102 struct bt_dependencies {
 103         GArray *deps;                   /* Array of GQuarks */
 104         int refcount;                   /* free when decremented to 0 */
 105 };
 106
 107 static
 108 struct bt_dependencies *_babeltrace_dependencies_create(const char *first,
 109                                                         va_list ap)
 110 {
 111         const char *iter;
 112         struct bt_dependencies *dep;
 113
 114         dep = g_new0(struct bt_dependencies, 1);
 115         dep->refcount = 1;
 116         dep->deps = g_array_new(FALSE, TRUE, sizeof(GQuark));
 117         iter = first;
 118         while (iter) {
 119                 GQuark q = g_quark_from_string(iter);
 120                 g_array_append_val(dep->deps, q);
 121                 iter = va_arg(ap, const char *);
 122         }
 123         return dep;
 124 }
 125
 126 struct bt_dependencies *babeltrace_dependencies_create(const char *first, ...)
 127 {
 128         va_list ap;
 129         struct bt_dependencies *deps;
 130
 131         va_start(ap, first);
 132         deps = _babeltrace_dependencies_create(first, ap);
 133         va_end(ap);
 134         return deps;
 135 }
 136
 137 /*
 138  * babeltrace_iter_add_callback: Add a callback to iterator.
 139  */
 140 int babeltrace_iter_add_callback(struct babeltrace_iter *iter,
 141                 bt_event_name event, void *private_data, int flags,
 142                 enum bt_cb_ret (*callback)(struct bt_ctf_data *ctf_data,
 143                                            void *private_data),
 144                 struct bt_dependencies *depends,
 145                 struct bt_dependencies *weak_depends,
 146                 struct bt_dependencies *provides)
 147 {
 148         int i, stream_id;
 149         gpointer *event_id_ptr;
 150         unsigned long event_id;
 151         struct trace_collection *tc = iter->tc;
 152
 153         for (i = 0; i < tc->array->len; i++) {
 154                 struct ctf_trace *tin;
 155                 struct trace_descriptor *td_read;
 156
 157                 td_read = g_ptr_array_index(tc->array, i);
 158                 tin = container_of(td_read, struct ctf_trace, parent);
 159
 160                 for (stream_id = 0; stream_id < tin->streams->len; stream_id++) {
 161                         struct ctf_stream_class *stream;
 162                         struct bt_stream_callbacks *bt_stream_cb = NULL;
 163                         struct bt_callback_chain *bt_chain = NULL;
 164                         struct bt_callback new_callback;
 165
 166                         stream = g_ptr_array_index(tin->streams, stream_id);
 167
 168                         /* find or create the bt_stream_callbacks for this stream */
 169                         if (iter->callbacks->len >= stream_id) {
 170                                 bt_stream_cb = &g_array_index(iter->callbacks,
 171                                                 struct bt_stream_callbacks, stream->stream_id);
 172                         } else {
 173                                 g_array_set_size(iter->callbacks, stream->stream_id);
 174                         }
 175                         if (!bt_stream_cb || !bt_stream_cb->per_id_callbacks) {
 176                                 struct bt_stream_callbacks new_stream_cb;
 177                                 new_stream_cb.per_id_callbacks = g_array_new(1, 1,
 178                                                 sizeof(struct bt_callback_chain));
 179                                 g_array_insert_val(iter->callbacks, stream->stream_id, new_stream_cb);
 180                                 bt_stream_cb = &g_array_index(iter->callbacks,
 181                                                 struct bt_stream_callbacks, stream->stream_id);
 182                         }
 183
 184                         if (event) {
 185                                 /* find the event id */
 186                                 event_id_ptr = g_hash_table_lookup(stream->event_quark_to_id,
 187                                                 (gconstpointer) (unsigned long) event);
 188                                 /* event not found in this stream class */
 189                                 if (!event_id_ptr) {
 190                                         printf("event not found\n");
 191                                         continue;
 192                                 }
 193                                 event_id = (uint64_t)(unsigned long)*event_id_ptr;
 194
 195                                 /* find or create the bt_callback_chain for this event */
 196                                 if (bt_stream_cb->per_id_callbacks->len >= event_id) {
 197                                         bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
 198                                                         struct bt_callback_chain, event_id);
 199                                 } else {
 200                                         g_array_set_size(bt_stream_cb->per_id_callbacks, event_id);
 201                                 }
 202                                 if (!bt_chain || !bt_chain->callback) {
 203                                         struct bt_callback_chain new_chain;
 204                                         new_chain.callback = g_array_new(1, 1, sizeof(struct bt_callback));
 205                                         g_array_insert_val(bt_stream_cb->per_id_callbacks, event_id,
 206                                                         new_chain);
 207                                         bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
 208                                                         struct bt_callback_chain, event_id);
 209                                 }
 210                         } else {
 211                                 /* callback for all events */
 212                                 if (!iter->main_callbacks.callback) {
 213                                         iter->main_callbacks.callback = g_array_new(1, 1,
 214                                                         sizeof(struct bt_callback));
 215                                 }
 216                                 bt_chain = &iter->main_callbacks;
 217                         }
 218
 219                         new_callback.private_data = private_data;
 220                         new_callback.flags = flags;
 221                         new_callback.callback = callback;
 222                         new_callback.depends = depends;
 223                         new_callback.weak_depends = weak_depends;
 224                         new_callback.provides = provides;
 225
 226                         /* TODO : take care of priority, for now just FIFO */
 227                         g_array_append_val(bt_chain->callback, new_callback);
 228                 }
 229         }
 230
 231         return 0;
 232 }
 233
 234 static int stream_read_event(struct ctf_file_stream *sin)
 235 {
 236         int ret;
 237
 238         ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
 239         if (ret == EOF)
 240                 return EOF;
 241         else if (ret) {
 242                 fprintf(stdout, "[error] Reading event failed.\n");
 243                 return ret;
 244         }
 245         return 0;
 246 }
 247
 248 /*
 249  * returns true if a < b, false otherwise.
 250  */
 251 int stream_compare(void *a, void *b)
 252 {
 253         struct ctf_file_stream *s_a = a, *s_b = b;
 254
 255         if (s_a->parent.timestamp < s_b->parent.timestamp)
 256                 return 1;
 257         else
 258                 return 0;
 259 }
 260
 261 /*
 262  * babeltrace_filestream_seek: seek a filestream to given position.
 263  *
 264  * The stream_id parameter is only useful for BT_SEEK_RESTORE.
 265  */
 266 static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
 267                 const struct trace_collection_pos *begin_pos,
 268                 unsigned long stream_id)
 269 {
 270         int ret = 0;
 271
 272         switch (begin_pos->type) {
 273         case BT_SEEK_CUR:
 274                 /*
 275                  * just insert into the heap we should already know
 276                  * the timestamps
 277                  */
 278                 break;
 279         case BT_SEEK_BEGIN:
 280                 ctf_move_pos_slow(&file_stream->pos, 0, SEEK_SET);
 281                 ret = stream_read_event(file_stream);
 282                 break;
 283         case BT_SEEK_TIME:
 284         case BT_SEEK_RESTORE:
 285         case BT_SEEK_END:
 286         default:
 287                 assert(0); /* Not yet defined */
 288         }
 289
 290         return ret;
 291 }
 292
 293 /*
 294  * babeltrace_iter_seek: seek iterator to given position.
 295  */
 296 int babeltrace_iter_seek(struct babeltrace_iter *iter,
 297                 const struct trace_collection_pos *begin_pos)
 298 {
 299     int i, stream_id;
 300         int ret = 0;
 301         struct trace_collection *tc = iter->tc;
 302
 303         for (i = 0; i < tc->array->len; i++) {
 304                 struct ctf_trace *tin;
 305                 struct trace_descriptor *td_read;
 306
 307                 td_read = g_ptr_array_index(tc->array, i);
 308                 tin = container_of(td_read, struct ctf_trace, parent);
 309
 310                 /* Populate heap with each stream */
 311                 for (stream_id = 0; stream_id < tin->streams->len;
 312                                 stream_id++) {
 313                         struct ctf_stream_class *stream;
 314                         int filenr;
 315
 316                         stream = g_ptr_array_index(tin->streams, stream_id);
 317                         for (filenr = 0; filenr < stream->streams->len;
 318                                         filenr++) {
 319                                 struct ctf_file_stream *file_stream;
 320
 321                                 file_stream = g_ptr_array_index(stream->streams,
 322                                                 filenr);
 323                                 ret = babeltrace_filestream_seek(file_stream, begin_pos,
 324                                                 stream_id);
 325                                 if (ret < 0)
 326                                         goto end;
 327                         }
 328                 }
 329         }
 330 end:
 331         return ret;
 332 }
 333
 334 struct babeltrace_iter *babeltrace_iter_create(struct trace_collection *tc,
 335                 struct trace_collection_pos *begin_pos,
 336                 struct trace_collection_pos *end_pos)
 337 {
 338         int i, stream_id;
 339         int ret = 0;
 340         struct babeltrace_iter *iter;
 341
 342         iter = malloc(sizeof(struct babeltrace_iter));
 343         if (!iter)
 344                 goto error_malloc;
 345         iter->stream_heap = g_new(struct ptr_heap, 1);
 346         iter->tc = tc;
 347         iter->end_pos = end_pos;
 348         iter->callbacks = g_array_new(0, 1, sizeof(struct bt_stream_callbacks));
 349         iter->recalculate_dep_graph = 0;
 350         iter->main_callbacks.callback = NULL;
 351         iter->dep_gc = g_ptr_array_new();
 352
 353         ret = heap_init(iter->stream_heap, 0, stream_compare);
 354         if (ret < 0)
 355                 goto error_heap_init;
 356
 357         for (i = 0; i < tc->array->len; i++) {
 358                 struct ctf_trace *tin;
 359                 struct trace_descriptor *td_read;
 360
 361                 td_read = g_ptr_array_index(tc->array, i);
 362                 tin = container_of(td_read, struct ctf_trace, parent);
 363
 364                 /* Populate heap with each stream */
 365                 for (stream_id = 0; stream_id < tin->streams->len;
 366                                 stream_id++) {
 367                         struct ctf_stream_class *stream;
 368                         int filenr;
 369
 370                         stream = g_ptr_array_index(tin->streams, stream_id);
 371                         if (!stream)
 372                                 continue;
 373                         for (filenr = 0; filenr < stream->streams->len;
 374                                         filenr++) {
 375                                 struct ctf_file_stream *file_stream;
 376
 377                                 file_stream = g_ptr_array_index(stream->streams,
 378                                                 filenr);
 379
 380                                 if (begin_pos) {
 381                                     ret = babeltrace_filestream_seek(file_stream, begin_pos,
 382                                                         stream_id);
 383                                     if (ret == EOF) {
 384                                            ret = 0;
 385                                            continue;
 386                                     } else if (ret) {
 387                                            goto error;
 388                                     }
 389                                 }
 390                                 /* Add to heap */
 391                                 ret = heap_insert(iter->stream_heap, file_stream);
 392                                 if (ret)
 393                                         goto error;
 394                         }
 395                 }
 396         }
 397
 398         return iter;
 399
 400 error:
 401         heap_free(iter->stream_heap);
 402 error_heap_init:
 403         g_free(iter->stream_heap);
 404         free(iter);
 405 error_malloc:
 406         return NULL;
 407 }
 408
 409 void babeltrace_iter_destroy(struct babeltrace_iter *iter)
 410 {
 411         struct bt_stream_callbacks *bt_stream_cb;
 412         struct bt_callback_chain *bt_chain;
 413         int i, j;
 414
 415         heap_free(iter->stream_heap);
 416         g_free(iter->stream_heap);
 417
 418         /* free all events callbacks */
 419         if (iter->main_callbacks.callback)
 420                 g_array_free(iter->main_callbacks.callback, TRUE);
 421
 422         /* free per-event callbacks */
 423         for (i = 0; i < iter->callbacks->len; i++) {
 424                 bt_stream_cb = &g_array_index(iter->callbacks,
 425                                 struct bt_stream_callbacks, i);
 426                 if (!bt_stream_cb || !bt_stream_cb->per_id_callbacks)
 427                         continue;
 428                 for (j = 0; j < bt_stream_cb->per_id_callbacks->len; j++) {
 429                         bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
 430                                         struct bt_callback_chain, j);
 431                         if (bt_chain->callback) {
 432                                 g_array_free(bt_chain->callback, TRUE);
 433                         }
 434                 }
 435                 g_array_free(bt_stream_cb->per_id_callbacks, TRUE);
 436         }
 437
 438         free(iter);
 439 }
 440
 441 int babeltrace_iter_next(struct babeltrace_iter *iter)
 442 {
 443         struct ctf_file_stream *file_stream, *removed;
 444         int ret;
 445
 446         file_stream = heap_maximum(iter->stream_heap);
 447         if (!file_stream) {
 448                 /* end of file for all streams */
 449                 ret = 0;
 450                 goto end;
 451         }
 452
 453         ret = stream_read_event(file_stream);
 454         if (ret == EOF) {
 455                 removed = heap_remove(iter->stream_heap);
 456                 assert(removed == file_stream);
 457                 ret = 0;
 458                 goto end;
 459         } else if (ret) {
 460                 goto end;
 461         }
 462         /* Reinsert the file stream into the heap, and rebalance. */
 463         removed = heap_replace_max(iter->stream_heap, file_stream);
 464         assert(removed == file_stream);
 465
 466 end:
 467         return ret;
 468 }
 469
 470 static
 471 struct ctf_stream_event *extract_ctf_stream_event(struct ctf_stream *stream)
 472 {
 473         struct ctf_stream_class *stream_class = stream->stream_class;
 474         struct ctf_event *event_class;
 475         struct ctf_stream_event *event;
 476         uint64_t id = stream->event_id;
 477
 478         if (id >= stream_class->events_by_id->len) {
 479                 fprintf(stdout, "[error] Event id %" PRIu64 " is outside range.\n", id);
 480                 return NULL;
 481         }
 482         event = g_ptr_array_index(stream->events_by_id, id);
 483         if (!event) {
 484                 fprintf(stdout, "[error] Event id %" PRIu64 " is unknown.\n", id);
 485                 return NULL;
 486         }
 487         event_class = g_ptr_array_index(stream_class->events_by_id, id);
 488         if (!event_class) {
 489                 fprintf(stdout, "[error] Event id %" PRIu64 " is unknown.\n", id);
 490                 return NULL;
 491         }
 492
 493         return event;
 494 }
 495
 496 static
 497 void process_callbacks(struct babeltrace_iter *iter,
 498                        struct ctf_stream *stream)
 499 {
 500         struct bt_stream_callbacks *bt_stream_cb;
 501         struct bt_callback_chain *bt_chain;
 502         struct bt_callback *cb;
 503         int i;
 504         enum bt_cb_ret ret;
 505         struct bt_ctf_data ctf_data;
 506
 507         ctf_data.event = extract_ctf_stream_event(stream);
 508
 509         /* process all events callback first */
 510         if (iter->main_callbacks.callback) {
 511                 for (i = 0; i < iter->main_callbacks.callback->len; i++) {
 512                         cb = &g_array_index(iter->main_callbacks.callback, struct bt_callback, i);
 513                         if (!cb)
 514                                 goto end;
 515                         ret = cb->callback(&ctf_data, cb->private_data);
 516                         switch (ret) {
 517                         case BT_CB_OK_STOP:
 518                         case BT_CB_ERROR_STOP:
 519                                 goto end;
 520                         default:
 521                                 break;
 522                         }
 523                 }
 524         }
 525
 526         /* process per event callbacks */
 527         bt_stream_cb = &g_array_index(iter->callbacks,
 528                         struct bt_stream_callbacks, stream->stream_id);
 529         if (!bt_stream_cb || !bt_stream_cb->per_id_callbacks)
 530                 goto end;
 531
 532         if (stream->event_id > bt_stream_cb->per_id_callbacks->len)
 533                 goto end;
 534         bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
 535                         struct bt_callback_chain, stream->event_id);
 536         if (!bt_chain || !bt_chain->callback)
 537                 goto end;
 538
 539         for (i = 0; i < bt_chain->callback->len; i++) {
 540                 cb = &g_array_index(bt_chain->callback, struct bt_callback, i);
 541                 if (!cb)
 542                         goto end;
 543                 ret = cb->callback(&ctf_data, cb->private_data);
 544                 switch (ret) {
 545                 case BT_CB_OK_STOP:
 546                 case BT_CB_ERROR_STOP:
 547                         goto end;
 548                 default:
 549                         break;
 550                 }
 551         }
 552
 553 end:
 554         return;
 555 }
 556
 557 int babeltrace_iter_read_event(struct babeltrace_iter *iter,
 558                 struct ctf_stream **stream,
 559                 struct ctf_stream_event **event)
 560 {
 561         struct ctf_file_stream *file_stream;
 562         int ret = 0;
 563
 564         file_stream = heap_maximum(iter->stream_heap);
 565         if (!file_stream) {
 566                 /* end of file for all streams */
 567                 ret = EOF;
 568                 goto end;
 569         }
 570         *stream = &file_stream->parent;
 571         *event = g_ptr_array_index((*stream)->events_by_id, (*stream)->event_id);
 572
 573         if ((*stream)->stream_id > iter->callbacks->len)
 574                 goto end;
 575
 576         process_callbacks(iter, *stream);
 577
 578 end:
 579         return ret;
 580 }
 581
 582 int convert_trace(struct trace_descriptor *td_write,
 583                   struct trace_collection *trace_collection_read)
 584 {
 585         struct babeltrace_iter *iter;
 586         struct ctf_stream *stream;
 587         struct ctf_stream_event *event;
 588         struct ctf_text_stream_pos *sout;
 589         struct trace_collection_pos begin_pos;
 590         int ret = 0;
 591
 592         sout = container_of(td_write, struct ctf_text_stream_pos,
 593                         trace_descriptor);
 594
 595         begin_pos.type = BT_SEEK_BEGIN;
 596         iter = babeltrace_iter_create(trace_collection_read, &begin_pos, NULL);
 597         while (babeltrace_iter_read_event(iter, &stream, &event) == 0) {
 598                 ret = sout->parent.event_cb(&sout->parent, stream);
 599                 if (ret) {
 600                         fprintf(stdout, "[error] Writing event failed.\n");
 601                         goto end;
 602                 }
 603                 ret = babeltrace_iter_next(iter);
 604                 if (ret < 0)
 605                         goto end;
 606         }
 607 end:
 608         babeltrace_iter_destroy(iter);
 609         return ret;
 610 }