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);
* 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 */
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;
}
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);
goto end;
} else if (ret == EAGAIN) {
/*
- * The stream is inactive for now, we just updated the timestamp_end
- * to skip over this stream up to a certain point in time.
+ * 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 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);
- if (file_stream->pos.content_size == 0) {
- ret = EAGAIN;
- } else {
- ret = 0;
- }
-
end:
return ret;
}