#include <babeltrace/types.h>
#include <babeltrace/ctf/types.h>
#include <babeltrace/ctf-ir/metadata.h>
+#include <stdarg.h>
+
+struct stream_saved_pos {
+ /*
+ * Use file_stream pointer to check if the trace collection we
+ * restore to match the one we saved from, for each stream.
+ */
+ struct ctf_file_stream *file_stream;
+ size_t cur_index; /* current index in packet index */
+ ssize_t offset; /* offset from base, in bits. EOF for end of file. */
+};
+
+struct babeltrace_saved_pos {
+ struct trace_collection *tc;
+ GArray *stream_saved_pos; /* Contains struct stream_saved_pos */
+};
+
+struct bt_callback {
+ int prio; /* Callback order priority. Lower first. Dynamically assigned from dependency graph. */
+ void *private_data;
+ enum bt_cb_ret (*callback)(void *private_data, void *caller_data);
+};
+
+struct bt_callback_chain {
+ GArray *callback; /* Array of struct bt_callback, ordered by priority */
+};
+
+/*
+ * per id callbacks need to be per stream class because event ID vs
+ * event name mapping can vary from stream to stream.
+ */
+struct bt_stream_callbacks {
+ GArray *per_id_callbacks; /* Array of struct bt_callback_chain */
+};
/*
* struct babeltrace_iter: data structure representing an iterator on a trace
struct babeltrace_iter {
struct ptr_heap *stream_heap;
struct trace_collection *tc;
+ struct trace_collection_pos *end_pos;
+ GArray *callbacks; /* Array of struct bt_stream_hooks */
+ struct bt_callback_chain main_callbacks; /* For all events */
+ /*
+ * Flag indicating if dependency graph needs to be recalculated.
+ * Set by babeltrace_iter_add_callback(), and checked (and
+ * cleared) by upon entry into babeltrace_iter_read_event().
+ * babeltrace_iter_read_event() is responsible for calling dep
+ * graph calculation if it sees this flag set.
+ */
+ int recalculate_dep_graph;
+ /*
+ * Array of pointers to struct bt_dependencies, for garbage
+ * collection. We're not using a linked list here because each
+ * struct bt_dependencies can belong to more than one
+ * babeltrace_iter.
+ */
+ GPtrArray *dep_gc;
};
+struct bt_dependencies {
+ GArray *deps; /* Array of GQuarks */
+ int refcount; /* free when decremented to 0 */
+};
+
+static
+struct bt_dependencies *_babeltrace_dependencies_create(const char *first,
+ va_list ap)
+{
+ const char *iter;
+ struct bt_dependencies *dep;
+
+ dep = g_new0(struct bt_dependencies, 1);
+ dep->refcount = 1;
+ dep->deps = g_array_new(FALSE, TRUE, sizeof(GQuark));
+ iter = first;
+ while (iter) {
+ GQuark q = g_quark_from_string(iter);
+ g_array_append_val(dep->deps, q);
+ iter = va_arg(ap, const char *);
+ }
+ return dep;
+}
+
+struct bt_dependencies *babeltrace_dependencies_create(const char *first, ...)
+{
+ va_list ap;
+ struct bt_dependencies *deps;
+
+ va_start(ap, first);
+ deps = _babeltrace_dependencies_create(first, ap);
+ va_end(ap);
+ return deps;
+}
+
static int stream_read_event(struct ctf_file_stream *sin)
{
int ret;
return 0;
}
-struct babeltrace_iter *babeltrace_iter_create(struct trace_collection *tc)
+/*
+ * babeltrace_filestream_seek: seek a filestream to given position.
+ *
+ * The stream_id parameter is only useful for BT_SEEK_RESTORE.
+ */
+static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
+ const struct trace_collection_pos *begin_pos,
+ unsigned long stream_id)
+{
+ int ret = 0;
+
+ switch (begin_pos->type) {
+ case BT_SEEK_CUR:
+ /*
+ * just insert into the heap we should already know
+ * the timestamps
+ */
+ break;
+ case BT_SEEK_BEGIN:
+ ctf_move_pos_slow(&file_stream->pos, 0, SEEK_SET);
+ ret = stream_read_event(file_stream);
+ break;
+ case BT_SEEK_TIME:
+ case BT_SEEK_RESTORE:
+ case BT_SEEK_END:
+ default:
+ assert(0); /* Not yet defined */
+ }
+
+ return ret;
+}
+
+/*
+ * babeltrace_iter_seek: seek iterator to given position.
+ */
+int babeltrace_iter_seek(struct babeltrace_iter *iter,
+ const struct trace_collection_pos *begin_pos)
+{
+ int i, stream_id;
+ int ret = 0;
+ struct trace_collection *tc = iter->tc;
+
+ for (i = 0; i < tc->array->len; i++) {
+ struct ctf_trace *tin;
+ struct trace_descriptor *td_read;
+
+ td_read = g_ptr_array_index(tc->array, i);
+ tin = container_of(td_read, struct ctf_trace, parent);
+
+ /* Populate heap with each stream */
+ for (stream_id = 0; stream_id < tin->streams->len;
+ stream_id++) {
+ struct ctf_stream_class *stream;
+ int filenr;
+
+ stream = g_ptr_array_index(tin->streams, stream_id);
+ for (filenr = 0; filenr < stream->streams->len;
+ filenr++) {
+ struct ctf_file_stream *file_stream;
+
+ file_stream = g_ptr_array_index(stream->streams,
+ filenr);
+ ret = babeltrace_filestream_seek(file_stream, begin_pos,
+ stream_id);
+ if (ret < 0)
+ goto end;
+ }
+ }
+ }
+end:
+ return ret;
+}
+
+struct babeltrace_iter *babeltrace_iter_create(struct trace_collection *tc,
+ struct trace_collection_pos *begin_pos,
+ struct trace_collection_pos *end_pos)
{
int i, stream_id;
int ret = 0;
goto error_malloc;
iter->stream_heap = g_new(struct ptr_heap, 1);
iter->tc = tc;
+ iter->end_pos = end_pos;
ret = heap_init(iter->stream_heap, 0, stream_compare);
if (ret < 0)
file_stream = g_ptr_array_index(stream->streams,
filenr);
- ret = stream_read_event(file_stream);
- if (ret == EOF) {
- ret = 0;
- continue;
- } else if (ret) {
- goto error;
+ if (begin_pos) {
+ ret = babeltrace_filestream_seek(file_stream, begin_pos,
+ stream_id);
+ if (ret == EOF) {
+ ret = 0;
+ continue;
+ } else if (ret) {
+ goto error;
+ }
}
/* Add to heap */
ret = heap_insert(iter->stream_heap, file_stream);
struct ctf_stream *stream;
struct ctf_stream_event *event;
struct ctf_text_stream_pos *sout;
+ struct trace_collection_pos begin_pos;
int ret = 0;
sout = container_of(td_write, struct ctf_text_stream_pos,
trace_descriptor);
- iter = babeltrace_iter_create(trace_collection_read);
+ begin_pos.type = BT_SEEK_BEGIN;
+ iter = babeltrace_iter_create(trace_collection_read, &begin_pos, NULL);
while (babeltrace_iter_read_event(iter, &stream, &event) == 0) {
ret = sout->parent.event_cb(&sout->parent, stream);
if (ret) {
projects / babeltrace.git / blobdiff