Live: let read API detect inactive streams, allow streams without fd

[babeltrace.git] / lib / iterator.c

diff --git a/lib/iterator.c b/lib/iterator.c
index 966a00010e3e82a8393987f72f78b9e4d820c70f..853898d33e995e7004c177bff0d5782272c9898c 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;
@@ -716,7 +719,7 @@ int bt_iter_init(struct bt_iter *iter,
                                if (ret == EOF) {
                                        ret = 0;
                                        continue;
-                               } else if (ret) {
+                               } else if (ret != 0 && ret != EAGAIN) {
                                        goto error;
                                }
                                /* Add to heap */
@@ -801,13 +804,32 @@ 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);
 
+       file_stream = bt_heap_maximum(iter->stream_heap);
+
 end:
        return ret;
 }