X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=lib%2Fiterator.c;h=70f25edd9f0a0948a7820b894fec2bb7143426a3;hp=41a46020f6c690b4cc253044d845c0919e34cf0f;hb=c862350b39e4d95750bbb627497d3768d87817ff;hpb=71dd417a2b700c2315ef9919104572df1bc83f7a diff --git a/lib/iterator.c b/lib/iterator.c index 41a46020..70f25edd 100644 --- a/lib/iterator.c +++ b/lib/iterator.c @@ -28,6 +28,7 @@ #include #include +#include #include #include #include @@ -65,24 +66,36 @@ 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; } + return 0; } /* - * 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 +137,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); @@ -159,6 +172,16 @@ static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin, { int i, j, ret; int found = 0; + struct bt_trace_descriptor *td = &tin->parent; + + if (td->interval_set) { + /* + * If this trace has an interval selected, don't allow seeks + * before the selected interval. We seek to the start of the + * interval, thereby presenting a shorter "virtual" trace. + */ + timestamp = max(timestamp, td->interval_real.timestamp_begin); + } /* for each stream_class */ for (i = 0; i < tin->streams->len; i++) { @@ -180,7 +203,7 @@ static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin, ret = seek_file_stream_by_timestamp(cfs, timestamp); if (ret == 0) { /* Add to heap */ - ret = heap_insert(stream_heap, cfs); + ret = bt_heap_insert(stream_heap, cfs); if (ret) { /* Return positive error. */ return -ret; @@ -218,7 +241,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 +332,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) @@ -358,8 +381,8 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos) if (!iter_pos->u.restore) return -EINVAL; - heap_free(iter->stream_heap); - ret = heap_init(iter->stream_heap, 0, stream_compare); + bt_heap_free(iter->stream_heap); + ret = bt_heap_init(iter->stream_heap, 0, stream_compare); if (ret < 0) goto error_heap_init; @@ -388,10 +411,15 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos) stream_pos->offset = saved_pos->offset; stream_pos->last_offset = LAST_OFFSET_POISON; - stream->prev_real_timestamp = 0; - stream->prev_real_timestamp_end = 0; - stream->prev_cycles_timestamp = 0; - stream->prev_cycles_timestamp_end = 0; + stream->current.real.begin = 0; + stream->current.real.end = 0; + stream->current.cycles.begin = 0; + stream->current.cycles.end = 0; + + stream->prev.real.begin = 0; + stream->prev.real.end = 0; + stream->prev.cycles.begin = 0; + stream->prev.cycles.end = 0; printf_debug("restored to cur_index = %" PRId64 " and " "offset = %" PRId64 ", timestamp = %" PRIu64 "\n", @@ -404,7 +432,7 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos) } /* Add to heap */ - ret = heap_insert(iter->stream_heap, + ret = bt_heap_insert(iter->stream_heap, saved_pos->file_stream); if (ret) goto error; @@ -413,15 +441,15 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos) case BT_SEEK_TIME: tc = iter->ctx->tc; - heap_free(iter->stream_heap); - ret = heap_init(iter->stream_heap, 0, stream_compare); + bt_heap_free(iter->stream_heap); + ret = bt_heap_init(iter->stream_heap, 0, stream_compare); if (ret < 0) goto error_heap_init; /* 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) @@ -444,14 +472,14 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos) return 0; case BT_SEEK_BEGIN: tc = iter->ctx->tc; - heap_free(iter->stream_heap); - ret = heap_init(iter->stream_heap, 0, stream_compare); + bt_heap_free(iter->stream_heap); + ret = bt_heap_init(iter->stream_heap, 0, stream_compare); if (ret < 0) goto error_heap_init; 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); @@ -487,7 +515,7 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos) /* Do not add EOF streams */ continue; } - ret = heap_insert(iter->stream_heap, file_stream); + ret = bt_heap_insert(iter->stream_heap, file_stream); if (ret) goto error; } @@ -503,13 +531,13 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos) if (ret != 0 || !cfs) goto error; /* remove all streams from the heap */ - heap_free(iter->stream_heap); + bt_heap_free(iter->stream_heap); /* Create a new empty heap */ - ret = heap_init(iter->stream_heap, 0, stream_compare); + ret = bt_heap_init(iter->stream_heap, 0, stream_compare); if (ret < 0) goto error; /* Insert the stream that contains the last event */ - ret = heap_insert(iter->stream_heap, cfs); + ret = bt_heap_insert(iter->stream_heap, cfs); if (ret) goto error; break; @@ -522,10 +550,10 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos) return 0; error: - heap_free(iter->stream_heap); + bt_heap_free(iter->stream_heap); error_heap_init: - if (heap_init(iter->stream_heap, 0, stream_compare) < 0) { - heap_free(iter->stream_heap); + if (bt_heap_init(iter->stream_heap, 0, stream_compare) < 0) { + bt_heap_free(iter->stream_heap); g_free(iter->stream_heap); iter->stream_heap = NULL; ret = -ENOMEM; @@ -555,12 +583,12 @@ struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter) if (!pos->u.restore->stream_saved_pos) goto error; - ret = heap_copy(&iter_heap_copy, iter->stream_heap); + ret = bt_heap_copy(&iter_heap_copy, iter->stream_heap); if (ret < 0) goto error_heap; /* iterate over each stream in the heap */ - file_stream = heap_maximum(&iter_heap_copy); + file_stream = bt_heap_maximum(&iter_heap_copy); while (file_stream != NULL) { struct stream_saved_pos saved_pos; @@ -584,12 +612,12 @@ struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter) saved_pos.current_real_timestamp); /* remove the stream from the heap copy */ - removed = heap_remove(&iter_heap_copy); + removed = bt_heap_remove(&iter_heap_copy); assert(removed == file_stream); - file_stream = heap_maximum(&iter_heap_copy); + file_stream = bt_heap_maximum(&iter_heap_copy); } - heap_free(&iter_heap_copy); + bt_heap_free(&iter_heap_copy); return pos; error_heap: @@ -599,14 +627,11 @@ error: return NULL; } -struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *iter, +struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *unused, uint64_t timestamp) { struct bt_iter_pos *pos; - if (!iter) - return NULL; - pos = g_new0(struct bt_iter_pos, 1); pos->type = BT_SEEK_TIME; pos->u.seek_time = timestamp; @@ -648,15 +673,63 @@ static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream, return ret; } +int bt_iter_add_trace(struct bt_iter *iter, + struct bt_trace_descriptor *td_read) +{ + struct ctf_trace *tin; + int stream_id, ret = 0; + + tin = container_of(td_read, struct ctf_trace, parent); + + /* Populate heap with each stream */ + for (stream_id = 0; stream_id < tin->streams->len; + stream_id++) { + struct ctf_stream_declaration *stream; + int filenr; + + stream = g_ptr_array_index(tin->streams, stream_id); + if (!stream) + continue; + 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; + + pos.type = BT_SEEK_BEGIN; + ret = babeltrace_filestream_seek(file_stream, + &pos, stream_id); + + if (ret == EOF) { + ret = 0; + continue; + } else if (ret != 0 && ret != EAGAIN) { + goto error; + } + /* Add to heap */ + ret = bt_heap_insert(iter->stream_heap, file_stream); + if (ret) + goto error; + } + } + +error: + return ret; +} + int bt_iter_init(struct bt_iter *iter, struct bt_context *ctx, const struct bt_iter_pos *begin_pos, const struct bt_iter_pos *end_pos) { - int i, stream_id; + int i; int ret = 0; - if (!iter || !ctx) + if (!iter || !ctx || !ctx->tc || !ctx->tc->array) return -EINVAL; if (ctx->current_iterator) { @@ -669,67 +742,33 @@ int bt_iter_init(struct bt_iter *iter, bt_context_get(ctx); iter->ctx = ctx; - ret = heap_init(iter->stream_heap, 0, stream_compare); + ret = bt_heap_init(iter->stream_heap, 0, stream_compare); if (ret < 0) goto error_heap_init; 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) continue; - tin = container_of(td_read, struct ctf_trace, parent); - - /* Populate heap with each stream */ - for (stream_id = 0; stream_id < tin->streams->len; - stream_id++) { - struct ctf_stream_declaration *stream; - int filenr; - - stream = g_ptr_array_index(tin->streams, stream_id); - if (!stream) - continue; - for (filenr = 0; filenr < stream->streams->len; - filenr++) { - struct ctf_file_stream *file_stream; + ret = bt_iter_add_trace(iter, td_read); + if (ret < 0) + goto error; + } - 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); - } - if (ret == EOF) { - ret = 0; - continue; - } else if (ret) { - goto error; - } - /* Add to heap */ - ret = heap_insert(iter->stream_heap, file_stream); - if (ret) - goto error; - } + ctx->current_iterator = iter; + if (begin_pos && begin_pos->type != BT_SEEK_BEGIN) { + ret = bt_iter_set_pos(iter, begin_pos); + if (ret) { + goto error; } } - ctx->current_iterator = iter; - return 0; + return ret; error: - heap_free(iter->stream_heap); + bt_heap_free(iter->stream_heap); error_heap_init: g_free(iter->stream_heap); iter->stream_heap = NULL; @@ -760,7 +799,7 @@ void bt_iter_fini(struct bt_iter *iter) { assert(iter); if (iter->stream_heap) { - heap_free(iter->stream_heap); + bt_heap_free(iter->stream_heap); g_free(iter->stream_heap); } iter->ctx->current_iterator = NULL; @@ -778,11 +817,12 @@ int bt_iter_next(struct bt_iter *iter) { struct ctf_file_stream *file_stream, *removed; int ret; + bool event_outside_interval = false; if (!iter) return -EINVAL; - file_stream = heap_maximum(iter->stream_heap); + file_stream = bt_heap_maximum(iter->stream_heap); if (!file_stream) { /* end of file for all streams */ ret = 0; @@ -790,18 +830,39 @@ int bt_iter_next(struct bt_iter *iter) } ret = stream_read_event(file_stream); - if (ret == EOF) { - removed = heap_remove(iter->stream_heap); + if (file_stream->pos.parent.trace->interval_set) { + event_outside_interval = + file_stream->parent.real_timestamp > + file_stream->pos.parent.trace->interval_real.timestamp_end; + } + if (ret == EOF || event_outside_interval) { + removed = bt_heap_remove(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 = heap_replace_max(iter->stream_heap, file_stream); + removed = bt_heap_replace_max(iter->stream_heap, file_stream); assert(removed == file_stream); - end: return ret; }