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         enum bt_cb_ret (*callback)(void *private_data, void *caller_data);
  56 };
  57
  58 struct bt_callback_chain {
  59         GArray *callback;       /* Array of struct bt_callback, ordered by priority */
  60 };
  61
  62 /*
  63  * per id callbacks need to be per stream class because event ID vs
  64  * event name mapping can vary from stream to stream.
  65  */
  66 struct bt_stream_callbacks {
  67         GArray *per_id_callbacks;       /* Array of struct bt_callback_chain */
  68 };
  69
  70 /*
  71  * struct babeltrace_iter: data structure representing an iterator on a trace
  72  * collection.
  73  */
  74 struct babeltrace_iter {
  75         struct ptr_heap *stream_heap;
  76         struct trace_collection *tc;
  77         struct trace_collection_pos *end_pos;
  78         GArray *callbacks;                              /* Array of struct bt_stream_hooks */
  79         struct bt_callback_chain main_callbacks;        /* For all events */
  80         /*
  81          * Flag indicating if dependency graph needs to be recalculated.
  82          * Set by babeltrace_iter_add_callback(), and checked (and
  83          * cleared) by upon entry into babeltrace_iter_read_event().
  84          * babeltrace_iter_read_event() is responsible for calling dep
  85          * graph calculation if it sees this flag set.
  86          */
  87         int recalculate_dep_graph;
  88         /*
  89          * Array of pointers to struct bt_dependencies, for garbage
  90          * collection. We're not using a linked list here because each
  91          * struct bt_dependencies can belong to more than one
  92          * babeltrace_iter.
  93          */
  94         GPtrArray *dep_gc;
  95 };
  96
  97 struct bt_dependencies {
  98         GArray *deps;                   /* Array of GQuarks */
  99         int refcount;                   /* free when decremented to 0 */
 100 };
 101
 102 static
 103 struct bt_dependencies *_babeltrace_dependencies_create(const char *first,
 104                                                         va_list ap)
 105 {
 106         const char *iter;
 107         struct bt_dependencies *dep;
 108
 109         dep = g_new0(struct bt_dependencies, 1);
 110         dep->refcount = 1;
 111         dep->deps = g_array_new(FALSE, TRUE, sizeof(GQuark));
 112         iter = first;
 113         while (iter) {
 114                 GQuark q = g_quark_from_string(iter);
 115                 g_array_append_val(dep->deps, q);
 116                 iter = va_arg(ap, const char *);
 117         }
 118         return dep;
 119 }
 120
 121 struct bt_dependencies *babeltrace_dependencies_create(const char *first, ...)
 122 {
 123         va_list ap;
 124         struct bt_dependencies *deps;
 125
 126         va_start(ap, first);
 127         deps = _babeltrace_dependencies_create(first, ap);
 128         va_end(ap);
 129         return deps;
 130 }
 131
 132 static int stream_read_event(struct ctf_file_stream *sin)
 133 {
 134         int ret;
 135
 136         ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
 137         if (ret == EOF)
 138                 return EOF;
 139         else if (ret) {
 140                 fprintf(stdout, "[error] Reading event failed.\n");
 141                 return ret;
 142         }
 143         return 0;
 144 }
 145
 146 /*
 147  * returns true if a < b, false otherwise.
 148  */
 149 int stream_compare(void *a, void *b)
 150 {
 151         struct ctf_file_stream *s_a = a, *s_b = b;
 152
 153         if (s_a->parent.timestamp < s_b->parent.timestamp)
 154                 return 1;
 155         else
 156                 return 0;
 157 }
 158
 159 /*
 160  * babeltrace_filestream_seek: seek a filestream to given position.
 161  *
 162  * The stream_id parameter is only useful for BT_SEEK_RESTORE.
 163  */
 164 static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
 165                 const struct trace_collection_pos *begin_pos,
 166                 unsigned long stream_id)
 167 {
 168         int ret = 0;
 169
 170         switch (begin_pos->type) {
 171         case BT_SEEK_CUR:
 172                 /*
 173                  * just insert into the heap we should already know
 174                  * the timestamps
 175                  */
 176                 break;
 177         case BT_SEEK_BEGIN:
 178                 ctf_move_pos_slow(&file_stream->pos, 0, SEEK_SET);
 179                 ret = stream_read_event(file_stream);
 180                 break;
 181         case BT_SEEK_TIME:
 182         case BT_SEEK_RESTORE:
 183         case BT_SEEK_END:
 184         default:
 185                 assert(0); /* Not yet defined */
 186         }
 187
 188         return ret;
 189 }
 190
 191 /*
 192  * babeltrace_iter_seek: seek iterator to given position.
 193  */
 194 int babeltrace_iter_seek(struct babeltrace_iter *iter,
 195                 const struct trace_collection_pos *begin_pos)
 196 {
 197     int i, stream_id;
 198         int ret = 0;
 199         struct trace_collection *tc = iter->tc;
 200
 201         for (i = 0; i < tc->array->len; i++) {
 202                 struct ctf_trace *tin;
 203                 struct trace_descriptor *td_read;
 204
 205                 td_read = g_ptr_array_index(tc->array, i);
 206                 tin = container_of(td_read, struct ctf_trace, parent);
 207
 208                 /* Populate heap with each stream */
 209                 for (stream_id = 0; stream_id < tin->streams->len;
 210                                 stream_id++) {
 211                         struct ctf_stream_class *stream;
 212                         int filenr;
 213
 214                         stream = g_ptr_array_index(tin->streams, stream_id);
 215                         for (filenr = 0; filenr < stream->streams->len;
 216                                         filenr++) {
 217                                 struct ctf_file_stream *file_stream;
 218
 219                                 file_stream = g_ptr_array_index(stream->streams,
 220                                                 filenr);
 221                                 ret = babeltrace_filestream_seek(file_stream, begin_pos,
 222                                                 stream_id);
 223                                 if (ret < 0)
 224                                         goto end;
 225                         }
 226                 }
 227         }
 228 end:
 229         return ret;
 230 }
 231
 232 struct babeltrace_iter *babeltrace_iter_create(struct trace_collection *tc,
 233                 struct trace_collection_pos *begin_pos,
 234                 struct trace_collection_pos *end_pos)
 235 {
 236         int i, stream_id;
 237         int ret = 0;
 238         struct babeltrace_iter *iter;
 239
 240         iter = malloc(sizeof(struct babeltrace_iter));
 241         if (!iter)
 242                 goto error_malloc;
 243         iter->stream_heap = g_new(struct ptr_heap, 1);
 244         iter->tc = tc;
 245         iter->end_pos = end_pos;
 246
 247         ret = heap_init(iter->stream_heap, 0, stream_compare);
 248         if (ret < 0)
 249                 goto error_heap_init;
 250
 251         for (i = 0; i < tc->array->len; i++) {
 252                 struct ctf_trace *tin;
 253                 struct trace_descriptor *td_read;
 254
 255                 td_read = g_ptr_array_index(tc->array, i);
 256                 tin = container_of(td_read, struct ctf_trace, parent);
 257
 258                 /* Populate heap with each stream */
 259                 for (stream_id = 0; stream_id < tin->streams->len;
 260                                 stream_id++) {
 261                         struct ctf_stream_class *stream;
 262                         int filenr;
 263
 264                         stream = g_ptr_array_index(tin->streams, stream_id);
 265                         if (!stream)
 266                                 continue;
 267                         for (filenr = 0; filenr < stream->streams->len;
 268                                         filenr++) {
 269                                 struct ctf_file_stream *file_stream;
 270
 271                                 file_stream = g_ptr_array_index(stream->streams,
 272                                                 filenr);
 273
 274                                 if (begin_pos) {
 275                                     ret = babeltrace_filestream_seek(file_stream, begin_pos,
 276                                                         stream_id);
 277                                     if (ret == EOF) {
 278                                            ret = 0;
 279                                            continue;
 280                                     } else if (ret) {
 281                                            goto error;
 282                                     }
 283                                 }
 284                                 /* Add to heap */
 285                                 ret = heap_insert(iter->stream_heap, file_stream);
 286                                 if (ret)
 287                                         goto error;
 288                         }
 289                 }
 290         }
 291
 292         return iter;
 293
 294 error:
 295         heap_free(iter->stream_heap);
 296 error_heap_init:
 297         g_free(iter->stream_heap);
 298         free(iter);
 299 error_malloc:
 300         return NULL;
 301 }
 302
 303 void babeltrace_iter_destroy(struct babeltrace_iter *iter)
 304 {
 305         heap_free(iter->stream_heap);
 306         g_free(iter->stream_heap);
 307         free(iter);
 308 }
 309
 310 int babeltrace_iter_next(struct babeltrace_iter *iter)
 311 {
 312         struct ctf_file_stream *file_stream, *removed;
 313         int ret;
 314
 315         file_stream = heap_maximum(iter->stream_heap);
 316         if (!file_stream) {
 317                 /* end of file for all streams */
 318                 ret = 0;
 319                 goto end;
 320         }
 321
 322         ret = stream_read_event(file_stream);
 323         if (ret == EOF) {
 324                 removed = heap_remove(iter->stream_heap);
 325                 assert(removed == file_stream);
 326                 ret = 0;
 327                 goto end;
 328         } else if (ret) {
 329                 goto end;
 330         }
 331         /* Reinsert the file stream into the heap, and rebalance. */
 332         removed = heap_replace_max(iter->stream_heap, file_stream);
 333         assert(removed == file_stream);
 334
 335 end:
 336         return ret;
 337 }
 338
 339 int babeltrace_iter_read_event(struct babeltrace_iter *iter,
 340                 struct ctf_stream **stream,
 341                 struct ctf_stream_event **event)
 342 {
 343         struct ctf_file_stream *file_stream;
 344         int ret = 0;
 345
 346         file_stream = heap_maximum(iter->stream_heap);
 347         if (!file_stream) {
 348                 /* end of file for all streams */
 349                 ret = EOF;
 350                 goto end;
 351         }
 352         *stream = &file_stream->parent;
 353         *event = g_ptr_array_index((*stream)->events_by_id, (*stream)->event_id);
 354 end:
 355         return ret;
 356 }
 357
 358 int convert_trace(struct trace_descriptor *td_write,
 359                   struct trace_collection *trace_collection_read)
 360 {
 361         struct babeltrace_iter *iter;
 362         struct ctf_stream *stream;
 363         struct ctf_stream_event *event;
 364         struct ctf_text_stream_pos *sout;
 365         struct trace_collection_pos begin_pos;
 366         int ret = 0;
 367
 368         sout = container_of(td_write, struct ctf_text_stream_pos,
 369                         trace_descriptor);
 370
 371         begin_pos.type = BT_SEEK_BEGIN;
 372         iter = babeltrace_iter_create(trace_collection_read, &begin_pos, NULL);
 373         while (babeltrace_iter_read_event(iter, &stream, &event) == 0) {
 374                 ret = sout->parent.event_cb(&sout->parent, stream);
 375                 if (ret) {
 376                         fprintf(stdout, "[error] Writing event failed.\n");
 377                         goto end;
 378                 }
 379                 ret = babeltrace_iter_next(iter);
 380                 if (ret < 0)
 381                         goto end;
 382         }
 383 end:
 384         babeltrace_iter_destroy(iter);
 385         return ret;
 386 }