Merge real index and cycles index into a single index

[babeltrace.git] / lib / iterator.c

diff --git a/lib/iterator.c b/lib/iterator.c
index dc12a2f41020d34fa4903bc61494bddb092e53c7..fd58dec99ce45dc4e8a682220d20ebf3a6edb3dc 100644 (file)
--- a/lib/iterator.c
+++ b/lib/iterator.c
@@ -65,6 +65,9 @@ static int stream_read_event(struct ctf_file_stream *sin)
        ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
        if (ret == EOF)
                return EOF;
+       else if (ret == EAGAIN)
+               /* Stream is inactive for now (live reading). */
+               return EAGAIN;
        else if (ret) {
                fprintf(stderr, "[error] Reading event failed.\n");
                return ret;
@@ -73,16 +76,24 @@ static int stream_read_event(struct ctf_file_stream *sin)
 }
 
 /*
- * returns true if a < b, false otherwise.
+ * Return true if a < b, false otherwise.
+ * If time stamps are exactly the same, compare by stream path. This
+ * ensures we get the same result between runs on the same trace
+ * collection on different environments.
+ * The result will be random for memory-mapped traces since there is no
+ * fixed path leading to those (they have empty path string).
  */
 static int stream_compare(void *a, void *b)
 {
        struct ctf_file_stream *s_a = a, *s_b = b;
 
-       if (s_a->parent.real_timestamp < s_b->parent.real_timestamp)
+       if (s_a->parent.real_timestamp < s_b->parent.real_timestamp) {
                return 1;
-       else
+       } else if (likely(s_a->parent.real_timestamp > s_b->parent.real_timestamp)) {
                return 0;
+       } else {
+               return strcmp(s_a->parent.path, s_b->parent.path);
+       }
 }
 
 void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
@@ -124,10 +135,10 @@ static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
        int i, ret;
 
        stream_pos = &cfs->pos;
-       for (i = 0; i < stream_pos->packet_real_index->len; i++) {
-               index = &g_array_index(stream_pos->packet_real_index,
+       for (i = 0; i < stream_pos->packet_index->len; i++) {
+               index = &g_array_index(stream_pos->packet_index,
                                struct packet_index, i);
-               if (index->timestamp_end < timestamp)
+               if (index->ts_real.timestamp_end < timestamp)
                        continue;
 
                stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
@@ -218,7 +229,7 @@ static int find_max_timestamp_ctf_file_stream(struct ctf_file_stream *cfs,
         * either find at least one event, or we reach the first packet
         * (some packets can be empty).
         */
-       for (i = stream_pos->packet_real_index->len - 1; i >= 0; i--) {
+       for (i = stream_pos->packet_index->len - 1; i >= 0; i--) {
                stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
                count = 0;
                /* read each event until we reach the end of the stream */
@@ -309,7 +320,7 @@ static int seek_last_ctf_trace_collection(struct trace_collection *tc,
        /* For each trace in the trace_collection */
        for (i = 0; i < tc->array->len; i++) {
                struct ctf_trace *tin;
-               struct trace_descriptor *td_read;
+               struct bt_trace_descriptor *td_read;
 
                td_read = g_ptr_array_index(tc->array, i);
                if (!td_read)
@@ -421,7 +432,7 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
                /* for each trace in the trace_collection */
                for (i = 0; i < tc->array->len; i++) {
                        struct ctf_trace *tin;
-                       struct trace_descriptor *td_read;
+                       struct bt_trace_descriptor *td_read;
 
                        td_read = g_ptr_array_index(tc->array, i);
                        if (!td_read)
@@ -451,7 +462,7 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 
                for (i = 0; i < tc->array->len; i++) {
                        struct ctf_trace *tin;
-                       struct trace_descriptor *td_read;
+                       struct bt_trace_descriptor *td_read;
                        int stream_id;
 
                        td_read = g_ptr_array_index(tc->array, i);
@@ -675,7 +686,7 @@ int bt_iter_init(struct bt_iter *iter,
 
        for (i = 0; i < ctx->tc->array->len; i++) {
                struct ctf_trace *tin;
-               struct trace_descriptor *td_read;
+               struct bt_trace_descriptor *td_read;
 
                td_read = g_ptr_array_index(ctx->tc->array, i);
                if (!td_read)
@@ -694,27 +705,21 @@ int bt_iter_init(struct bt_iter *iter,
                        for (filenr = 0; filenr < stream->streams->len;
                                        filenr++) {
                                struct ctf_file_stream *file_stream;
+                               struct bt_iter_pos pos;
 
                                file_stream = g_ptr_array_index(stream->streams,
                                                filenr);
                                if (!file_stream)
                                        continue;
-                               if (begin_pos) {
-                                       ret = babeltrace_filestream_seek(
-                                                       file_stream,
-                                                       begin_pos,
-                                                       stream_id);
-                               } else {
-                                       struct bt_iter_pos pos;
-                                       pos.type = BT_SEEK_BEGIN;
-                                       ret = babeltrace_filestream_seek(
-                                                       file_stream, &pos,
-                                                       stream_id);
-                               }
+
+                               pos.type = BT_SEEK_BEGIN;
+                               ret = babeltrace_filestream_seek(file_stream,
+                                       &pos, stream_id);
+
                                if (ret == EOF) {
                                        ret = 0;
                                        continue;
-                               } else if (ret) {
+                               } else if (ret != 0 && ret != EAGAIN) {
                                        goto error;
                                }
                                /* Add to heap */
@@ -726,7 +731,11 @@ int bt_iter_init(struct bt_iter *iter,
        }
 
        ctx->current_iterator = iter;
-       return 0;
+       if (begin_pos && begin_pos->type != BT_SEEK_BEGIN) {
+               ret = bt_iter_set_pos(iter, begin_pos);
+       }
+
+       return ret;
 
 error:
        bt_heap_free(iter->stream_heap);
@@ -795,13 +804,29 @@ int bt_iter_next(struct bt_iter *iter)
                assert(removed == file_stream);
                ret = 0;
                goto end;
+       } else if (ret == EAGAIN) {
+               /*
+                * Live streaming: the stream is inactive for now, we
+                * just updated the timestamp_end to skip over this
+                * stream up to a certain point in time.
+                *
+                * Since we can't guarantee that a stream will ever have
+                * any activity, we can't rely on the fact that
+                * bt_iter_next will be called for each stream and deal
+                * with inactive streams. So instead, we return 0 here
+                * to the caller and let the read API handle the
+                * retry case.
+                */
+               ret = 0;
+               goto reinsert;
        } else if (ret) {
                goto end;
        }
+
+reinsert:
        /* Reinsert the file stream into the heap, and rebalance. */
        removed = bt_heap_replace_max(iter->stream_heap, file_stream);
        assert(removed == file_stream);
-
 end:
        return ret;
 }