struct ctf_file_stream *file_stream;
size_t cur_index; /* current index in packet index */
ssize_t offset; /* offset from base, in bits. EOF for end of file. */
- uint64_t current_timestamp;
+ uint64_t current_real_timestamp;
+ uint64_t current_cycles_timestamp;
};
struct bt_saved_pos {
/*
* returns true if a < b, false otherwise.
*/
-int stream_compare(void *a, void *b)
+static int stream_compare(void *a, void *b)
{
struct ctf_file_stream *s_a = a, *s_b = b;
- if (s_a->parent.timestamp < s_b->parent.timestamp)
+ if (s_a->parent.real_timestamp < s_b->parent.real_timestamp)
return 1;
else
return 0;
int i, ret;
stream_pos = &cfs->pos;
- for (i = 0; i < stream_pos->packet_index->len; i++) {
- index = &g_array_index(stream_pos->packet_index,
+ for (i = 0; i < stream_pos->packet_real_index->len; i++) {
+ index = &g_array_index(stream_pos->packet_real_index,
struct packet_index, i);
if (index->timestamp_end < timestamp)
continue;
stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
do {
ret = stream_read_event(cfs);
- } while (cfs->parent.timestamp < timestamp && ret == 0);
+ } while (cfs->parent.real_timestamp < timestamp && ret == 0);
/* Can return either EOF, 0, or error (> 0). */
return ret;
struct trace_collection *tc;
int i, ret;
+ if (!iter || !iter_pos)
+ return -EINVAL;
+
switch (iter_pos->type) {
case BT_SEEK_RESTORE:
if (!iter_pos->u.restore)
heap_free(iter->stream_heap);
ret = heap_init(iter->stream_heap, 0, stream_compare);
if (ret < 0)
- goto error;
+ goto error_heap_init;
for (i = 0; i < iter_pos->u.restore->stream_saved_pos->len;
i++) {
* packet_seek, because this function resets
* the timestamp to the beginning of the packet
*/
- stream->timestamp = saved_pos->current_timestamp;
+ stream->real_timestamp = saved_pos->current_real_timestamp;
+ stream->cycles_timestamp = saved_pos->current_cycles_timestamp;
stream_pos->offset = saved_pos->offset;
- stream_pos->last_offset = saved_pos->offset;
+ stream_pos->last_offset = LAST_OFFSET_POISON;
- stream->prev_timestamp = 0;
- stream->prev_timestamp_end = 0;
- stream->consumed = 0;
+ stream->prev_real_timestamp = 0;
+ stream->prev_real_timestamp_end = 0;
+ stream->prev_cycles_timestamp = 0;
+ stream->prev_cycles_timestamp_end = 0;
printf_debug("restored to cur_index = %zd and "
"offset = %zd, timestamp = %" PRIu64 "\n",
stream_pos->cur_index,
- stream_pos->offset, stream->timestamp);
+ stream_pos->offset, stream->real_timestamp);
- stream_read_event(saved_pos->file_stream);
+ ret = stream_read_event(saved_pos->file_stream);
+ if (ret != 0) {
+ goto error;
+ }
/* Add to heap */
ret = heap_insert(iter->stream_heap,
heap_free(iter->stream_heap);
ret = heap_init(iter->stream_heap, 0, stream_compare);
if (ret < 0)
- goto error;
+ goto error_heap_init;
/* for each trace in the trace_collection */
for (i = 0; i < tc->array->len; i++) {
return 0;
case BT_SEEK_BEGIN:
tc = iter->ctx->tc;
+ heap_free(iter->stream_heap);
+ ret = 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;
if (ret != 0 && ret != EOF) {
goto error;
}
+ ret = heap_insert(iter->stream_heap, file_stream);
+ if (ret)
+ goto error;
}
}
}
error:
heap_free(iter->stream_heap);
+error_heap_init:
if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
heap_free(iter->stream_heap);
g_free(iter->stream_heap);
iter->stream_heap = NULL;
ret = -ENOMEM;
}
+
return ret;
}
struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
{
struct bt_iter_pos *pos;
- struct trace_collection *tc = iter->ctx->tc;
- int i, stream_class_id, stream_id;
+ struct trace_collection *tc;
+ struct ctf_file_stream *file_stream = NULL, *removed;
+ struct ptr_heap iter_heap_copy;
+ int ret;
+
+ if (!iter)
+ return NULL;
+ tc = iter->ctx->tc;
pos = g_new0(struct bt_iter_pos, 1);
pos->type = BT_SEEK_RESTORE;
pos->u.restore = g_new0(struct bt_saved_pos, 1);
if (!pos->u.restore->stream_saved_pos)
goto error;
- for (i = 0; i < tc->array->len; i++) {
- struct ctf_trace *tin;
- struct trace_descriptor *td_read;
-
- td_read = g_ptr_array_index(tc->array, i);
- if (!td_read)
- continue;
- tin = container_of(td_read, struct ctf_trace, parent);
-
- for (stream_class_id = 0; stream_class_id < tin->streams->len;
- stream_class_id++) {
- struct ctf_stream_declaration *stream_class;
-
- stream_class = g_ptr_array_index(tin->streams,
- stream_class_id);
- if (!stream_class)
- continue;
- for (stream_id = 0;
- stream_id < stream_class->streams->len;
- stream_id++) {
- struct ctf_stream_definition *stream;
- struct ctf_file_stream *cfs;
- struct stream_saved_pos saved_pos;
-
- stream = g_ptr_array_index(
- stream_class->streams,
- stream_id);
- if (!stream)
- continue;
- cfs = container_of(stream,
- struct ctf_file_stream,
- parent);
-
- saved_pos.file_stream = cfs;
- saved_pos.cur_index = cfs->pos.cur_index;
+ ret = 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);
+ while (file_stream != NULL) {
+ struct stream_saved_pos saved_pos;
+
+ assert(file_stream->pos.last_offset != LAST_OFFSET_POISON);
+ saved_pos.offset = file_stream->pos.last_offset;
+ saved_pos.file_stream = file_stream;
+ saved_pos.cur_index = file_stream->pos.cur_index;
+
+ saved_pos.current_real_timestamp = file_stream->parent.real_timestamp;
+ saved_pos.current_cycles_timestamp = file_stream->parent.cycles_timestamp;
+
+ g_array_append_val(
+ pos->u.restore->stream_saved_pos,
+ saved_pos);
+
+ printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
+ "offset : %zd, "
+ "timestamp = %" PRIu64 "\n",
+ file_stream->parent.stream_id,
+ saved_pos.cur_index, saved_pos.offset,
+ saved_pos.current_real_timestamp);
+
+ /* remove the stream from the heap copy */
+ removed = heap_remove(&iter_heap_copy);
+ assert(removed == file_stream);
- /*
- * It is possible that an event was read during
- * the last restore, never consumed and its
- * position saved again. For this case, we
- * need to check if the event really was
- * consumed by the caller otherwise it is lost.
- */
- if (stream->consumed)
- saved_pos.offset = cfs->pos.offset;
- else
- saved_pos.offset = cfs->pos.last_offset;
-
- saved_pos.current_timestamp = stream->timestamp;
-
- g_array_append_val(
- pos->u.restore->stream_saved_pos,
- saved_pos);
-
- printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
- "offset : %zd, "
- "timestamp = %" PRIu64 "\n",
- stream->stream_id, saved_pos.cur_index,
- saved_pos.offset,
- saved_pos.current_timestamp);
- }
- }
+ file_stream = heap_maximum(&iter_heap_copy);
}
-
+ heap_free(&iter_heap_copy);
return pos;
+error_heap:
+ g_array_free(pos->u.restore->stream_saved_pos, TRUE);
error:
+ g_free(pos);
return NULL;
}
{
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;
{
int ret = 0;
+ if (!file_stream || !begin_pos)
+ return -EINVAL;
+
switch (begin_pos->type) {
case BT_SEEK_CUR:
/*
return ret;
}
-/*
- * bt_iter_seek: seek iterator to given position.
- */
-int bt_iter_seek(struct bt_iter *iter,
- const struct bt_iter_pos *begin_pos)
-{
- int i, stream_id;
- int ret = 0;
- struct trace_collection *tc = iter->ctx->tc;
-
- for (i = 0; i < tc->array->len; i++) {
- struct ctf_trace *tin;
- struct trace_descriptor *td_read;
-
- td_read = g_ptr_array_index(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;
-
- file_stream = g_ptr_array_index(stream->streams,
- filenr);
- if (!file_stream)
- continue;
- ret = babeltrace_filestream_seek(file_stream, begin_pos,
- stream_id);
- if (ret < 0)
- goto end;
- }
- }
- }
-end:
- return ret;
-}
-
int bt_iter_init(struct bt_iter *iter,
struct bt_context *ctx,
const struct bt_iter_pos *begin_pos,
int i, stream_id;
int ret = 0;
+ if (!iter || !ctx)
+ return -EINVAL;
+
if (ctx->current_iterator) {
ret = -1;
goto error_ctx;
heap_free(iter->stream_heap);
error_heap_init:
g_free(iter->stream_heap);
+ iter->stream_heap = NULL;
error_ctx:
return ret;
}
struct bt_iter *iter;
int ret;
+ if (!ctx)
+ return NULL;
+
iter = g_new0(struct bt_iter, 1);
ret = bt_iter_init(iter, ctx, begin_pos, end_pos);
if (ret) {
void bt_iter_fini(struct bt_iter *iter)
{
+ assert(iter);
if (iter->stream_heap) {
heap_free(iter->stream_heap);
g_free(iter->stream_heap);
void bt_iter_destroy(struct bt_iter *iter)
{
+ assert(iter);
bt_iter_fini(iter);
g_free(iter);
}
struct ctf_file_stream *file_stream, *removed;
int ret;
+ if (!iter)
+ return -EINVAL;
+
file_stream = heap_maximum(iter->stream_heap);
if (!file_stream) {
/* end of file for all streams */
projects / babeltrace.git / blobdiff