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/ctf/metadata.h>
  26 #include <babeltrace/iterator-internal.h>
  27 #include <babeltrace/prio_heap.h>
  28
  29 struct stream_saved_pos {
  30         /*
  31          * Use file_stream pointer to check if the trace collection we
  32          * restore to match the one we saved from, for each stream.
  33          */
  34         struct ctf_file_stream *file_stream;
  35         size_t cur_index;       /* current index in packet index */
  36         ssize_t offset;         /* offset from base, in bits. EOF for end of file. */
  37 };
  38
  39 struct babeltrace_saved_pos {
  40         struct trace_collection *tc;
  41         GArray *stream_saved_pos;       /* Contains struct stream_saved_pos */
  42 };
  43
  44 static int stream_read_event(struct ctf_file_stream *sin)
  45 {
  46         int ret;
  47
  48         ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
  49         if (ret == EOF)
  50                 return EOF;
  51         else if (ret) {
  52                 fprintf(stdout, "[error] Reading event failed.\n");
  53                 return ret;
  54         }
  55         return 0;
  56 }
  57
  58 /*
  59  * returns true if a < b, false otherwise.
  60  */
  61 int stream_compare(void *a, void *b)
  62 {
  63         struct ctf_file_stream *s_a = a, *s_b = b;
  64
  65         if (s_a->parent.timestamp < s_b->parent.timestamp)
  66                 return 1;
  67         else
  68                 return 0;
  69 }
  70
  71 /*
  72  * babeltrace_filestream_seek: seek a filestream to given position.
  73  *
  74  * The stream_id parameter is only useful for BT_SEEK_RESTORE.
  75  */
  76 static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
  77                 const struct trace_collection_pos *begin_pos,
  78                 unsigned long stream_id)
  79 {
  80         int ret = 0;
  81
  82         switch (begin_pos->type) {
  83         case BT_SEEK_CUR:
  84                 /*
  85                  * just insert into the heap we should already know
  86                  * the timestamps
  87                  */
  88                 break;
  89         case BT_SEEK_BEGIN:
  90                 file_stream->pos.move_pos_slow(&file_stream->pos, 0, SEEK_SET);
  91                 ret = stream_read_event(file_stream);
  92                 break;
  93         case BT_SEEK_TIME:
  94         case BT_SEEK_RESTORE:
  95         case BT_SEEK_END:
  96         default:
  97                 assert(0); /* Not yet defined */
  98         }
  99
 100         return ret;
 101 }
 102
 103 /*
 104  * babeltrace_iter_seek: seek iterator to given position.
 105  */
 106 int babeltrace_iter_seek(struct babeltrace_iter *iter,
 107                 const struct trace_collection_pos *begin_pos)
 108 {
 109         int i, stream_id;
 110         int ret = 0;
 111         struct trace_collection *tc = iter->ctx->tc;
 112
 113         for (i = 0; i < tc->array->len; i++) {
 114                 struct ctf_trace *tin;
 115                 struct trace_descriptor *td_read;
 116
 117                 td_read = g_ptr_array_index(tc->array, i);
 118                 tin = container_of(td_read, struct ctf_trace, parent);
 119
 120                 /* Populate heap with each stream */
 121                 for (stream_id = 0; stream_id < tin->streams->len;
 122                                 stream_id++) {
 123                         struct ctf_stream_class *stream;
 124                         int filenr;
 125
 126                         stream = g_ptr_array_index(tin->streams, stream_id);
 127                         for (filenr = 0; filenr < stream->streams->len;
 128                                         filenr++) {
 129                                 struct ctf_file_stream *file_stream;
 130
 131                                 file_stream = g_ptr_array_index(stream->streams,
 132                                                 filenr);
 133                                 ret = babeltrace_filestream_seek(file_stream, begin_pos,
 134                                                 stream_id);
 135                                 if (ret < 0)
 136                                         goto end;
 137                         }
 138                 }
 139         }
 140 end:
 141         return ret;
 142 }
 143
 144 struct babeltrace_iter *babeltrace_iter_create(struct bt_context *ctx,
 145                 struct trace_collection_pos *begin_pos,
 146                 struct trace_collection_pos *end_pos)
 147 {
 148         int i, stream_id;
 149         int ret = 0;
 150         struct babeltrace_iter *iter;
 151
 152         iter = malloc(sizeof(struct babeltrace_iter));
 153         if (!iter)
 154                 goto error_malloc;
 155         iter->stream_heap = g_new(struct ptr_heap, 1);
 156         iter->end_pos = end_pos;
 157         iter->callbacks = g_array_new(0, 1, sizeof(struct bt_stream_callbacks));
 158         iter->recalculate_dep_graph = 0;
 159         iter->main_callbacks.callback = NULL;
 160         iter->dep_gc = g_ptr_array_new();
 161         if (bt_context_get(ctx) != 0)
 162                 goto error_ctx;
 163         iter->ctx = ctx;
 164
 165         ret = heap_init(iter->stream_heap, 0, stream_compare);
 166         if (ret < 0)
 167                 goto error_heap_init;
 168
 169         for (i = 0; i < ctx->tc->array->len; i++) {
 170                 struct ctf_trace *tin;
 171                 struct trace_descriptor *td_read;
 172
 173                 td_read = g_ptr_array_index(ctx->tc->array, i);
 174                 tin = container_of(td_read, struct ctf_trace, parent);
 175
 176                 /* Populate heap with each stream */
 177                 for (stream_id = 0; stream_id < tin->streams->len;
 178                                 stream_id++) {
 179                         struct ctf_stream_class *stream;
 180                         int filenr;
 181
 182                         stream = g_ptr_array_index(tin->streams, stream_id);
 183                         if (!stream)
 184                                 continue;
 185                         for (filenr = 0; filenr < stream->streams->len;
 186                                         filenr++) {
 187                                 struct ctf_file_stream *file_stream;
 188
 189                                 file_stream = g_ptr_array_index(stream->streams,
 190                                                 filenr);
 191
 192                                 if (begin_pos) {
 193                                     ret = babeltrace_filestream_seek(file_stream, begin_pos,
 194                                                         stream_id);
 195                                     if (ret == EOF) {
 196                                            ret = 0;
 197                                            continue;
 198                                     } else if (ret) {
 199                                            goto error;
 200                                     }
 201                                 }
 202                                 /* Add to heap */
 203                                 ret = heap_insert(iter->stream_heap, file_stream);
 204                                 if (ret)
 205                                         goto error;
 206                         }
 207                 }
 208         }
 209
 210         return iter;
 211
 212 error:
 213         heap_free(iter->stream_heap);
 214 error_heap_init:
 215         g_free(iter->stream_heap);
 216 error_ctx:
 217         free(iter);
 218 error_malloc:
 219         return NULL;
 220 }
 221
 222 void babeltrace_iter_destroy(struct babeltrace_iter *iter)
 223 {
 224         struct bt_stream_callbacks *bt_stream_cb;
 225         struct bt_callback_chain *bt_chain;
 226         int i, j;
 227
 228         heap_free(iter->stream_heap);
 229         g_free(iter->stream_heap);
 230
 231         /* free all events callbacks */
 232         if (iter->main_callbacks.callback)
 233                 g_array_free(iter->main_callbacks.callback, TRUE);
 234
 235         /* free per-event callbacks */
 236         for (i = 0; i < iter->callbacks->len; i++) {
 237                 bt_stream_cb = &g_array_index(iter->callbacks,
 238                                 struct bt_stream_callbacks, i);
 239                 if (!bt_stream_cb || !bt_stream_cb->per_id_callbacks)
 240                         continue;
 241                 for (j = 0; j < bt_stream_cb->per_id_callbacks->len; j++) {
 242                         bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
 243                                         struct bt_callback_chain, j);
 244                         if (bt_chain->callback) {
 245                                 g_array_free(bt_chain->callback, TRUE);
 246                         }
 247                 }
 248                 g_array_free(bt_stream_cb->per_id_callbacks, TRUE);
 249         }
 250
 251         bt_context_put(iter->ctx);
 252
 253         free(iter);
 254 }
 255
 256 int babeltrace_iter_next(struct babeltrace_iter *iter)
 257 {
 258         struct ctf_file_stream *file_stream, *removed;
 259         int ret;
 260
 261         file_stream = heap_maximum(iter->stream_heap);
 262         if (!file_stream) {
 263                 /* end of file for all streams */
 264                 ret = 0;
 265                 goto end;
 266         }
 267
 268         ret = stream_read_event(file_stream);
 269         if (ret == EOF) {
 270                 removed = heap_remove(iter->stream_heap);
 271                 assert(removed == file_stream);
 272                 ret = 0;
 273                 goto end;
 274         } else if (ret) {
 275                 goto end;
 276         }
 277         /* Reinsert the file stream into the heap, and rebalance. */
 278         removed = heap_replace_max(iter->stream_heap, file_stream);
 279         assert(removed == file_stream);
 280
 281 end:
 282         return ret;
 283 }
 284
 285 int babeltrace_iter_read_event(struct babeltrace_iter *iter,
 286                 struct ctf_stream **stream,
 287                 struct ctf_stream_event **event)
 288 {
 289         struct ctf_file_stream *file_stream;
 290         int ret = 0;
 291
 292         file_stream = heap_maximum(iter->stream_heap);
 293         if (!file_stream) {
 294                 /* end of file for all streams */
 295                 ret = EOF;
 296                 goto end;
 297         }
 298         *stream = &file_stream->parent;
 299         *event = g_ptr_array_index((*stream)->events_by_id, (*stream)->event_id);
 300
 301         if ((*stream)->stream_id > iter->callbacks->len)
 302                 goto end;
 303
 304         process_callbacks(iter, *stream);
 305
 306 end:
 307         return ret;
 308 }