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;
}
/*
- * 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)
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 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)
/* 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)
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);
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)
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);
-
end:
return ret;
}