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;
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 */
}
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);
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;
}